diff --git a/agent/models_dev.py b/agent/models_dev.py index f583262445..61483b6a10 100644 --- a/agent/models_dev.py +++ b/agent/models_dev.py @@ -1,11 +1,21 @@ -"""Models.dev registry integration for provider-aware context length detection. +"""Models.dev registry integration — primary database for providers and models. -Fetches model metadata from https://models.dev/api.json — a community-maintained -database of 3800+ models across 100+ providers, including per-provider context -windows, pricing, and capabilities. +Fetches from https://models.dev/api.json — a community-maintained database +of 4000+ models across 109+ providers. Provides: -Data is cached in memory (1hr TTL) and on disk (~/.hermes/models_dev_cache.json) -to avoid cold-start network latency. +- **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 (like TypeScript OpenCode): + 1. Bundled snapshot (ships with the package — offline-first) + 2. Disk cache (~/.hermes/models_dev_cache.json) + 3. Network fetch (https://models.dev/api.json) + 4. Background refresh every 60 minutes + +Other modules should import the dataclasses and query functions from here +rather than parsing the raw JSON themselves. """ import difflib @@ -15,7 +25,7 @@ import os import time from dataclasses import dataclass, field from pathlib import Path -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple, Union from utils import atomic_json_write @@ -30,7 +40,110 @@ _MODELS_DEV_CACHE_TTL = 3600 # 1 hour in-memory _models_dev_cache: Dict[str, Any] = {} _models_dev_cache_time: float = 0 -# Provider ID mapping: Hermes provider names → models.dev provider IDs + +# --------------------------------------------------------------------------- +# 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 + + def has_api_url(self) -> bool: + return bool(self.api) + + +# --------------------------------------------------------------------------- +# Provider ID mapping: Hermes ↔ models.dev +# --------------------------------------------------------------------------- + +# Hermes provider names → models.dev provider IDs PROVIDER_TO_MODELS_DEV: Dict[str, str] = { "openrouter": "openrouter", "anthropic": "anthropic", @@ -46,8 +159,28 @@ PROVIDER_TO_MODELS_DEV: Dict[str, str] = { "opencode-go": "opencode-go", "kilocode": "kilo", "fireworks": "fireworks-ai", + "huggingface": "huggingface", + "google": "google", + "xai": "xai", + "nvidia": "nvidia", + "groq": "groq", + "mistral": "mistral", + "togetherai": "togetherai", + "perplexity": "perplexity", + "cohere": "cohere", } +# Reverse mapping: models.dev → Hermes (built lazily) +_MODELS_DEV_TO_PROVIDER: Optional[Dict[str, str]] = None + + +def _get_reverse_mapping() -> Dict[str, str]: + """Return models.dev ID → Hermes provider ID mapping.""" + global _MODELS_DEV_TO_PROVIDER + if _MODELS_DEV_TO_PROVIDER is None: + _MODELS_DEV_TO_PROVIDER = {v: k for k, v in PROVIDER_TO_MODELS_DEV.items()} + return _MODELS_DEV_TO_PROVIDER + def _get_cache_path() -> Path: """Return path to disk cache file.""" @@ -375,3 +508,240 @@ def search_models_dev( return results return results + + +# --------------------------------------------------------------------------- +# 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) -> 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) + + data = fetch_models_dev() + raw = data.get(mdev_id) + if not isinstance(raw, dict): + return None + + return _parse_provider_info(mdev_id, raw) + + +def list_all_providers() -> Dict[str, ProviderInfo]: + """Return all providers from models.dev as {provider_id: ProviderInfo}. + + Returns the full catalog — 109+ providers. For providers that have + a Hermes alias, both the models.dev ID and the Hermes ID are included. + """ + data = fetch_models_dev() + result: Dict[str, ProviderInfo] = {} + + for pid, pdata in data.items(): + if isinstance(pdata, dict): + info = _parse_provider_info(pid, pdata) + result[pid] = info + + return result + + +def get_providers_for_env_var(env_var: str) -> List[str]: + """Reverse lookup: find all providers that use a given env var. + + Useful for auto-detection: "user has ANTHROPIC_API_KEY set, which + providers does that enable?" + + Returns list of models.dev provider IDs. + """ + data = fetch_models_dev() + matches: List[str] = [] + + for pid, pdata in data.items(): + if isinstance(pdata, dict): + env = pdata.get("env", []) + if isinstance(env, list) and env_var in env: + matches.append(pid) + + return matches + + +# --------------------------------------------------------------------------- +# 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 + + +def get_model_info_any_provider(model_id: str) -> Optional[ModelInfo]: + """Search all providers for a model by ID. + + Useful when you have a full slug like "anthropic/claude-sonnet-4.6" or + a bare name and want to find it anywhere. Checks Hermes-mapped providers + first, then falls back to all models.dev providers. + """ + data = fetch_models_dev() + + # Try Hermes-mapped providers first (more likely what the user wants) + for hermes_id, mdev_id in PROVIDER_TO_MODELS_DEV.items(): + pdata = data.get(mdev_id) + if not isinstance(pdata, dict): + continue + models = pdata.get("models", {}) + if not isinstance(models, dict): + continue + + raw = models.get(model_id) + if isinstance(raw, dict): + return _parse_model_info(model_id, raw, mdev_id) + + # Case-insensitive + 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) + + # Fall back to ALL providers + for pid, pdata in data.items(): + if pid in _get_reverse_mapping(): + continue # already checked + if not isinstance(pdata, dict): + continue + models = pdata.get("models", {}) + if not isinstance(models, dict): + continue + + raw = models.get(model_id) + if isinstance(raw, dict): + return _parse_model_info(model_id, raw, pid) + + return None + + +def list_provider_model_infos(provider_id: str) -> List[ModelInfo]: + """Return all models for a provider as ModelInfo objects. + + Filters out deprecated models by default. + """ + 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 [] + + models = pdata.get("models", {}) + if not isinstance(models, dict): + return [] + + result: List[ModelInfo] = [] + for mid, mdata in models.items(): + if not isinstance(mdata, dict): + continue + status = mdata.get("status", "") + if status == "deprecated": + continue + result.append(_parse_model_info(mid, mdata, mdev_id)) + + return result diff --git a/cli.py b/cli.py index 0cddb60211..f67e37441d 100644 --- a/cli.py +++ b/cli.py @@ -3523,51 +3523,63 @@ class HermesCLI: """Handle /model command — switch model for this session. Supports: - /model — show current model + usage hints - /model — switch for this session only - /model --global — switch and persist to config.yaml - /model provider:model — switch provider + model + /model — show current model + usage hints + /model — switch for this session only + /model --global — switch and persist to config.yaml + /model --provider — switch provider + model + /model --provider — switch to provider, auto-detect model """ - from hermes_cli.model_switch import switch_model, ModelSwitchResult - from hermes_cli.models import _PROVIDER_LABELS + from hermes_cli.model_switch import switch_model, parse_model_flags + from hermes_cli.providers import get_label # Parse args from the original command parts = cmd_original.split(None, 1) # split off '/model' raw_args = parts[1].strip() if len(parts) > 1 else "" - # Check for --global flag - persist_global = "--global" in raw_args - if persist_global: - raw_args = raw_args.replace("--global", "").strip() + # Parse --provider and --global flags + model_input, explicit_provider, persist_global = parse_model_flags(raw_args) - # No args: show current model info and usage - if not raw_args: + # No args at all: show current model info and usage + if not model_input and not explicit_provider: model_display = self.model or "unknown" - provider_display = _PROVIDER_LABELS.get(self.provider, self.provider) if self.provider else "unknown" + provider_display = get_label(self.provider) if self.provider else "unknown" _cprint(f" Current model: {model_display}") _cprint(f" Provider: {provider_display} ({self.provider})") if self.base_url: _cprint(f" Endpoint: {self.base_url}") + + # Show model metadata if available + try: + from agent.models_dev import get_model_info + mi = get_model_info(self.provider, self.model) + if mi: + _cprint(f" Context: {mi.context_window:,} tokens") + if mi.max_output: + _cprint(f" Max output: {mi.max_output:,} tokens") + if mi.has_cost_data(): + _cprint(f" Cost: {mi.format_cost()}") + _cprint(f" Capabilities: {mi.format_capabilities()}") + except Exception: + pass + _cprint("") - _cprint(" Aliases: /model sonnet") - _cprint(" /model opus") - _cprint(" /model gpt5") - _cprint(" /model gemini") - _cprint("") - _cprint(" Full name: /model anthropic/claude-sonnet-4.5") - _cprint(" Provider: /model anthropic:sonnet (switch to native Anthropic)") - _cprint(" /model deepseek:deepseek-chat (switch to native DeepSeek)") - _cprint(" Persist: /model sonnet --global (save to config)") - _cprint(" Custom: /model custom:my-local-model") + _cprint(" Usage:") + _cprint(" /model sonnet (alias)") + _cprint(" /model anthropic/claude-sonnet-4.6 (full name)") + _cprint(" /model sonnet --provider anthropic (switch provider)") + _cprint(" /model --provider my-ollama (auto-detect model)") + _cprint(" /model sonnet --global (persist to config)") return # Perform the switch result = switch_model( - raw_input=raw_args, + raw_input=model_input, current_provider=self.provider or "", current_model=self.model or "", current_base_url=self.base_url or "", current_api_key=self.api_key or "", + is_global=persist_global, + explicit_provider=explicit_provider, ) if not result.success: @@ -3598,23 +3610,34 @@ class HermesCLI: except Exception as exc: _cprint(f" ⚠ Agent swap failed ({exc}); change applied to next session.") - # Display confirmation + # Display confirmation with full metadata provider_label = result.provider_label or result.target_provider _cprint(f" ✓ Model switched: {result.new_model}") _cprint(f" Provider: {provider_label}") - # Show context window info - try: - from agent.model_metadata import get_model_context_length - ctx = get_model_context_length( - result.new_model, - base_url=result.base_url or self.base_url, - api_key=result.api_key or self.api_key, - provider=result.target_provider, - ) - _cprint(f" Context: {ctx:,} tokens") - except Exception: - pass + # Rich metadata from models.dev + mi = result.model_info + if mi: + if mi.context_window: + _cprint(f" Context: {mi.context_window:,} tokens") + if mi.max_output: + _cprint(f" Max output: {mi.max_output:,} tokens") + if mi.has_cost_data(): + _cprint(f" Cost: {mi.format_cost()}") + _cprint(f" Capabilities: {mi.format_capabilities()}") + else: + # Fallback to old context length lookup + try: + from agent.model_metadata import get_model_context_length + ctx = get_model_context_length( + result.new_model, + base_url=result.base_url or self.base_url, + api_key=result.api_key or self.api_key, + provider=result.target_provider, + ) + _cprint(f" Context: {ctx:,} tokens") + except Exception: + pass # Cache notice cache_enabled = ( diff --git a/gateway/run.py b/gateway/run.py index fb39550a8a..1111c141bb 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -3250,21 +3250,20 @@ class GatewayRunner: """Handle /model command — switch model for this session. Supports: - /model — show current model info - /model — switch for this session only - /model --global — switch and persist to config.yaml - /model provider:model — switch provider + model + /model — show current model info + /model — switch for this session only + /model --global — switch and persist to config.yaml + /model --provider — switch provider + model + /model --provider — switch to provider, auto-detect model """ import yaml - from hermes_cli.model_switch import switch_model as _switch_model - from hermes_cli.models import _PROVIDER_LABELS + from hermes_cli.model_switch import switch_model as _switch_model, parse_model_flags + from hermes_cli.providers import get_label raw_args = event.get_command_args().strip() - # Check for --global flag - persist_global = "--global" in raw_args - if persist_global: - raw_args = raw_args.replace("--global", "").strip() + # Parse --provider and --global flags + model_input, explicit_provider, persist_global = parse_model_flags(raw_args) # Read current model/provider from config current_model = "" @@ -3287,7 +3286,7 @@ class GatewayRunner: # Check for session override source = event.source session_key = self._session_key_for_source(source) - override = self._session_model_overrides.get(session_key, {}) + override = getattr(self, "_session_model_overrides", {}).get(session_key, {}) if override: current_model = override.get("model", current_model) current_provider = override.get("provider", current_provider) @@ -3295,37 +3294,50 @@ class GatewayRunner: current_api_key = override.get("api_key", current_api_key) # No args: show current model info - if not raw_args: - provider_label = _PROVIDER_LABELS.get(current_provider, current_provider) + if not model_input and not explicit_provider: + provider_label = get_label(current_provider) lines = [ - f"🤖 **Current model:** `{current_model or 'unknown'}`", - f"🔌 **Provider:** {provider_label} (`{current_provider}`)", + f"Current model: `{current_model or 'unknown'}`", + f"Provider: {provider_label} (`{current_provider}`)", ] if current_base_url: - lines.append(f"🌐 **Endpoint:** `{current_base_url}`") + lines.append(f"Endpoint: `{current_base_url}`") + + # Show model metadata from models.dev + try: + from agent.models_dev import get_model_info + mi = get_model_info(current_provider, current_model) + if mi: + lines.append(f"Context: {mi.context_window:,} tokens") + if mi.max_output: + lines.append(f"Max output: {mi.max_output:,} tokens") + if mi.has_cost_data(): + lines.append(f"Cost: {mi.format_cost()}") + lines.append(f"Capabilities: {mi.format_capabilities()}") + except Exception: + pass + lines.append("") lines.append("**Usage:**") - lines.append("`/model ` — switch for this session") - lines.append("`/model provider:model` — switch provider + model") - lines.append("`/model --global` — persist to config") - lines.append("") - lines.append("**Examples:**") - lines.append("`/model claude-sonnet-4`") - lines.append("`/model openai:gpt-4.1`") - lines.append("`/model anthropic:claude-sonnet-4 --global`") + lines.append("`/model sonnet` — alias") + lines.append("`/model sonnet --provider anthropic` — switch provider") + lines.append("`/model --provider my-ollama` — auto-detect model") + lines.append("`/model sonnet --global` — persist to config") return "\n".join(lines) # Perform the switch result = _switch_model( - raw_input=raw_args, + raw_input=model_input, current_provider=current_provider, current_model=current_model, current_base_url=current_base_url, current_api_key=current_api_key, + is_global=persist_global, + explicit_provider=explicit_provider, ) if not result.success: - return f"❌ {result.error_message}" + return f"Error: {result.error_message}" # If there's a cached agent, update it in-place cached_entry = None @@ -3348,6 +3360,8 @@ class GatewayRunner: logger.warning("In-place model switch failed for cached agent: %s", exc) # Store session override so next agent creation uses the new model + if not hasattr(self, "_session_model_overrides"): + self._session_model_overrides = {} self._session_model_overrides[session_key] = { "model": result.new_model, "provider": result.target_provider, @@ -3374,23 +3388,33 @@ class GatewayRunner: except Exception as e: logger.warning("Failed to persist model switch: %s", e) - # Build confirmation message + # Build confirmation message with full metadata provider_label = result.provider_label or result.target_provider - lines = [f"✅ Model switched to `{result.new_model}`"] + lines = [f"Model switched to `{result.new_model}`"] lines.append(f"Provider: {provider_label}") - # Context window info - try: - from agent.model_metadata import get_model_context_length - ctx = get_model_context_length( - result.new_model, - base_url=result.base_url or current_base_url, - api_key=result.api_key or current_api_key, - provider=result.target_provider, - ) - lines.append(f"📏 Context: {ctx:,} tokens") - except Exception: - pass + # Rich metadata from models.dev + mi = result.model_info + if mi: + if mi.context_window: + lines.append(f"Context: {mi.context_window:,} tokens") + if mi.max_output: + lines.append(f"Max output: {mi.max_output:,} tokens") + if mi.has_cost_data(): + lines.append(f"Cost: {mi.format_cost()}") + lines.append(f"Capabilities: {mi.format_capabilities()}") + else: + try: + from agent.model_metadata import get_model_context_length + ctx = get_model_context_length( + result.new_model, + base_url=result.base_url or current_base_url, + api_key=result.api_key or current_api_key, + provider=result.target_provider, + ) + lines.append(f"Context: {ctx:,} tokens") + except Exception: + pass # Cache notice cache_enabled = ( @@ -3398,15 +3422,15 @@ class GatewayRunner: or result.api_mode == "anthropic_messages" ) if cache_enabled: - lines.append("💾 Prompt caching: enabled") + lines.append("Prompt caching: enabled") if result.warning_message: - lines.append(f"⚠️ {result.warning_message}") + lines.append(f"Warning: {result.warning_message}") if persist_global: - lines.append("💾 Saved to config.yaml (`--global`)") + lines.append("Saved to config.yaml (`--global`)") else: - lines.append("_(session only — add `--global` to persist)_") + lines.append("_(session only -- add `--global` to persist)_") return "\n".join(lines) diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 00d0923d2f..abdd4a396d 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -199,6 +199,7 @@ def ensure_hermes_home(): DEFAULT_CONFIG = { "model": "", + "providers": {}, "fallback_providers": [], "credential_pool_strategies": {}, "toolsets": ["hermes-cli"], diff --git a/hermes_cli/model_switch.py b/hermes_cli/model_switch.py index a4d2bd4907..20f8060dd0 100644 --- a/hermes_cli/model_switch.py +++ b/hermes_cli/model_switch.py @@ -3,18 +3,19 @@ Both the CLI (cli.py) and gateway (gateway/run.py) /model handlers share the same core pipeline: - alias resolution -> parse_model_input -> aggregator slug fixup -> - detect_provider -> credential resolution -> normalize model name -> - capability lookup -> build result + parse flags -> alias resolution -> provider resolution -> + credential resolution -> normalize model name -> + metadata lookup -> build result This module ties together the foundation layers: -- ``hermes_cli.providers`` -- canonical provider identity +- ``agent.models_dev`` -- models.dev catalog, ModelInfo, ProviderInfo +- ``hermes_cli.providers`` -- canonical provider identity + overlays - ``hermes_cli.model_normalize`` -- per-provider name formatting -- ``agent.models_dev`` -- models.dev catalog & capabilities -Callers handle all platform-specific concerns: state mutation, config -persistence, output formatting. +Provider switching uses the ``--provider`` flag exclusively. +No colon-based ``provider:model`` syntax — colons are reserved for +OpenRouter variant suffixes (``:free``, ``:extended``, ``:fast``). """ from __future__ import annotations @@ -26,11 +27,13 @@ from typing import List, NamedTuple, Optional from hermes_cli.providers import ( ALIASES, LABELS, - PROVIDERS, + TRANSPORT_TO_API_MODE, determine_api_mode, + get_label, get_provider, is_aggregator, normalize_provider, + resolve_provider_full, ) from hermes_cli.model_normalize import ( detect_vendor, @@ -38,7 +41,9 @@ from hermes_cli.model_normalize import ( ) from agent.models_dev import ( ModelCapabilities, + ModelInfo, get_model_capabilities, + get_model_info, list_provider_models, search_models_dev, ) @@ -129,6 +134,7 @@ class ModelSwitchResult: provider_label: str = "" resolved_via_alias: str = "" capabilities: Optional[ModelCapabilities] = None + model_info: Optional[ModelInfo] = None is_global: bool = False @@ -144,14 +150,44 @@ class CustomAutoResult: # --------------------------------------------------------------------------- -# Known provider names (for parse_model_input colon-splitting) +# Flag parsing # --------------------------------------------------------------------------- -_KNOWN_PROVIDER_NAMES: set[str] = ( - set(PROVIDERS.keys()) - | set(ALIASES.keys()) - | {"custom"} -) +def parse_model_flags(raw_args: str) -> tuple[str, str, bool]: + """Parse --provider and --global flags from /model command args. + + Returns (model_input, explicit_provider, is_global). + + Examples:: + + "sonnet" -> ("sonnet", "", False) + "sonnet --global" -> ("sonnet", "", True) + "sonnet --provider anthropic" -> ("sonnet", "anthropic", False) + "--provider my-ollama" -> ("", "my-ollama", False) + "sonnet --provider anthropic --global" -> ("sonnet", "anthropic", True) + """ + is_global = False + explicit_provider = "" + + # Extract --global + if "--global" in raw_args: + is_global = True + raw_args = raw_args.replace("--global", "").strip() + + # Extract --provider + parts = raw_args.split() + i = 0 + filtered: list[str] = [] + while i < len(parts): + if parts[i] == "--provider" and i + 1 < len(parts): + explicit_provider = parts[i + 1] + i += 2 + else: + filtered.append(parts[i]) + i += 1 + + model_input = " ".join(filtered).strip() + return (model_input, explicit_provider, is_global) # --------------------------------------------------------------------------- @@ -169,10 +205,6 @@ def resolve_alias( starts with ``vendor/family`` (or just ``family`` for non-aggregator providers). - Args: - raw_input: The user's raw model input (e.g. ``"sonnet"``). - current_provider: The currently active Hermes provider id. - Returns: ``(provider, resolved_model_id, alias_name)`` if a match is found on the current provider, or ``None`` if the alias doesn't @@ -202,7 +234,6 @@ def resolve_alias( return (current_provider, model_id, key) else: # Non-aggregator: bare names -- e.g. "claude-sonnet-4-6" - # Match family prefix (dots/hyphens may vary) family_lower = family.lower() if mid_lower.startswith(family_lower): return (current_provider, model_id, key) @@ -214,14 +245,7 @@ def _resolve_alias_fallback( raw_input: str, fallback_providers: tuple[str, ...] = ("openrouter", "nous"), ) -> Optional[tuple[str, str, str]]: - """Try to resolve an alias on fallback providers. - - Called when the alias exists in MODEL_ALIASES but no matching model - was found on the current provider. - - Returns: - ``(provider, resolved_model_id, alias_name)`` or ``None``. - """ + """Try to resolve an alias on fallback providers.""" for provider in fallback_providers: result = resolve_alias(raw_input, provider) if result is not None: @@ -229,40 +253,6 @@ def _resolve_alias_fallback( return None -# --------------------------------------------------------------------------- -# parse_model_input -- shared with models.py but using our provider set -# --------------------------------------------------------------------------- - -def parse_model_input(raw: str, current_provider: str) -> tuple[str, str]: - """Parse ``/model`` input into ``(provider, model)``. - - Supports ``provider:model`` syntax to switch providers at runtime:: - - openrouter:anthropic/claude-sonnet-4.5 -> ("openrouter", "anthropic/claude-sonnet-4.5") - nous:hermes-3 -> ("nous", "hermes-3") - anthropic/claude-sonnet-4.5 -> (current_provider, "anthropic/claude-sonnet-4.5") - gpt-5.4 -> (current_provider, "gpt-5.4") - - The colon is only treated as a provider delimiter if the left side - is a recognized provider name or alias. - """ - stripped = raw.strip() - colon = stripped.find(":") - if colon > 0: - provider_part = stripped[:colon].strip().lower() - model_part = stripped[colon + 1:].strip() - if provider_part and model_part and provider_part in _KNOWN_PROVIDER_NAMES: - # Support custom:name:model triple syntax - if provider_part == "custom" and ":" in model_part: - second_colon = model_part.find(":") - custom_name = model_part[:second_colon].strip() - actual_model = model_part[second_colon + 1:].strip() - if custom_name and actual_model: - return (f"custom:{custom_name}", actual_model) - return (normalize_provider(provider_part), model_part) - return (current_provider, stripped) - - # --------------------------------------------------------------------------- # Core model-switching pipeline # --------------------------------------------------------------------------- @@ -274,31 +264,41 @@ def switch_model( current_base_url: str = "", current_api_key: str = "", is_global: bool = False, + explicit_provider: str = "", + user_providers: dict = None, ) -> ModelSwitchResult: """Core model-switching pipeline shared between CLI and gateway. Resolution chain: - a. Try alias resolution on current provider - b. If alias exists but not on current provider -> try fallback - providers (openrouter, nous) - c. If on aggregator and input has vendor:model where vendor is NOT - a known hermes provider -> convert to vendor/model slug - d. Standard parse_model_input() - e. If on aggregator, try to resolve within aggregator catalog first - f. Fall back to detect_provider_for_model() - g. Resolve credentials via resolve_runtime_provider() - h. Normalize model name for target provider - i. Get capabilities from get_model_capabilities() - j. Build result + + If --provider given: + a. Resolve provider via resolve_provider_full() + b. Resolve credentials + c. If model given, resolve alias on target provider or use as-is + d. If no model, auto-detect from endpoint + + If no --provider: + a. Try alias resolution on current provider + b. If alias exists but not on current provider -> fallback + c. On aggregator, try vendor/model slug conversion + d. Aggregator catalog search + e. detect_provider_for_model() as last resort + f. Resolve credentials + g. Normalize model name for target provider + + Finally: + h. Get full model metadata from models.dev + i. Build result Args: - raw_input: The user's model input (e.g. "sonnet", "claude-sonnet-4", - "zai:glm-5", "custom:local:qwen"). + raw_input: The model name (after flag parsing). current_provider: The currently active provider. current_model: The currently active model name. current_base_url: The currently active base URL. current_api_key: The currently active API key. - is_global: Whether this switch should be persisted globally. + is_global: Whether to persist the switch. + explicit_provider: From --provider flag (empty = no explicit provider). + user_providers: The ``providers:`` dict from config.yaml (for user endpoints). Returns: ModelSwitchResult with all information the caller needs. @@ -311,142 +311,163 @@ def switch_model( from hermes_cli.runtime_provider import resolve_runtime_provider resolved_alias = "" + new_model = raw_input.strip() + target_provider = current_provider - # --- Step a: Try alias resolution on current provider --- - alias_result = resolve_alias(raw_input, current_provider) + # ================================================================= + # PATH A: Explicit --provider given + # ================================================================= + if explicit_provider: + # Resolve the provider + pdef = resolve_provider_full(explicit_provider, user_providers) + if pdef is None: + return ModelSwitchResult( + success=False, + is_global=is_global, + error_message=( + f"Unknown provider '{explicit_provider}'. " + f"Check 'hermes model' for available providers, or define it " + f"in config.yaml under 'providers:'." + ), + ) - if alias_result is not None: - target_provider, new_model, resolved_alias = alias_result - logger.debug( - "Alias '%s' resolved to %s on %s", - resolved_alias, new_model, target_provider, - ) - else: - # --- Step b: Alias exists but not on current provider -> fallback --- - key = raw_input.strip().lower() - if key in MODEL_ALIASES: - fallback_result = _resolve_alias_fallback(raw_input) - if fallback_result is not None: - target_provider, new_model, resolved_alias = fallback_result - logger.debug( - "Alias '%s' resolved via fallback to %s on %s", - resolved_alias, new_model, target_provider, - ) + target_provider = pdef.id + + # If no model specified, try auto-detect from endpoint + if not new_model: + if pdef.base_url: + from hermes_cli.runtime_provider import _auto_detect_local_model + detected = _auto_detect_local_model(pdef.base_url) + if detected: + new_model = detected + else: + return ModelSwitchResult( + success=False, + target_provider=target_provider, + provider_label=pdef.name, + is_global=is_global, + error_message=( + f"No model detected on {pdef.name} ({pdef.base_url}). " + f"Specify the model explicitly: /model --provider {explicit_provider}" + ), + ) else: - # Alias exists but no model found anywhere - identity = MODEL_ALIASES[key] return ModelSwitchResult( success=False, + target_provider=target_provider, + provider_label=pdef.name, is_global=is_global, error_message=( - f"Alias '{key}' maps to {identity.vendor}/{identity.family} " - f"but no matching model was found in any provider catalog. " - f"Try specifying the full model name." + f"Provider '{pdef.name}' has no base URL configured. " + f"Specify a model: /model --provider {explicit_provider}" ), ) + + # Resolve alias on the TARGET provider + alias_result = resolve_alias(new_model, target_provider) + if alias_result is not None: + _, new_model, resolved_alias = alias_result + + # ================================================================= + # PATH B: No explicit provider — resolve from model input + # ================================================================= + else: + # --- Step a: Try alias resolution on current provider --- + alias_result = resolve_alias(raw_input, current_provider) + + if alias_result is not None: + target_provider, new_model, resolved_alias = alias_result + logger.debug( + "Alias '%s' resolved to %s on %s", + resolved_alias, new_model, target_provider, + ) else: - # --- Step c: aggregator vendor:model fixup --- - # If on an aggregator and the input looks like vendor:model - # where vendor is NOT a known Hermes provider, treat the colon - # as a vendor/model separator instead of provider:model. - target_provider = current_provider - new_model = raw_input.strip() - - colon_pos = raw_input.find(":") - if colon_pos > 0 and is_aggregator(current_provider): - left = raw_input[:colon_pos].strip().lower() - right = raw_input[colon_pos + 1:].strip() - if left and right and left not in _KNOWN_PROVIDER_NAMES: - # Convert vendor:model -> vendor/model for aggregator - new_model = f"{left}/{right}" - target_provider = current_provider + # --- Step b: Alias exists but not on current provider -> fallback --- + key = raw_input.strip().lower() + if key in MODEL_ALIASES: + fallback_result = _resolve_alias_fallback(raw_input) + if fallback_result is not None: + target_provider, new_model, resolved_alias = fallback_result logger.debug( - "Converted vendor:model '%s' to aggregator slug '%s'", - raw_input, new_model, + "Alias '%s' resolved via fallback to %s on %s", + resolved_alias, new_model, target_provider, ) - - # --- Step d: Standard parse_model_input() --- - if new_model == raw_input.strip(): - target_provider, new_model = parse_model_input( - raw_input, current_provider, - ) - - # --- Step d2: If parsed model is a known alias, resolve it - # on the TARGET provider. This handles "anthropic:sonnet" - # -> resolve "sonnet" alias on anthropic provider. - model_as_alias = new_model.strip().lower() - if model_as_alias in MODEL_ALIASES and target_provider != current_provider: - alias_on_target = resolve_alias(new_model, target_provider) - if alias_on_target is not None: - _, new_model, resolved_alias = alias_on_target - logger.debug( - "Resolved alias '%s' on target provider %s -> %s", - model_as_alias, target_provider, new_model, + else: + identity = MODEL_ALIASES[key] + return ModelSwitchResult( + success=False, + is_global=is_global, + error_message=( + f"Alias '{key}' maps to {identity.vendor}/{identity.family} " + f"but no matching model was found in any provider catalog. " + f"Try specifying the full model name." + ), ) - - # --- Step e: If on aggregator, try to resolve within aggregator catalog --- - if is_aggregator(target_provider) and not resolved_alias: - catalog = list_provider_models(target_provider) - if catalog: - new_model_lower = new_model.lower() - # Exact match - for mid in catalog: - if mid.lower() == new_model_lower: - new_model = mid - break else: - # Try matching bare name against catalog entries + # --- Step c: On aggregator, convert vendor:model to vendor/model --- + colon_pos = raw_input.find(":") + if colon_pos > 0 and is_aggregator(current_provider): + left = raw_input[:colon_pos].strip().lower() + right = raw_input[colon_pos + 1:].strip() + if left and right: + # Colons become slashes for aggregator slugs + new_model = f"{left}/{right}" + logger.debug( + "Converted vendor:model '%s' to aggregator slug '%s'", + raw_input, new_model, + ) + + # --- Step d: Aggregator catalog search --- + if is_aggregator(target_provider) and not resolved_alias: + catalog = list_provider_models(target_provider) + if catalog: + new_model_lower = new_model.lower() for mid in catalog: - if "/" in mid: - _, bare = mid.split("/", 1) - if bare.lower() == new_model_lower: - new_model = mid - break + if mid.lower() == new_model_lower: + new_model = mid + break + else: + for mid in catalog: + if "/" in mid: + _, bare = mid.split("/", 1) + if bare.lower() == new_model_lower: + new_model = mid + break - # --- Step f: Fall back to detect_provider_for_model() --- - # Only auto-detect when no explicit provider was given and we're not - # on a custom endpoint. - _base = current_base_url or "" - is_custom = current_provider == "custom" or ( - "localhost" in _base or "127.0.0.1" in _base - ) + # --- Step e: detect_provider_for_model() as last resort --- + _base = current_base_url or "" + is_custom = current_provider in ("custom", "local") or ( + "localhost" in _base or "127.0.0.1" in _base + ) - if ( - target_provider == current_provider - and not is_custom - and not resolved_alias - ): - detected = detect_provider_for_model(new_model, current_provider) - if detected: - target_provider, new_model = detected + if ( + target_provider == current_provider + and not is_custom + and not resolved_alias + ): + detected = detect_provider_for_model(new_model, current_provider) + if detected: + target_provider, new_model = detected + + # ================================================================= + # COMMON PATH: Resolve credentials, normalize, get metadata + # ================================================================= provider_changed = target_provider != current_provider + provider_label = get_label(target_provider) - # --- Step g: Resolve credentials --- + # --- Resolve credentials --- api_key = current_api_key base_url = current_base_url api_mode = "" - provider_label = LABELS.get(target_provider, target_provider) - if provider_changed: + if provider_changed or explicit_provider: try: runtime = resolve_runtime_provider(requested=target_provider) api_key = runtime.get("api_key", "") base_url = runtime.get("base_url", "") api_mode = runtime.get("api_mode", "") except Exception as e: - if target_provider == "custom": - return ModelSwitchResult( - success=False, - target_provider=target_provider, - provider_label=provider_label, - is_global=is_global, - error_message=( - "No custom endpoint configured. Set model.base_url " - "in config.yaml, or set OPENAI_BASE_URL in .env, " - "or run: hermes setup -> Custom OpenAI-compatible endpoint" - ), - ) return ModelSwitchResult( success=False, target_provider=target_provider, @@ -458,7 +479,6 @@ def switch_model( ), ) else: - # Re-resolve for current provider to get accurate base_url try: runtime = resolve_runtime_provider(requested=current_provider) api_key = runtime.get("api_key", "") @@ -467,7 +487,7 @@ def switch_model( except Exception: pass - # --- Step h: Normalize model name for target provider --- + # --- Normalize model name for target provider --- new_model = normalize_model_for_provider(new_model, target_provider) # --- Validate --- @@ -498,17 +518,20 @@ def switch_model( ) # --- OpenCode api_mode override --- - if target_provider in {"opencode-zen", "opencode-go"}: + if target_provider in {"opencode-zen", "opencode-go", "opencode", "opencode-go"}: api_mode = opencode_model_api_mode(target_provider, new_model) # --- Determine api_mode if not already set --- if not api_mode: api_mode = determine_api_mode(target_provider, base_url) - # --- Step i: Get capabilities from models.dev --- + # --- Get capabilities (legacy) --- capabilities = get_model_capabilities(target_provider, new_model) - # --- Step j: Build result --- + # --- Get full model info from models.dev --- + model_info = get_model_info(target_provider, new_model) + + # --- Build result --- return ModelSwitchResult( success=True, new_model=new_model, @@ -521,6 +544,7 @@ def switch_model( provider_label=provider_label, resolved_via_alias=resolved_alias, capabilities=capabilities, + model_info=model_info, is_global=is_global, ) @@ -530,18 +554,7 @@ def switch_model( # --------------------------------------------------------------------------- def suggest_models(raw_input: str, limit: int = 3) -> List[str]: - """Return fuzzy model suggestions for a (possibly misspelled) input. - - Searches the models.dev catalog across all providers and returns up - to *limit* model IDs that best match *raw_input*. - - Args: - raw_input: The user's raw model input. - limit: Maximum number of suggestions. - - Returns: - List of model ID strings (may be empty). - """ + """Return fuzzy model suggestions for a (possibly misspelled) input.""" query = raw_input.strip() if not query: return [] @@ -550,15 +563,8 @@ def suggest_models(raw_input: str, limit: int = 3) -> List[str]: suggestions: list[str] = [] for r in results: mid = r.get("model_id", "") - prov = r.get("provider", "") if mid: - # For aggregator providers, model_id already has vendor/ prefix - if prov and is_aggregator(prov): - suggestions.append(mid) - elif prov: - suggestions.append(f"{prov}:{mid}") - else: - suggestions.append(mid) + suggestions.append(mid) return suggestions[:limit] @@ -568,10 +574,7 @@ def suggest_models(raw_input: str, limit: int = 3) -> List[str]: # --------------------------------------------------------------------------- def switch_to_custom_provider() -> CustomAutoResult: - """Handle bare '/model custom' -- resolve endpoint and auto-detect model. - - Returns a result object; the caller handles persistence and output. - """ + """Handle bare '/model --provider custom' — resolve endpoint and auto-detect model.""" from hermes_cli.runtime_provider import ( resolve_runtime_provider, _auto_detect_local_model, @@ -607,7 +610,7 @@ def switch_to_custom_provider() -> CustomAutoResult: error_message=( f"Custom endpoint at {cust_base} is reachable but no single " f"model was auto-detected. Specify the model explicitly: " - f"/model custom:" + f"/model --provider custom" ), ) diff --git a/hermes_cli/providers.py b/hermes_cli/providers.py index 215acb0eeb..890927884c 100644 --- a/hermes_cli/providers.py +++ b/hermes_cli/providers.py @@ -1,196 +1,160 @@ """ Single source of truth for provider identity in Hermes Agent. -Every provider known to the system is defined here exactly once. Other -modules (auth, models, inference) import from this file rather than -maintaining their own parallel registries. +Two data sources, merged at runtime: -Concepts --------- -- **ProviderDef** -- immutable description of one inference provider. -- **PROVIDERS** -- canonical ``{id: ProviderDef}`` mapping. -- **ALIASES** -- maps human-friendly or legacy names to a canonical id. -- **LABELS** -- short display names for UI / CLI menus. -- **transport** -- wire protocol the provider speaks - (``openai_chat``, ``anthropic_messages``, ``codex_responses``). +1. **models.dev catalog** — 109+ providers with base URLs, env vars, display + names, and full model metadata (context, cost, capabilities). This is + the primary database. + +2. **Hermes overlays** — transport type, auth patterns, aggregator flags, + and additional env vars that models.dev doesn't track. Small dict, + maintained here. + +3. **User config** (``providers:`` section in config.yaml) — user-defined + endpoints and overrides. Merged on top of everything else. + +Other modules import from this file. No parallel registries. """ from __future__ import annotations -from dataclasses import dataclass -from typing import Optional +import logging +import os +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional, Tuple + +logger = logging.getLogger(__name__) -# -- Provider definition ------------------------------------------------------ +# -- Hermes overlay ---------------------------------------------------------- +# Hermes-specific metadata that models.dev doesn't provide. @dataclass(frozen=True) -class ProviderDef: - """Immutable description of a known inference provider.""" +class HermesOverlay: + """Hermes-specific provider metadata layered on top of models.dev.""" - id: str - name: str - transport: str # 'openai_chat' | 'anthropic_messages' | 'codex_responses' - api_key_env_vars: tuple # env vars to check, in priority order - base_url: str = "" - base_url_env_var: str = "" + transport: str = "openai_chat" # openai_chat | anthropic_messages | codex_responses is_aggregator: bool = False - auth_type: str = "api_key" # api_key | oauth_device_code | oauth_external | external_process + auth_type: str = "api_key" # api_key | oauth_device_code | oauth_external | external_process + extra_env_vars: Tuple[str, ...] = () # env vars models.dev doesn't list + base_url_override: str = "" # override if models.dev URL is wrong/missing + base_url_env_var: str = "" # env var for user-custom base URL -# -- Canonical provider table ------------------------------------------------- - -PROVIDERS: dict[str, ProviderDef] = { - "openrouter": ProviderDef( - id="openrouter", - name="OpenRouter", +HERMES_OVERLAYS: Dict[str, HermesOverlay] = { + "openrouter": HermesOverlay( transport="openai_chat", - api_key_env_vars=("OPENROUTER_API_KEY", "OPENAI_API_KEY"), - base_url="https://openrouter.ai/api/v1", - base_url_env_var="OPENROUTER_BASE_URL", is_aggregator=True, + extra_env_vars=("OPENAI_API_KEY",), + base_url_env_var="OPENROUTER_BASE_URL", ), - "nous": ProviderDef( - id="nous", - name="Nous Portal", + "nous": HermesOverlay( transport="openai_chat", - api_key_env_vars=(), - base_url="https://inference-api.nousresearch.com/v1", auth_type="oauth_device_code", + base_url_override="https://inference-api.nousresearch.com/v1", ), - "openai-codex": ProviderDef( - id="openai-codex", - name="OpenAI Codex", + "openai-codex": HermesOverlay( transport="codex_responses", - api_key_env_vars=(), - base_url="https://chatgpt.com/backend-api/codex", auth_type="oauth_external", + base_url_override="https://chatgpt.com/backend-api/codex", ), - "copilot": ProviderDef( - id="copilot", - name="GitHub Copilot", - transport="openai_chat", - api_key_env_vars=("COPILOT_GITHUB_TOKEN", "GH_TOKEN", "GITHUB_TOKEN"), - base_url="https://api.githubcopilot.com", - ), - "copilot-acp": ProviderDef( - id="copilot-acp", - name="GitHub Copilot ACP", + "copilot-acp": HermesOverlay( transport="codex_responses", - api_key_env_vars=(), - base_url="acp://copilot", - base_url_env_var="COPILOT_ACP_BASE_URL", auth_type="external_process", + base_url_override="acp://copilot", + base_url_env_var="COPILOT_ACP_BASE_URL", ), - "zai": ProviderDef( - id="zai", - name="Z.AI / GLM", + "github-copilot": HermesOverlay( transport="openai_chat", - api_key_env_vars=("GLM_API_KEY", "ZAI_API_KEY", "Z_AI_API_KEY"), - base_url="https://api.z.ai/api/paas/v4", + extra_env_vars=("COPILOT_GITHUB_TOKEN", "GH_TOKEN"), + ), + "anthropic": HermesOverlay( + transport="anthropic_messages", + extra_env_vars=("ANTHROPIC_TOKEN", "CLAUDE_CODE_OAUTH_TOKEN"), + ), + "zai": HermesOverlay( + transport="openai_chat", + extra_env_vars=("GLM_API_KEY", "ZAI_API_KEY", "Z_AI_API_KEY"), base_url_env_var="GLM_BASE_URL", ), - "kimi-coding": ProviderDef( - id="kimi-coding", - name="Kimi / Moonshot", + "kimi-for-coding": HermesOverlay( transport="openai_chat", - api_key_env_vars=("KIMI_API_KEY",), - base_url="https://api.moonshot.ai/v1", base_url_env_var="KIMI_BASE_URL", ), - "minimax": ProviderDef( - id="minimax", - name="MiniMax", + "minimax": HermesOverlay( transport="openai_chat", - api_key_env_vars=("MINIMAX_API_KEY",), - base_url="https://api.minimax.io/anthropic", base_url_env_var="MINIMAX_BASE_URL", ), - "minimax-cn": ProviderDef( - id="minimax-cn", - name="MiniMax (China)", + "minimax-cn": HermesOverlay( transport="openai_chat", - api_key_env_vars=("MINIMAX_CN_API_KEY",), - base_url="https://api.minimaxi.com/anthropic", base_url_env_var="MINIMAX_CN_BASE_URL", ), - "anthropic": ProviderDef( - id="anthropic", - name="Anthropic", - transport="anthropic_messages", - api_key_env_vars=("ANTHROPIC_API_KEY", "ANTHROPIC_TOKEN", "CLAUDE_CODE_OAUTH_TOKEN"), - base_url="https://api.anthropic.com", - ), - "deepseek": ProviderDef( - id="deepseek", - name="DeepSeek", + "deepseek": HermesOverlay( transport="openai_chat", - api_key_env_vars=("DEEPSEEK_API_KEY",), - base_url="https://api.deepseek.com/v1", base_url_env_var="DEEPSEEK_BASE_URL", ), - "ai-gateway": ProviderDef( - id="ai-gateway", - name="AI Gateway", + "alibaba": HermesOverlay( transport="openai_chat", - api_key_env_vars=("AI_GATEWAY_API_KEY",), - base_url="https://ai-gateway.vercel.sh/v1", - base_url_env_var="AI_GATEWAY_BASE_URL", - is_aggregator=True, - ), - "opencode-zen": ProviderDef( - id="opencode-zen", - name="OpenCode Zen", - transport="openai_chat", - api_key_env_vars=("OPENCODE_ZEN_API_KEY",), - base_url="https://opencode.ai/zen/v1", - base_url_env_var="OPENCODE_ZEN_BASE_URL", - is_aggregator=True, - ), - "opencode-go": ProviderDef( - id="opencode-go", - name="OpenCode Go", - transport="openai_chat", - api_key_env_vars=("OPENCODE_GO_API_KEY",), - base_url="https://opencode.ai/zen/go/v1", - base_url_env_var="OPENCODE_GO_BASE_URL", - is_aggregator=True, - ), - "kilocode": ProviderDef( - id="kilocode", - name="Kilo Code", - transport="openai_chat", - api_key_env_vars=("KILOCODE_API_KEY",), - base_url="https://api.kilo.ai/api/gateway", - base_url_env_var="KILOCODE_BASE_URL", - is_aggregator=True, - ), - "alibaba": ProviderDef( - id="alibaba", - name="Alibaba Cloud (DashScope)", - transport="openai_chat", - api_key_env_vars=("DASHSCOPE_API_KEY",), - base_url="https://dashscope-intl.aliyuncs.com/compatible-mode/v1", base_url_env_var="DASHSCOPE_BASE_URL", ), - "huggingface": ProviderDef( - id="huggingface", - name="Hugging Face", + "vercel": HermesOverlay( transport="openai_chat", - api_key_env_vars=("HF_TOKEN",), - base_url="https://router.huggingface.co/v1", - base_url_env_var="HF_BASE_URL", is_aggregator=True, ), + "opencode": HermesOverlay( + transport="openai_chat", + is_aggregator=True, + base_url_env_var="OPENCODE_ZEN_BASE_URL", + ), + "opencode-go": HermesOverlay( + transport="openai_chat", + is_aggregator=True, + base_url_env_var="OPENCODE_GO_BASE_URL", + ), + "kilo": HermesOverlay( + transport="openai_chat", + is_aggregator=True, + base_url_env_var="KILOCODE_BASE_URL", + ), + "huggingface": HermesOverlay( + transport="openai_chat", + is_aggregator=True, + base_url_env_var="HF_BASE_URL", + ), } -# -- Aliases ------------------------------------------------------------------ -# Merged from auth.py _PROVIDER_ALIASES and models.py _PROVIDER_ALIASES. -# Every human-friendly / legacy name maps to exactly one canonical id. +# -- Resolved provider ------------------------------------------------------- +# The merged result of models.dev + overlay + user config. -ALIASES: dict[str, str] = { +@dataclass +class ProviderDef: + """Complete provider definition — merged from all sources.""" + + id: str + name: str + transport: str # openai_chat | anthropic_messages | codex_responses + api_key_env_vars: Tuple[str, ...] # all env vars to check for API key + base_url: str = "" + base_url_env_var: str = "" + is_aggregator: bool = False + auth_type: str = "api_key" + doc: str = "" + source: str = "" # "models.dev", "hermes", "user-config" + + @property + def is_user_defined(self) -> bool: + return self.source == "user-config" + + +# -- Aliases ------------------------------------------------------------------ +# Maps human-friendly / legacy names to canonical provider IDs. +# Uses models.dev IDs where possible. + +ALIASES: Dict[str, str] = { # openrouter - "openai": "openrouter", + "openai": "openrouter", # bare "openai" → route through aggregator # zai "glm": "zai", @@ -198,9 +162,10 @@ ALIASES: dict[str, str] = { "z.ai": "zai", "zhipu": "zai", - # kimi-coding - "kimi": "kimi-coding", - "moonshot": "kimi-coding", + # kimi-for-coding (models.dev ID) + "kimi": "kimi-for-coding", + "kimi-coding": "kimi-for-coding", + "moonshot": "kimi-for-coding", # minimax-cn "minimax-china": "minimax-cn", @@ -210,33 +175,28 @@ ALIASES: dict[str, str] = { "claude": "anthropic", "claude-code": "anthropic", - # copilot - "github": "copilot", - "github-copilot": "copilot", - "github-models": "copilot", - "github-model": "copilot", - - # copilot-acp + # github-copilot (models.dev ID) + "copilot": "github-copilot", + "github": "github-copilot", "github-copilot-acp": "copilot-acp", - "copilot-acp-agent": "copilot-acp", - # ai-gateway - "aigateway": "ai-gateway", - "vercel": "ai-gateway", - "vercel-ai-gateway": "ai-gateway", + # vercel (models.dev ID for AI Gateway) + "ai-gateway": "vercel", + "aigateway": "vercel", + "vercel-ai-gateway": "vercel", - # opencode-zen - "opencode": "opencode-zen", - "zen": "opencode-zen", + # opencode (models.dev ID for OpenCode Zen) + "opencode-zen": "opencode", + "zen": "opencode", # opencode-go "go": "opencode-go", "opencode-go-sub": "opencode-go", - # kilocode - "kilo": "kilocode", - "kilo-code": "kilocode", - "kilo-gateway": "kilocode", + # kilo (models.dev ID for KiloCode) + "kilocode": "kilo", + "kilo-code": "kilo", + "kilo-gateway": "kilo", # deepseek "deep-seek": "deepseek", @@ -252,29 +212,33 @@ ALIASES: dict[str, str] = { "hugging-face": "huggingface", "huggingface-hub": "huggingface", - # Local server aliases -- route to the virtual "custom" provider - "lmstudio": "custom", - "lm-studio": "custom", - "lm_studio": "custom", - "ollama": "custom", - "vllm": "custom", - "llamacpp": "custom", - "llama.cpp": "custom", - "llama-cpp": "custom", + # Local server aliases → virtual "local" concept (resolved via user config) + "lmstudio": "lmstudio", + "lm-studio": "lmstudio", + "lm_studio": "lmstudio", + "ollama": "ollama-cloud", + "vllm": "local", + "llamacpp": "local", + "llama.cpp": "local", + "llama-cpp": "local", } # -- Display labels ----------------------------------------------------------- -# Short human-readable names for CLI menus and status output. +# Built dynamically from models.dev + overlays. Fallback for providers +# not in the catalog. -LABELS: dict[str, str] = {pid: pdef.name for pid, pdef in PROVIDERS.items()} -# Virtual "custom" provider is not in PROVIDERS but needs a label. -LABELS["custom"] = "Custom endpoint" +_LABEL_OVERRIDES: Dict[str, str] = { + "nous": "Nous Portal", + "openai-codex": "OpenAI Codex", + "copilot-acp": "GitHub Copilot ACP", + "local": "Local endpoint", +} -# -- Transport -> API mode mapping -------------------------------------------- +# -- Transport → API mode mapping --------------------------------------------- -TRANSPORT_TO_API_MODE: dict[str, str] = { +TRANSPORT_TO_API_MODE: Dict[str, str] = { "openai_chat": "chat_completions", "anthropic_messages": "anthropic_messages", "codex_responses": "codex_responses", @@ -287,28 +251,154 @@ def normalize_provider(name: str) -> str: """Resolve aliases and normalise casing to a canonical provider id. Returns the canonical id string. Does *not* validate that the id - corresponds to a known provider -- callers should check ``PROVIDERS`` - if they need that guarantee. + corresponds to a known provider. """ key = name.strip().lower() return ALIASES.get(key, key) -def get_provider(name: str) -> Optional[ProviderDef]: - """Look up a provider by id or alias. +def get_overlay(provider_id: str) -> Optional[HermesOverlay]: + """Get Hermes overlay for a provider, if one exists.""" + canonical = normalize_provider(provider_id) + return HERMES_OVERLAYS.get(canonical) - Returns the ``ProviderDef`` or ``None`` if the provider is unknown. + +def get_provider(name: str) -> Optional[ProviderDef]: + """Look up a provider by id or alias, merging all data sources. + + Resolution order: + 1. Hermes overlays (for providers not in models.dev: nous, openai-codex, etc.) + 2. models.dev catalog + Hermes overlay + 3. User-defined providers from config (TODO: Phase 4) + + Returns a fully-resolved ProviderDef or None. """ canonical = normalize_provider(name) - return PROVIDERS.get(canonical) + + # Try to get models.dev data + try: + from agent.models_dev import get_provider_info as _mdev_provider + mdev_info = _mdev_provider(canonical) + except Exception: + mdev_info = None + + overlay = HERMES_OVERLAYS.get(canonical) + + if mdev_info is not None: + # Merge models.dev + overlay + transport = overlay.transport if overlay else "openai_chat" + is_agg = overlay.is_aggregator if overlay else False + auth = overlay.auth_type if overlay else "api_key" + base_url_env = overlay.base_url_env_var if overlay else "" + base_url_override = overlay.base_url_override if overlay else "" + + # Combine env vars: models.dev env + hermes extra + env_vars = list(mdev_info.env) + if overlay and overlay.extra_env_vars: + for ev in overlay.extra_env_vars: + if ev not in env_vars: + env_vars.append(ev) + + return ProviderDef( + id=canonical, + name=mdev_info.name, + transport=transport, + api_key_env_vars=tuple(env_vars), + base_url=base_url_override or mdev_info.api, + base_url_env_var=base_url_env, + is_aggregator=is_agg, + auth_type=auth, + doc=mdev_info.doc, + source="models.dev", + ) + + if overlay is not None: + # Hermes-only provider (not in models.dev) + return ProviderDef( + id=canonical, + name=_LABEL_OVERRIDES.get(canonical, canonical), + transport=overlay.transport, + api_key_env_vars=overlay.extra_env_vars, + base_url=overlay.base_url_override, + base_url_env_var=overlay.base_url_env_var, + is_aggregator=overlay.is_aggregator, + auth_type=overlay.auth_type, + source="hermes", + ) + + return None + + +def get_label(provider_id: str) -> str: + """Get a human-readable display name for a provider.""" + canonical = normalize_provider(provider_id) + + # Check label overrides first + if canonical in _LABEL_OVERRIDES: + return _LABEL_OVERRIDES[canonical] + + # Try models.dev + pdef = get_provider(canonical) + if pdef: + return pdef.name + + return canonical + + +# Build LABELS dict for backward compat +def _build_labels() -> Dict[str, str]: + """Build labels dict from overlays + overrides. Lazy, cached.""" + labels: Dict[str, str] = {} + for pid in HERMES_OVERLAYS: + labels[pid] = get_label(pid) + labels.update(_LABEL_OVERRIDES) + return labels + +# Lazy-built on first access +_labels_cache: Optional[Dict[str, str]] = None + +@property +def LABELS() -> Dict[str, str]: + """Backward-compatible labels dict.""" + global _labels_cache + if _labels_cache is None: + _labels_cache = _build_labels() + return _labels_cache + +# For direct import compat, expose as module-level dict +# Built on demand by get_label() calls +LABELS: Dict[str, str] = { + # Static entries for backward compat — get_label() is the proper API + "openrouter": "OpenRouter", + "nous": "Nous Portal", + "openai-codex": "OpenAI Codex", + "copilot-acp": "GitHub Copilot ACP", + "github-copilot": "GitHub Copilot", + "anthropic": "Anthropic", + "zai": "Z.AI / GLM", + "kimi-for-coding": "Kimi / Moonshot", + "minimax": "MiniMax", + "minimax-cn": "MiniMax (China)", + "deepseek": "DeepSeek", + "alibaba": "Alibaba Cloud (DashScope)", + "vercel": "Vercel AI Gateway", + "opencode": "OpenCode Zen", + "opencode-go": "OpenCode Go", + "kilo": "Kilo Gateway", + "huggingface": "Hugging Face", + "local": "Local endpoint", + "custom": "Custom endpoint", + # Legacy Hermes IDs (point to same providers) + "ai-gateway": "Vercel AI Gateway", + "kilocode": "Kilo Gateway", + "copilot": "GitHub Copilot", + "kimi-coding": "Kimi / Moonshot", + "opencode-zen": "OpenCode Zen", +} def is_aggregator(provider: str) -> bool: - """Return True when the provider is a multi-model aggregator. - - Aggregators (OpenRouter, AI Gateway, etc.) expose models from many - upstream providers through a single API key and endpoint. - """ + """Return True when the provider is a multi-model aggregator.""" pdef = get_provider(provider) return pdef.is_aggregator if pdef else False @@ -317,17 +407,15 @@ def determine_api_mode(provider: str, base_url: str = "") -> str: """Determine the API mode (wire protocol) for a provider/endpoint. Resolution order: - - 1. Known provider -> transport -> ``TRANSPORT_TO_API_MODE``. - 2. URL heuristics for unknown / custom providers. - 3. Default: ``'chat_completions'``. + 1. Known provider → transport → TRANSPORT_TO_API_MODE. + 2. URL heuristics for unknown / custom providers. + 3. Default: 'chat_completions'. """ - # 1. Known provider lookup pdef = get_provider(provider) if pdef is not None: return TRANSPORT_TO_API_MODE.get(pdef.transport, "chat_completions") - # 2. URL-based heuristics for custom / unknown providers + # URL-based heuristics for custom / unknown providers if base_url: url_lower = base_url.rstrip("/").lower() if url_lower.endswith("/anthropic") or "api.anthropic.com" in url_lower: @@ -335,5 +423,97 @@ def determine_api_mode(provider: str, base_url: str = "") -> str: if "api.openai.com" in url_lower: return "codex_responses" - # 3. Default return "chat_completions" + + +# -- Provider from user config ------------------------------------------------ + +def resolve_user_provider(name: str, user_config: Dict[str, Any]) -> Optional[ProviderDef]: + """Resolve a provider from the user's config.yaml ``providers:`` section. + + Args: + name: Provider name as given by the user. + user_config: The ``providers:`` dict from config.yaml. + + Returns: + ProviderDef if found, else None. + """ + if not user_config or not isinstance(user_config, dict): + return None + + entry = user_config.get(name) + if not isinstance(entry, dict): + return None + + # Extract fields + display_name = entry.get("name", "") or name + api_url = entry.get("api", "") or entry.get("url", "") or entry.get("base_url", "") or "" + key_env = entry.get("key_env", "") or "" + transport = entry.get("transport", "openai_chat") or "openai_chat" + + env_vars: List[str] = [] + if key_env: + env_vars.append(key_env) + + return ProviderDef( + id=name, + name=display_name, + transport=transport, + api_key_env_vars=tuple(env_vars), + base_url=api_url, + is_aggregator=False, + auth_type="api_key", + source="user-config", + ) + + +def resolve_provider_full( + name: str, + user_providers: Optional[Dict[str, Any]] = None, +) -> Optional[ProviderDef]: + """Full resolution chain: built-in → models.dev → user config. + + This is the main entry point for --provider flag resolution. + + Args: + name: Provider name or alias. + user_providers: The ``providers:`` dict from config.yaml (optional). + + Returns: + ProviderDef if found, else None. + """ + canonical = normalize_provider(name) + + # 1. Built-in (models.dev + overlays) + pdef = get_provider(canonical) + if pdef is not None: + return pdef + + # 2. User-defined providers from config + if user_providers: + # Try canonical name + user_pdef = resolve_user_provider(canonical, user_providers) + if user_pdef is not None: + return user_pdef + # Try original name (in case alias didn't match) + user_pdef = resolve_user_provider(name.strip().lower(), user_providers) + if user_pdef is not None: + return user_pdef + + # 3. Try models.dev directly (for providers not in our ALIASES) + try: + from agent.models_dev import get_provider_info as _mdev_provider + mdev_info = _mdev_provider(canonical) + if mdev_info is not None: + return ProviderDef( + id=canonical, + name=mdev_info.name, + transport="openai_chat", + api_key_env_vars=mdev_info.env, + base_url=mdev_info.api, + source="models.dev", + ) + except Exception: + pass + + return None