Python 88.3%
TypeScript 7.6%
Shell 4.1%
1#!/usr/bin/env python32"""Export the API Atlas capability graph from the merged files.34 python3 scripts/export_capability_graph.py # generated/{models,endpoints,tools}.json → generated/capability-graph.{json,mmd,dot}5 python3 scripts/export_capability_graph.py --models X --endpoints Y --tools Z --out-dir DIR [--max-mermaid-nodes N]67Inputs may be partially missing or malformed: each is skipped gracefully (recorded in `meta.inputs`).8Node types: provider · api_family · endpoint · model · capability · tool9Edge types: provider→api_family HAS_FAMILY · api_family→endpoint HAS_ENDPOINT · provider→model OFFERS ·10 model→endpoint AVAILABLE_ON · model→capability SUPPORTS (value true) / SUPPORTS_UNKNOWN (value "unknown") ·11 model→tool SUPPORTS_TOOL · tool→endpoint USABLE_ON · tool→model COMPATIBLE_WITH · model→model ALIAS_OF / SNAPSHOT_OF /12 REDIRECTS_TO (xAI `kind: "retired_redirect"` records whose live target is in verification.request_note)13Capabilities that are `false` are NOT edges (absence = false when the model has a capabilities object; see graph meta).1415Provider-agnostic (openai · anthropic · xai · gemini): every provider found in the records becomes a node. Alias entries may be16strings (optionally with a trailing parenthetical note, stripped) or dicts `{alias, resolves_to_live}` (Gemini `-latest`17records → ALIAS_OF the live target when it exists); model `tools[]` may be strings or dicts `{type, category, support}`18(Gemini; `support: false` entries produce no edge, string values such as "Supported (Preview)" count as supported).1920Record schemas: CLAUDE.md (model, endpoint, tool). Accepts a list of records or {"records"|"models"|"endpoints"|"tools": [...]}.21Stdlib only. Tested offline by tests/shared/test_capability_graph.py against tests/shared/fixtures/*.json (4 providers).22"""23from __future__ import annotations2425import argparse26import json27import re28import sys29from datetime import datetime, timezone30from pathlib import Path31from typing import Any, Optional3233ROOT = Path(__file__).resolve().parent.parent34GEN = ROOT / "generated"353637def load_records(path: Optional[Path], list_keys: tuple[str, ...]) -> tuple[list[dict[str, Any]], str]:38 """Return (records, status) where status ∈ {loaded, missing, invalid, empty}."""39 if path is None or not path.exists():40 return [], "missing"41 try:42 doc = json.loads(path.read_text())43 except (json.JSONDecodeError, OSError):44 return [], "invalid"45 if isinstance(doc, dict):46 for k in list_keys:47 if isinstance(doc.get(k), list):48 doc = doc[k]49 break50 else:51 doc = [({"id": k, **v} if isinstance(v, dict) and "id" not in v and "name" not in v else v) for k, v in doc.items() if isinstance(v, dict)]52 if not isinstance(doc, list):53 return [], "invalid"54 recs = [r for r in doc if isinstance(r, dict)]55 return recs, ("loaded" if recs else "empty")565758def _slug(s: str) -> str:59 return re.sub(r"[^A-Za-z0-9_]", "_", s)606162_PAREN_NOTE = re.compile(r"\s*\(.*\)\s*$")63_REDIRECT_NOTE = re.compile(r"with id ['\"]([^'\"]+)['\"]")646566def _alias_name(a: Any) -> Optional[str]:67 """'grok-voice-latest (routes here since …)' → 'grok-voice-latest'; {alias: …} dicts → their alias."""68 if isinstance(a, dict):69 a = a.get("alias") or a.get("id")70 if not isinstance(a, str):71 return None72 a = _PAREN_NOTE.sub("", a).strip()73 return a or None747576def _redirect_target(m: dict[str, Any]) -> Optional[str]:77 for k in ("resolves_to", "redirects_to", "redirect_to", "alias_of"):78 if isinstance(m.get(k), str) and m[k].strip():79 return _alias_name(m[k])80 if m.get("kind") == "retired_redirect":81 mm = _REDIRECT_NOTE.search(str((m.get("verification") or {}).get("request_note") or ""))82 if mm:83 return mm.group(1)84 return None858687def _model_tool_types(m: dict[str, Any]) -> list[str]:88 """Model tools[] as type strings; dict entries (Gemini) with support == false are dropped."""89 out = []90 for t in m.get("tools") or []:91 if isinstance(t, str):92 out.append(t)93 elif isinstance(t, dict):94 typ = t.get("type") or t.get("name")95 sup = t.get("support", True)96 if isinstance(sup, str):97 sup = not (sup.lower().startswith("not") or sup.lower() in ("no", "unsupported"))98 if isinstance(typ, str) and sup is not False:99 out.append(typ)100 return out101102103class Graph:104 def __init__(self) -> None:105 self.nodes: dict[str, dict[str, Any]] = {}106 self.edges: list[dict[str, Any]] = []107 self._edge_keys: set[tuple[str, str, str]] = set()108109 def node(self, ntype: str, key: str, label: Optional[str] = None, **attrs: Any) -> str:110 nid = f"{ntype}:{key}"111 if nid not in self.nodes:112 self.nodes[nid] = {"id": nid, "type": ntype, "label": label or key, **attrs}113 else:114 for k, v in attrs.items():115 self.nodes[nid].setdefault(k, v)116 return nid117118 def edge(self, src: str, etype: str, dst: str, **attrs: Any) -> None:119 k = (src, etype, dst)120 if k in self._edge_keys:121 return122 self._edge_keys.add(k)123 self.edges.append({"source": src, "type": etype, "target": dst, **attrs})124125126def build_graph(models: list[dict[str, Any]], endpoints: list[dict[str, Any]], tools: list[dict[str, Any]]) -> Graph:127 g = Graph()128 endpoint_ids: dict[tuple[str, str], str] = {} # (provider, "METHOD /path") -> node id129130 # --- endpoints (provider → api_family → endpoint)131 for e in endpoints:132 prov, path = str(e.get("provider", "")).lower(), e.get("path")133 if not prov or not isinstance(path, str):134 continue135 method = str(e.get("method", "GET")).upper()136 fam = str(e.get("api_family") or "unknown")137 p = g.node("provider", prov)138 f = g.node("api_family", f"{prov}/{fam}", label=fam, provider=prov)139 key = f"{method} {path}"140 ep = g.node("endpoint", f"{prov}/{key}", label=key, provider=prov, api_family=fam, method=method, path=path,141 status=e.get("status") or [], streaming=(e.get("streaming") or {}).get("supported"), idempotency=e.get("idempotency"))142 g.edge(p, "HAS_FAMILY", f)143 g.edge(f, "HAS_ENDPOINT", ep)144 endpoint_ids[(prov, key)] = ep145 endpoint_ids[(prov, path)] = ep # tolerate path-only references146147 def endpoint_ref(prov: str, ref: Any) -> Optional[str]:148 if not isinstance(ref, str):149 return None150 ref = ref.strip()151 if (prov, ref) in endpoint_ids:152 return endpoint_ids[(prov, ref)]153 # create a placeholder endpoint node when endpoints.json is missing or incomplete154 method, _, path = ref.partition(" ") if " " in ref else ("POST", "", ref)155 key = f"{method.upper()} {path}" if path else ref156 ep = g.node("endpoint", f"{prov}/{key}", label=key, provider=prov, method=method.upper(), path=path or ref, placeholder=True)157 endpoint_ids[(prov, key)] = ep158 return ep159160 # --- models161 model_ids: dict[tuple[str, str], str] = {}162 for m in models:163 prov, mid = str(m.get("provider", "")).lower(), m.get("id")164 if not prov or not isinstance(mid, str):165 continue166 p = g.node("provider", prov)167 # record_kind: OpenAI emits it (model | snapshot | id_only | alias); xAI/Gemini use `kind` (alias / retired_redirect are pointers).168 # Anthropic's `kind: "snapshot"` denotes a real dated model record and stays "model".169 kind = m.get("record_kind") or {"retired_redirect": "redirect", "alias": "alias"}.get(str(m.get("kind")), "model")170 mn = g.node("model", f"{prov}/{mid}", label=mid, provider=prov, status=m.get("status") or [], family=m.get("family"),171 record_kind=kind, context_window=m.get("context_window"), max_output=m.get("max_output"))172 model_ids[(prov, mid.lower())] = mn173 g.edge(p, "OFFERS", mn)174 for m in models:175 prov, mid = str(m.get("provider", "")).lower(), m.get("id")176 if not prov or not isinstance(mid, str):177 continue178 mn = model_ids[(prov, mid.lower())]179 canon = m.get("canonical_model")180 if isinstance(canon, str) and canon.lower() != mid.lower() and (prov, canon.lower()) in model_ids:181 g.edge(mn, "SNAPSHOT_OF", model_ids[(prov, canon.lower())])182 tgt = _redirect_target(m)183 if tgt and tgt.lower() != mid.lower() and (prov, tgt.lower()) in model_ids:184 g.edge(mn, "REDIRECTS_TO", model_ids[(prov, tgt.lower())])185 for a in m.get("aliases") or []:186 name = _alias_name(a)187 if not name:188 continue189 if isinstance(a, dict) and name.lower() == mid.lower():190 # Gemini `-latest` alias records: {alias: <this id>, resolves_to_live: <target>} → this node ALIAS_OF the live target191 live = a.get("resolves_to_live") or a.get("resolves_to")192 if isinstance(live, str) and (prov, _alias_name(live).lower()) in model_ids and _alias_name(live).lower() != mid.lower():193 g.nodes[mn].setdefault("record_kind", "alias")194 g.nodes[mn]["record_kind"] = "alias"195 g.edge(mn, "ALIAS_OF", model_ids[(prov, _alias_name(live).lower())])196 continue197 if name.lower() == mid.lower():198 continue199 an = model_ids.get((prov, name.lower())) or g.node("model", f"{prov}/{name}", label=name, provider=prov, record_kind="alias")200 g.edge(an, "ALIAS_OF", mn)201 for s in m.get("snapshots") or []:202 name = _alias_name(s)203 if name and name.lower() != mid.lower():204 sn = model_ids.get((prov, name.lower())) or g.node("model", f"{prov}/{name}", label=name, provider=prov, record_kind="snapshot")205 g.edge(sn, "SNAPSHOT_OF", mn)206 for ref in m.get("endpoints") or []:207 ep = endpoint_ref(prov, ref)208 if ep:209 g.edge(mn, "AVAILABLE_ON", ep)210 caps = m.get("capabilities")211 if isinstance(caps, dict):212 for cap, val in caps.items():213 if val is True:214 g.edge(mn, "SUPPORTS", g.node("capability", cap))215 elif val == "unknown":216 g.edge(mn, "SUPPORTS_UNKNOWN", g.node("capability", cap))217 # False → no edge; non-boolean metadata (lists, notes) → ignored218 for t in _model_tool_types(m):219 g.edge(mn, "SUPPORTS_TOOL", g.node("tool", f"{prov}/{t}", label=t, provider=prov))220221 # --- tools222 for t in tools:223 prov = str(t.get("provider", "")).lower()224 typ = t.get("type") or t.get("name")225 if not prov or not isinstance(typ, str):226 continue227 tn = g.node("tool", f"{prov}/{typ}", label=typ, provider=prov, name=t.get("name"), category=t.get("category"),228 status=t.get("status") or [], beta_header=t.get("beta_header"), security=t.get("security"))229 g.edge(g.node("provider", prov), "PROVIDES_TOOL", tn)230 for ref in t.get("compatible_endpoints") or []:231 ep = endpoint_ref(prov, ref)232 if ep:233 g.edge(tn, "USABLE_ON", ep)234 for mref in t.get("compatible_models") or []:235 if isinstance(mref, str):236 mn = model_ids.get((prov, mref.lower())) or g.node("model", f"{prov}/{mref}", label=mref, provider=prov, placeholder=True)237 g.edge(tn, "COMPATIBLE_WITH", mn)238 return g239240241# ----------------------------------------------------------------------------- exporters242243def to_json(g: Graph, meta: dict[str, Any]) -> dict[str, Any]:244 counts: dict[str, int] = {}245 for n in g.nodes.values():246 counts[n["type"]] = counts.get(n["type"], 0) + 1247 ecounts: dict[str, int] = {}248 for e in g.edges:249 ecounts[e["type"]] = ecounts.get(e["type"], 0) + 1250 providers = sorted(n["label"] for n in g.nodes.values() if n["type"] == "provider")251 per_provider = {p: {"models": sum(1 for n in g.nodes.values() if n["type"] == "model" and n.get("provider") == p252 and n.get("record_kind", "model") not in ("alias", "snapshot", "redirect") and not n.get("placeholder")),253 "endpoints": sum(1 for n in g.nodes.values() if n["type"] == "endpoint" and n.get("provider") == p),254 "tools": sum(1 for n in g.nodes.values() if n["type"] == "tool" and n.get("provider") == p)} for p in providers}255 return {"meta": {**meta, "providers": providers, "per_provider": per_provider, "node_counts": counts, "edge_counts": ecounts,256 "semantics": {"SUPPORTS": "capability value true", "SUPPORTS_UNKNOWN": "capability value \"unknown\"",257 "absent": "capability false, or model without a capabilities object",258 "REDIRECTS_TO": "retired id whose requests are served by the target (xAI retired_redirect)",259 "ALIAS_OF": "alias id → canonical record (incl. Gemini -latest records → live target)"}},260 "nodes": list(g.nodes.values()), "edges": g.edges}261262263_MMD_SHAPE = {"provider": ("[[", "]]"), "api_family": ("(", ")"), "endpoint": ("[", "]"), "model": ("([", "])"),264 "capability": ("{{", "}}"), "tool": (">", "]")}265266267def to_mermaid(g: Graph, max_nodes: int = 400) -> str:268 """Mermaid gets unreadable past a few hundred nodes: skip alias/snapshot/redirect-only model nodes first, then truncate."""269 lines = ["graph LR"]270 nodes = [n for n in g.nodes.values() if n.get("record_kind") not in ("alias", "snapshot", "redirect")]271 keep = {n["id"] for n in nodes[:max_nodes]}272 truncated = len(g.nodes) - len(keep)273 for n in nodes[:max_nodes]:274 o, c = _MMD_SHAPE.get(n["type"], ("[", "]"))275 label = n["label"].replace('"', "'")276 lines.append(f' {_slug(n["id"])}{o}"{label}"{c}')277 for e in g.edges:278 if e["source"] in keep and e["target"] in keep:279 lines.append(f' {_slug(e["source"])} -->|{e["type"]}| {_slug(e["target"])}')280 for t in _MMD_SHAPE:281 ids = [_slug(n["id"]) for n in nodes[:max_nodes] if n["type"] == t]282 if ids:283 lines.append(f" classDef {t} stroke-width:2px;")284 lines.append(f" class {','.join(ids)} {t};")285 if truncated > 0:286 lines.append(f" %% {truncated} node(s) omitted (aliases/snapshots or beyond --max-mermaid-nodes); see capability-graph.json")287 return "\n".join(lines) + "\n"288289290_DOT_SHAPE = {"provider": "doubleoctagon", "api_family": "ellipse", "endpoint": "box", "model": "component",291 "capability": "hexagon", "tool": "cds"}292293294def to_dot(g: Graph) -> str:295 lines = ["digraph capability_graph {", " rankdir=LR;", " node [fontname=Helvetica fontsize=10];", " edge [fontname=Helvetica fontsize=8];"]296 for n in g.nodes.values():297 label = n["label"].replace('"', '\\"')298 lines.append(f' "{n["id"]}" [label="{label}" shape={_DOT_SHAPE.get(n["type"], "box")} class="{n["type"]}"];')299 for e in g.edges:300 lines.append(f' "{e["source"]}" -> "{e["target"]}" [label="{e["type"]}"];')301 lines.append("}")302 return "\n".join(lines) + "\n"303304305def export(models_path: Optional[Path], endpoints_path: Optional[Path], tools_path: Optional[Path], out_dir: Path,306 max_mermaid_nodes: int = 400) -> dict[str, Any]:307 models, ms = load_records(models_path, ("records", "models"))308 endpoints, es = load_records(endpoints_path, ("records", "endpoints"))309 tools, ts = load_records(tools_path, ("records", "tools"))310 g = build_graph(models, endpoints, tools)311 meta = {"generated_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),312 "inputs": {"models": {"path": str(models_path), "status": ms, "records": len(models)},313 "endpoints": {"path": str(endpoints_path), "status": es, "records": len(endpoints)},314 "tools": {"path": str(tools_path), "status": ts, "records": len(tools)}}}315 out_dir.mkdir(parents=True, exist_ok=True)316 doc = to_json(g, meta)317 (out_dir / "capability-graph.json").write_text(json.dumps(doc, indent=1, ensure_ascii=False) + "\n")318 (out_dir / "capability-graph.mmd").write_text(to_mermaid(g, max_mermaid_nodes))319 (out_dir / "capability-graph.dot").write_text(to_dot(g))320 return doc321322323def main(argv: Optional[list[str]] = None) -> int:324 ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])325 ap.add_argument("--models", type=Path, default=GEN / "models.json")326 ap.add_argument("--endpoints", type=Path, default=GEN / "endpoints.json")327 ap.add_argument("--tools", type=Path, default=GEN / "tools.json")328 ap.add_argument("--out-dir", type=Path, default=GEN)329 ap.add_argument("--max-mermaid-nodes", type=int, default=400)330 a = ap.parse_args(argv)331 doc = export(a.models, a.endpoints, a.tools, a.out_dir, a.max_mermaid_nodes)332 m = doc["meta"]333 print(f"inputs: " + ", ".join(f"{k}={v['status']}({v['records']})" for k, v in m["inputs"].items()), file=sys.stderr)334 print(f"providers: {m['providers']} per_provider: {m['per_provider']}", file=sys.stderr)335 print(f"nodes: {m['node_counts']}\nedges: {m['edge_counts']}\n→ {a.out_dir}/capability-graph.{{json,mmd,dot}}", file=sys.stderr)336 return 0337338339if __name__ == "__main__":340 sys.exit(main())341