Files

614 lines
21 KiB
Python
Raw Permalink Normal View History

2026-08-24 08:11:59 -07:00
#!/usr/bin/env python3
"""Generate the bundled route/model catalog and upstream inventory.
The inventories are intentionally explicit and reviewable. `codex-mmo catalog
verify --remote` compares dynamic provider listings against this baseline; it
does not silently add models whose protocol and capabilities are unknown.
"""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
from typing import Any
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "libexec"))
from mmo_inventory_snapshot import load_inventory_snapshots # noqa: E402
from mmo_profiles import validate_catalog_data # noqa: E402
from mmo_util import atomic_write_json, atomic_write_text, toml_dumps # noqa: E402
from mmo_version import MMO_SCHEMA_VERSION # noqa: E402
INVENTORY_SNAPSHOT_ROOT = ROOT / "config" / "inventory-snapshots"
def route(**values: Any) -> dict[str, Any]:
return values
def model(
route_key: str,
upstream_id: str,
maker: str,
display_name: str,
*,
description: str,
context: int,
output: int | None = None,
reasoning: list[str] | None = None,
default_reasoning: str | None = None,
modalities: list[str] | None = None,
output_modalities: list[str] | None = None,
tool_calling: bool = True,
parallel_tool_calls: bool = True,
summaries: bool = False,
structured: bool = True,
resource: str | None = None,
kind: str = "chat",
agent_compatible: bool = True,
availability: str = "current",
confidence: str = "documented",
source: str,
inventory: str | None = None,
input_cost: float | None = None,
output_cost: float | None = None,
cached_input_cost: float | None = None,
cache_write_input_cost: float | None = None,
unit_cost: float | None = None,
) -> dict[str, Any]:
values: dict[str, Any] = {
"route": route_key,
"upstream_id": upstream_id,
"maker": maker,
"display_name": display_name,
"description": description,
"kind": kind,
"agent_compatible": agent_compatible,
"context_window": context,
"reasoning_levels": reasoning or (["none"] if not agent_compatible else ["high"]),
"default_reasoning": default_reasoning
or ("none" if not agent_compatible else (reasoning or ["high"])[-1]),
"modalities": modalities or ["text"],
"output_modalities": output_modalities or ["text"],
"tool_calling": tool_calling,
"parallel_tool_calls": parallel_tool_calls,
"supports_reasoning_summaries": summaries,
"structured_output": structured,
"availability": availability,
"capability_confidence": confidence,
"source": source,
}
if output is not None:
values["max_output_tokens"] = output
if resource:
values["resource_group"] = resource
if inventory:
values["inventory"] = inventory
if input_cost is not None:
values["input_cost_per_million"] = input_cost
if output_cost is not None:
values["output_cost_per_million"] = output_cost
if cached_input_cost is not None:
values["cached_input_cost_per_million"] = cached_input_cost
if cache_write_input_cost is not None:
values["cache_write_input_cost_per_million"] = cache_write_input_cost
if unit_cost is not None:
values["unit_cost_usd"] = unit_cost
return values
routes: dict[str, dict[str, Any]] = {
"zai_coding_responses": route(
driver="switchyard",
name="Z.AI Coding Plan Responses",
api_operator="zai",
access_product="zai_coding_plan",
wire_protocol="openai_responses",
billing_mode="subscription",
base_url="https://api.z.ai/api/v1",
credential_envs=["ZAI_CODING_API_KEY"],
max_retries=1,
transport_modalities=["text"],
tool_calling=True,
parallel_tool_calls=True,
resource_group="zai_coding_plan",
inventory="zai-coding-plan",
),
"zai_coding_openai_chat": route(
driver="switchyard",
name="Z.AI Coding Plan OpenAI-compatible",
api_operator="zai",
access_product="zai_coding_plan",
wire_protocol="openai_chat",
billing_mode="subscription",
base_url="https://api.z.ai/api/coding/paas/v4",
credential_envs=["ZAI_CODING_API_KEY"],
max_retries=1,
transport_modalities=["text"],
tool_calling=True,
parallel_tool_calls=True,
resource_group="zai_coding_plan",
inventory="zai-coding-plan",
),
"zai_coding_anthropic_messages": route(
driver="catalog_only",
name="Z.AI Coding Plan Anthropic-compatible (catalog only)",
api_operator="zai",
access_product="zai_coding_plan",
wire_protocol="anthropic_messages",
billing_mode="subscription",
base_url="https://api.z.ai/api/anthropic",
credential_envs=["ZAI_CODING_API_KEY"],
auth="bearer",
transport_modalities=["text"],
transport_output_modalities=["text"],
tool_calling=False,
parallel_tool_calls=False,
resource_group="zai_coding_plan",
inventory="zai-coding-plan",
),
"zai_general_openai_chat": route(
driver="switchyard",
name="Z.AI General API",
api_operator="zai",
access_product="zai_general_api",
wire_protocol="openai_chat",
billing_mode="api",
base_url="https://api.z.ai/api/paas/v4",
credential_envs=["ZAI_API_KEY"],
max_retries=1,
transport_modalities=["text", "image", "video", "file"],
tool_calling=True,
parallel_tool_calls=True,
resource_group="zai_api",
inventory="zai-api",
),
"zai_general_catalog": route(
driver="catalog_only",
name="Z.AI media and specialist API catalog",
api_operator="zai",
access_product="zai_general_api",
wire_protocol="catalog_only",
billing_mode="catalog_only",
transport_modalities=["text", "image", "video", "audio", "file"],
tool_calling=False,
parallel_tool_calls=False,
resource_group="zai_api",
inventory="zai-api",
),
"opencode_zen_openai_chat": route(
driver="switchyard",
name="OpenCode Zen Chat Completions",
api_operator="opencode",
access_product="opencode_zen",
wire_protocol="openai_chat",
billing_mode="api",
base_url="https://opencode.ai/zen/v1",
credential_envs=["OPENCODE_API_KEY"],
max_retries=1,
transport_modalities=["text"],
tool_calling=True,
parallel_tool_calls=False,
resource_group="opencode_zen",
inventory="opencode-zen",
),
"opencode_zen_responses": route(
driver="switchyard",
name="OpenCode Zen Responses",
api_operator="opencode",
access_product="opencode_zen",
wire_protocol="openai_responses",
billing_mode="api",
base_url="https://opencode.ai/zen/v1",
credential_envs=["OPENCODE_API_KEY"],
max_retries=1,
transport_modalities=["text"],
tool_calling=True,
parallel_tool_calls=False,
resource_group="opencode_zen",
inventory="opencode-zen",
),
"opencode_zen_anthropic_messages": route(
driver="switchyard",
name="OpenCode Zen Anthropic Messages",
api_operator="opencode",
access_product="opencode_zen",
wire_protocol="anthropic_messages",
billing_mode="api",
base_url="https://opencode.ai/zen/v1",
credential_envs=["OPENCODE_API_KEY"],
max_retries=1,
transport_modalities=["text"],
tool_calling=True,
parallel_tool_calls=False,
resource_group="opencode_zen",
inventory="opencode-zen",
),
"opencode_zen_google_catalog": route(
driver="catalog_only",
name="OpenCode Zen Google-native models (catalog only)",
api_operator="opencode",
access_product="opencode_zen",
wire_protocol="catalog_only",
billing_mode="catalog_only",
transport_modalities=["text", "image", "video", "audio", "file"],
transport_output_modalities=["text"],
tool_calling=True,
parallel_tool_calls=False,
resource_group="opencode_zen",
inventory="opencode-zen",
),
# OpenCode Go uses different endpoints for different model families.
"opencode_go_openai_chat": route(
driver="switchyard",
name="OpenCode Go Chat Completions",
api_operator="opencode",
access_product="opencode_go",
wire_protocol="openai_chat",
billing_mode="subscription",
base_url="https://opencode.ai/zen/go/v1",
credential_envs=["OPENCODE_API_KEY"],
max_retries=1,
transport_modalities=["text"],
tool_calling=True,
parallel_tool_calls=True,
resource_group="opencode_go",
inventory="opencode-go",
),
"opencode_go_responses": route(
driver="switchyard",
name="OpenCode Go Responses",
api_operator="opencode",
access_product="opencode_go",
wire_protocol="openai_responses",
billing_mode="subscription",
base_url="https://opencode.ai/zen/go/v1",
credential_envs=["OPENCODE_API_KEY"],
max_retries=1,
transport_modalities=["text"],
tool_calling=True,
parallel_tool_calls=True,
resource_group="opencode_go",
inventory="opencode-go",
),
"opencode_go_anthropic_messages": route(
driver="switchyard",
name="OpenCode Go Anthropic Messages",
api_operator="opencode",
access_product="opencode_go",
wire_protocol="anthropic_messages",
billing_mode="subscription",
base_url="https://opencode.ai/zen/go/v1",
credential_envs=["OPENCODE_API_KEY"],
max_retries=1,
transport_modalities=["text"],
tool_calling=True,
parallel_tool_calls=True,
resource_group="opencode_go",
inventory="opencode-go",
),
"openrouter_openai_chat": route(
driver="switchyard",
name="OpenRouter Chat Completions",
api_operator="openrouter",
access_product="openrouter_api",
wire_protocol="openai_chat",
billing_mode="api",
base_url="https://openrouter.ai/api/v1",
credential_envs=["OPENROUTER_API_KEY"],
extra_headers={"X-OpenRouter-Metadata": "enabled"},
max_retries=1,
transport_modalities=["text"],
tool_calling=True,
parallel_tool_calls=True,
resource_group="openrouter",
inventory="openrouter",
),
"llama_cpp_local_openai_chat": route(
driver="switchyard",
name="Local llama.cpp",
api_operator="local",
access_product="llama_cpp",
wire_protocol="openai_chat",
billing_mode="local",
base_url="http://127.0.0.1:8001/v1",
max_retries=0,
transport_modalities=["text"],
tool_calling=True,
parallel_tool_calls=False,
resource_group="local_gpu_0",
),
"codex_chatgpt_builtin": route(
driver="codex_builtin",
name="Built-in Codex with ChatGPT authentication",
api_operator="openai",
access_product="chatgpt_codex",
wire_protocol="codex_builtin",
billing_mode="chatgpt_subscription",
provider_id="openai",
auth="chatgpt",
transport_modalities=["text", "image"],
tool_calling=True,
parallel_tool_calls=True,
resource_group="chatgpt_subscription",
inventory="openai-codex",
),
"openai_api_responses": route(
driver="switchyard",
name="OpenAI API through Switchyard",
api_operator="openai",
access_product="openai_api",
wire_protocol="openai_responses",
billing_mode="api",
base_url="https://api.openai.com/v1",
credential_envs=["OPENAI_API_KEY"],
max_retries=1,
transport_modalities=["text", "image"],
tool_calling=True,
parallel_tool_calls=True,
resource_group="openai_api",
),
"anthropic_api_messages": route(
driver="switchyard",
name="Anthropic API through Switchyard",
api_operator="anthropic",
access_product="anthropic_api",
wire_protocol="anthropic_messages",
billing_mode="api",
base_url="https://api.anthropic.com/v1",
credential_envs=["ANTHROPIC_API_KEY"],
max_retries=1,
transport_modalities=["text"],
tool_calling=True,
parallel_tool_calls=True,
resource_group="anthropic_api",
),
"ollama_local_openai_chat": route(
driver="switchyard",
name="Local Ollama through Switchyard",
api_operator="local",
access_product="ollama",
wire_protocol="openai_chat",
billing_mode="local",
base_url="http://127.0.0.1:11434/v1",
max_retries=0,
transport_modalities=["text"],
tool_calling=True,
parallel_tool_calls=False,
resource_group="local_gpu_0",
),
"lmstudio_local_openai_chat": route(
driver="switchyard",
name="Local LM Studio through Switchyard",
api_operator="local",
access_product="lmstudio",
wire_protocol="openai_chat",
billing_mode="local",
base_url="http://127.0.0.1:1234/v1",
max_retries=0,
transport_modalities=["text"],
tool_calling=True,
parallel_tool_calls=False,
resource_group="local_gpu_0",
),
"ollama_codex_oss": route(
driver="codex_oss",
name="Codex native Ollama OSS mode",
api_operator="local",
access_product="ollama",
wire_protocol="codex_oss",
billing_mode="local",
provider_id="ollama",
transport_modalities=["text"],
tool_calling=True,
parallel_tool_calls=False,
resource_group="local_gpu_0",
),
"lmstudio_codex_oss": route(
driver="codex_oss",
name="Codex native LM Studio OSS mode",
api_operator="local",
access_product="lmstudio",
wire_protocol="codex_oss",
billing_mode="local",
provider_id="lmstudio",
transport_modalities=["text"],
tool_calling=True,
parallel_tool_calls=False,
resource_group="local_gpu_0",
),
}
resources: dict[str, dict[str, Any]] = {
"opencode_zen": {
"description": "OpenCode Zen request capacity",
"lock_key": "provider:opencode-zen",
"max_active": 8,
},
"zai_coding_plan": {
"description": "Z.AI Coding Plan request capacity",
"lock_key": "provider:zai-coding-plan",
"max_active": 4,
},
"zai_api": {
"description": "Z.AI general API request capacity",
"lock_key": "provider:zai-api",
"max_active": 8,
},
"opencode_go": {
"description": "OpenCode Go request capacity",
"lock_key": "provider:opencode-go",
"max_active": 6,
},
"openrouter": {
"description": "OpenRouter request capacity",
"lock_key": "provider:openrouter",
"max_active": 8,
},
"local_gpu_0": {
"description": "One local GPU/model-server slot",
"lock_key": "local-gpu:0",
"max_active": 1,
},
"chatgpt_subscription": {
"description": "Built-in ChatGPT/Codex account concurrency",
"lock_key": "provider:chatgpt",
"max_active": 4,
},
"openai_api": {
"description": "OpenAI API concurrency",
"lock_key": "provider:openai-api",
"max_active": 8,
},
"anthropic_api": {
"description": "Anthropic API concurrency",
"lock_key": "provider:anthropic-api",
"max_active": 8,
},
}
models: dict[str, dict[str, Any]] = {}
binding_keys: dict[tuple[str, str], str] = {}
inventory_snapshots = load_inventory_snapshots(INVENTORY_SNAPSHOT_ROOT)
inventory_as_of = max(str(snapshot["as_of"]) for snapshot in inventory_snapshots)
for snapshot in inventory_snapshots:
inventory_id = str(snapshot["inventory"])
for key, record in snapshot["models"].items():
if key in models:
raise ValueError(f"duplicate catalog model key across inventory snapshots: {key}")
catalog_record = record["catalog"]
if catalog_record.get("inventory") != inventory_id:
raise ValueError(f"snapshot {inventory_id} model {key} has mismatched inventory")
binding = (str(catalog_record["route"]), str(catalog_record["upstream_id"]))
if binding in binding_keys:
raise ValueError(
f"duplicate route/upstream binding across inventory snapshots: "
f"{binding_keys[binding]!r} and {key!r} both select {binding!r}"
)
binding_keys[binding] = key
models[key] = dict(catalog_record)
# Project-local deployments are curated configuration, not upstream inventory.
models["llama_cpp_local_openai_chat__qwen3_5_9b"] = model(
"llama_cpp_local_openai_chat",
"qwen3.5-9b",
"qwen",
"Qwen3.5-9B local",
description="Project-capped text-only 32K/8K local evidence deployment; the upstream model is natively 262K and multimodal",
context=32_768,
output=8_192,
reasoning=["none"],
default_reasoning="none",
parallel_tool_calls=False,
summaries=False,
structured=False,
resource="local_gpu_0",
source="qwen35-model-card",
confidence="project-capped-deployment",
)
catalog = {
"schema_version": MMO_SCHEMA_VERSION,
"routes": routes,
"models": models,
"resources": resources,
}
validate_catalog_data(catalog, label="generated catalog")
inventory: dict[str, Any] = {
"schema_version": MMO_SCHEMA_VERSION,
"as_of": inventory_as_of,
"sources": {
"openai-api-models": "https://developers.openai.com/api/docs/models",
"qwen35-model-card": "https://huggingface.co/Qwen/Qwen3.5-9B",
"llama-cpp-server": "https://github.com/ggml-org/llama.cpp/blob/master/tools/server/README.md",
},
"inventories": {},
}
for snapshot in inventory_snapshots:
inventory_id = str(snapshot["inventory"])
for source_id, source_url in snapshot["sources"].items():
existing = inventory["sources"].get(source_id)
if existing is not None and existing != source_url:
raise ValueError(f"conflicting URL for inventory source {source_id}")
inventory["sources"][source_id] = source_url
catalog_keys = sorted(snapshot["models"])
model_ids = sorted(
{str(snapshot["models"][key]["catalog"]["upstream_id"]) for key in catalog_keys}
)
entry: dict[str, Any] = {
"as_of": snapshot["as_of"],
"snapshot": f"config/inventory-snapshots/{inventory_id}.json",
"adapter": snapshot["adapter"],
"fingerprint_fields": snapshot["fingerprint_fields"],
"models_sha256": snapshot["models_sha256"],
"dynamic": snapshot["dynamic"],
"expected_count": len(model_ids),
"models": model_ids,
"catalog_keys": catalog_keys,
}
entry.update(snapshot["discovery"])
if snapshot["captures"]:
entry["captures"] = snapshot["captures"]
inventory["inventories"][inventory_id] = entry
HEADER = f"""# Generated by scripts/generate_catalog.py. Do not edit this file directly.\n# Add local routes/models under ~/.config/codex-mmo/catalog.d/*.toml.\n# Inventory baseline: config/upstream-inventory.json (as of {inventory_as_of}).\n\n"""
def _rendered_outputs() -> dict[Path, str]:
return {
ROOT / "config" / "catalog.toml": HEADER + toml_dumps(catalog),
ROOT / "config" / "upstream-inventory.json": json.dumps(
inventory, ensure_ascii=False, indent=2, sort_keys=True, allow_nan=False
)
+ "\n",
}
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Generate the bundled model catalog")
parser.add_argument(
"--check",
action="store_true",
help="fail if generated files differ without modifying the source tree",
)
args = parser.parse_args(argv)
outputs = _rendered_outputs()
changed = [
path.relative_to(ROOT).as_posix()
for path, expected in outputs.items()
if not path.is_file() or path.read_text(encoding="utf-8") != expected
]
if not args.check:
atomic_write_text(
ROOT / "config" / "catalog.toml",
outputs[ROOT / "config" / "catalog.toml"],
0o644,
)
atomic_write_json(ROOT / "config" / "upstream-inventory.json", inventory, 0o644)
print(
json.dumps(
{
"routes": len(routes),
"models": len(models),
"resources": len(resources),
"inventory_models": {
key: len(value["models"])
for key, value in sorted(inventory["inventories"].items())
},
"changed": changed,
"passed": not changed if args.check else True,
},
indent=2,
sort_keys=True,
allow_nan=False,
)
)
return 1 if args.check and changed else 0
if __name__ == "__main__":
raise SystemExit(main())