mirror of
https://github.com/mims-harvard/ToolUniverse.git
synced 2026-09-19 07:31:47 +08:00
406add8bf0
Pinned third-party MCP tools were stored only as a contract_sha256. A hash can show that an upstream server changed a tool but never what it changed, so a maintainer asked to re-pin had nothing to review and could only accept blind -- which is what re-pinning the nine drifted Folklore and Genomic Intelligence contracts actually amounted to. Adds src/tooluniverse/data/mcp_contracts.lock.json recording the reviewed inputSchema/outputSchema of all 27 pinned tools across the four loaders, so upstream drift shows up as a readable diff instead of two hex strings. scripts/sync_mcp_contracts.py has three modes: --check-config (offline; the lockfile and the configs must agree), --check-live (diff the servers against the lockfile) and --update (re-record, producing the diff to review). The new workflow moves detection off the user's critical path. Contract verification currently fires inside the user's process, where nobody can review a third-party schema or re-pin it; Folklore and Genomic Intelligence were each serving zero tools for an unknown period, found only by accident, because the weekly health check cannot see them -- MCP loader URLs are opt-in env vars CI never sets. check-config gates every PR; check-live runs daily and fails on real drift while skipping servers it cannot reach, so a vendor outage or rate limit does not read as drift. The drift report calls out new input fields specifically: the agent fills in any parameter a server advertises, so an added field is a new path for data to leave the session.
151 lines
5.3 KiB
Python
151 lines
5.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Generate the API-key catalog and .env.template from the tool configs.
|
|
|
|
Key names come from each tool's required_api_keys / optional_api_keys arrays.
|
|
The human metadata (domain, registration link, purpose, without-it note, type)
|
|
is declared inline in the same tool configs via an ``api_key_info`` block:
|
|
|
|
{
|
|
"name": "my_tool",
|
|
"required_api_keys": ["MY_KEY"],
|
|
"api_key_info": {
|
|
"MY_KEY": {
|
|
"domain": "Drugs & Chemistry",
|
|
"type": "secret",
|
|
"register_url": "https://...",
|
|
"purpose": "What this key unlocks.",
|
|
"without": "What happens without it."
|
|
}
|
|
}
|
|
}
|
|
|
|
A key needs exactly one api_key_info entry across all configs (define it once,
|
|
even if several tools share the key).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
REPO = Path(__file__).resolve().parents[1]
|
|
DATA = REPO / "src" / "tooluniverse" / "data"
|
|
CATALOG = DATA / "api_keys_catalog.json"
|
|
ENV_TEMPLATE = REPO / ".env.template"
|
|
_SKIP = {"api_keys_catalog.json", "mcp_contracts.lock.json"}
|
|
|
|
DOMAIN_ORDER = [
|
|
"Genomics & Variants",
|
|
"Drugs & Chemistry",
|
|
"Proteins & Structure",
|
|
"Literature & Patents",
|
|
"Clinical & Safety",
|
|
"Models & Infrastructure",
|
|
]
|
|
|
|
|
|
def scan_configs():
|
|
"""Single pass over tool configs.
|
|
|
|
Returns (found, info, conflicts):
|
|
found: {name: {"requirement": str, "used_by": set}}
|
|
info: {name: {domain, type, register_url, purpose, without}}
|
|
conflicts: [(name, file), ...] where api_key_info disagreed across files
|
|
"""
|
|
found: dict = {}
|
|
info: dict = {}
|
|
conflicts: list = []
|
|
for path in sorted(DATA.glob("*.json")):
|
|
if path.name in _SKIP:
|
|
continue
|
|
try:
|
|
doc = json.loads(path.read_text())
|
|
except (json.JSONDecodeError, OSError):
|
|
continue
|
|
tools = doc if isinstance(doc, list) else [doc]
|
|
for tool in tools:
|
|
if not isinstance(tool, dict):
|
|
continue
|
|
for name in tool.get("required_api_keys") or []:
|
|
e = found.setdefault(name, {"requirement": "optional", "used_by": set()})
|
|
e["requirement"] = "required"
|
|
e["used_by"].add(path.name)
|
|
for name in tool.get("optional_api_keys") or []:
|
|
e = found.setdefault(name, {"requirement": "optional", "used_by": set()})
|
|
e["used_by"].add(path.name)
|
|
for name, meta in (tool.get("api_key_info") or {}).items():
|
|
if name in info and info[name] != meta:
|
|
conflicts.append((name, path.name))
|
|
else:
|
|
info.setdefault(name, meta)
|
|
return found, info, conflicts
|
|
|
|
|
|
def build_catalog(found: dict, info: dict):
|
|
"""Merge declared keys with their inline info. Returns (catalog, missing, unused)."""
|
|
catalog, missing = [], []
|
|
for name in sorted(found):
|
|
meta = info.get(name)
|
|
if meta is None:
|
|
missing.append(name)
|
|
continue
|
|
catalog.append({
|
|
"name": name,
|
|
"requirement": found[name]["requirement"],
|
|
"type": meta.get("type", "secret"),
|
|
"domain": meta.get("domain", "Other"),
|
|
"register_url": meta["register_url"],
|
|
"purpose": meta["purpose"],
|
|
"without": meta["without"],
|
|
"used_by": sorted(found[name]["used_by"]),
|
|
})
|
|
unused = sorted(set(info) - set(found))
|
|
return catalog, missing, unused
|
|
|
|
|
|
def render_env_template(catalog: list) -> str:
|
|
lines = [
|
|
"# API Keys for ToolUniverse",
|
|
"# AUTO-GENERATED by scripts/gen_api_key_catalog.py - DO NOT EDIT BY HAND.",
|
|
"# Edit the api_key_info block in the relevant tool config under",
|
|
"# src/tooluniverse/data/, then re-run the generator.",
|
|
"# Copy to .env or ~/.tooluniverse/.env and fill in the keys you need.",
|
|
"",
|
|
]
|
|
seen = dict.fromkeys(e["domain"] for e in catalog)
|
|
domains = [d for d in DOMAIN_ORDER if d in seen] + [d for d in seen if d not in DOMAIN_ORDER]
|
|
for domain in domains:
|
|
items = [e for e in catalog if e["domain"] == domain]
|
|
if not items:
|
|
continue
|
|
lines += [f"# === {domain} ===", ""]
|
|
for e in items:
|
|
lines += [
|
|
f"# {e['name']} ({e['requirement']}) — {e['purpose']} ({e['register_url']})",
|
|
f"# Without: {e['without']}",
|
|
f"{e['name']}=",
|
|
"",
|
|
]
|
|
return "\n".join(lines).rstrip() + "\n"
|
|
|
|
|
|
def main(argv=None) -> int:
|
|
found, info, conflicts = scan_configs()
|
|
for name, fname in conflicts:
|
|
print(f"warning: conflicting api_key_info for '{name}' in {fname} (kept first seen)", file=sys.stderr)
|
|
catalog, missing, unused = build_catalog(found, info)
|
|
for name in unused:
|
|
print(f"warning: api_key_info '{name}' is not used by any tool config", file=sys.stderr)
|
|
if missing:
|
|
for name in missing:
|
|
print(f"error: api-key '{name}' has no api_key_info block in any tool config", file=sys.stderr)
|
|
return 1
|
|
CATALOG.write_text(json.dumps(catalog, indent=2) + "\n")
|
|
ENV_TEMPLATE.write_text(render_env_template(catalog))
|
|
print(f"Wrote {len(catalog)} keys to {CATALOG.relative_to(REPO)} and {ENV_TEMPLATE.name}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|