hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 4 Packages Projects Releases Wiki Activity

Merge branch 'simp/r3-23-g3' into simp/r3-23

Browse Source
This commit is contained in:
Teknium
2026-09-02 23:30:51 -07:00
parent 4f8f967a03 8f800b3a92
commit 4e1338bd19
6 changed files with 777 additions and 1557 deletions
Download Patch File Download Diff File Expand all files Collapse all files

110
hermes_cli/provider_catalog.py
View File

@@ -1,35 +1,26 @@
"""Unified provider catalog — one source of truth for the provider universe.
The provider list shown by ``hermes model`` (CLI/TUI) and the desktop Settings → Providers tabs
(Accounts + API keys) **must be the same set**. Every provider added after those lists were written
silently went missing from the GUI — e.g.
* ``auth_type`` / ``api_key_env_vars`` / ``base_url_env_var`` from
:data:`hermes_cli.auth.PROVIDER_REGISTRY` (credential truth), and * ``display_name`` /
``description`` / ``signup_url`` from the provider's :class:`providers.base.ProviderProfile` when
one exists, falling back to the ``CANONICAL_PROVIDERS`` entry's ``label`` / ``tui_desc`` and the
``OPTIONAL_ENV_VARS`` signup URL otherwise (many profiles leave these blank, and four canonical
providers have no profile at all — lmstudio, openai-api, tencent-tokenhub, xai-oauth — so the
fallbacks are load-bearing).
(Accounts + API keys) **must be the same set**; providers added after those lists were written
silently went missing from the GUI. ``auth_type`` / ``api_key_env_vars`` / ``base_url_env_var``
come from :data:`hermes_cli.auth.PROVIDER_REGISTRY` (credential truth); ``display_name`` /
``description`` / ``signup_url`` from the provider's :class:`providers.base.ProviderProfile`, falling
back to the ``CANONICAL_PROVIDERS`` entry's ``label`` / ``tui_desc`` and the ``OPTIONAL_ENV_VARS``
signup URL (many profiles leave these blank, and lmstudio, openai-api, tencent-tokenhub, xai-oauth
have no profile at all — the fallbacks are load-bearing).
"""
from __future__ import annotations
from dataclasses import dataclass
# Auth types that authenticate via an account / sign-in flow rather than a
# pasted API key. These route to the desktop "Accounts" tab; everything else
# (api_key, and aws_sdk which is configured via AWS_REGION/AWS_PROFILE) routes
# to the "API keys" tab. Mirrors the auth_type strings used in
# hermes_cli.auth.PROVIDER_REGISTRY and providers.base.ProviderProfile.
# Auth types that authenticate via an account / sign-in flow rather than a pasted API key; these
# route to the desktop "Accounts" tab, everything else (api_key, and aws_sdk configured via
# AWS_REGION/AWS_PROFILE) to "API keys". Mirrors the auth_type strings in PROVIDER_REGISTRY and
# ProviderProfile: external_process = copilot-acp (spawns `copilot --acp --stdio`), copilot = GitHub
# Copilot token / gh auth.
_ACCOUNTS_AUTH_TYPES: frozenset[str] = frozenset(
{
"oauth_device_code",
"oauth_external",
"oauth_minimax",
"external_process", # copilot-acp: spawns `copilot --acp --stdio`
"copilot", # GitHub Copilot token / gh auth
}
{"oauth_device_code", "oauth_external", "oauth_minimax", "external_process", "copilot"}
)
@@ -54,19 +45,19 @@ def tab_for_auth_type(auth_type: str) -> str:
return "accounts" if auth_type in _ACCOUNTS_AUTH_TYPES else "keys"
def _is_url_var(name: str) -> bool:
return name.endswith("_BASE_URL") or name.endswith("_URL")
def _split_env_vars(env_vars: tuple[str, ...]) -> tuple[tuple[str, ...], str]:
"""Split a profile's ``env_vars`` into (api_key_vars, base_url_var)."""
keys = tuple(v for v in env_vars if not (v.endswith("_BASE_URL") or v.endswith("_URL")))
base = next((v for v in env_vars if v.endswith("_BASE_URL") or v.endswith("_URL")), "")
return keys, base
return tuple(v for v in env_vars if not _is_url_var(v)), next((v for v in env_vars if _is_url_var(v)), "")
def _safe_import(module: str, attr: str, default):
"""Import ``attr`` from ``module``; return ``default`` on ANY failure.
This module is on the import path of the web server and the CLI, and a
provider-plugin import error must never blank the whole catalog.
"""
"""Import ``attr`` from ``module``; ``default`` on ANY failure — this module is on the import
path of the web server and the CLI, and a provider-plugin import error must never blank the
whole catalog."""
try:
return getattr(__import__(module, fromlist=[attr]), attr)
except Exception:
@@ -74,74 +65,47 @@ def _safe_import(module: str, attr: str, default):
def provider_catalog() -> list[ProviderDescriptor]:
"""Return one descriptor per provider in the ``hermes model`` universe.
Membership is :data:`CANONICAL_PROVIDERS` (auto-extended by provider plugins). Auth/env come
from ``PROVIDER_REGISTRY``; display metadata from ``ProviderProfile`` with canonical/env
fallbacks so providers without a profile still resolve sensibly.
"""
"""One descriptor per provider in the ``hermes model`` universe (:data:`CANONICAL_PROVIDERS`,
auto-extended by provider plugins). Auth/env from ``PROVIDER_REGISTRY``; display metadata from
``ProviderProfile`` with canonical/env fallbacks so profile-less providers still resolve."""
from hermes_cli.models import CANONICAL_PROVIDERS
PROVIDER_REGISTRY = _safe_import("hermes_cli.auth", "PROVIDER_REGISTRY", {})
OPTIONAL_ENV_VARS = _safe_import("hermes_cli.config", "OPTIONAL_ENV_VARS", {})
# Hermes overlays carry auth_type for providers that have no registry/profile
# entry of their own — notably the ``moa`` virtual provider (auth_type
# "virtual"), which has no real credential and no network endpoint.
# Overlays carry auth_type for providers with no registry/profile entry — notably the ``moa``
# virtual provider (auth_type "virtual"), which has no credential and no network endpoint.
HERMES_OVERLAYS = _safe_import("hermes_cli.providers", "HERMES_OVERLAYS", {})
try:
from providers import list_providers
profiles = {p.name: p for p in list_providers()}
except Exception:
profiles = {}
out: list[ProviderDescriptor] = []
for order, entry in enumerate(CANONICAL_PROVIDERS):
slug = entry.slug
cfg = PROVIDER_REGISTRY.get(slug)
prof = profiles.get(slug)
overlay = HERMES_OVERLAYS.get(slug)
# auth_type: registry is authoritative; fall back to profile, then the
# Hermes overlay (e.g. moa → "virtual"), then api_key.
auth_type = (
(cfg.auth_type if cfg else "")
or (prof.auth_type if prof else "")
or (overlay.auth_type if overlay else "")
or "api_key"
)
# Credential env vars: registry first (it already normalizes these),
# else derive from the profile's env_vars tuple.
# auth_type: registry is authoritative; then profile, then overlay (moa → "virtual"), then api_key.
auth_type = ((cfg.auth_type if cfg else "") or (prof.auth_type if prof else "")
or (overlay.auth_type if overlay else "") or "api_key")
# Credential env vars: registry first (already normalized), else derived from the profile.
if cfg and cfg.api_key_env_vars:
api_key_vars = tuple(cfg.api_key_env_vars)
base_url_var = cfg.base_url_env_var or ""
api_key_vars, base_url_var = tuple(cfg.api_key_env_vars), cfg.base_url_env_var or ""
elif prof and prof.env_vars:
api_key_vars, base_url_var = _split_env_vars(tuple(prof.env_vars))
else:
api_key_vars, base_url_var = (), ""
label = (prof.display_name if prof else "") or entry.label or slug
description = (prof.description if prof else "") or entry.tui_desc or label
signup_url = (prof.signup_url if prof else "") or ""
if not signup_url and api_key_vars:
info = OPTIONAL_ENV_VARS.get(api_key_vars[0]) or {}
signup_url = info.get("url") or ""
signup_url = (OPTIONAL_ENV_VARS.get(api_key_vars[0]) or {}).get("url") or ""
out.append(
ProviderDescriptor(
slug=slug,
label=label,
description=description,
auth_type=auth_type,
tab=tab_for_auth_type(auth_type),
api_key_env_vars=api_key_vars,
base_url_env_var=base_url_var,
signup_url=signup_url,
order=order,
# Keyless providers (e.g. opencode-free) are served
# anonymously: there is no credential to configure, so the
# GUI renders no key card and contract tests exempt them.
slug=slug, label=label, description=(prof.description if prof else "") or entry.tui_desc or label,
auth_type=auth_type, tab=tab_for_auth_type(auth_type), api_key_env_vars=api_key_vars,
base_url_env_var=base_url_var, signup_url=signup_url, order=order,
# Keyless providers (opencode-free) are served anonymously: no key card in the GUI,
# and contract tests exempt them.
keyless=bool(overlay.keyless) if overlay else False,
)
)

737
hermes_cli/providers.py
View File

