|
| 1 | +#!/usr/bin/env python3 |
| 2 | +"""Refresh the vendored models.dev pricing snapshot used by the gatekeeper.""" |
| 3 | + |
| 4 | +import json |
| 5 | +import sys |
| 6 | +import urllib.request |
| 7 | + |
| 8 | +from dataclasses import dataclass |
| 9 | +from pathlib import Path |
| 10 | +from typing import Any |
| 11 | + |
| 12 | + |
| 13 | +REPO_ROOT = Path(__file__).resolve().parents[1] |
| 14 | +OUTPUT_PATH = REPO_ROOT / "src/linux_mcp_server/gatekeeper/data/models_dev_fallback.json" |
| 15 | +MODELS_DEV_URL = "https://models.dev/api.json" |
| 16 | + |
| 17 | +# Provider -> model IDs used by eval/gatekeeper/standard-evals.sh |
| 18 | +WANTED: dict[str, list[str]] = { |
| 19 | + "anthropic": ["claude-sonnet-4-6", "claude-opus-4-6", "claude-haiku-4-5"], |
| 20 | + "openai": ["gpt-5.4", "gpt-oss-20b", "gpt-oss-120b"], |
| 21 | + "google": ["gemini-2.0-flash", "gemini-3.1-pro-preview"], |
| 22 | + "openrouter": [ |
| 23 | + "openai/gpt-oss-120b", |
| 24 | + "anthropic/claude-sonnet-4-6", |
| 25 | + "google/gemma-4-26b-a4b-it", |
| 26 | + "ibm-granite/granite-4.0-h-small", |
| 27 | + "qwen/qwen3.5-35b-a3b", |
| 28 | + ], |
| 29 | +} |
| 30 | + |
| 31 | + |
| 32 | +@dataclass(frozen=True) |
| 33 | +class TokenCostPerMillion: |
| 34 | + """USD per million tokens, matching models.dev's cost.input / cost.output fields.""" |
| 35 | + |
| 36 | + input: float |
| 37 | + output: float |
| 38 | + |
| 39 | + def to_dict(self) -> dict[str, float]: |
| 40 | + return {"input": self.input, "output": self.output} |
| 41 | + |
| 42 | + |
| 43 | +@dataclass(frozen=True) |
| 44 | +class ModelPricing: |
| 45 | + """Pricing structure for a single model. Contains the cost per million tokens.""" |
| 46 | + |
| 47 | + cost: TokenCostPerMillion |
| 48 | + |
| 49 | + def to_dict(self) -> dict[str, Any]: |
| 50 | + return {"cost": self.cost.to_dict()} |
| 51 | + |
| 52 | + @classmethod |
| 53 | + def from_models_dev_entry(cls, entry: object) -> "ModelPricing | None": |
| 54 | + if not isinstance(entry, dict): |
| 55 | + return None |
| 56 | + cost = entry.get("cost") |
| 57 | + if not isinstance(cost, dict): |
| 58 | + return None |
| 59 | + input_mtok = cost.get("input") |
| 60 | + output_mtok = cost.get("output") |
| 61 | + if not isinstance(input_mtok, (int, float)) or not isinstance(output_mtok, (int, float)): |
| 62 | + return None |
| 63 | + return cls(cost=TokenCostPerMillion(input=float(input_mtok), output=float(output_mtok))) |
| 64 | + |
| 65 | + |
| 66 | +@dataclass |
| 67 | +class ProviderPricing: |
| 68 | + """Pricing structure for a single provider. Maps model ID to its pricing.""" |
| 69 | + |
| 70 | + models: dict[str, ModelPricing] |
| 71 | + |
| 72 | + def to_dict(self) -> dict[str, Any]: |
| 73 | + return {"models": {model_id: pricing.to_dict() for model_id, pricing in self.models.items()}} |
| 74 | + |
| 75 | + |
| 76 | +@dataclass |
| 77 | +class PricingSnapshot: |
| 78 | + """Snapshot of models.dev pricing for the gatekeeper. Maps provider name to a pricing structure.""" |
| 79 | + |
| 80 | + providers: dict[str, ProviderPricing] |
| 81 | + |
| 82 | + def to_dict(self) -> dict[str, Any]: |
| 83 | + return {provider: pricing.to_dict() for provider, pricing in self.providers.items()} |
| 84 | + |
| 85 | + |
| 86 | +def _models_dev_catalog(data: dict[str, Any], provider: str) -> dict[str, Any]: |
| 87 | + """Extract the models catalog for a given provider from the models.dev data.""" |
| 88 | + provider_data = data.get(provider, {}) |
| 89 | + if not isinstance(provider_data, dict): |
| 90 | + return {} |
| 91 | + models = provider_data.get("models", {}) |
| 92 | + return models if isinstance(models, dict) else {} |
| 93 | + |
| 94 | + |
| 95 | +def build_snapshot(data: dict[str, Any]) -> tuple[PricingSnapshot, list[str]]: |
| 96 | + """Build a pricing snapshot from the models.dev data.""" |
| 97 | + providers: dict[str, ProviderPricing] = {} |
| 98 | + missing: list[str] = [] |
| 99 | + |
| 100 | + for provider, model_ids in WANTED.items(): |
| 101 | + src_models = _models_dev_catalog(data, provider) |
| 102 | + picked: dict[str, ModelPricing] = {} |
| 103 | + for model_id in model_ids: |
| 104 | + pricing = ModelPricing.from_models_dev_entry(src_models.get(model_id)) |
| 105 | + if pricing is None: |
| 106 | + missing.append(f"{provider}/{model_id}") |
| 107 | + else: |
| 108 | + picked[model_id] = pricing |
| 109 | + if picked: |
| 110 | + providers[provider] = ProviderPricing(models=picked) |
| 111 | + |
| 112 | + return PricingSnapshot(providers=providers), missing |
| 113 | + |
| 114 | + |
| 115 | +def main() -> int: |
| 116 | + with urllib.request.urlopen(MODELS_DEV_URL, timeout=30) as response: |
| 117 | + data = json.load(response) |
| 118 | + |
| 119 | + snapshot, missing = build_snapshot(data) |
| 120 | + |
| 121 | + OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True) |
| 122 | + OUTPUT_PATH.write_text(json.dumps(snapshot.to_dict(), indent=2) + "\n", encoding="utf-8") |
| 123 | + print(f"Wrote {OUTPUT_PATH}") |
| 124 | + |
| 125 | + if missing: |
| 126 | + print("Missing pricing for:", ", ".join(missing), file=sys.stderr) |
| 127 | + return 1 |
| 128 | + return 0 |
| 129 | + |
| 130 | + |
| 131 | +if __name__ == "__main__": |
| 132 | + raise SystemExit(main()) |
0 commit comments