"""Models.dev registry integration — primary database for providers and models. Fetches from https://models.dev/api.json — a community-maintained database of 4000+ models across 109+ providers. Provides: - **Provider metadata**: name, base URL, env vars, documentation link - **Model metadata**: context window, max output, cost/M tokens, capabilities (reasoning, tools, vision, PDF, audio), modalities, knowledge cutoff, open-weights flag, family grouping, deprecation status Data resolution order: 1. In-memory cache (fresh, or stale served immediately while a single background daemon thread refreshes) 2. Disk cache (~/.hermes/models_dev_cache.json — any age; stale data is served rather than blocking callers on the network) 3. Network fetch (https://models.dev/api.json) — only when no cache exists at all; failed refreshes back off for 5 minutes process-wide Latency-sensitive callers (gateway route-identity checks) pass ``allow_network=False`` and never touch the network. Other modules should import the dataclasses and query functions from here rather than parsing the raw JSON themselves. """ import json import logging import threading import time from dataclasses import dataclass from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from utils import atomic_json_write import requests logger = logging.getLogger(__name__) MODELS_DEV_URL = "https://models.dev/api.json" _MODELS_DEV_CACHE_TTL = 3600 # 1 hour in-memory _MODELS_DEV_RETRY_DELAY = 300 # 5 minutes after a failed refresh # In-memory cache _models_dev_cache: Dict[str, Any] = {} _models_dev_cache_time: float = 0 _models_dev_retry_after: float = 0 _models_dev_fetch_lock = threading.Lock() _models_dev_refresh_lock = threading.Lock() _models_dev_refresh_in_flight = False # --------------------------------------------------------------------------- # Dataclasses — rich metadata for providers and models # --------------------------------------------------------------------------- @dataclass class ModelInfo: """Full metadata for a single model from models.dev.""" id: str name: str family: str provider_id: str # models.dev provider ID (e.g. "anthropic") # Capabilities reasoning: bool = False tool_call: bool = False attachment: bool = False # supports image/file attachments (vision) temperature: bool = False structured_output: bool = False open_weights: bool = False # Modalities input_modalities: Tuple[str, ...] = () # ("text", "image", "pdf", ...) output_modalities: Tuple[str, ...] = () # Limits context_window: int = 0 max_output: int = 0 max_input: Optional[int] = None # Cost (per million tokens, USD) cost_input: float = 0.0 cost_output: float = 0.0 cost_cache_read: Optional[float] = None cost_cache_write: Optional[float] = None # Metadata knowledge_cutoff: str = "" release_date: str = "" status: str = "" # "alpha", "beta", "deprecated", or "" interleaved: Any = False # True or {"field": "reasoning_content"} def has_cost_data(self) -> bool: return self.cost_input > 0 or self.cost_output > 0 def supports_vision(self) -> bool: return self.attachment or "image" in self.input_modalities def supports_pdf(self) -> bool: return "pdf" in self.input_modalities def supports_audio_input(self) -> bool: return "audio" in self.input_modalities def format_cost(self) -> str: """Human-readable cost string, e.g. '$3.00/M in, $15.00/M out'.""" if not self.has_cost_data(): return "unknown" parts = [f"${self.cost_input:.2f}/M in", f"${self.cost_output:.2f}/M out"] if self.cost_cache_read is not None: parts.append(f"cache read ${self.cost_cache_read:.2f}/M") return ", ".join(parts) def format_capabilities(self) -> str: """Human-readable capabilities, e.g. 'reasoning, tools, vision, PDF'.""" caps = [] if self.reasoning: caps.append("reasoning") if self.tool_call: caps.append("tools") if self.supports_vision(): caps.append("vision") if self.supports_pdf(): caps.append("PDF") if self.supports_audio_input(): caps.append("audio") if self.structured_output: caps.append("structured output") if self.open_weights: caps.append("open weights") return ", ".join(caps) if caps else "basic" @dataclass class ProviderInfo: """Full metadata for a provider from models.dev.""" id: str # models.dev provider ID name: str # display name env: Tuple[str, ...] # env var names for API key api: str # base URL doc: str = "" # documentation URL model_count: int = 0 # --------------------------------------------------------------------------- # Provider ID mapping: Hermes ↔ models.dev # --------------------------------------------------------------------------- # Hermes provider names → models.dev provider IDs PROVIDER_TO_MODELS_DEV: Dict[str, str] = { "openrouter": "openrouter", "novita": "novita-ai", "anthropic": "anthropic", "openai": "openai", "openai-codex": "openai", "zai": "zai", "kimi": "kimi-for-coding", "kimi-coding": "kimi-for-coding", "moonshot": "kimi-for-coding", "stepfun": "stepfun", "kimi-coding-cn": "kimi-for-coding", "minimax": "minimax", "minimax-oauth": "minimax", "minimax-cn": "minimax-cn", "deepseek": "deepseek", "alibaba": "alibaba", "qwen-oauth": "alibaba", "copilot": "github-copilot", "opencode-zen": "opencode", "opencode-go": "opencode-go", "kilocode": "kilo", "fireworks": "fireworks-ai", "huggingface": "huggingface", "gemini": "google", "google": "google", "xai": "xai", # xAI OAuth is an authentication/transport path for the same xAI model # catalog, so model metadata should resolve through the xAI provider. "xai-oauth": "xai", "xiaomi": "xiaomi", "nvidia": "nvidia", "groq": "groq", "mistral": "mistral", "togetherai": "togetherai", "perplexity": "perplexity", "cohere": "cohere", "ollama-cloud": "ollama-cloud", } # Reverse mapping: models.dev → Hermes (built lazily) _MODELS_DEV_TO_PROVIDER: Optional[Dict[str, str]] = None def _get_cache_path() -> Path: """Return path to disk cache file.""" from hermes_constants import get_hermes_home return get_hermes_home() / "models_dev_cache.json" def _load_disk_cache() -> Dict[str, Any]: """Load models.dev data from disk cache.""" try: cache_path = _get_cache_path() if cache_path.exists(): with open(cache_path, encoding="utf-8") as f: return json.load(f) except Exception as e: logger.debug("Failed to load models.dev disk cache: %s", e) return {} def _disk_cache_age_seconds() -> Optional[float]: """Return age (in seconds) of the disk cache file, or None if missing. Used by ``fetch_models_dev`` to short-circuit the network probe when a recent on-disk cache exists. Errors (missing file, permission denied, weird filesystem) all return None — callers fall through to the network fetch path. """ try: cache_path = _get_cache_path() if not cache_path.exists(): return None mtime = cache_path.stat().st_mtime age = time.time() - mtime # Negative age means the file's mtime is in the future (clock skew # or system clock reset). Treat as "unknown freshness" → fall # through to network so we don't serve potentially-bad data # forever. if age < 0: return None return age except Exception as e: logger.debug("Failed to stat models.dev disk cache: %s", e) return None def _save_disk_cache(data: Dict[str, Any]) -> None: """Save models.dev data to disk cache atomically.""" try: cache_path = _get_cache_path() atomic_json_write(cache_path, data, indent=None, separators=(",", ":")) except Exception as e: logger.debug("Failed to save models.dev disk cache: %s", e) def _fetch_models_dev_from_network() -> Dict[str, Any]: """Fetch the live models.dev registry without touching local caches. Raises on network errors and on an empty/invalid registry payload. """ # Tuple (connect, read): a flat timeout=15 let a blackholed connect # stall the first-turn critical path for the full 15 s. 5 s connect # fails fast on unreachable hosts; 10 s read still tolerates a slow # registry response (matches the OpenRouter fetch convention in # agent/model_metadata.py). response = requests.get(MODELS_DEV_URL, timeout=(5, 10)) response.raise_for_status() data = response.json() if not isinstance(data, dict) or not data: raise ValueError("models.dev returned an empty or invalid registry") return data def _mark_stale_cache_grace() -> None: """Give stale cache data a short in-memory grace before retrying refresh. Only ever moves the timestamp forward: if a background refresh completed between the caller's staleness check and this call, the fresh timestamp is preserved instead of being rewound to a 5-minute grace. """ global _models_dev_cache_time grace_time = time.time() - _MODELS_DEV_CACHE_TTL + _MODELS_DEV_RETRY_DELAY if grace_time > _models_dev_cache_time: _models_dev_cache_time = grace_time def _commit_registry(data: Dict[str, Any], *, where: str) -> None: """Persist a freshly fetched registry: disk + in-mem + clear backoff. Callers must hold ``_models_dev_fetch_lock`` so a failing refresh on one path can never stomp the state a succeeding refresh on the other path just committed (e.g. a failing background worker re-arming the backoff immediately after a successful ``force_refresh``). """ global _models_dev_cache, _models_dev_cache_time, _models_dev_retry_after _save_disk_cache(data) _models_dev_cache = data _models_dev_cache_time = time.time() _models_dev_retry_after = 0 logger.debug( "Refreshed models.dev registry (%s): %d providers, %d total models", where, len(data), sum(len(p.get("models", {})) for p in data.values() if isinstance(p, dict)), ) def _note_refresh_failure(exc: Exception, *, where: str) -> None: """Record a failed refresh: arm the process-wide 5-minute backoff. Callers must hold ``_models_dev_fetch_lock`` (see ``_commit_registry``). """ global _models_dev_retry_after _models_dev_retry_after = time.time() + _MODELS_DEV_RETRY_DELAY logger.debug( "models.dev refresh failed (%s); retry suppressed for %ds: %s", where, _MODELS_DEV_RETRY_DELAY, exc, ) def _background_refresh_models_dev() -> None: """Best-effort refresh after serving stale cache data.""" global _models_dev_refresh_in_flight try: data = _fetch_models_dev_from_network() with _models_dev_fetch_lock: _commit_registry(data, where="background") except Exception as e: with _models_dev_fetch_lock: _note_refresh_failure(e, where="background") finally: with _models_dev_refresh_lock: _models_dev_refresh_in_flight = False def _start_background_refresh_models_dev() -> None: """Start one daemon refresh worker if none is already running. Honors the process-wide failure backoff: after a failed refresh, no new background worker is spawned until ``_models_dev_retry_after``. """ global _models_dev_refresh_in_flight if time.time() < _models_dev_retry_after: return with _models_dev_refresh_lock: if _models_dev_refresh_in_flight: return _models_dev_refresh_in_flight = True thread = threading.Thread( target=_background_refresh_models_dev, name="models-dev-refresh", daemon=True, ) try: thread.start() except Exception as e: # Thread/fd exhaustion: clear the flag so refresh isn't disabled # for the rest of the process lifetime. Callers still get stale data. with _models_dev_refresh_lock: _models_dev_refresh_in_flight = False logger.debug("Failed to start models.dev refresh thread: %s", e) def fetch_models_dev( force_refresh: bool = False, *, allow_network: bool = True ) -> Dict[str, Any]: """Fetch models.dev registry. Cache hierarchy: in-mem → disk → network. Returns the full registry dict keyed by provider ID, or empty dict on failure. Cache hierarchy (when ``force_refresh=False``): 1. Fresh in-memory cache → return immediately. 2. Stale in-memory cache → return immediately and refresh in a single background daemon thread. Callers never block on the network while any cache exists; ``models.dev`` only changes when providers add new models, so stale data is preferable to a foreground timeout. 3. Disk cache file (any age) → load, populate in-mem, return immediately. Stale disk caches trigger the same background refresh. 4. No cache at all → singleflight foreground network fetch. On success, save to disk + in-mem and return. 5. Any failed refresh (foreground or background) suppresses further automatic refreshes for 5 minutes process-wide. When ``force_refresh=True`` (used by ``hermes config refresh``, the \"refresh model catalog\" code path), cache fast paths and the failure backoff are bypassed; the function hits the network and only falls back to cached data if the call fails. When ``allow_network=False``, any memory or disk cache is returned regardless of age and no request is made — used by latency-sensitive paths (gateway route-identity checks) that must never wait on the network. """ global _models_dev_cache, _models_dev_cache_time, _models_dev_retry_after if not allow_network: if _models_dev_cache: return _models_dev_cache disk_data = _load_disk_cache() if disk_data: _models_dev_cache = disk_data disk_age = _disk_cache_age_seconds() _models_dev_cache_time = ( time.time() - disk_age if disk_age is not None else 0 ) return _models_dev_cache # Stage 1: fresh in-memory cache wins. This is the hot path on # long-lived processes — no I/O, no system calls. if ( not force_refresh and _models_dev_cache and (time.time() - _models_dev_cache_time) < _MODELS_DEV_CACHE_TTL ): return _models_dev_cache # Stage 2: stale in-memory cache is still better than blocking provider # resolution on a foreground network timeout. Refresh it in the background. if not force_refresh and _models_dev_cache: _mark_stale_cache_grace() _start_background_refresh_models_dev() logger.debug( "Using stale in-memory models.dev cache; refreshing in background" ) return _models_dev_cache # Stage 3: disk cache short-circuits the network call. # Only kicks in on cold-start processes (in-mem cache is empty) and only # when the user hasn't asked for a forced refresh. A stale disk cache is # deliberately usable: provider/model resolution should not hang just # because models.dev is unreachable. if not force_refresh: disk_age = _disk_cache_age_seconds() if disk_age is not None: disk_data = _load_disk_cache() if disk_data: _models_dev_cache = disk_data if disk_age < _MODELS_DEV_CACHE_TTL: # Anchor in-mem TTL to the disk file's age so we don't # extend an already-aging cache by another full hour. _models_dev_cache_time = time.time() - disk_age logger.debug( "Loaded models.dev from fresh disk cache " "(%d providers, age=%.0fs)", len(disk_data), disk_age, ) else: _mark_stale_cache_grace() _start_background_refresh_models_dev() logger.debug( "Using stale models.dev disk cache (age=%.0fs); " "refreshing in background", disk_age, ) return _models_dev_cache # Failed automatic refreshes are process-wide. Avoid making every caller # retry the same unreachable endpoint while no usable cache exists. if not force_refresh and time.time() < _models_dev_retry_after: return _models_dev_cache # Stage 4: singleflight foreground network fetch — only reached when no # memory or disk cache exists (or on force_refresh). Recheck state after # acquiring the lock because another caller may have refreshed or # established backoff while we waited. with _models_dev_fetch_lock: now = time.time() if not force_refresh: if _models_dev_cache: return _models_dev_cache if now < _models_dev_retry_after: return _models_dev_cache try: data = _fetch_models_dev_from_network() _commit_registry(data, where="foreground") return data except Exception as e: _note_refresh_failure(e, where="foreground") # Stage 5: network failed — return any stale memory/disk cache. Cache # freshness remains expired; the retry-after timestamp controls when # the next automatic request is allowed. if not _models_dev_cache: _models_dev_cache = _load_disk_cache() _models_dev_cache_time = 0 if _models_dev_cache: logger.debug( "Loaded stale models.dev disk cache (%d providers)", len(_models_dev_cache), ) return _models_dev_cache def lookup_models_dev_context(provider: str, model: str) -> Optional[int]: """Look up context_length for a provider+model combo in models.dev. Returns the context window in tokens, or None if not found. Handles case-insensitive matching and filters out context=0 entries. """ mdev_provider_id = PROVIDER_TO_MODELS_DEV.get(provider) if not mdev_provider_id: return None data = fetch_models_dev() provider_data = data.get(mdev_provider_id) if not isinstance(provider_data, dict): return None models = provider_data.get("models", {}) if not isinstance(models, dict): return None # Exact match entry = models.get(model) if entry: ctx = _extract_context(entry) if ctx: return ctx # Case-insensitive match model_lower = model.lower() for mid, mdata in models.items(): if mid.lower() == model_lower: ctx = _extract_context(mdata) if ctx: return ctx # Suffix-aware fallback: some providers (e.g. ollama-cloud) store # model IDs with :cloud / -cloud suffixes in models.dev while the # live API returns bare names. Without this, kimi-k2.6 misses the # kimi-k2.6:cloud entry and falls through to stale OpenRouter metadata # reporting 32768 — tripping the 64k minimum-context guard. # The suffix-stripping in fetch_ollama_cloud_models() handles the # model-picker UX; this handles the context-length lookup path. for suffix in (":cloud", "-cloud"): suffixed_key = model + suffix entry = models.get(suffixed_key) if entry: ctx = _extract_context(entry) if ctx: return ctx # Also try case-insensitive suffixed_lower = model_lower + suffix for mid, mdata in models.items(): if mid.lower() == suffixed_lower: ctx = _extract_context(mdata) if ctx: return ctx return None def _extract_context(entry: Dict[str, Any]) -> Optional[int]: """Extract context_length from a models.dev model entry. Returns None for invalid/zero values (some audio/image models have context=0). """ if not isinstance(entry, dict): return None limit = entry.get("limit") if not isinstance(limit, dict): return None ctx = limit.get("context") if isinstance(ctx, (int, float)) and ctx > 0: return int(ctx) return None # --------------------------------------------------------------------------- # Model capability metadata # --------------------------------------------------------------------------- @dataclass class ModelCapabilities: """Structured capability metadata for a model from models.dev.""" supports_tools: bool = True supports_vision: bool = False supports_reasoning: bool = False context_window: int = 200000 max_output_tokens: int = 8192 model_family: str = "" def _get_provider_models(provider: str) -> Optional[Dict[str, Any]]: """Resolve a Hermes provider ID to its models dict from models.dev. Returns the models dict or None if the provider is unknown or has no data. """ mdev_provider_id = PROVIDER_TO_MODELS_DEV.get(provider) if not mdev_provider_id: return None data = fetch_models_dev() provider_data = data.get(mdev_provider_id) if not isinstance(provider_data, dict): return None models = provider_data.get("models", {}) if not isinstance(models, dict): return None return models def _find_model_entry(models: Dict[str, Any], model: str) -> Optional[Dict[str, Any]]: """Find a model entry by exact match, then case-insensitive fallback.""" # Exact match entry = models.get(model) if isinstance(entry, dict): return entry # Case-insensitive match model_lower = model.lower() for mid, mdata in models.items(): if mid.lower() == model_lower and isinstance(mdata, dict): return mdata return None def get_model_capabilities(provider: str, model: str) -> Optional[ModelCapabilities]: """Look up full capability metadata from models.dev cache. Uses the existing fetch_models_dev() and PROVIDER_TO_MODELS_DEV mapping. Returns None if model not found. Extracts from model entry fields: - reasoning (bool) → supports_reasoning - tool_call (bool) → supports_tools - attachment (bool) → supports_vision - limit.context (int) → context_window - limit.output (int) → max_output_tokens - family (str) → model_family """ models = _get_provider_models(provider) if models is None: return None entry = _find_model_entry(models, model) if entry is None: return None # Extract capability flags (default to False if missing) supports_tools = bool(entry.get("tool_call", False)) # Vision: prefer explicit `modalities.input` when models.dev provides it. # The older `attachment` flag can be stale or too broad for image routing; # fall back to it only when the input modalities are absent/invalid. input_mods = entry.get("modalities", {}) if isinstance(input_mods, dict): input_mods = input_mods.get("input") else: input_mods = None if isinstance(input_mods, list): supports_vision = "image" in input_mods else: supports_vision = bool(entry.get("attachment", False)) supports_reasoning = bool(entry.get("reasoning", False)) # Extract limits limit = entry.get("limit", {}) if not isinstance(limit, dict): limit = {} ctx = limit.get("context") context_window = int(ctx) if isinstance(ctx, (int, float)) and ctx > 0 else 200000 out = limit.get("output") max_output_tokens = int(out) if isinstance(out, (int, float)) and out > 0 else 8192 model_family = entry.get("family", "") or "" return ModelCapabilities( supports_tools=supports_tools, supports_vision=supports_vision, supports_reasoning=supports_reasoning, context_window=context_window, max_output_tokens=max_output_tokens, model_family=model_family, ) def list_provider_models(provider: str) -> List[str]: """Return all model IDs for a provider from models.dev. Returns an empty list if the provider is unknown or has no data. """ from hermes_cli.models import normalize_provider provider = normalize_provider(provider) or provider models = _get_provider_models(provider) if models is None: return [] return [ mid for mid in models.keys() if not _should_hide_from_provider_catalog(provider, mid) ] # Patterns that indicate non-agentic or noise models (TTS, embedding, # dated preview snapshots, live/streaming-only, image-only). import re _NOISE_PATTERNS: re.Pattern = re.compile( r"-tts\b|embedding|live-|-(preview|exp)-\d{2,4}[-_]|" r"-image\b|-image-preview\b|-customtools\b", re.IGNORECASE, ) # Google's live Gemini catalogs currently include a mix of stale slugs and # Gemma models whose TPM quotas are too small for normal Hermes agent traffic. # Keep capability metadata available for direct/manual use, but hide these from # the Gemini model catalogs we surface in setup and model selection. _GOOGLE_HIDDEN_MODELS = frozenset({ # Low-TPM Gemma models that trip Google input-token quota walls under # agent-style traffic despite advertising large context windows. "gemma-4-31b-it", "gemma-4-26b-it", "gemma-4-26b-a4b-it", "gemma-3-1b", "gemma-3-1b-it", "gemma-3-2b", "gemma-3-2b-it", "gemma-3-4b", "gemma-3-4b-it", "gemma-3-12b", "gemma-3-12b-it", "gemma-3-27b", "gemma-3-27b-it", # Stale/retired Google slugs that still surface through models.dev-backed # Gemini selection but 404 on the current Google endpoints. "gemini-1.5-flash", "gemini-1.5-pro", "gemini-1.5-flash-8b", "gemini-2.0-flash", "gemini-2.0-flash-lite", }) def _should_hide_from_provider_catalog(provider: str, model_id: str) -> bool: provider_lower = (provider or "").strip().lower() model_lower = (model_id or "").strip().lower() if provider_lower in {"gemini", "google"} and model_lower in _GOOGLE_HIDDEN_MODELS: return True return False def list_agentic_models(provider: str) -> List[str]: """Return model IDs suitable for agentic use from models.dev. Filters for tool_call=True and excludes noise (TTS, embedding, dated preview snapshots, live/streaming, image-only models). Returns an empty list on any failure. """ models = _get_provider_models(provider) if models is None: return [] result = [] for mid, entry in models.items(): if not isinstance(entry, dict): continue if _should_hide_from_provider_catalog(provider, mid): continue if not entry.get("tool_call", False): continue if _NOISE_PATTERNS.search(mid): continue result.append(mid) return result # --------------------------------------------------------------------------- # Rich dataclass constructors — parse raw models.dev JSON into dataclasses # --------------------------------------------------------------------------- def _parse_model_info(model_id: str, raw: Dict[str, Any], provider_id: str) -> ModelInfo: """Convert a raw models.dev model entry dict into a ModelInfo dataclass.""" limit = raw.get("limit") or {} if not isinstance(limit, dict): limit = {} cost = raw.get("cost") or {} if not isinstance(cost, dict): cost = {} modalities = raw.get("modalities") or {} if not isinstance(modalities, dict): modalities = {} input_mods = modalities.get("input") or [] output_mods = modalities.get("output") or [] ctx = limit.get("context") ctx_int = int(ctx) if isinstance(ctx, (int, float)) and ctx > 0 else 0 out = limit.get("output") out_int = int(out) if isinstance(out, (int, float)) and out > 0 else 0 inp = limit.get("input") inp_int = int(inp) if isinstance(inp, (int, float)) and inp > 0 else None return ModelInfo( id=model_id, name=raw.get("name", "") or model_id, family=raw.get("family", "") or "", provider_id=provider_id, reasoning=bool(raw.get("reasoning", False)), tool_call=bool(raw.get("tool_call", False)), attachment=bool(raw.get("attachment", False)), temperature=bool(raw.get("temperature", False)), structured_output=bool(raw.get("structured_output", False)), open_weights=bool(raw.get("open_weights", False)), input_modalities=tuple(input_mods) if isinstance(input_mods, list) else (), output_modalities=tuple(output_mods) if isinstance(output_mods, list) else (), context_window=ctx_int, max_output=out_int, max_input=inp_int, cost_input=float(cost.get("input", 0) or 0), cost_output=float(cost.get("output", 0) or 0), cost_cache_read=float(cost["cache_read"]) if "cache_read" in cost and cost["cache_read"] is not None else None, cost_cache_write=float(cost["cache_write"]) if "cache_write" in cost and cost["cache_write"] is not None else None, knowledge_cutoff=raw.get("knowledge", "") or "", release_date=raw.get("release_date", "") or "", status=raw.get("status", "") or "", interleaved=raw.get("interleaved", False), ) def _parse_provider_info(provider_id: str, raw: Dict[str, Any]) -> ProviderInfo: """Convert a raw models.dev provider entry dict into a ProviderInfo.""" env = raw.get("env") or [] models = raw.get("models") or {} return ProviderInfo( id=provider_id, name=raw.get("name", "") or provider_id, env=tuple(env) if isinstance(env, list) else (), api=raw.get("api", "") or "", doc=raw.get("doc", "") or "", model_count=len(models) if isinstance(models, dict) else 0, ) # --------------------------------------------------------------------------- # Provider-level queries # --------------------------------------------------------------------------- def get_provider_info( provider_id: str, *, allow_network: bool = True ) -> Optional[ProviderInfo]: """Get full provider metadata from models.dev. Accepts either a Hermes provider ID (e.g. "kilocode") or a models.dev ID (e.g. "kilo"). Returns None if the provider is not in the catalog. """ # Resolve Hermes ID → models.dev ID mdev_id = PROVIDER_TO_MODELS_DEV.get(provider_id, provider_id) # NOTE: keep the zero-argument call on the default path. Dozens of test # sites monkeypatch fetch_models_dev with zero-arg lambdas; passing the # kwarg unconditionally would break them all (they raise TypeError). data = ( fetch_models_dev() if allow_network else fetch_models_dev(allow_network=False) ) raw = data.get(mdev_id) if not isinstance(raw, dict): return None return _parse_provider_info(mdev_id, raw) # --------------------------------------------------------------------------- # Model-level queries (rich ModelInfo) # --------------------------------------------------------------------------- def get_model_info( provider_id: str, model_id: str ) -> Optional[ModelInfo]: """Get full model metadata from models.dev. Accepts Hermes or models.dev provider ID. Tries exact match then case-insensitive fallback. Returns None if not found. """ mdev_id = PROVIDER_TO_MODELS_DEV.get(provider_id, provider_id) data = fetch_models_dev() pdata = data.get(mdev_id) if not isinstance(pdata, dict): return None models = pdata.get("models", {}) if not isinstance(models, dict): return None # Exact match raw = models.get(model_id) if isinstance(raw, dict): return _parse_model_info(model_id, raw, mdev_id) # Case-insensitive fallback model_lower = model_id.lower() for mid, mdata in models.items(): if mid.lower() == model_lower and isinstance(mdata, dict): return _parse_model_info(mid, mdata, mdev_id) return None