@@ -11,8 +11,7 @@ from utils import base_url_host_matches, base_url_hostname
logger = logging.getLogger(__name__)
# -- Hermes overlay ----------------------------------------------------------
# Hermes-specific metadata that models.dev doesn't provide.
# -- Hermes overlay: metadata models.dev doesn't provide ----------------------
@dataclass(frozen=True)
class HermesOverlay:
@@ -30,153 +29,76 @@ class HermesOverlay:
HERMES_OVERLAYS: Dict[str, HermesOverlay] = {
"moa": HermesOverlay(auth_type="virtual", base_url_override="moa://local"),
"openrouter": HermesOverlay(is_aggregator=True, base_url_env_var="OPENROUTER_BASE_URL"),
"nous": HermesOverlay(
auth_type="oauth_device_code",
base_url_override="https://inference-api.nousresearch.com/v1",
),
"openai-codex": HermesOverlay(
transport="codex_responses",
auth_type="oauth_external",
base_url_override="https://chatgpt.com/backend-api/codex",
),
"openai-api": HermesOverlay(
transport="codex_responses",
base_url_override="https://api.openai.com/v1",
base_url_env_var="OPENAI_BASE_URL",
),
"xai-oauth": HermesOverlay(
transport="codex_responses",
auth_type="oauth_external",
base_url_override="https://api.x.ai/v1",
base_url_env_var="XAI_BASE_URL",
),
"qwen-oauth": HermesOverlay(
auth_type="oauth_external",
base_url_override="https://portal.qwen.ai/v1",
base_url_env_var="HERMES_QWEN_BASE_URL",
),
"lmstudio": HermesOverlay(
extra_env_vars=("LM_API_KEY",),
base_url_override="http://127.0.0.1:1234/v1",
base_url_env_var="LM_BASE_URL",
),
"copilot-acp": HermesOverlay(
transport="codex_responses",
auth_type="external_process",
base_url_override="acp://copilot",
base_url_env_var="COPILOT_ACP_BASE_URL",
),
"nous": HermesOverlay(auth_type="oauth_device_code", base_url_override="https://inference-api.nousresearch.com/v1"),
"openai-codex": HermesOverlay(transport="codex_responses", auth_type="oauth_external",
base_url_override="https://chatgpt.com/backend-api/codex"),
"openai-api": HermesOverlay(transport="codex_responses", base_url_override="https://api.openai.com/v1",
base_url_env_var="OPENAI_BASE_URL"),
"xai-oauth": HermesOverlay(transport="codex_responses", auth_type="oauth_external",
base_url_override="https://api.x.ai/v1", base_url_env_var="XAI_BASE_URL"),
"qwen-oauth": HermesOverlay(auth_type="oauth_external", base_url_override="https://portal.qwen.ai/v1",
base_url_env_var="HERMES_QWEN_BASE_URL"),
"lmstudio": HermesOverlay(extra_env_vars=("LM_API_KEY",), base_url_override="http://127.0.0.1:1234/v1",
base_url_env_var="LM_BASE_URL"),
"copilot-acp": HermesOverlay(transport="codex_responses", auth_type="external_process",
base_url_override="acp://copilot", base_url_env_var="COPILOT_ACP_BASE_URL"),
"github-copilot": HermesOverlay(extra_env_vars=("COPILOT_GITHUB_TOKEN", "GH_TOKEN")),
"anthropic": HermesOverlay(
transport="anthropic_messages",
extra_env_vars=("ANTHROPIC_TOKEN", "CLAUDE_CODE_OAUTH_TOKEN"),
),
"zai": HermesOverlay(
extra_env_vars=("GLM_API_KEY", "ZAI_API_KEY", "Z_AI_API_KEY"),
base_url_env_var="GLM_BASE_URL",
),
"anthropic": HermesOverlay(transport="anthropic_messages", extra_env_vars=("ANTHROPIC_TOKEN", "CLAUDE_CODE_OAUTH_TOKEN")),
"zai": HermesOverlay(extra_env_vars=("GLM_API_KEY", "ZAI_API_KEY", "Z_AI_API_KEY"), base_url_env_var="GLM_BASE_URL"),
"kimi-for-coding": HermesOverlay(base_url_env_var="KIMI_BASE_URL"),
"stepfun": HermesOverlay(
extra_env_vars=("STEPFUN_API_KEY",),
base_url_override="https://api.stepfun.ai/step_plan/v1",
base_url_env_var="STEPFUN_BASE_URL",
),
"stepfun": HermesOverlay(extra_env_vars=("STEPFUN_API_KEY",),
base_url_override="https://api.stepfun.ai/step_plan/v1",
base_url_env_var="STEPFUN_BASE_URL"),
"minimax": HermesOverlay(transport="anthropic_messages", base_url_env_var="MINIMAX_BASE_URL"),
"minimax-oauth": HermesOverlay(
transport="anthropic_messages",
auth_type="oauth_external",
base_url_override="https://api.minimax.io/anthropic",
),
"minimax-cn": HermesOverlay(
transport="anthropic_messages",
base_url_env_var="MINIMAX_CN_BASE_URL",
),
"minimax-oauth": HermesOverlay(transport="anthropic_messages", auth_type="oauth_external",
base_url_override="https://api.minimax.io/anthropic"),
"minimax-cn": HermesOverlay(transport="anthropic_messages", base_url_env_var="MINIMAX_CN_BASE_URL"),
"deepseek": HermesOverlay(base_url_env_var="DEEPSEEK_BASE_URL"),
"alibaba": HermesOverlay(base_url_env_var="DASHSCOPE_BASE_URL"),
"alibaba-coding-plan": HermesOverlay(base_url_env_var="ALIBABA_CODING_PLAN_BASE_URL"),
"vercel": HermesOverlay(is_aggregator=True),
"opencode": HermesOverlay(is_aggregator=True, base_url_env_var="OPENCODE_ZEN_BASE_URL"),
"opencode-go": HermesOverlay(is_aggregator=True, base_url_env_var="OPENCODE_GO_BASE_URL"),
"opencode-free": HermesOverlay(
is_aggregator=True,
base_url_override="https://opencode.ai/zen/v1",
keyless=True,
),
"opencode-free": HermesOverlay(is_aggregator=True, base_url_override="https://opencode.ai/zen/v1", keyless=True),
"kilo": HermesOverlay(is_aggregator=True, base_url_env_var="KILOCODE_BASE_URL"),
"huggingface": HermesOverlay(is_aggregator=True, base_url_env_var="HF_BASE_URL"),
"novita": HermesOverlay(is_aggregator=True, base_url_env_var="NOVITA_BASE_URL"),
"xai": HermesOverlay(
transport="codex_responses",
base_url_override="https://api.x.ai/v1",
base_url_env_var="XAI_BASE_URL",
),
"nvidia": HermesOverlay(
base_url_override="https://integrate.api.nvidia.com/v1",
base_url_env_var="NVIDIA_BASE_URL",
),
"xai": HermesOverlay(transport="codex_responses", base_url_override="https://api.x.ai/v1", base_url_env_var="XAI_BASE_URL"),
"nvidia": HermesOverlay(base_url_override="https://integrate.api.nvidia.com/v1", base_url_env_var="NVIDIA_BASE_URL"),
"xiaomi": HermesOverlay(base_url_env_var="XIAOMI_BASE_URL"),
"tencent-tokenhub": HermesOverlay(base_url_env_var="TOKENHUB_BASE_URL"),
"tencent-tokenplan": HermesOverlay(
transport="anthropic_messages",
base_url_override="https://api.lkeap.cloud.tencent.com/plan/anthropic",
base_url_env_var="TOKENPLAN_BASE_URL",
),
"arcee": HermesOverlay(
base_url_override="https://api.arcee.ai/api/v1",
base_url_env_var="ARCEE_BASE_URL",
),
"gmi": HermesOverlay(
extra_env_vars=("GMI_API_KEY",),
base_url_override="https://api.gmi-serving.com/v1",
base_url_env_var="GMI_BASE_URL",
),
"fireworks": HermesOverlay(
extra_env_vars=("FIREWORKS_API_KEY",),
base_url_override="https://api.fireworks.ai/inference/v1",
),
"actual": HermesOverlay(
transport="codex_responses",
extra_env_vars=("ACTUAL_API_KEY", "ACTUAL_BASE_URL"),
base_url_override="https://api.actual.inc/v1",
base_url_env_var="ACTUAL_BASE_URL",
),
"upstage": HermesOverlay(
extra_env_vars=("UPSTAGE_API_KEY",),
base_url_override="https://api.upstage.ai/v1",
base_url_env_var="UPSTAGE_BASE_URL",
),
"nebius-token-factory": HermesOverlay(
extra_env_vars=("NEBIUS_API_KEY", "NEBIUS_TOKEN_FACTORY_API_KEY"),
base_url_override="https://api.tokenfactory.nebius.com/v1",
base_url_env_var="NEBIUS_BASE_URL",
),
"ollama-cloud": HermesOverlay(
base_url_override="https://ollama.com/v1",
base_url_env_var="OLLAMA_BASE_URL",
),
# Azure Foundry: supports both OpenAI-style and Anthropic-style endpoints.
# The transport is determined at runtime from config.yaml model.api_mode.
"azure-foundry": HermesOverlay(base_url_env_var="AZURE_FOUNDRY_BASE_URL"), # openai_chat default; api_mode overrides
"tencent-tokenplan": HermesOverlay(transport="anthropic_messages",
base_url_override="https://api.lkeap.cloud.tencent.com/plan/anthropic",
base_url_env_var="TOKENPLAN_BASE_URL"),
"arcee": HermesOverlay(base_url_override="https://api.arcee.ai/api/v1", base_url_env_var="ARCEE_BASE_URL"),
"gmi": HermesOverlay(extra_env_vars=("GMI_API_KEY",), base_url_override="https://api.gmi-serving.com/v1",
base_url_env_var="GMI_BASE_URL"),
"fireworks": HermesOverlay(extra_env_vars=("FIREWORKS_API_KEY",),
base_url_override="https://api.fireworks.ai/inference/v1"),
"actual": HermesOverlay(transport="codex_responses", extra_env_vars=("ACTUAL_API_KEY", "ACTUAL_BASE_URL"),
base_url_override="https://api.actual.inc/v1", base_url_env_var="ACTUAL_BASE_URL"),
"upstage": HermesOverlay(extra_env_vars=("UPSTAGE_API_KEY",), base_url_override="https://api.upstage.ai/v1",
base_url_env_var="UPSTAGE_BASE_URL"),
"nebius-token-factory": HermesOverlay(extra_env_vars=("NEBIUS_API_KEY", "NEBIUS_TOKEN_FACTORY_API_KEY"),
base_url_override="https://api.tokenfactory.nebius.com/v1",
base_url_env_var="NEBIUS_BASE_URL"),
"ollama-cloud": HermesOverlay(base_url_override="https://ollama.com/v1", base_url_env_var="OLLAMA_BASE_URL"),
# Azure Foundry serves OpenAI- and Anthropic-style endpoints; transport comes from model.api_mode.
"azure-foundry": HermesOverlay(base_url_env_var="AZURE_FOUNDRY_BASE_URL"),
"bedrock": HermesOverlay(transport="bedrock_converse", auth_type="aws_sdk"),
# Vertex authenticates via OAuth2 (service-account JSON / ADC), not a
# static API key or models.dev entry — resolved specially by
# agent/vertex_adapter.py, like bedrock's aws_sdk. Without an overlay
# entry get_provider("vertex") returns None, which makes
# _preserve_provider_with_base_url() in agent/auxiliary_client.py treat
# a Vertex MoA slot's resolved (base_url, api_key) pair as an unknown
# custom endpoint instead of "vertex" — losing the provider identity
# that _refresh_provider_credentials() needs to re-mint an expired
# OAuth2 token on a 401.
# Vertex is OAuth2 (service-account JSON / ADC), resolved by agent/vertex_adapter.py. Without an
# overlay get_provider("vertex") is None and auxiliary_client._preserve_provider_with_base_url
# would treat a Vertex MoA slot as an unknown custom endpoint, losing the identity
# _refresh_provider_credentials() needs to re-mint an expired token on 401.
"vertex": HermesOverlay(auth_type="vertex"),
}
# -- Resolved provider -------------------------------------------------------
# The merged result of models.dev + overlay + user config.
@dataclass
class ProviderDef:
"""Complete provider definition — merged from all sources."""
"""Complete provider definition — merged from models.dev + overlay + user config."""
id: str
name: str
@@ -190,86 +112,50 @@ class ProviderDef:
source: str = "" # "models.dev", "hermes", "user-config"
# -- Aliases ------------------------------------------------------------------
# Maps human-friendly / legacy names to canonical provider IDs.
# Uses models.dev IDs where possible.
# Aliases grouped by canonical provider id; ``ALIASES`` is the inverted lookup table.
# -- Aliases: human-friendly / legacy names grouped by canonical (models.dev where possible) id;
# ``ALIASES`` is the inverted lookup table. ---------------------------------------------------
_ALIAS_GROUPS: Dict[str, Tuple[str, ...]] = {
"openrouter": ("openai",),
"zai": ("glm", "z-ai", "z.ai", "zhipu"),
"xai": ("x-ai", "x.ai", "grok"),
"openrouter": ("openai",), "zai": ("glm", "z-ai", "z.ai", "zhipu"), "xai": ("x-ai", "x.ai", "grok"),
"xai-oauth": ("grok-oauth", "xai-oauth", "x-ai-oauth", "xai-grok-oauth"),
"nvidia": ("nim", "nvidia-nim", "build-nvidia", "nemotron"),
"kimi-for-coding": ("kimi", "kimi-coding", "kimi-coding-cn", "moonshot"),
"stepfun": ("step", "stepfun-coding-plan"),
"minimax-cn": ("minimax-china", "minimax_cn"),
"anthropic": ("claude", "claude-code"),
"github-copilot": ("copilot", "github"),
"copilot-acp": ("github-copilot-acp",),
"vercel": ("ai-gateway", "aigateway", "vercel-ai-gateway"),
"opencode": ("opencode-zen", "zen"),
"opencode-go": ("go", "opencode-go-sub"),
"opencode-free": ("free", "opencode_free"),
"kilo": ("kilocode", "kilo-code", "kilo-gateway"),
"deepseek": ("deep-seek",),
"alibaba": ("dashscope", "aliyun", "qwen", "alibaba-cloud"),
"stepfun": ("step", "stepfun-coding-plan"), "minimax-cn": ("minimax-china", "minimax_cn"),
"anthropic": ("claude", "claude-code"), "github-copilot": ("copilot", "github"),
"copilot-acp": ("github-copilot-acp",), "vercel": ("ai-gateway", "aigateway", "vercel-ai-gateway"),
"opencode": ("opencode-zen", "zen"), "opencode-go": ("go", "opencode-go-sub"),
"opencode-free": ("free", "opencode_free"), "kilo": ("kilocode", "kilo-code", "kilo-gateway"),
"deepseek": ("deep-seek",), "alibaba": ("dashscope", "aliyun", "qwen", "alibaba-cloud"),
"alibaba-coding-plan": ("alibaba_coding", "alibaba-coding", "alibaba_coding_plan"),
"huggingface": ("hf", "hugging-face", "huggingface-hub"),
"novita": ("novita-ai", "novitaai"),
"xiaomi": ("mimo", "xiaomi-mimo"),
"tencent-tokenhub": ("tencent", "tokenhub", "tencent-cloud", "tencentmaas"),
"huggingface": ("hf", "hugging-face", "huggingface-hub"), "novita": ("novita-ai", "novitaai"),
"xiaomi": ("mimo", "xiaomi-mimo"), "tencent-tokenhub": ("tencent", "tokenhub", "tencent-cloud", "tencentmaas"),
"tencent-tokenplan": ("tokenplan", "tencent-lkeap"),
"bedrock": ("aws", "aws-bedrock", "amazon-bedrock", "amazon"),
"arcee": ("arcee-ai", "arceeai"),
"gmi": ("gmi-cloud", "gmicloud"),
"fireworks": ("fireworks-ai", "fw"),
"upstage": ("solar",),
"bedrock": ("aws", "aws-bedrock", "amazon-bedrock", "amazon"), "arcee": ("arcee-ai", "arceeai"),
"gmi": ("gmi-cloud", "gmicloud"), "fireworks": ("fireworks-ai", "fw"), "upstage": ("solar",),
"actual": ("actual-computer", "actualcomputer", "aci"),
"nebius-token-factory": (
"nebius", "nebius-tokenfactory", "nebius-tf", "token-factory", "tokenfactory",
),
"lmstudio": ("lmstudio", "lm-studio", "lm_studio"),
"custom": ("ollama",),
"nebius-token-factory": ("nebius", "nebius-tokenfactory", "nebius-tf", "token-factory", "tokenfactory"),
"lmstudio": ("lmstudio", "lm-studio", "lm_studio"), "custom": ("ollama",),
"local": ("vllm", "llamacpp", "llama.cpp", "llama-cpp"),
}
ALIASES: Dict[str, str] = {alias: canon for canon, aliases in _ALIAS_GROUPS.items() for alias in aliases}
# -- Display labels -----------------------------------------------------------
# Built dynamically from models.dev + overlays. Fallback for providers
# not in the catalog.
# -- Display labels for providers not in the models.dev catalog ---------------
_LABEL_OVERRIDES: Dict[str, str] = {
"moa": "Mixture of Agents",
"nous": "Nous Portal",
"openai-codex": "ChatGPT or Codex Subscription",
"copilot-acp": "GitHub Copilot ACP",
"stepfun": "StepFun Step Plan",
"xiaomi": "Xiaomi MiMo",
"gmi": "GMI Cloud",
"upstage": "Upstage Solar",
"actual": "Actual Computer",
"tencent-tokenhub": "Tencent TokenHub",
"nebius-token-factory": "Nebius Token Factory",
"tencent-tokenplan": "Tencent TokenPlan",
"lmstudio": "LM Studio",
"local": "Local endpoint",
"bedrock": "AWS Bedrock",
"vertex": "Google Vertex AI",
"ollama-cloud": "Ollama Cloud",
"xai-oauth": "xAI Grok OAuth (SuperGrok / Premium+)",
"opencode-free": "OpenCode Free",
"moa": "Mixture of Agents", "nous": "Nous Portal", "openai-codex": "ChatGPT or Codex Subscription",
"copilot-acp": "GitHub Copilot ACP", "stepfun": "StepFun Step Plan", "xiaomi": "Xiaomi MiMo", "gmi": "GMI Cloud",
"upstage": "Upstage Solar", "actual": "Actual Computer", "tencent-tokenhub": "Tencent TokenHub",
"nebius-token-factory": "Nebius Token Factory", "tencent-tokenplan": "Tencent TokenPlan", "lmstudio": "LM Studio",
"local": "Local endpoint", "bedrock": "AWS Bedrock", "vertex": "Google Vertex AI", "ollama-cloud": "Ollama Cloud",
"xai-oauth": "xAI Grok OAuth (SuperGrok / Premium+)", "opencode-free": "OpenCode Free",
}
# -- Transport → API mode mapping ---------------------------------------------
TRANSPORT_TO_API_MODE: Dict[str, str] = {
"openai_chat": "chat_completions",
"anthropic_messages": "anthropic_messages",
"codex_responses": "codex_responses",
"bedrock_converse": "bedrock_converse",
"openai_chat": "chat_completions", "anthropic_messages": "anthropic_messages",
"codex_responses": "codex_responses", "bedrock_converse": "bedrock_converse",
}
@@ -281,111 +167,66 @@ def normalize_provider(name: str) -> str:
return ALIASES.get(key, key)
def get_provider(name: str, *, allow_network: bool = True) -> Optional[ProviderDef]:
"""Look up a built-in provider by id or alias.
Resolution order: 1. Hermes overlays (for providers not in models.dev: nous, openai-codex, etc.)
2. models.dev catalog + Hermes overlay
"""
canonical = normalize_provider(name)
# Try to get models.dev data
def _models_dev_info(canonical: str, allow_network: bool = True):
"""models.dev entry or None. Single-arg call on the default path: test sites monkeypatch
``get_provider_info`` with single-arg lambdas."""
try:
from agent.models_dev import get_provider_info as _mdev_provider
# Keep the single-argument call on the default path: test sites
# monkeypatch get_provider_info with single-arg lambdas.
mdev_info = (
_mdev_provider(canonical)
if allow_network
else _mdev_provider(canonical, allow_network=False)
)
return _mdev_provider(canonical) if allow_network else _mdev_provider(canonical, allow_network=False)
except Exception:
mdev_info = None
return None
def _overlay_pdef(canonical, ov: HermesOverlay, name, env_vars, base_url, doc, source) -> ProviderDef:
return ProviderDef(id=canonical, name=name, transport=ov.transport, api_key_env_vars=env_vars, base_url=base_url,
base_url_env_var=ov.base_url_env_var, is_aggregator=ov.is_aggregator, auth_type=ov.auth_type, doc=doc,
source=source)
def get_provider(name: str, *, allow_network: bool = True) -> Optional[ProviderDef]:
"""Look up a built-in provider by id or alias: models.dev catalog merged with the Hermes overlay;
Hermes-only overlay (nous, openai-codex, …); plugin provider profiles with a concrete endpoint."""
canonical = normalize_provider(name)
mdev_info = _models_dev_info(canonical, allow_network)
overlay = HERMES_OVERLAYS.get(canonical)
if mdev_info is not None:
# Merge models.dev + overlay (defaults when no overlay); env vars = models.dev + hermes extra
ov = overlay or HermesOverlay()
env_vars = list(mdev_info.env)
for ev in ov.extra_env_vars:
if ev not in env_vars:
env_vars.append(ev)
return ProviderDef(
id=canonical,
name=mdev_info.name,
transport=ov.transport,
api_key_env_vars=tuple(env_vars),
base_url=ov.base_url_override or mdev_info.api,
base_url_env_var=ov.base_url_env_var,
is_aggregator=ov.is_aggregator,
auth_type=ov.auth_type,
doc=mdev_info.doc,
source="models.dev",
)
return _overlay_pdef(canonical, ov, mdev_info.name, tuple(env_vars), ov.base_url_override or mdev_info.api,
mdev_info.doc, "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",
)
# Plugin-registered provider profiles (plugins/model-providers/<name>/).
# Providers that ship only as plugin profiles (e.g. commandcode,
# tencent-tokenhub) are absent from models.dev and HERMES_OVERLAYS, so
# without this fallback they resolve as "Unknown provider" in /model,
# --provider, and the model-switch path even though the picker lists them
# (CANONICAL_PROVIDERS auto-extends from the same plugin registry).
return _overlay_pdef(canonical, overlay, _LABEL_OVERRIDES.get(canonical, canonical), overlay.extra_env_vars,
overlay.base_url_override, "", "hermes")
# Plugin-registered profiles (plugins/model-providers/<name>/) absent from models.dev and
# HERMES_OVERLAYS would otherwise be "Unknown provider" in /model, --provider and model-switch
# even though the picker lists them. Only profiles with a concrete endpoint resolve here:
# placeholder profiles like ``custom`` (aliases ollama/local/vllm) ship an empty base_url and
# are completed by config.yaml custom_providers — resolving them would preempt
# resolve_provider_full's custom step and collapse keyed ``custom:<name>`` ids to bare custom.
try:
from providers import get_provider_profile as _profile
_prof = _profile(canonical)
# Only profiles with a concrete endpoint resolve here. Placeholder
# profiles like ``custom`` (aliases: ollama/local/vllm) ship with an
# empty base_url and are completed by config.yaml custom_providers —
# resolving them here would preempt resolve_provider_full's
# custom-provider step and collapse keyed IDs
# (``custom:local-...``) back to a bare, endpoint-less ``custom``.
if _prof is not None and (_prof.base_url or "").strip():
_api_mode_to_transport = {v: k for k, v in TRANSPORT_TO_API_MODE.items()}
_transport = _api_mode_to_transport.get(_prof.api_mode, "openai_chat")
return ProviderDef(
id=canonical,
name=_prof.display_name or _prof.name or canonical,
transport=_transport,
api_key_env_vars=tuple(_prof.env_vars or ()),
base_url=_prof.base_url or "",
auth_type=_prof.auth_type or "api_key",
source="plugin-profile",
)
return ProviderDef(id=canonical, name=_prof.display_name or _prof.name or canonical,
transport=_api_mode_to_transport.get(_prof.api_mode, "openai_chat"),
api_key_env_vars=tuple(_prof.env_vars or ()), base_url=_prof.base_url or "",
auth_type=_prof.auth_type or "api_key", source="plugin-profile")
except Exception:
pass
return None
def get_label(provider_id: str) -> str:
"""Get a human-readable display name for a provider."""
"""Human-readable display name: label override, else models.dev name, else the id."""
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
return pdef.name if pdef else canonical
def is_aggregator(provider: str) -> bool:
@@ -397,30 +238,19 @@ def is_aggregator(provider: str) -> bool:
return pdef.is_aggregator if pdef else False
# Flat-namespace resellers (e.g. opencode-go, opencode-zen) are flagged
# ``is_aggregator=True`` because their live ``/v1/models`` returns bare model
# IDs ("deepseek-v4-flash") rather than ``vendor/model`` routing slugs — the
# model-switch resolver relies on that flag to search their flat catalog
# (see model_switch.py step d). But they are NOT routing aggregators: every
# model they list is a first-party model served under their own subscription,
# not a passthrough route to another provider's endpoint. The picker dedup
# (build_models_payload) must treat them differently from true routers like
# OpenRouter — a reseller's first-party "minimax-m3" must never be stripped
# just because a user's custom proxy also happens to serve a same-named model.
_FLAT_NAMESPACE_RESELLERS: frozenset[str] = frozenset({
# Use normalized provider IDs: normalize_provider("opencode-zen") -> "opencode".
"opencode-go",
"opencode",
})
# Flat-namespace resellers (opencode-go, opencode-zen) are flagged ``is_aggregator=True`` because
# their live ``/v1/models`` returns bare model IDs ("deepseek-v4-flash") rather than
# ``vendor/model`` routing slugs — model_switch searches their flat catalog on that flag. But they
# are NOT routing aggregators: every listed model is first-party under their own subscription, so
# picker dedup (build_models_payload) must not strip a reseller's "minimax-m3" just because a
# user's custom proxy serves a same-named model. Normalized ids: "opencode-zen" -> "opencode".
_FLAT_NAMESPACE_RESELLERS: frozenset[str] = frozenset({"opencode-go", "opencode"})
def is_routing_aggregator(provider: str) -> bool:
"""True only for TRUE routing aggregators (OpenRouter, named ``custom:*`` proxies).
Unlike ``is_aggregator``, excludes flat-namespace resellers (opencode-go/zen) whose catalog is
first-party. Use for "would selecting this model silently re-route away from the intended
provider?" -- i.e. picker dedup; reseller rows must not be deduped against user proxies.
"""
"""True only for TRUE routing aggregators (OpenRouter, named ``custom:*`` proxies) — excludes
flat-namespace resellers whose catalog is first-party. Use for "would selecting this model
silently re-route away from the intended provider?" (picker dedup)."""
provider_norm = normalize_provider(provider or "")
if provider_norm in _FLAT_NAMESPACE_RESELLERS:
return False
@@ -428,154 +258,99 @@ def is_routing_aggregator(provider: str) -> bool:
def is_official_openai_host(base_url: str) -> bool:
"""True when *base_url* points at OpenAI's official API host family.
Hostname-parsed matching only — never substring — so lookalike hosts
(``api.openai.com.attacker.test``) and path-segment spoofs (``proxy.test/api.openai.com/v1``)
are rejected. A genuine ``*.api.openai.com`` subdomain requires control of openai.com DNS, so
the dot-suffix match does not reopen the #32243 spoofing hole.
"""
"""True when *base_url* points at OpenAI's official API host family. Hostname-parsed matching
only — never substring — so lookalike hosts (``api.openai.com.attacker.test``) and path-segment
spoofs (``proxy.test/api.openai.com/v1``) are rejected; a genuine ``*.api.openai.com``
subdomain requires control of openai.com DNS."""
return base_url_host_matches(base_url, "api.openai.com")
# Exact hostnames that are Responses-API-native: api.meta.ai only achieves prompt-cache hits on
# Responses with prompt_cache_retention (chat/completions stays cache-cold); api.router.com (Ramp
# Router) keeps reasoning validation/summaries and prompt caching on /v1/responses and serves
# /v1/chat/completions as a minimal shim.
_RESPONSES_NATIVE_HOSTS: frozenset[str] = frozenset({"api.meta.ai", "api.router.com"})
def host_mandated_api_mode(base_url: str = "") -> Optional[str]:
"""Return the wire protocol a specific endpoint *requires*, or None.
Some hosts only accept one API mode and reject the others outright: - api.openai.com only
accepts the Responses API for its (reasoning) models when tools + reasoning are in play
(chat/completions 400s).
These are *mandatory* — a session carrying a stale api_mode (e.g. a /model switch that kept the
previous provider's ``chat_completions``) must be overridden to the host's required mode, not
merely filled in when empty.
"""
"""Return the wire protocol a specific endpoint *requires*, or None. Some hosts accept exactly
one API mode (api.openai.com 400s chat/completions for reasoning models with tools); these are
*mandatory*: a session carrying a stale api_mode (a /model switch that kept the previous
provider's ``chat_completions``) must be overridden, not merely filled in when empty.
Exact-hostname matching only — never substring — so lookalike hosts and path-segment spoofs are
not treated as the real endpoint."""
if not base_url:
return None
url_lower = base_url.rstrip("/").lower()
hostname = base_url_hostname(base_url)
# Exact-hostname matching only — never bare substring — so lookalike hosts
# (api.openai.com.attacker.test) and path-segment spoofs
# (proxy.test/api.openai.com/v1) are NOT treated as the real endpoint. (#32243)
if hostname == "api.kimi.com" and "/coding" in url_lower:
return "anthropic_messages"
if hostname == "api.anthropic.com" or url_lower.endswith("/anthropic"):
return "anthropic_messages"
# Official OpenAI host family: canonical + data-residency regional hosts
# (us./eu.api.openai.com) all mandate the Responses API for reasoning
# models with tools. Shared predicate keeps this lane in lockstep with
# catalog filtering and listing authority.
if is_official_openai_host(base_url):
return "codex_responses"
if hostname in _RESPONSES_NATIVE_HOSTS:
# Official OpenAI host family (canonical + us./eu. data-residency hosts) mandates Responses;
# the shared predicate keeps this in lockstep with catalog filtering and listing authority.
if is_official_openai_host(base_url) or hostname in _RESPONSES_NATIVE_HOSTS:
return "codex_responses"
if hostname.startswith("bedrock-runtime.") and base_url_host_matches(base_url, "amazonaws.com"):
return "bedrock_converse"
return None
# Exact hostnames (#32243) that are Responses-API-native:
# - api.meta.ai: Meta Model API only achieves prompt-cache hits on the Responses API with
# prompt_cache_retention; chat/completions stays cache-cold (0% vs 93-99% measured).
# - api.router.com: Ramp Router keeps reasoning-effort validation, reasoning summaries and prompt
# caching on /v1/responses; /v1/chat/completions is a minimal shim (docs.router.com/api/endpoint).
_RESPONSES_NATIVE_HOSTS: frozenset[str] = frozenset({"api.meta.ai", "api.router.com"})
def nous_api_mode(model: str = "") -> str:
"""Resolve the wire protocol for a Nous Portal model.
Portal serves its ``anthropic/*`` catalog on a native Anthropic Messages route
(``/v1/messages``) alongside the OpenAI-compatible ``/v1/chat/completions`` used by every other
model it proxies.
When *model* is empty/unknown, defaults to ``chat_completions`` — the historical Nous transport
— so callers that don't yet know the model stay on the safer OpenAI-compatible path.
"""
"""Wire protocol for a Nous Portal model: Portal serves its ``anthropic/*`` catalog on a native
Messages route alongside OpenAI-compatible chat/completions for everything else. Empty/unknown
model defaults to ``chat_completions`` (the historical Nous transport) as the safer path."""
if str(model or "").strip().lower().startswith("anthropic/"):
return "anthropic_messages"
return "chat_completions"
def determine_api_mode(provider: str, base_url: str = "", model: str = "") -> str:
"""Determine the API mode (wire protocol) for a provider/endpoint.
Resolution order: 1. Host-mandated mode (special endpoints that only accept one protocol). 2.
Nous Portal dual-wire (model-derived; overlay alone is openai_chat). 3. Known provider →
transport → TRANSPORT_TO_API_MODE. 4. Direct provider checks (bedrock). 5. Default:
'chat_completions'.
"""
"""API mode (wire protocol) for a provider/endpoint: host-mandated mode, then Nous dual-wire
(model-derived — the overlay alone says openai_chat and would pin Claude on the wrong wire),
then the known provider's transport, then bedrock, else ``chat_completions``."""
mandated = host_mandated_api_mode(base_url)
if mandated is not None:
return mandated
# Nous is dual-wire: anthropic/* → Messages, everything else →
# chat_completions. The Hermes overlay still advertises openai_chat
# (the majority of the Portal catalog), so the transport lookup below
# would pin Claude on the wrong wire without this carve-out.
provider_norm = (provider or "").strip().lower()
if provider_norm in {"nous", "nous-portal", "nousresearch"}:
if (provider or "").strip().lower() in {"nous", "nous-portal", "nousresearch"}:
return nous_api_mode(model)
pdef = get_provider(provider)
if pdef is not None:
return TRANSPORT_TO_API_MODE.get(pdef.transport, "chat_completions")
# Direct provider checks for providers not in HERMES_OVERLAYS
if provider == "bedrock":
return "bedrock_converse"
return "chat_completions"
# -- Provider from user config ------------------------------------------------
def _user_pdef(pid: str, name: str, base_url: str, key_env: str, transport: str = "openai_chat") -> ProviderDef:
"""``source="user-config"`` ProviderDef shared by ``providers:`` and ``custom_providers:`` entries."""
return ProviderDef(id=pid, name=name, transport=transport, api_key_env_vars=(key_env,) if key_env else (),
base_url=base_url, is_aggregator=False, auth_type="api_key", source="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."""
if not user_config or not isinstance(user_config, dict):
return None
entry = user_config.get(name)
entry = user_config.get(name) if isinstance(user_config, dict) and user_config else None
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 entry.get("api_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",
)
return _user_pdef(name, entry.get("name", "") or name,
entry.get("api", "") or entry.get("url", "") or entry.get("base_url", "") or "",
entry.get("key_env") or entry.get("api_key_env") or "",
entry.get("transport", "openai_chat") or "openai_chat")
def custom_provider_slug(display_name: str, provider_key: str = "") -> str:
"""Build the stable ``custom:`` identity for a configured provider.
Keyed ``providers:`` entries use their config key so the identity survives display-name
changes; legacy ``custom_providers:`` entries have no key, so their normalized display name
remains the identity.
"""
"""Stable ``custom:`` identity for a configured provider: keyed ``providers:`` entries use their
config key (survives display-name changes); legacy ``custom_providers:`` entries have no key,
so their normalized display name is the identity."""
identity = str(provider_key or "").strip() or str(display_name or "").strip()
normalized = identity.lower().replace(" ", "-")
return normalized if normalized.startswith("custom:") else f"custom:{normalized}"
def custom_provider_aliases(
display_name: str,
provider_key: str = "",
) -> frozenset[str]:
def custom_provider_aliases(display_name: str, provider_key: str = "") -> frozenset[str]:
"""Return every current and legacy identity accepted for one endpoint."""
aliases: set[str] = set()
for value in (display_name, provider_key):
@@ -591,175 +366,105 @@ def custom_provider_aliases(
return frozenset(aliases)
def resolve_custom_provider(
name: str,
custom_providers: Optional[List[Dict[str, Any]]],
) -> Optional[ProviderDef]:
"""Resolve a provider from the user's config.yaml ``custom_providers`` list."""
if not custom_providers or not isinstance(custom_providers, list):
return None
def resolve_custom_provider(name: str, custom_providers: Optional[List[Dict[str, Any]]]) -> Optional[ProviderDef]:
"""Resolve a provider from the user's config.yaml ``custom_providers`` list. A stored bare
``"custom"`` (corrupt state from a prior model-switch bug) falls back to the first valid entry
so existing configs self-heal."""
requested = (name or "").strip().lower()
if not requested:
if not requested or not custom_providers or not isinstance(custom_providers, list):
return None
# If the stored provider is the bare string "custom" (corrupt state
# from a prior model-switch bug), fall back to the first custom
# provider entry so existing configs self-heal. (GH #17478)
bare_custom_fallback = requested == "custom"
first_valid: Optional[ProviderDef] = None
for entry in custom_providers:
if not isinstance(entry, dict):
continue
display_name = (entry.get("name") or "").strip()
api_url = (
entry.get("base_url", "")
or entry.get("url", "")
or entry.get("api", "")
or ""
).strip()
api_url = (entry.get("base_url", "") or entry.get("url", "") or entry.get("api", "") or "").strip()
if not display_name or not api_url:
continue
key_env = (entry.get("key_env") or "").strip()
provider_key = (entry.get("provider_key") or "").strip()
pdef = ProviderDef(
id=custom_provider_slug(display_name, provider_key),
name=display_name,
transport="openai_chat",
api_key_env_vars=(key_env,) if key_env else (),
base_url=api_url,
is_aggregator=False,
auth_type="api_key",
source="user-config",
)
# Stash the first valid entry for bare-"custom" fallback
pdef = _user_pdef(custom_provider_slug(display_name, provider_key), display_name, api_url,
(entry.get("key_env") or "").strip())
if first_valid is None:
first_valid = pdef
if requested in custom_provider_aliases(display_name, provider_key):
return pdef
# Self-heal: bare "custom" matched nothing — return first valid entry
if bare_custom_fallback and first_valid:
if requested == "custom" and first_valid:
return first_valid
return None
def resolve_provider_full(
name: str,
user_providers: Optional[Dict[str, Any]] = None,
custom_providers: Optional[List[Dict[str, Any]]] = None,
) -> Optional[ProviderDef]:
"""Full resolution chain: built-in → models.dev → user config."""
def _lossy_alias_registry_pdef(raw: str, canonical: str) -> Optional[ProviderDef]:
"""Exact Hermes registry ids win over LOSSY alias collapsing (kimi-coding-cn must stay distinct
from kimi-coding instead of collapsing through the shared models.dev alias "kimi-for-coding").
A collapse is lossy only when MULTIPLE registry providers normalize to the same canonical name;
single-entry rewrites ("copilot" -> "github-copilot") are correct routing and keep resolving
through the built-in chain so overlay transports apply."""
try:
from hermes_cli.auth import PROVIDER_REGISTRY as _AUTH_PROVIDER_REGISTRY
_pcfg = _AUTH_PROVIDER_REGISTRY.get(raw)
if _pcfg is None:
return None
if sum(1 for _rid in _AUTH_PROVIDER_REGISTRY if normalize_provider(_rid) == canonical) > 1:
return ProviderDef(id=_pcfg.id, name=_pcfg.name, transport="openai_chat",
api_key_env_vars=tuple(_pcfg.api_key_env_vars or ()), base_url=_pcfg.inference_base_url or "",
source="hermes-auth-registry")
except Exception:
pass
return None
def _llamacpp_pdef() -> Optional[ProviderDef]:
"""The llamacpp aliases are a real provider whenever the managed server (or a detected external
one) resolves — reachability is the credential. Without this rung model-switch rejected the very
provider the Local Models 'Use' flow writes to config."""
try:
from hermes_cli.local_runtime.endpoint import resolve_llamacpp_endpoint
endpoint = resolve_llamacpp_endpoint(wait_for_boot_s=0)
except Exception:
endpoint = None
if not endpoint:
return None
return ProviderDef(id="llamacpp", name="Local", transport="openai_chat", api_key_env_vars=(), base_url=endpoint["base_url"],
source="local-runtime")
def resolve_provider_full(name: str, user_providers: Optional[Dict[str, Any]] = None,
custom_providers: Optional[List[Dict[str, Any]]] = None) -> Optional[ProviderDef]:
"""Full resolution chain: user ``providers.<raw name>`` -> lossy-alias registry id -> built-in
(models.dev + overlays) -> user providers (canonical, then raw) -> ``custom_providers`` ->
managed llamacpp -> models.dev directly. User-defined ``providers.<name>`` is tried FIRST on
the raw (pre-alias) name: a configured ``providers.openai`` pointing at api.openai.com must not
be hijacked by the legacy "openai" -> "openrouter" alias."""
canonical = normalize_provider(name)
raw = name.strip().lower()
# 0. User-defined config providers win over the built-in alias table.
# A user who declares ``providers.<name>`` in config.yaml has stated
# explicit intent for that name — it must not be hijacked by a legacy
# vendor alias (e.g. bare "openai" → "openrouter"). Resolve the raw
# name against user config FIRST so a configured ``providers.openai``
# (pointing at api.openai.com) beats the alias that would otherwise
# silently route to OpenRouter. Only the raw (pre-alias) name is tried
# here; canonical/alias resolution still happens below.
if user_providers:
user_pdef = resolve_user_provider(raw, user_providers)
if user_pdef is not None:
return user_pdef
# 0.5 Exact Hermes provider IDs must win over LOSSY alias collapsing.
# Example: kimi-coding-cn should stay distinct from kimi-coding instead of
# normalizing through the shared models.dev alias "kimi-for-coding".
# A collapse is lossy only when MULTIPLE distinct registry providers
# normalize to the same canonical name — resolving through the alias
# would then lose which one the caller meant. Single-entry rewrites
# (e.g. "copilot" → "github-copilot") are correct routing and must keep
# resolving through the built-in chain below so overlay transports apply.
if canonical != raw:
try:
from hermes_cli.auth import PROVIDER_REGISTRY as _AUTH_PROVIDER_REGISTRY
_pcfg = _AUTH_PROVIDER_REGISTRY.get(raw)
if _pcfg is not None:
_collapsed_siblings = [
_rid
for _rid in _AUTH_PROVIDER_REGISTRY
if normalize_provider(_rid) == canonical
]
if len(_collapsed_siblings) > 1:
return ProviderDef(
id=_pcfg.id,
name=_pcfg.name,
transport="openai_chat",
api_key_env_vars=tuple(_pcfg.api_key_env_vars or ()),
base_url=_pcfg.inference_base_url or "",
source="hermes-auth-registry",
)
except Exception:
pass
# 1. Built-in (models.dev + overlays)
pdef = _lossy_alias_registry_pdef(raw, canonical)
if pdef is not None:
return pdef
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(raw, user_providers)
if user_pdef is not None:
return user_pdef
# 2b. Saved custom providers from config
for candidate in (canonical, raw):
user_pdef = resolve_user_provider(candidate, user_providers)
if user_pdef is not None:
return user_pdef
custom_pdef = resolve_custom_provider(name, custom_providers)
if custom_pdef is not None:
return custom_pdef
# 2c. Managed local runtime: the llamacpp aliases are a real provider
# whenever the managed server (or a detected external one) resolves —
# no credential and no providers: entry required, the credential is
# reachability. Without this rung the model-switch path rejected the
# very provider the Local Models 'Use' flow writes to config
# ("Unknown provider 'llamacpp'" from the desktop dropdown).
if raw in ("llamacpp", "llama.cpp", "llama-cpp"):
try:
from hermes_cli.local_runtime.endpoint import resolve_llamacpp_endpoint
endpoint = resolve_llamacpp_endpoint(wait_for_boot_s=0)
except Exception:
endpoint = None
if endpoint:
return ProviderDef(
id="llamacpp",
name="Local",
transport="openai_chat",
api_key_env_vars=(),
base_url=endpoint["base_url"],
source="local-runtime",
)
# 3. Try models.dev directly (for providers not in our ALIASES)
pdef = _llamacpp_pdef()
if pdef is not None:
return pdef
try:
from agent.models_dev import get_provider_info as _mdev_provider
mdev_info = _mdev_provider(canonical)
mdev_info = _models_dev_info(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",
)
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

54
hermes_cli/route_identity.py
View File

@@ -28,7 +28,6 @@ def normalize_route_base_url(base_url: Any) -> str:
port = parsed.port
except (TypeError, ValueError):
return raw
route_host = parsed.netloc.rsplit("@", 1)[-1]
if route_host.startswith("[") or ":" in host:
host = f"[{host}]"
@@ -36,68 +35,33 @@ def normalize_route_base_url(base_url: Any) -> str:
host = f"{host}:{port}"
if "@" in parsed.netloc:
host = f"{parsed.netloc.rsplit('@', 1)[0]}@{host}"
path = parsed.path
if path.endswith("/") and not had_query_delimiter:
path = path[:-1]
normalized = urlunsplit((scheme, host, path, parsed.query, ""))
if had_query_delimiter and not parsed.query:
normalized += "?"
return normalized
def should_clear_context_pin(
configured_model: Any,
active_model: Any,
configured_base_url: Any,
active_base_url: Any,
configured_provider: Any,
active_provider: Any,
) -> bool:
def should_clear_context_pin(configured_model: Any, active_model: Any, configured_base_url: Any, active_base_url: Any,
configured_provider: Any, active_provider: Any) -> bool:
"""True when a configured ``model.context_length`` pin no longer matches its runtime route.
Fail-closed: any error during route comparison returns ``True`` (drop the pin) so a stale window
never silently inflates the compression threshold.
"""
never silently inflates the compression threshold."""
configured_model = str(configured_model or "").strip()
if configured_model and configured_model != str(active_model or "").strip():
return True
try:
from agent.agent_init import _context_route_mismatch
return _context_route_mismatch(
configured_base_url,
active_base_url,
configured_provider,
active_provider,
)
return _context_route_mismatch(configured_base_url, active_base_url, configured_provider, active_provider)
except Exception:
return True
async def should_clear_context_pin_async(
configured_model: Any,
active_model: Any,
configured_base_url: Any,
active_base_url: Any,
configured_provider: Any,
active_provider: Any,
) -> bool:
"""Async wrapper for ``should_clear_context_pin``.
Offloads the route comparison to a worker thread so async gateway handlers never run it on the
event loop — the resolution chain is cache-only (``allow_network=False``) but can still do cold-
start disk I/O. Shares all logic with the sync version — no code duplication.
"""
async def should_clear_context_pin_async(*args: Any) -> bool:
"""``should_clear_context_pin`` on a worker thread so async gateway handlers never run it on the
event loop — the resolution chain is cache-only (``allow_network=False``) but can still do
cold-start disk I/O."""
import asyncio
return await asyncio.to_thread(
should_clear_context_pin,
configured_model,
active_model,
configured_base_url,
active_base_url,
configured_provider,
active_provider,
)
return await asyncio.to_thread(should_clear_context_pin, *args)

806
hermes_cli/runtime_provider.py
View File

File diff suppressed because it is too large Load Diff

251
hermes_cli/runtime_provider_backends.py
View File

@@ -1,10 +1,8 @@
"""Provider-specific runtime builders for :mod:`hermes_cli.runtime_provider`.
Azure Foundry, the OpenRouter / bare-custom fallback resolver, Bedrock, and external-process
providers. Origin-internal collaborators are resolved on the origin module at call time via
:func:`_rp` so test patches on ``hermes_cli.runtime_provider.*`` (``_get_model_config``,
``load_config``, ``has_usable_secret``, ``_try_resolve_from_custom_pool``, …) still apply.
"""
"""Provider-specific runtime builders for :mod:`hermes_cli.runtime_provider`: Azure Foundry, the
OpenRouter / bare-custom fallback resolver, Bedrock, and external-process providers. Origin-internal
collaborators are resolved on the origin module at call time via :func:`_rp` so test patches on
``hermes_cli.runtime_provider.*`` (``_get_model_config``, ``load_config``, ``has_usable_secret``,
``_try_resolve_from_custom_pool``, …) still apply."""
from __future__ import annotations
@@ -18,15 +16,9 @@ from utils import base_url_host_matches
def _rp():
import hermes_cli.runtime_provider as origin
return origin
def _strip_v1(base_url: str) -> str:
"""Anthropic SDK appends /v1/messages itself — drop an inherited trailing /v1."""
return re.sub(r"/v1/?$", "", base_url)
# ── Azure Foundry ──────────────────────────────────────────────────────────────────────────
@@ -35,11 +27,7 @@ def _azure_entra_credentials(cfg_entra: Dict[str, Any]) -> Any:
``build_anthropic_client`` injects the bearer via an httpx hook)."""
AuthError = _rp().AuthError
try:
from agent.azure_identity_adapter import (
SCOPE_AI_AZURE_DEFAULT,
EntraIdentityConfig,
build_token_provider,
)
from agent.azure_identity_adapter import SCOPE_AI_AZURE_DEFAULT, EntraIdentityConfig, build_token_provider
except Exception as exc:
raise AuthError(
"Azure Foundry Entra ID auth requires the 'azure-identity' "
@@ -53,68 +41,15 @@ def _azure_entra_credentials(cfg_entra: Dict[str, Any]) -> Any:
raise AuthError(str(exc)) from exc
def _resolve_azure_foundry_runtime(
*,
requested_provider: str,
model_cfg: Dict[str, Any],
explicit_api_key: Optional[str] = None,
explicit_base_url: Optional[str] = None,
target_model: Optional[str] = None,
) -> Dict[str, Any]:
"""Azure Foundry: ``model.base_url`` + ``model.api_mode`` (or explicit overrides), API key from
``.env``/env or a per-request Entra ID token, trailing ``/v1`` stripped for Anthropic-style
endpoints."""
rp = _rp()
explicit_api_key = str(explicit_api_key or "").strip()
explicit_base_url_clean = str(explicit_base_url or "").strip().rstrip("/")
cfg_base_url, cfg_api_mode, cfg_auth_mode, cfg_entra = "", "chat_completions", "api_key", {}
if rp._cfg_provider(model_cfg) == "azure-foundry":
cfg_base_url = rp._config_base_url_for_provider(model_cfg, "azure-foundry")
cfg_api_mode = rp._parse_api_mode(model_cfg.get("api_mode")) or "chat_completions"
cfg_auth_mode = str(model_cfg.get("auth_mode") or "api_key").strip().lower() or "api_key"
if isinstance(model_cfg.get("entra"), dict):
cfg_entra = model_cfg["entra"]
# GPT-5.x / codex / o1-o4 deployments are Responses-API-only on Foundry.
effective_model = str(target_model or model_cfg.get("default") or "").strip()
cfg_api_mode = rp._azure_inferred_api_mode(effective_model, cfg_api_mode)
env_base_url = rp._getenv("AZURE_FOUNDRY_BASE_URL", "").strip().rstrip("/")
base_url = explicit_base_url_clean or cfg_base_url or env_base_url
if not base_url:
raise rp.AuthError(
"Azure Foundry requires a base URL. Set it via 'hermes model' or "
"the AZURE_FOUNDRY_BASE_URL environment variable."
)
if cfg_api_mode == "anthropic_messages":
base_url = _strip_v1(base_url)
if cfg_auth_mode == "entra_id":
if explicit_api_key:
# --api-key on the CLI while config says entra_id: honour the explicit string
# (escape hatch for one-off testing).
api_key, source, auth_mode = explicit_api_key, "explicit", "api_key"
else:
api_key, source, auth_mode = _azure_entra_credentials(cfg_entra), "entra_id", "entra_id"
clean_entra = {}
configured_scope = str(cfg_entra.get("scope") or "").strip()
if auth_mode == "entra_id" and configured_scope:
clean_entra["scope"] = configured_scope
return rp._runtime(
"azure-foundry", cfg_api_mode, base_url, api_key,
auth_mode=auth_mode, entra=clean_entra, source=source, requested_provider=requested_provider,
)
api_key = explicit_api_key
if not api_key:
try:
from hermes_cli.config import get_env_value
api_key = get_env_value("AZURE_FOUNDRY_API_KEY") or ""
except Exception:
api_key = ""
api_key = api_key or rp._getenv("AZURE_FOUNDRY_API_KEY", "").strip()
def _azure_foundry_api_key(rp, explicit_api_key: str) -> str:
if explicit_api_key:
return explicit_api_key
try:
from hermes_cli.config import get_env_value
api_key = get_env_value("AZURE_FOUNDRY_API_KEY") or ""
except Exception:
api_key = ""
api_key = api_key or rp._getenv("AZURE_FOUNDRY_API_KEY", "").strip()
if not api_key:
raise rp.AuthError(
"Azure Foundry requires an API key. Set AZURE_FOUNDRY_API_KEY in "
@@ -123,92 +58,103 @@ def _resolve_azure_foundry_runtime(
"model.auth_mode: entra_id in config.yaml (or pick "
"'Microsoft Entra ID' in 'hermes model')."
)
return rp._runtime(
"azure-foundry", cfg_api_mode, base_url, api_key,
auth_mode="api_key",
source="explicit" if (explicit_api_key or explicit_base_url) else "config",
requested_provider=requested_provider,
)
return api_key
def _resolve_azure_foundry_runtime(*, requested_provider: str, model_cfg: Dict[str, Any],
explicit_api_key: Optional[str] = None, explicit_base_url: Optional[str] = None,
target_model: Optional[str] = None) -> Dict[str, Any]:
"""Azure Foundry: ``model.base_url`` + ``model.api_mode`` (or explicit overrides), API key from
``.env``/env or a per-request Entra ID token, trailing ``/v1`` stripped for Anthropic-style
endpoints (the Anthropic SDK appends /v1/messages itself)."""
rp = _rp()
explicit_api_key = str(explicit_api_key or "").strip()
explicit_base_url_clean = str(explicit_base_url or "").strip().rstrip("/")
cfg_base_url, cfg_api_mode, cfg_auth_mode, cfg_entra = "", "chat_completions", "api_key", {}
if rp._cfg_provider(model_cfg) == "azure-foundry":
cfg_base_url = rp._config_base_url_for_provider(model_cfg, "azure-foundry")
cfg_api_mode = rp._parse_api_mode(model_cfg.get("api_mode")) or "chat_completions"
cfg_auth_mode = str(model_cfg.get("auth_mode") or "api_key").strip().lower() or "api_key"
if isinstance(model_cfg.get("entra"), dict):
cfg_entra = model_cfg["entra"]
# GPT-5.x / codex / o1-o4 deployments are Responses-API-only on Foundry.
effective_model = str(target_model or model_cfg.get("default") or "").strip()
cfg_api_mode = rp._azure_inferred_api_mode(effective_model, cfg_api_mode)
env_base_url = rp._getenv("AZURE_FOUNDRY_BASE_URL", "").strip().rstrip("/")
base_url = explicit_base_url_clean or cfg_base_url or env_base_url
if not base_url:
raise rp.AuthError(
"Azure Foundry requires a base URL. Set it via 'hermes model' or "
"the AZURE_FOUNDRY_BASE_URL environment variable."
)
if cfg_api_mode == "anthropic_messages":
base_url = re.sub(r"/v1/?$", "", base_url)
if cfg_auth_mode == "entra_id":
# --api-key on the CLI while config says entra_id: honour the explicit string (escape hatch
# for one-off testing).
if explicit_api_key:
api_key, source, auth_mode, entra = explicit_api_key, "explicit", "api_key", {}
else:
scope = str(cfg_entra.get("scope") or "").strip()
api_key, source, auth_mode, entra = _azure_entra_credentials(cfg_entra), "entra_id", "entra_id", (
{"scope": scope} if scope else {}
)
return rp._runtime("azure-foundry", cfg_api_mode, base_url, api_key, auth_mode=auth_mode, entra=entra, source=source,
requested_provider=requested_provider)
return rp._runtime("azure-foundry", cfg_api_mode, base_url, _azure_foundry_api_key(rp, explicit_api_key),
auth_mode="api_key", source="explicit" if (explicit_api_key or explicit_base_url) else "config",
requested_provider=requested_provider)
# ── OpenRouter / bare custom fallback ──────────────────────────────────────────────────────
def _resolve_openrouter_runtime(
*,
requested_provider: str,
explicit_api_key: Optional[str] = None,
explicit_base_url: Optional[str] = None,
*, requested_provider: str, explicit_api_key: Optional[str] = None, explicit_base_url: Optional[str] = None
) -> Dict[str, Any]:
"""Terminal resolver: OpenRouter, or a bare/aliased ``custom`` endpoint.
base_url precedence: explicit > CUSTOM_BASE_URL > trusted ``model.base_url`` > OPENROUTER_BASE_URL
> default. OPENAI_BASE_URL is deliberately NOT consulted — config.yaml is the single source of
truth for endpoint URLs. OpenRouter contexts prefer OPENROUTER_API_KEY; custom endpoints never
receive the OpenRouter key and only get env keys gated on their authoritative hosts.
"""
"""Terminal resolver: OpenRouter, or a bare/aliased ``custom`` endpoint. base_url precedence:
explicit > CUSTOM_BASE_URL > trusted ``model.base_url`` > OPENROUTER_BASE_URL > default.
OPENAI_BASE_URL is deliberately NOT consulted — config.yaml is the single source of truth for
endpoint URLs. OpenRouter contexts prefer OPENROUTER_API_KEY; custom endpoints never receive the
OpenRouter key and only get env keys gated on their authoritative hosts."""
rp = _rp()
model_cfg = rp._get_model_config()
cfg_base_url = model_cfg.get("base_url") if isinstance(model_cfg.get("base_url"), str) else ""
cfg_provider = model_cfg.get("provider") if isinstance(model_cfg.get("provider"), str) else ""
cfg_api_key = next(
(v.strip() for v in (model_cfg.get("api_key"), model_cfg.get("api")) if isinstance(v, str) and v.strip()),
"",
)
cfg_provider = (model_cfg.get("provider") if isinstance(model_cfg.get("provider"), str) else "").strip().lower()
cfg_api_key = next((v.strip() for v in (model_cfg.get("api_key"), model_cfg.get("api")) if isinstance(v, str) and v.strip()), "")
requested_norm = (requested_provider or "").strip().lower()
cfg_provider = cfg_provider.strip().lower()
# Aliases resolving to "custom" (ollama, vllm, …) follow bare-custom trust + routing rules.
if requested_norm and requested_norm != "custom" and rp._resolves_to_custom(requested_norm):
requested_norm = "custom"
env_openrouter_base_url = rp._getenv("OPENROUTER_BASE_URL", "").strip()
env_custom_base_url = rp._getenv("CUSTOM_BASE_URL", "").strip()
use_config_base_url = bool(cfg_base_url.strip()) and not explicit_base_url and (
(requested_norm == "auto" and cfg_provider in ("", "auto"))
or (requested_norm == "custom" and rp._config_base_url_trustworthy_for_bare_custom(cfg_base_url, cfg_provider))
)
base_url = (
(explicit_base_url or "").strip()
or env_custom_base_url
or (cfg_base_url.strip() if use_config_base_url else "")
or env_openrouter_base_url
or OPENROUTER_BASE_URL
).rstrip("/")
base_url = ((explicit_base_url or "").strip() or env_custom_base_url or (cfg_base_url.strip() if use_config_base_url else "")
or env_openrouter_base_url or OPENROUTER_BASE_URL).rstrip("/")
is_openrouter_url = base_url_host_matches(base_url, "openrouter.ai")
# Explicitly-configured OpenRouter mirrors (OPENROUTER_BASE_URL + provider=openrouter) still
# count as OpenRouter for key selection.
is_openrouter_context = is_openrouter_url or (
requested_norm == "openrouter"
and (env_openrouter_base_url or base_url == env_openrouter_base_url)
requested_norm == "openrouter" and (env_openrouter_base_url or base_url == env_openrouter_base_url)
and base_url == (env_openrouter_base_url or "").rstrip("/")
)
if is_openrouter_context:
candidates = [explicit_api_key, rp._getenv("OPENROUTER_API_KEY"), rp._getenv("OPENAI_API_KEY")]
else:
candidates = [
explicit_api_key,
(cfg_api_key if use_config_base_url else ""),
*rp._host_gated_env_key_candidates(base_url, ollama=True),
]
candidates = [explicit_api_key, (cfg_api_key if use_config_base_url else ""),
*rp._host_gated_env_key_candidates(base_url, ollama=True)]
api_key = next((str(c or "").strip() for c in candidates if rp.has_usable_secret(c)), "")
source = "explicit" if (explicit_api_key or explicit_base_url) else "env/config"
cfg_api_mode = rp._parse_api_mode(model_cfg.get("api_mode"))
# Explicit "custom" stays "custom" rather than relabeling to "openrouter".
if requested_norm != "custom":
return rp._runtime(
"openrouter",
rp._parse_api_mode(model_cfg.get("api_mode")) or rp._detect_api_mode_for_url(base_url) or "chat_completions",
base_url, api_key, source=source,
)
return rp._runtime("openrouter", cfg_api_mode or rp._detect_api_mode_for_url(base_url) or "chat_completions", base_url,
api_key, source=source)
if base_url:
# provider_name makes pool lookup prefer name match over base_url (fixes credential
# mix-ups when multiple custom providers share a base_url).
pool_result = rp._try_resolve_from_custom_pool(
base_url, "custom", rp._parse_api_mode(model_cfg.get("api_mode")),
provider_name=requested_provider if requested_norm != "custom" else None,
)
pool_result = rp._try_resolve_from_custom_pool(base_url, "custom", cfg_api_mode, provider_name=None)
if pool_result:
return pool_result
# Local no-auth servers get a placeholder key — the OpenAI SDK requires a non-empty string.
@@ -236,17 +182,10 @@ def _resolve_bedrock_runtime(requested_provider: str, model_cfg: Dict[str, Any],
Claude → AnthropicBedrock SDK (prompt caching, thinking budgets); others → Converse API.
AWS_BEARER_TOKEN_BEDROCK auth is unsupported by AnthropicBedrock (SigV4 only), so bearer users
go through Converse regardless of model."""
from agent.bedrock_adapter import (
bedrock_openai_base_url,
has_aws_credentials,
is_anthropic_bedrock_model,
is_openai_bedrock_model,
resolve_aws_auth_env_var,
resolve_bedrock_bearer_token,
resolve_bedrock_runtime_region,
)
from agent.bedrock_adapter import (bedrock_openai_base_url, has_aws_credentials, is_anthropic_bedrock_model,
is_openai_bedrock_model, resolve_aws_auth_env_var, resolve_bedrock_bearer_token,
resolve_bedrock_runtime_region)
from hermes_cli.config import load_config # direct (not the origin delegate), as before
rp = _rp()
# Explicitly selected bedrock trusts boto3's credential chain (IMDS, ECS/Lambda roles, SSO)
# which the env-var check can't detect.
@@ -267,20 +206,12 @@ def _resolve_bedrock_runtime(requested_provider: str, model_cfg: Dict[str, Any],
guardrail_config = _bedrock_guardrail_config(bedrock_cfg)
current_model = str(target_model or model_cfg.get("default") or "").strip()
has_bearer_token = bool(os.environ.get("AWS_BEARER_TOKEN_BEDROCK", "").strip())
runtime = rp._runtime(
"bedrock", "bedrock_converse", f"https://bedrock-runtime.{region}.amazonaws.com", "aws-sdk",
source=auth_source, region=region, requested_provider=requested_provider,
)
runtime = rp._runtime("bedrock", "bedrock_converse", f"https://bedrock-runtime.{region}.amazonaws.com", "aws-sdk",
source=auth_source, region=region, requested_provider=requested_provider)
if is_openai_bedrock_model(current_model):
bearer = resolve_bedrock_bearer_token()
runtime.update(
api_mode="codex_responses",
base_url=bedrock_openai_base_url(region),
api_key=bearer or "aws-sdk",
source="AWS_BEARER_TOKEN_BEDROCK" if bearer else auth_source,
model=current_model,
bedrock_openai=True,
)
runtime.update(api_mode="codex_responses", base_url=bedrock_openai_base_url(region), api_key=bearer or "aws-sdk",
source="AWS_BEARER_TOKEN_BEDROCK" if bearer else auth_source, model=current_model, bedrock_openai=True)
elif is_anthropic_bedrock_model(current_model) and not has_bearer_token:
runtime.update(api_mode="anthropic_messages", bedrock_anthropic=True)
if guardrail_config:
@@ -298,9 +229,7 @@ def _is_external_process_provider(provider: str) -> bool:
if not name:
return False
try:
from hermes_cli.auth import PROVIDER_REGISTRY
pconfig = PROVIDER_REGISTRY.get(name)
pconfig = _rp().PROVIDER_REGISTRY.get(name)
if pconfig is not None:
return pconfig.auth_type == "external_process"
except Exception:
@@ -317,8 +246,6 @@ def _is_external_process_provider(provider: str) -> bool:
def _resolve_external_process_runtime(provider: str, requested_provider: str) -> Dict[str, Any]:
rp = _rp()
creds = rp.resolve_external_process_provider_credentials(provider)
return rp._runtime(
provider, "chat_completions", creds.get("base_url", "").rstrip("/"), creds.get("api_key", ""),
command=creds.get("command", ""), args=list(creds.get("args") or []),
source=creds.get("source", "process"), requested_provider=requested_provider,
)
return rp._runtime(provider, "chat_completions", creds.get("base_url", "").rstrip("/"), creds.get("api_key", ""),
command=creds.get("command", ""), args=list(creds.get("args") or []),
source=creds.get("source", "process"), requested_provider=requested_provider)

376
hermes_cli/runtime_provider_custom.py
View File

@@ -1,12 +1,9 @@
"""Custom-provider resolution: ``providers:`` / ``custom_providers:`` lookup, identity
recovery, custom credential pools, and the named-custom runtime builder.
Extracted from :mod:`hermes_cli.runtime_provider`; every public/private name here is
re-exported there. Origin-internal collaborators (``load_config``, ``_get_model_config``,
``load_pool``, ``has_usable_secret``, ``custom_provider_pool_key_candidates``, …) are looked up
on the origin module AT CALL TIME via :func:`_rp` so ``monkeypatch.setattr(runtime_provider,
name, …)`` in tests keeps working for moved bodies.
"""
"""Custom-provider resolution: ``providers:`` / ``custom_providers:`` lookup, identity recovery,
custom credential pools, and the named-custom runtime builder. Extracted from
:mod:`hermes_cli.runtime_provider` (every name re-exported there); origin-internal collaborators
(``load_config``, ``_get_model_config``, ``load_pool``, ``has_usable_secret``, …) are looked up on
the origin module AT CALL TIME via :func:`_rp` so ``monkeypatch.setattr(runtime_provider, name, …)``
keeps working for moved bodies."""
from __future__ import annotations
@@ -19,11 +16,12 @@ from utils import base_url_hostname
logger = logging.getLogger("hermes_cli.runtime_provider")
_LLAMACPP_ALIASES = ("llamacpp", "llama.cpp", "llama-cpp")
def _rp():
"""Origin module, late-bound so test patches on ``hermes_cli.runtime_provider.*`` apply."""
import hermes_cli.runtime_provider as origin
return origin
@@ -65,11 +63,9 @@ def _lift_model_capabilities(entry: Dict[str, Any], model: Optional[str], result
def _lift_max_output_tokens(entry: Dict[str, Any], result: Dict[str, Any]) -> None:
"""``max_output_tokens`` or ``max_tokens`` on a provider entry pins its own output limit.
Gateway/CLI map it onto ``AIAgent.max_tokens`` only when top-level ``model.max_tokens`` is
unset, so the documented global key still wins.
"""
"""``max_output_tokens`` or ``max_tokens`` on a provider entry pins its own output limit;
gateway/CLI map it onto ``AIAgent.max_tokens`` only when top-level ``model.max_tokens`` is
unset, so the documented global key still wins."""
for key in ("max_output_tokens", "max_tokens"):
value = entry.get(key)
if isinstance(value, int) and value > 0:
@@ -84,14 +80,8 @@ def _lift_extra_headers(entry: Dict[str, Any], result: Dict[str, Any]) -> None:
result["extra_headers"] = extra_headers
def _lift_common_custom_fields(
entry: Dict[str, Any],
result: Dict[str, Any],
*,
provider_key: str,
key_env: str,
api_mode: Optional[str],
) -> None:
def _lift_common_custom_fields(entry: Dict[str, Any], result: Dict[str, Any], *, provider_key: str, key_env: str,
api_mode: Optional[str]) -> None:
"""Copy the optional fields shared by ``providers:`` and legacy ``custom_providers:`` entries."""
if key_env:
result["key_env"] = key_env
@@ -104,23 +94,19 @@ def _lift_common_custom_fields(
if api_mode:
result["api_mode"] = api_mode
_lift_max_output_tokens(entry, result)
capabilities = _filter_capabilities(entry.get("capabilities"))
if capabilities:
result["capabilities"] = capabilities
_lift_model_capabilities(entry, None, result)
# ── config lookup ──────────────────────────────────────────────────────────────────────────
def _shadowed_by_builtin(requested_norm: str) -> bool:
"""Raw names map to custom providers only when they are not canonical built-ins.
Explicit ``custom:<name>`` keys always target the saved entry, and bare ``custom`` is
exempt: a user may literally name a ``providers:`` entry "custom" (returning None before
the config scan made such cron jobs fail with ``auth_unavailable``). Defer to the built-in
only when the raw name IS the canonical provider (``nous``); an entry matching merely an
alias (``kimi`` → ``kimi-coding``) is the user's target.
"""
"""Raw names map to custom providers only when they are not canonical built-ins. Explicit
``custom:<name>`` keys always target the saved entry, and bare ``custom`` is exempt: a user may
literally name a ``providers:`` entry "custom" (returning None before the config scan made such
cron jobs fail with ``auth_unavailable``). Defer to the built-in only when the raw name IS the
canonical provider (``nous``); an entry matching merely an alias (``kimi`` → ``kimi-coding``)
is the user's target."""
if requested_norm == "custom" or requested_norm.startswith("custom:"):
return False
rp = _rp()
@@ -134,7 +120,6 @@ def _shadowed_by_builtin(requested_norm: str) -> bool:
def _match_new_style_provider(requested_norm: str, providers: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""Scan ``providers:`` (new-style, keyed) for ``requested_norm``."""
from hermes_cli.config import is_provider_enabled
rp = _rp()
for ep_name, entry in providers.items():
# ``providers.<name>.enabled: false`` entries stay in config but are invisible here.
@@ -149,12 +134,8 @@ def _match_new_style_provider(requested_norm: str, providers: Dict[str, Any]) ->
base_url = _entry_url(entry)
if not base_url:
continue
result: Dict[str, Any] = {
"name": entry.get("name", ep_name),
"base_url": base_url.strip(),
"api_key": api_key or _clean(entry.get("api_key", "")),
"model": entry.get("default_model", ""),
}
result: Dict[str, Any] = {"name": entry.get("name", ep_name), "base_url": base_url.strip(),
"api_key": api_key or _clean(entry.get("api_key", "")), "model": entry.get("default_model", "")}
# Command that PRINTS a short-lived credential; wrapped in a per-request token provider.
key_cmd = _clean(entry.get("key_cmd", ""))
if key_cmd:
@@ -162,9 +143,7 @@ def _match_new_style_provider(requested_norm: str, providers: Dict[str, Any]) ->
# v12 migration writes ``transport``; hand-edited configs may still use ``api_mode``.
# Accept both or migrated configs silently downgrade to chat_completions.
_lift_common_custom_fields(
entry, result,
provider_key=_clean(ep_name),
key_env=key_env,
entry, result, provider_key=_clean(ep_name), key_env=key_env,
api_mode=rp._parse_api_mode(entry.get("api_mode") or entry.get("transport")),
)
return result
@@ -174,9 +153,7 @@ def _match_new_style_provider(requested_norm: str, providers: Dict[str, Any]) ->
def _match_legacy_custom_provider(requested_norm: str, custom_providers) -> Optional[Dict[str, Any]]:
"""Scan the legacy ``custom_providers:`` list for ``requested_norm``."""
for entry in custom_providers:
if not isinstance(entry, dict):
continue
name, base_url = entry.get("name"), entry.get("base_url")
name, base_url = (entry.get("name"), entry.get("base_url")) if isinstance(entry, dict) else (None, None)
if not isinstance(name, str) or not isinstance(base_url, str):
continue
provider_key = _clean(entry.get("provider_key", ""))
@@ -186,12 +163,8 @@ def _match_legacy_custom_provider(requested_norm: str, custom_providers) -> Opti
model_name = _clean(entry.get("model", ""))
if model_name:
result["model"] = model_name
_lift_common_custom_fields(
entry, result,
provider_key=provider_key,
key_env=_clean(entry.get("key_env", "")),
api_mode=_rp()._parse_api_mode(entry.get("api_mode")),
)
_lift_common_custom_fields(entry, result, provider_key=provider_key, key_env=_clean(entry.get("key_env", "")),
api_mode=_rp()._parse_api_mode(entry.get("api_mode")))
return result
return None
@@ -200,33 +173,24 @@ def _get_named_custom_provider(requested_provider: str) -> Optional[Dict[str, An
requested_norm = _normalize_custom_provider_name(requested_provider or "")
if not requested_norm or requested_norm == "auto" or _shadowed_by_builtin(requested_norm):
return None
rp = _rp()
config = rp.load_config()
providers = config.get("providers")
if isinstance(providers, dict):
found = _match_new_style_provider(requested_norm, providers)
if found:
return found
found = _match_new_style_provider(requested_norm, providers) if isinstance(providers, dict) else None
if found:
return found
if isinstance(config.get("custom_providers"), dict):
logger.warning(
"custom_providers in config.yaml is a dict, not a list. "
"Each entry must be prefixed with '-' in YAML. "
"Run 'hermes doctor' for details."
)
logger.warning("custom_providers in config.yaml is a dict, not a list. "
"Each entry must be prefixed with '-' in YAML. "
"Run 'hermes doctor' for details.")
return None
custom_providers = rp.get_compatible_custom_providers(config)
if not custom_providers:
return None
return _match_legacy_custom_provider(requested_norm, custom_providers)
return _match_legacy_custom_provider(requested_norm, custom_providers) if custom_providers else None
def has_named_custom_provider(requested_provider: str) -> bool:
"""True when config defines a ``providers:`` / ``custom_providers:`` entry matching the request.
Public wrapper so other modules (e.g. the cronjob tool) need not reach into a private helper.
"""
"""True when config defines a ``providers:`` / ``custom_providers:`` entry matching the request
(public wrapper so e.g. the cronjob tool need not reach into a private helper)."""
try:
return _rp()._get_named_custom_provider(requested_provider) is not None
except Exception:
@@ -244,96 +208,72 @@ def _find_custom_identity(matches: Callable[[Dict[str, Any]], bool]) -> Optional
config = rp.load_config()
except Exception:
return None
providers = config.get("providers")
if isinstance(providers, dict):
for ep_name, entry in providers.items():
if isinstance(entry, dict) and matches(entry):
return custom_provider_slug(str(ep_name), str(ep_name))
try:
custom_providers = rp.get_compatible_custom_providers(config)
except Exception:
custom_providers = None
for entry in custom_providers or []:
if not isinstance(entry, dict):
continue
name = entry.get("name")
name = entry.get("name") if isinstance(entry, dict) else None
if isinstance(name, str) and name.strip() and matches(entry):
return custom_provider_slug(name, str(entry.get("provider_key", "") or ""))
return None
def find_custom_provider_identity(base_url: str) -> Optional[str]:
"""Map an endpoint URL back to its canonical ``custom:<name>`` menu key.
Session persistence stores the agent's *resolved* provider, and for every named custom
endpoint that is the literal string ``"custom"`` — the entry name is lost, and the api_key is
deliberately never persisted.
"""
"""Map an endpoint URL back to its canonical ``custom:<name>`` menu key. Session persistence
stores the agent's *resolved* provider, which for every named custom endpoint is the literal
string ``"custom"`` — the entry name is lost, and the api_key is deliberately never persisted."""
target = _normalize_base_url_for_match(base_url)
if not target:
return None
return _find_custom_identity(lambda entry: _normalize_base_url_for_match(_entry_url(entry)) == target)
def find_custom_provider_identity_by_model(model: str) -> Optional[str]:
"""Map a model id back to the ``custom:<name>`` entry that serves it.
def _model_id_matches(value: Any, target: str) -> bool:
return isinstance(value, str) and value.strip().lower() == target
Companion to :func:`find_custom_provider_identity` for persistence paths where no base_url
survived the round-trip: the session row always stores the model name.
"""
def find_custom_provider_identity_by_model(model: str) -> Optional[str]:
"""Map a model id back to the ``custom:<name>`` entry that serves it — companion to
:func:`find_custom_provider_identity` for persistence paths where no base_url survived the
round-trip (the session row always stores the model name)."""
target = str(model or "").strip().lower()
if not target:
return None
def _entry_serves_model(entry: Dict[str, Any]) -> bool:
for key in ("model", "default_model"):
value = entry.get(key)
if isinstance(value, str) and value.strip().lower() == target:
return True
if any(_model_id_matches(entry.get(key), target) for key in ("model", "default_model")):
return True
models = entry.get("models")
if isinstance(models, dict):
return any(str(mid).strip().lower() == target for mid in models)
if isinstance(models, list):
for item in models:
if isinstance(item, str) and item.strip().lower() == target:
return True
if isinstance(item, dict):
mid = item.get("id") or item.get("name")
if isinstance(mid, str) and mid.strip().lower() == target:
return True
return any(_model_id_matches(item.get("id") or item.get("name") if isinstance(item, dict) else item, target)
for item in models)
return False
return _find_custom_identity(_entry_serves_model)
def canonical_custom_identity(
*,
base_url: Optional[str] = None,
config_provider: Optional[str] = None,
model: Optional[str] = None,
) -> Optional[str]:
"""Recover a routable ``custom:<name>`` identity for a bare custom provider.
Every path that persists or restores a session's provider override must run the resolved
provider through this so a bare ``"custom"`` is upgraded back to its durable
``custom:<name>`` menu key. Recovery sources, in priority order: (1) ``base_url`` reverse
lookup — the one fact that always survives the round-trip when a URL was recorded; (2)
``model`` reverse lookup (``model``/``default_model``/``models`` catalog); (3) the configured
provider (arg, then ``model.provider``, then ``HERMES_INFERENCE_PROVIDER``) when it names a
real entry.
"""
def canonical_custom_identity(*, base_url: Optional[str] = None, config_provider: Optional[str] = None,
model: Optional[str] = None) -> Optional[str]:
"""Recover a routable ``custom:<name>`` identity for a bare custom provider. Every path that
persists or restores a session's provider override must run the resolved provider through this
so a bare ``"custom"`` is upgraded back to its durable menu key. Sources in priority order:
(1) ``base_url`` reverse lookup — the one fact that always survives the round-trip when a URL
was recorded; (2) ``model`` reverse lookup (``model``/``default_model``/``models`` catalog);
(3) the configured provider (arg, ``model.provider``, ``HERMES_INFERENCE_PROVIDER``) when it
names a real entry."""
rp = _rp()
if base_url:
identity = find_custom_provider_identity(base_url)
if identity:
return identity
if model:
identity = find_custom_provider_identity_by_model(model)
if identity:
return identity
identity = (find_custom_provider_identity(base_url) if base_url else None) or (
find_custom_provider_identity_by_model(model) if model else None)
if identity:
return identity
candidate = str(config_provider or "").strip()
if not candidate:
try:
@@ -342,37 +282,33 @@ def canonical_custom_identity(
candidate = ""
if not candidate:
candidate = os.environ.get("HERMES_INFERENCE_PROVIDER", "").strip()
candidate_norm = _normalize_custom_provider_name(candidate)
# A bare/non-routable candidate cannot heal a bare custom override.
if not candidate_norm or candidate_norm in {"custom", "auto", "openrouter"}:
return None
# Only when it resolves to a configured entry — never invent a ``custom:<x>`` resolution
# can't honor.
# can't honor. ``candidate`` may be the entry's DISPLAY NAME, not the durable identity of a
# keyed ``providers:`` entry — re-resolve via its endpoint so every path returns the same
# config-key slug.
try:
entry = rp._get_named_custom_provider(candidate)
if entry is not None:
# ``candidate`` may be the entry's DISPLAY NAME, not the durable identity of a keyed
# ``providers:`` entry — re-resolve via its endpoint so every path returns the same
# config-key slug.
identity = find_custom_provider_identity(str(entry.get("base_url") or ""))
if identity:
return identity
return candidate_norm if candidate_norm.startswith("custom:") else f"custom:{candidate_norm}"
except Exception:
pass
return None
return None
if entry is None:
return None
try:
identity = find_custom_provider_identity(str(entry.get("base_url") or ""))
except Exception:
return None
return identity or custom_provider_slug(candidate_norm)
def is_routable_provider(provider: Optional[str]) -> bool:
"""Whether a provider name currently resolves to a routable route.
Empty/None/``auto`` is vacuously routable (agent build falls back to the configured
default). Bare ``custom`` is the resolved billing class shared by every named entry — not a
routable identity; restore paths must heal it (:func:`canonical_custom_identity`) or fall
back. Anything else is routable iff the full chain (built-in -> ``providers:`` ->
``custom_providers:`` -> models.dev) resolves it.
"""
"""Whether a provider name currently resolves to a routable route. Empty/None/``auto`` is
vacuously routable (agent build falls back to the configured default). Bare ``custom`` is the
resolved billing class shared by every named entry — not a routable identity; restore paths
must heal it (:func:`canonical_custom_identity`) or fall back. Anything else is routable iff the
full chain (built-in -> ``providers:`` -> ``custom_providers:`` -> models.dev) resolves it."""
name = str(provider or "").strip()
if not name or name.lower() == "auto":
return True
@@ -380,12 +316,9 @@ def is_routable_provider(provider: Optional[str]) -> bool:
return False
try:
from hermes_cli.providers import resolve_provider_full
rp = _rp()
config = rp.load_config()
return resolve_provider_full(
name, config.get("providers"), rp.get_compatible_custom_providers(config)
) is not None
return resolve_provider_full(name, config.get("providers"), rp.get_compatible_custom_providers(config)) is not None
except Exception:
return False
@@ -394,10 +327,7 @@ def is_routable_provider(provider: Optional[str]) -> bool:
def _try_resolve_from_custom_pool(
base_url: str,
provider_label: str,
api_mode_override: Optional[str] = None,
provider_name: Optional[str] = None,
base_url: str, provider_label: str, api_mode_override: Optional[str] = None, provider_name: Optional[str] = None
) -> Optional[Dict[str, Any]]:
"""Runtime dict from the first credential pool that owns this custom endpoint, else None."""
rp = _rp()
@@ -410,12 +340,8 @@ def _try_resolve_from_custom_pool(
for pool_key in candidates:
try:
pool = rp.load_pool(pool_key)
if not pool.has_credentials():
continue
entry = pool.select()
if entry is None:
continue
pool_api_key = rp._pool_entry_api_key(entry)
entry = pool.select() if pool.has_credentials() else None
pool_api_key = rp._pool_entry_api_key(entry) if entry is not None else ""
if not pool_api_key:
continue
if not rp.has_usable_secret(pool_api_key) and rp._loopback_hostname(base_url_hostname(base_url)):
@@ -423,14 +349,8 @@ def _try_resolve_from_custom_pool(
# services; has_usable_secret's 4-char floor rejects them. Every other path
# substitutes "no-key-required" for a loopback endpoint — this was the one gap.
pool_api_key = "no-key-required"
return rp._runtime(
provider_label,
api_mode_override or rp._detect_api_mode_for_url(base_url) or "chat_completions",
base_url,
pool_api_key,
source=f"pool:{pool_key}",
credential_pool=pool,
)
return rp._runtime(provider_label, api_mode_override or rp._detect_api_mode_for_url(base_url) or "chat_completions",
base_url, pool_api_key, source=f"pool:{pool_key}", credential_pool=pool)
except Exception:
continue
return None
@@ -443,16 +363,11 @@ def _custom_provider_request_overrides(custom_provider: Dict[str, Any]) -> Dict[
return {"extra_body": dict(extra_body)}
def _apply_custom_provider_extras(
custom_provider: Dict[str, Any], target_model: Optional[str], result: Dict[str, Any]
) -> None:
def _apply_custom_provider_extras(custom_provider: Dict[str, Any], target_model: Optional[str], result: Dict[str, Any]) -> None:
"""Copy model / capabilities / max_output_tokens / extra_headers / request_overrides onto a
resolved custom runtime.
An explicit ``target_model`` wins over the provider's configured default (auxiliary slots /
background-review resolve a concrete model and must not fall back to ``default_model``).
``extra_headers`` may carry credentials — NEVER log them.
"""
resolved custom runtime. An explicit ``target_model`` wins over the provider's configured
default (auxiliary slots / background-review resolve a concrete model and must not fall back to
``default_model``). ``extra_headers`` may carry credentials — NEVER log them."""
model_name = target_model or custom_provider.get("model")
if model_name:
result["model"] = model_name
@@ -463,52 +378,45 @@ def _apply_custom_provider_extras(
result["extra_headers"] = dict(custom_provider["extra_headers"])
request_overrides = _custom_provider_request_overrides(custom_provider)
if request_overrides:
result["request_overrides"] = {**dict(result.get("request_overrides") or {}), **request_overrides}
result["request_overrides"] = {**(result.get("request_overrides") or {}), **request_overrides}
def _resolve_llamacpp_runtime(requested_provider: str, explicit_api_key: Optional[str]) -> Dict[str, Any]:
"""Managed llama.cpp runtime: the supervised (or detected external) server, or a typed error.
No server => say so and stop; falling through to the generic custom path would surface "local
server is off" as OpenRouter's baffling "401 Invalid API key". The switch's state picks the
message (server off → point at the switch; else the setup pane).
"""
message (server off → point at the switch; else the setup pane)."""
rp = _rp()
try:
from hermes_cli.local_runtime.endpoint import resolve_llamacpp_endpoint
endpoint = resolve_llamacpp_endpoint()
except Exception: # noqa: BLE001 — resolution is best-effort
endpoint = None
if endpoint:
return rp._runtime(
"custom",
"chat_completions",
endpoint["base_url"],
(explicit_api_key or "").strip() or endpoint["api_key"] or "no-key-required",
source="local-runtime",
requested_provider=requested_provider,
)
return rp._runtime("custom", "chat_completions", endpoint["base_url"],
(explicit_api_key or "").strip() or endpoint["api_key"] or "no-key-required", source="local-runtime",
requested_provider=requested_provider)
try:
enabled = bool((rp.load_config().get("local_runtime") or {}).get("enabled"))
except Exception: # noqa: BLE001
enabled = False
if enabled:
raise ValueError(
"The local model server isn't running. It may still be "
"starting — try again in a moment, or check Settings → "
"Providers → Local models."
)
raise ValueError(
"The local model server is turned off. Turn it back on in "
"Settings → Providers → Local models, or switch to another "
"model."
)
raise ValueError("The local model server isn't running. It may still be "
"starting — try again in a moment, or check Settings → "
"Providers → Local models.")
raise ValueError("The local model server is turned off. Turn it back on in "
"Settings → Providers → Local models, or switch to another "
"model.")
def _resolve_direct_alias_runtime(
requested_provider: str, explicit_api_key: Optional[str], explicit_base_url: str
) -> Dict[str, Any]:
def _custom_runtime(rp, base_url: str, api_key: Any, api_mode: Optional[str], **extra: Any) -> Dict[str, Any]:
"""``custom`` runtime dict with URL-detected api_mode fallback and the no-auth placeholder."""
return rp._runtime("custom", api_mode or rp._detect_api_mode_for_url(base_url) or "chat_completions", base_url,
api_key or "no-key-required", **extra)
def _resolve_direct_alias_runtime(requested_provider: str, explicit_api_key: Optional[str],
explicit_base_url: str) -> Dict[str, Any]:
"""Bare ``custom`` + explicit base_url (e.g. a ``model_aliases:`` direct alias)."""
rp = _rp()
base_url = explicit_base_url.strip().rstrip("/")
@@ -521,21 +429,13 @@ def _resolve_direct_alias_runtime(
# OLLAMA_API_KEY gets its own gate here: without it a `model_aliases:` entry pointing at
# Ollama Cloud resolved no key at all.
candidates = [(explicit_api_key or "").strip(), *rp._host_gated_env_key_candidates(base_url, ollama=True)]
api_key = next((c for c in candidates if rp.has_usable_secret(c)), "") or "no-key-required"
return rp._runtime(
"custom",
rp._detect_api_mode_for_url(base_url) or "chat_completions",
base_url,
api_key,
source="direct-alias",
requested_provider=requested_provider,
)
api_key = next((c for c in candidates if rp.has_usable_secret(c)), "")
return _custom_runtime(rp, base_url, api_key, None, source="direct-alias", requested_provider=requested_provider)
def _opencode_family_for_custom(requested_provider: str, base_url: str) -> Optional[str]:
"""OpenCode family by provider name, else by opencode.ai host (``/zen/go`` => opencode-go)."""
from hermes_cli.models import opencode_provider_family
family = opencode_provider_family(requested_provider)
if family is not None:
return family
@@ -547,87 +447,63 @@ def _opencode_family_for_custom(requested_provider: str, base_url: str) -> Optio
return None
def _resolve_named_custom_runtime(
*,
requested_provider: str,
explicit_api_key: Optional[str] = None,
explicit_base_url: Optional[str] = None,
target_model: Optional[str] = None,
) -> Optional[Dict[str, Any]]:
def _resolve_named_custom_runtime(*, requested_provider: str, explicit_api_key: Optional[str] = None,
explicit_base_url: Optional[str] = None,
target_model: Optional[str] = None) -> Optional[Dict[str, Any]]:
"""Runtime for a llamacpp alias, a bare-custom direct alias, or a configured custom entry.
Aliases resolving to "custom" (ollama, vllm, llamacpp, …) are treated like bare ``custom``.
A llamacpp alias with no explicit base_url resolves to the managed server first; an explicit
base_url always wins.
"""
Aliases resolving to "custom" (ollama, vllm, llamacpp, …) are treated like bare ``custom``. A
llamacpp alias with no explicit base_url resolves to the managed server first; an explicit
base_url always wins."""
rp = _rp()
requested_norm = (requested_provider or "").strip().lower()
if requested_norm in ("llamacpp", "llama.cpp", "llama-cpp") and not explicit_base_url:
if requested_norm in _LLAMACPP_ALIASES and not explicit_base_url:
return _resolve_llamacpp_runtime(requested_provider, explicit_api_key)
if requested_norm and requested_norm != "custom" and rp._resolves_to_custom(requested_norm):
requested_norm = "custom"
if requested_norm == "custom" and explicit_base_url:
return _resolve_direct_alias_runtime(requested_provider, explicit_api_key, explicit_base_url)
custom_provider = rp._get_named_custom_provider(requested_provider)
if not custom_provider:
return None
base_url = ((explicit_base_url or "").strip() or custom_provider.get("base_url", "")).rstrip("/")
if not base_url:
return None
pool_result = rp._try_resolve_from_custom_pool(
base_url,
"custom",
custom_provider.get("api_mode"),
base_url, "custom", custom_provider.get("api_mode"),
provider_name=custom_provider.get("provider_key") or custom_provider.get("name"),
)
if pool_result:
# The pool doesn't know the custom_providers fields — propagate them here too.
_apply_custom_provider_extras(custom_provider, target_model, pool_result)
return pool_result
explicit_key = (explicit_api_key or "").strip()
candidates = [
(explicit_api_key or "").strip(),
explicit_key,
_clean(custom_provider.get("api_key", "")),
rp._getenv(_clean(custom_provider.get("key_env", "")), "").strip(),
*rp._host_gated_env_key_candidates(base_url, ollama=False),
]
api_key: Any = next((c for c in candidates if rp.has_usable_secret(c)), "")
# ``key_cmd`` credentials are minted per request (short-lived bearers would go stale
# mid-session); both wire clients accept a callable api_key (the Entra ID contract). An
# explicit --api-key still wins as the one-off recovery escape hatch.
key_cmd = _clean(custom_provider.get("key_cmd", ""))
if key_cmd and not rp.has_usable_secret((explicit_api_key or "").strip()):
if key_cmd and not rp.has_usable_secret(explicit_key):
from agent.command_token_source import build_command_token_provider
token_provider = build_command_token_provider(
key_cmd, str(custom_provider.get("name", requested_provider) or "custom")
)
token_provider = build_command_token_provider(key_cmd, str(custom_provider.get("name", requested_provider) or "custom"))
if token_provider is not None:
api_key = token_provider
result = rp._runtime(
"custom",
custom_provider.get("api_mode") or rp._detect_api_mode_for_url(base_url) or "chat_completions",
base_url,
api_key or "no-key-required",
source=f"custom_provider:{custom_provider.get('name', requested_provider)}",
requested_provider=requested_provider,
)
result = _custom_runtime(rp, base_url, api_key, custom_provider.get("api_mode"),
source=f"custom_provider:{custom_provider.get('name', requested_provider)}",
requested_provider=requested_provider)
_apply_custom_provider_extras(custom_provider, target_model, result)
# OpenCode-family custom providers (opencode-go/zen names, or opencode.ai hosts) serve models
# on different API surfaces — a static api_mode 503s for /v1/responses-only models. Re-derive
# api_mode from the model and normalize /v1 like the built-in paths.
family = _opencode_family_for_custom(requested_provider, base_url)
if family is not None and not custom_provider.get("api_mode"):
from hermes_cli.models import normalize_opencode_base_url, opencode_model_api_mode
effective_model = str(
target_model or custom_provider.get("model") or rp._get_model_config().get("default") or ""
).strip()
effective_model = str(target_model or custom_provider.get("model") or rp._get_model_config().get("default") or "").strip()
if effective_model:
result["api_mode"] = opencode_model_api_mode(family, effective_model)
result["base_url"] = normalize_opencode_base_url(family, result["api_mode"], result["base_url"])