2664 lines
119 KiB
Python
2664 lines
119 KiB
Python
"""Shared runtime provider resolution for CLI, gateway, cron, and helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import re
|
|
from urllib.parse import urlparse
|
|
from typing import Any, Dict, Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
from hermes_cli import auth as auth_mod
|
|
from agent.credential_pool import (
|
|
CredentialPool,
|
|
PooledCredential,
|
|
credential_pool_matches_provider,
|
|
custom_provider_pool_key_candidates,
|
|
load_pool,
|
|
)
|
|
from agent.secret_scope import get_secret as _get_secret
|
|
from hermes_cli.auth import (
|
|
ACTUAL_LOCAL_NOAUTH_PLACEHOLDER,
|
|
AuthError,
|
|
DEFAULT_CODEX_BASE_URL,
|
|
DEFAULT_QWEN_BASE_URL,
|
|
DEFAULT_XAI_OAUTH_BASE_URL,
|
|
PROVIDER_REGISTRY,
|
|
_agent_key_is_usable,
|
|
_nous_inference_env_override,
|
|
format_auth_error,
|
|
resolve_provider,
|
|
resolve_nous_runtime_credentials,
|
|
resolve_codex_runtime_credentials,
|
|
resolve_xai_oauth_runtime_credentials,
|
|
resolve_qwen_runtime_credentials,
|
|
resolve_api_key_provider_credentials,
|
|
resolve_external_process_provider_credentials,
|
|
has_usable_secret,
|
|
is_actual_local_base_url,
|
|
normalize_actual_base_url,
|
|
)
|
|
from hermes_cli import config as _config_mod
|
|
from hermes_cli.providers import custom_provider_aliases, custom_provider_slug
|
|
from hermes_constants import OPENROUTER_BASE_URL
|
|
from hermes_cli.providers import is_official_openai_host
|
|
|
|
|
|
def load_config():
|
|
"""Late-bound delegate to :func:`hermes_cli.config.load_config`.
|
|
|
|
Deliberately NOT a module-level ``from hermes_cli.config import
|
|
load_config``: this module is often imported lazily (inside functions),
|
|
so its first import can happen while a test has
|
|
``hermes_cli.config.load_config`` patched — a from-import would then
|
|
bind the MagicMock *permanently*, poisoning every later caller in the
|
|
process (the mock's fixed config shadows the real one long after the
|
|
patch exits). Delegating at call time keeps both patch targets working:
|
|
patching ``hermes_cli.config.load_config`` OR
|
|
``hermes_cli.runtime_provider.load_config`` behaves as expected.
|
|
"""
|
|
return _config_mod.load_config()
|
|
|
|
|
|
def get_compatible_custom_providers(config=None):
|
|
"""Late-bound delegate — see :func:`load_config` for why."""
|
|
return _config_mod.get_compatible_custom_providers(config)
|
|
|
|
|
|
def normalize_extra_headers(value):
|
|
"""Late-bound delegate — see :func:`load_config` for why."""
|
|
return _config_mod.normalize_extra_headers(value)
|
|
from utils import base_url_host_matches, base_url_hostname, env_int
|
|
|
|
|
|
def _getenv(name: str, default: str = "") -> str:
|
|
"""Profile-scoped replacement for ``os.getenv`` on credential/provider reads.
|
|
|
|
Routes through the secret scope (Workstream A): identical to ``os.getenv``
|
|
when multiplexing is off, scope-aware (and fail-closed on an unscoped read)
|
|
when on. Genuinely-global vars are handled inside ``get_secret`` and still
|
|
read ``os.environ``. Keeps the ``(name, default) -> str`` contract every
|
|
call site here already relies on.
|
|
"""
|
|
val = _get_secret(name, default)
|
|
return val if val is not None else default
|
|
|
|
|
|
def _normalize_custom_provider_name(value: str) -> str:
|
|
return value.strip().lower().replace(" ", "-")
|
|
|
|
|
|
def _loopback_hostname(host: str) -> bool:
|
|
h = (host or "").lower().rstrip(".")
|
|
return h in {"localhost", "127.0.0.1", "::1", "0.0.0.0"}
|
|
|
|
|
|
def _config_base_url_trustworthy_for_bare_custom(cfg_base_url: str, cfg_provider: str) -> bool:
|
|
"""Decide whether ``model.base_url`` may back bare ``custom`` runtime resolution.
|
|
|
|
GitHub #14676: the model picker can select Custom while ``model.provider`` still reflects a
|
|
previous provider. Reject non-loopback URLs unless the YAML provider is already ``custom``
|
|
(or one of the local-server aliases that resolve to ``custom`` — ollama, vllm, llamacpp, …),
|
|
so a stale OpenRouter/Z.ai base_url cannot hijack local ``custom`` sessions.
|
|
"""
|
|
cfg_provider_norm = (cfg_provider or "").strip().lower()
|
|
bu = (cfg_base_url or "").strip()
|
|
if not bu:
|
|
return False
|
|
if cfg_provider_norm == "custom":
|
|
return True
|
|
# GitHub #27132: provider aliases that resolve to "custom" at runtime
|
|
# (ollama, vllm, llamacpp, …) should be trusted the same way "custom"
|
|
# is, otherwise a legit LAN/WireGuard ollama endpoint silently falls
|
|
# through to OpenRouter.
|
|
try:
|
|
from hermes_cli.auth import resolve_provider as _resolve_provider
|
|
|
|
if _resolve_provider(cfg_provider_norm) == "custom":
|
|
return True
|
|
except Exception:
|
|
pass
|
|
if base_url_host_matches(bu, "openrouter.ai"):
|
|
return False
|
|
return _loopback_hostname(base_url_hostname(bu))
|
|
|
|
|
|
def _detect_api_mode_for_url(base_url: str) -> Optional[str]:
|
|
"""Auto-detect api_mode from the resolved base URL.
|
|
|
|
- Direct api.openai.com endpoints need the Responses API for GPT-5.x
|
|
tool calls with reasoning (chat/completions returns 400).
|
|
- Direct api.anthropic.com endpoints must use the native Messages
|
|
API (``/v1/messages``). Anthropic also exposes an OpenAI-compat
|
|
``/chat/completions`` shim on the same host, but Pro/Max OAuth
|
|
subscriptions are only billed against the native Messages route;
|
|
hitting the shim accounts against a separate "extra usage" pool
|
|
that is empty by default and surfaces as HTTP 400 "You're out of
|
|
extra usage." See issue #32243.
|
|
- Third-party Anthropic-compatible gateways (MiniMax, Zhipu GLM,
|
|
LiteLLM proxies, etc.) conventionally expose the native Anthropic
|
|
protocol under a ``/anthropic`` suffix — treat those as
|
|
``anthropic_messages`` transport instead of the default
|
|
``chat_completions``.
|
|
- Kimi Code's ``api.kimi.com/coding`` endpoint also speaks the
|
|
Anthropic Messages protocol (the /coding route accepts Claude
|
|
Code's native request shape).
|
|
"""
|
|
normalized = (base_url or "").strip().lower().rstrip("/")
|
|
hostname = base_url_hostname(base_url)
|
|
if hostname == "api.x.ai":
|
|
return "codex_responses"
|
|
# Official OpenAI host family: canonical api.openai.com plus the
|
|
# data-residency regional hosts (us./eu.api.openai.com). Same API
|
|
# surface, same Responses-API mandate. Shared predicate — see
|
|
# providers.is_official_openai_host for the spoof-rejection contract.
|
|
if is_official_openai_host(base_url):
|
|
return "codex_responses"
|
|
# Meta Model API: prompt caching only on Responses API (0% on
|
|
# chat/completions vs 93-99% on /responses with retention). Exact
|
|
# hostname per #32243.
|
|
if hostname == "api.meta.ai":
|
|
return "codex_responses"
|
|
if hostname == "api.actual.inc":
|
|
return "codex_responses"
|
|
# Ramp Router: Responses-native host — /v1/chat/completions is only a
|
|
# minimal compatibility shim, while reasoning and caching support live
|
|
# on /v1/responses (docs.router.com/api/endpoint). Mirrors the
|
|
# host_mandated_api_mode clause in hermes_cli/providers.py so the
|
|
# runtime resolver stays in lockstep. Exact hostname per #32243.
|
|
if hostname == "api.router.com":
|
|
return "codex_responses"
|
|
# Direct native Anthropic host: realign with providers.determine_api_mode,
|
|
# which already maps this host to anthropic_messages. The exact-hostname
|
|
# match rejects lookalike subdomains (api.anthropic.com.attacker.test) and
|
|
# path-segment spoofing (proxy.test/api.anthropic.com/v1). (#32243)
|
|
if hostname == "api.anthropic.com":
|
|
return "anthropic_messages"
|
|
path = urlparse(normalized).path.rstrip("/")
|
|
if path.endswith("/anthropic") or path.endswith("/anthropic/v1"):
|
|
return "anthropic_messages"
|
|
if hostname == "api.kimi.com" and "/coding" in normalized:
|
|
return "anthropic_messages"
|
|
return None
|
|
|
|
|
|
def _fallback_api_mode(provider: str, base_url: str, model: str = "") -> str:
|
|
"""Resolve api_mode when no explicit/persisted mode applies.
|
|
|
|
Precedence: URL detection (host-mandated wire shapes) first, then the
|
|
transport the provider overlay itself declares via
|
|
``providers.determine_api_mode`` — which already handles host mandates,
|
|
dual-wire providers, and the registry transport map — and only then the
|
|
``chat_completions`` default for genuinely unknown providers/endpoints.
|
|
|
|
Before this helper the runtime paths consulted URL detection ONLY and
|
|
silently landed reasoning providers on ``chat_completions`` whenever the
|
|
hostname wasn't literally recognized. That is how ``openai-api`` pointed
|
|
at OpenAI's data-residency hosts (``us.api.openai.com``) 400'd on every
|
|
tool-calling turn: the provider declares ``codex_responses`` but the
|
|
declaration was never consulted. Same latent class covered the other
|
|
non-chat overlays (MiniMax family, copilot-acp).
|
|
"""
|
|
detected = _detect_api_mode_for_url(base_url)
|
|
if detected:
|
|
return detected
|
|
from hermes_cli.providers import determine_api_mode
|
|
|
|
return determine_api_mode(provider, base_url, model) or "chat_completions"
|
|
|
|
|
|
def _resolve_plain_custom_api_mode(model_cfg: Dict[str, Any], base_url: str) -> str:
|
|
"""Resolve api_mode for legacy/plain ``provider: custom`` endpoints.
|
|
|
|
Custom endpoints should stay conservative by default. Only direct OpenAI/xAI
|
|
URLs imply Responses API automatically; named custom providers can opt in via
|
|
their own ``api_mode`` field. This also prevents a stale persisted
|
|
``model.api_mode: codex_responses`` from forcing generic relays onto the
|
|
Responses path after upgrades or /reset.
|
|
"""
|
|
configured_mode = _parse_api_mode(model_cfg.get("api_mode"))
|
|
# Note: api.meta.ai is handled by _detect_api_mode_for_url (returns codex_responses), so the suppression guard below does not fire for Meta.
|
|
detected_mode = _detect_api_mode_for_url(base_url)
|
|
|
|
if configured_mode == "codex_responses" and detected_mode != "codex_responses":
|
|
logger.info(
|
|
"Ignoring persisted custom api_mode=codex_responses for non-OpenAI endpoint %s",
|
|
base_url or "(unknown)",
|
|
)
|
|
configured_mode = None
|
|
|
|
return configured_mode or detected_mode or "chat_completions"
|
|
|
|
|
|
def _host_derived_api_key(base_url: str) -> str:
|
|
"""Look up `<VENDOR>_API_KEY` in the env, derived from the base URL host.
|
|
|
|
Examples:
|
|
https://api.deepseek.com/v1 → DEEPSEEK_API_KEY
|
|
https://api.groq.com/openai/v1 → GROQ_API_KEY
|
|
https://api.mistral.ai/v1 → MISTRAL_API_KEY
|
|
https://generativelanguage.googleapis.com/v1beta/openai/ → GOOGLEAPIS_API_KEY
|
|
|
|
Returns the env value (stripped) or "". Never returns env vars whose names
|
|
are already explicitly checked elsewhere — those are handled by their own
|
|
host-gated paths (OPENAI/OPENROUTER/OLLAMA).
|
|
|
|
The vendor label is the *registrable* portion of the hostname: strip
|
|
``api.`` / ``www.`` prefixes, then take the second-to-last label
|
|
(``api.deepseek.com`` → ``deepseek``). Falls back to "" for hostnames
|
|
that don't yield a usable vendor label (IPs, loopback, single-label
|
|
hosts).
|
|
"""
|
|
hostname = base_url_hostname(base_url)
|
|
if not hostname:
|
|
return ""
|
|
# Reject IPv4 / IPv6 / loopback — no meaningful vendor label.
|
|
if any(ch.isdigit() for ch in hostname.split(".")[-1]):
|
|
# Last label starts with a digit → likely IP. (TLDs are never numeric.)
|
|
return ""
|
|
if hostname in ("localhost",) or ":" in hostname:
|
|
return ""
|
|
labels = [lbl for lbl in hostname.split(".") if lbl]
|
|
# Strip common API/CDN prefixes.
|
|
while labels and labels[0] in ("api", "www"):
|
|
labels.pop(0)
|
|
if len(labels) < 2:
|
|
return ""
|
|
# Take the *registrable* label (second-to-last). For typical provider
|
|
# hosts this is what users intuitively call "the vendor":
|
|
# deepseek.com → labels[-2] = "deepseek" ✓
|
|
# api.groq.com → groq.com → labels[-2] = "groq" ✓
|
|
# api.mistral.ai → labels[-2] = "mistral" ✓
|
|
# Crucially, lookalike hosts pick the ATTACKER's label, not the spoofed
|
|
# vendor:
|
|
# api.deepseek.com.attacker.test → labels[-2] = "attacker"
|
|
# so DEEPSEEK_API_KEY stays put and the chain falls through to
|
|
# no-key-required. This mirrors how `base_url_host_matches` resists the
|
|
# same lookalike attack for explicit hosts.
|
|
vendor = labels[-2]
|
|
# Sanitize to env var charset: A-Z, 0-9, underscore.
|
|
sanitized = "".join(ch if ch.isalnum() else "_" for ch in vendor).upper()
|
|
if not sanitized or not sanitized[0].isalpha():
|
|
return ""
|
|
# Don't re-derive env vars already handled by explicit host-gated paths.
|
|
if sanitized in ("OPENAI", "OPENROUTER", "OLLAMA"):
|
|
return ""
|
|
env_name = f"{sanitized}_API_KEY"
|
|
return (_getenv(env_name, "") or "").strip()
|
|
|
|
|
|
def _anthropic_base_url_override_ok(base_url: str) -> bool:
|
|
"""Decide whether a configured ``model.base_url`` may back native Anthropic.
|
|
|
|
Native ``provider: anthropic`` resolution honors ``model.base_url`` so users
|
|
can point at Anthropic-compatible endpoints (official Anthropic/Claude hosts,
|
|
Azure Foundry, MiniMax/Zhipu/LiteLLM-style ``/anthropic`` proxies, Kimi's
|
|
``/coding`` route). But a config can carry a *stale* non-Anthropic URL — e.g.
|
|
``provider: anthropic`` left with ``base_url: https://openrouter.ai/api/v1``
|
|
after a provider switch — which would route Anthropic OAuth/setup-token
|
|
traffic to an OpenAI-compatible aggregator and 404. Ignore those.
|
|
|
|
Returns True only when the URL plausibly speaks the Anthropic Messages
|
|
protocol; otherwise the caller falls back to ``https://api.anthropic.com``.
|
|
"""
|
|
candidate = (base_url or "").strip()
|
|
if not candidate:
|
|
return False
|
|
|
|
hostname = (base_url_hostname(candidate) or "").lower()
|
|
if not hostname:
|
|
return False
|
|
|
|
# Official Anthropic / Claude hosts.
|
|
if hostname == "api.anthropic.com" or hostname.endswith(".anthropic.com") or hostname.endswith(".claude.com"):
|
|
return True
|
|
# Azure Foundry Anthropic endpoints (handled specially downstream).
|
|
if hostname.endswith(".azure.com"):
|
|
return True
|
|
# Anthropic-compatible proxies conventionally expose the native Messages
|
|
# protocol under a ``/anthropic`` suffix, and Kimi under ``/coding`` — same
|
|
# signal _detect_api_mode_for_url() uses to pick anthropic_messages.
|
|
if _detect_api_mode_for_url(candidate) == "anthropic_messages":
|
|
return True
|
|
# Bare api.kimi.com without the /coding path is not an Anthropic endpoint.
|
|
return False
|
|
|
|
|
|
def _auto_detect_local_model(base_url: str) -> str:
|
|
"""Query a local server for its model name when only one model is loaded."""
|
|
if not base_url:
|
|
return ""
|
|
try:
|
|
import requests
|
|
url = base_url.rstrip("/")
|
|
if not url.endswith("/v1"):
|
|
url += "/v1"
|
|
resp = requests.get(url + "/models", timeout=(2, 3))
|
|
if resp.ok:
|
|
models = resp.json().get("data", [])
|
|
if len(models) == 1:
|
|
model_id = models[0].get("id", "")
|
|
if model_id:
|
|
return model_id
|
|
except Exception as exc:
|
|
# Log instead of silently swallowing — aids debugging when
|
|
# local model auto-detection fails unexpectedly.
|
|
logger.debug("Auto-detect model from %s failed: %s", base_url, exc)
|
|
return ""
|
|
|
|
|
|
def _get_model_config() -> Dict[str, Any]:
|
|
config = load_config()
|
|
model_cfg = config.get("model")
|
|
if isinstance(model_cfg, dict):
|
|
cfg = dict(model_cfg)
|
|
# Accept "model" as alias for "default" (users intuitively write model.model)
|
|
if not cfg.get("default") and cfg.get("model"):
|
|
cfg["default"] = cfg["model"]
|
|
# Handle model.default being a dict {provider: ..., model: ...} rather than a string
|
|
_default = cfg.get("default")
|
|
if isinstance(_default, dict):
|
|
from hermes_cli.config import split_model_config_default
|
|
cfg_model, cfg_provider = split_model_config_default(_default)
|
|
cfg_provider = cfg_provider or str(model_cfg.get("provider") or "")
|
|
cfg["default"] = cfg_model
|
|
if cfg_provider and not cfg.get("provider"):
|
|
cfg["provider"] = cfg_provider
|
|
_default = cfg_model
|
|
default = (str(_default or "")).strip()
|
|
base_url = (cfg.get("base_url") or "").strip()
|
|
is_local = base_url_hostname(base_url) in ("localhost", "127.0.0.1")
|
|
is_fallback = not default
|
|
if is_local and is_fallback and base_url:
|
|
detected = _auto_detect_local_model(base_url)
|
|
if detected:
|
|
cfg["default"] = detected
|
|
return cfg
|
|
if isinstance(model_cfg, str) and model_cfg.strip():
|
|
return {"default": model_cfg.strip()}
|
|
return {}
|
|
|
|
|
|
def _provider_supports_explicit_api_mode(provider: Optional[str], configured_provider: Optional[str] = None) -> bool:
|
|
"""Check whether a persisted api_mode should be honored for a given provider.
|
|
|
|
Prevents stale api_mode from a previous provider leaking into a
|
|
different one after a model/provider switch. Only applies the
|
|
persisted mode when the config's provider matches the runtime
|
|
provider (or when no configured provider is recorded).
|
|
"""
|
|
normalized_provider = (provider or "").strip().lower()
|
|
normalized_configured = (configured_provider or "").strip().lower()
|
|
if not normalized_configured:
|
|
return True
|
|
if normalized_provider == "custom":
|
|
return normalized_configured == "custom" or normalized_configured.startswith("custom:")
|
|
return normalized_configured == normalized_provider
|
|
|
|
|
|
def _copilot_runtime_api_mode(
|
|
model_cfg: Dict[str, Any],
|
|
api_key: str,
|
|
*,
|
|
target_model: Optional[str] = None,
|
|
) -> str:
|
|
configured_provider = str(model_cfg.get("provider") or "").strip().lower()
|
|
configured_mode = _parse_api_mode(model_cfg.get("api_mode"))
|
|
if configured_mode and _provider_supports_explicit_api_mode("copilot", configured_provider):
|
|
return configured_mode
|
|
|
|
# Use the model being resolved for this runtime, not the persisted global
|
|
# default. MoA slots, fallback models, and mid-session model switches all
|
|
# resolve credentials for a target model that can differ from config.yaml's
|
|
# model.default. If we derive Copilot api_mode from the stale default, a
|
|
# Claude/Gemini MoA slot can inherit codex_responses from a GPT-5 default and
|
|
# fail with "model ... does not support Responses API".
|
|
model_name = str(target_model or model_cfg.get("default") or "").strip()
|
|
if not model_name:
|
|
return "chat_completions"
|
|
|
|
try:
|
|
from hermes_cli.models import copilot_model_api_mode
|
|
|
|
return copilot_model_api_mode(model_name, api_key=api_key)
|
|
except Exception:
|
|
return "chat_completions"
|
|
|
|
|
|
_VALID_API_MODES = {
|
|
"chat_completions",
|
|
"codex_responses",
|
|
"anthropic_messages",
|
|
"bedrock_converse",
|
|
# Optional opt-in: hand the entire turn to a `codex app-server` subprocess
|
|
# so terminal/file-ops/patching/sandboxing run inside Codex's own runtime
|
|
# instead of Hermes' tool dispatch. Gated behind config key
|
|
# `model.openai_runtime == "codex_app_server"` AND provider in
|
|
# {"openai", "openai-codex"}. Default is unchanged.
|
|
"codex_app_server",
|
|
}
|
|
|
|
|
|
def _parse_api_mode(raw: Any) -> Optional[str]:
|
|
"""Validate an api_mode value from config. Returns None if invalid.
|
|
|
|
Legacy/alias spellings (``openai``, ``anthropic``, ``responses``, …) are
|
|
canonicalized via the shared alias map before validation, so configs
|
|
written against older releases keep selecting the transport they named
|
|
instead of silently falling through to hostname-based detection.
|
|
"""
|
|
if isinstance(raw, str):
|
|
from hermes_cli.config import _canonical_api_mode
|
|
|
|
normalized = _canonical_api_mode(raw).lower()
|
|
if normalized in _VALID_API_MODES:
|
|
return normalized
|
|
return None
|
|
|
|
|
|
def _nous_inference_base_url_override() -> str:
|
|
"""Return the trusted Nous runtime base URL override, if configured.
|
|
|
|
Delegates to ``auth._nous_inference_env_override`` so every
|
|
``NOUS_INFERENCE_BASE_URL`` read shares one normalization path
|
|
(trailing-slash stripping, blank → empty). The env source is trusted
|
|
and intentionally bypasses the network host allowlist there.
|
|
"""
|
|
return _nous_inference_env_override() or ""
|
|
|
|
|
|
def _maybe_apply_codex_app_server_runtime(
|
|
*,
|
|
provider: str,
|
|
api_mode: str,
|
|
model_cfg: Optional[Dict[str, Any]],
|
|
) -> str:
|
|
"""Optional opt-in: rewrite api_mode → "codex_app_server" for OpenAI/Codex
|
|
providers when the user has explicitly enabled that runtime via
|
|
`model.openai_runtime: codex_app_server` in config.yaml.
|
|
|
|
Default behavior is preserved: when the key is unset, "auto", or empty,
|
|
this function is a no-op. Only providers in {"openai", "openai-codex"}
|
|
are eligible — other providers (anthropic, openrouter, etc.) cannot be
|
|
rerouted through codex.
|
|
|
|
Returns the (possibly-rewritten) api_mode."""
|
|
if not model_cfg:
|
|
return api_mode
|
|
if provider not in {"openai", "openai-codex"}:
|
|
return api_mode
|
|
runtime = str(model_cfg.get("openai_runtime") or "").strip().lower()
|
|
if runtime == "codex_app_server":
|
|
return "codex_app_server"
|
|
return api_mode
|
|
|
|
|
|
def _resolve_runtime_from_pool_entry(
|
|
*,
|
|
provider: str,
|
|
entry: PooledCredential,
|
|
requested_provider: str,
|
|
model_cfg: Optional[Dict[str, Any]] = None,
|
|
pool: Optional[CredentialPool] = None,
|
|
target_model: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
model_cfg = model_cfg or _get_model_config()
|
|
# When the caller is resolving for a specific target model (e.g. a /model
|
|
# mid-session switch), prefer that over the persisted model.default. This
|
|
# prevents api_mode being computed from a stale config default that no
|
|
# longer matches the model actually being used — the bug that caused
|
|
# opencode-zen /v1 to be stripped for chat_completions requests when
|
|
# config.default was still a Claude model.
|
|
effective_model = (target_model or model_cfg.get("default") or "")
|
|
base_url = (getattr(entry, "runtime_base_url", None) or getattr(entry, "base_url", None) or "").rstrip("/")
|
|
api_key = getattr(entry, "runtime_api_key", None) or getattr(entry, "access_token", "")
|
|
api_mode = "chat_completions"
|
|
if provider == "openai-codex":
|
|
api_mode = "codex_responses"
|
|
base_url = base_url or DEFAULT_CODEX_BASE_URL
|
|
elif provider == "xai-oauth":
|
|
api_mode = "codex_responses"
|
|
base_url = base_url or DEFAULT_XAI_OAUTH_BASE_URL
|
|
elif provider == "qwen-oauth":
|
|
api_mode = "chat_completions"
|
|
base_url = base_url or DEFAULT_QWEN_BASE_URL
|
|
elif provider == "minimax-oauth":
|
|
# MiniMax OAuth tokens are valid only against the Anthropic Messages
|
|
# compatible endpoint. Do not honor stale model.api_mode values from a
|
|
# prior OpenAI-compatible provider, or the client will hit
|
|
# /chat/completions under /anthropic and receive a bare nginx 404.
|
|
api_mode = "anthropic_messages"
|
|
pconfig = PROVIDER_REGISTRY.get(provider)
|
|
base_url = base_url or (pconfig.inference_base_url if pconfig else "")
|
|
elif provider == "anthropic":
|
|
api_mode = "anthropic_messages"
|
|
cfg_provider = str(model_cfg.get("provider") or "").strip().lower()
|
|
cfg_base_url = ""
|
|
if cfg_provider == "anthropic":
|
|
cfg_base_url = str(model_cfg.get("base_url") or "").strip().rstrip("/")
|
|
if not _anthropic_base_url_override_ok(cfg_base_url):
|
|
cfg_base_url = ""
|
|
base_url = cfg_base_url or base_url or "https://api.anthropic.com"
|
|
elif provider == "openrouter":
|
|
base_url = base_url or OPENROUTER_BASE_URL
|
|
elif provider == "xai":
|
|
api_mode = "codex_responses"
|
|
elif provider == "nous":
|
|
from hermes_cli.providers import nous_api_mode
|
|
|
|
api_mode = nous_api_mode(effective_model)
|
|
base_url = _nous_inference_base_url_override() or base_url
|
|
elif provider == "copilot":
|
|
api_mode = _copilot_runtime_api_mode(
|
|
model_cfg,
|
|
getattr(entry, "runtime_api_key", ""),
|
|
target_model=effective_model,
|
|
)
|
|
base_url = base_url or PROVIDER_REGISTRY["copilot"].inference_base_url
|
|
elif provider == "azure-foundry":
|
|
# Azure Foundry: read api_mode and base_url from config
|
|
cfg_provider = str(model_cfg.get("provider") or "").strip().lower()
|
|
if cfg_provider == "azure-foundry":
|
|
cfg_base_url = str(model_cfg.get("base_url") or "").strip().rstrip("/")
|
|
if cfg_base_url:
|
|
base_url = cfg_base_url
|
|
configured_mode = _parse_api_mode(model_cfg.get("api_mode"))
|
|
if configured_mode:
|
|
api_mode = configured_mode
|
|
# Model-family inference for GPT-5.x / codex / o1-o4: Azure rejects
|
|
# /chat/completions on these with 400 "operation unsupported" — see
|
|
# azure_foundry_model_api_mode() for rationale. Skip when the user
|
|
# explicitly picked anthropic_messages (Anthropic-style endpoint).
|
|
if effective_model and api_mode != "anthropic_messages":
|
|
try:
|
|
from hermes_cli.models import azure_foundry_model_api_mode
|
|
|
|
inferred = azure_foundry_model_api_mode(effective_model)
|
|
except Exception:
|
|
inferred = None
|
|
if inferred:
|
|
api_mode = inferred
|
|
# For Anthropic-style endpoints, strip /v1 suffix
|
|
if api_mode == "anthropic_messages":
|
|
base_url = re.sub(r"/v1/?$", "", base_url)
|
|
else:
|
|
configured_provider = str(model_cfg.get("provider") or "").strip().lower()
|
|
# Honour model.base_url from config.yaml when the configured provider
|
|
# matches this provider — same pattern as the Anthropic branch above.
|
|
# Only override when the pool entry has no explicit base_url (i.e. it
|
|
# fell back to the hardcoded default). Env var overrides win (#6039).
|
|
pconfig = PROVIDER_REGISTRY.get(provider)
|
|
pool_url_is_default = pconfig and base_url.rstrip("/") == pconfig.inference_base_url.rstrip("/")
|
|
if configured_provider == provider and pool_url_is_default:
|
|
cfg_base_url = str(model_cfg.get("base_url") or "").strip().rstrip("/")
|
|
if cfg_base_url:
|
|
base_url = cfg_base_url
|
|
configured_mode = _parse_api_mode(model_cfg.get("api_mode"))
|
|
from hermes_cli.models import opencode_provider_family
|
|
if opencode_provider_family(provider) is not None:
|
|
# Re-derive api_mode from the effective model rather than the
|
|
# persisted api_mode: the opencode providers serve both
|
|
# anthropic_messages and chat_completions models, so the previous
|
|
# session's mode must not leak across /model switches.
|
|
# Refs #16878.
|
|
from hermes_cli.models import opencode_model_api_mode
|
|
api_mode = opencode_model_api_mode(provider, effective_model)
|
|
elif configured_mode and _provider_supports_explicit_api_mode(provider, configured_provider):
|
|
api_mode = configured_mode
|
|
else:
|
|
# URL detection first (Anthropic /anthropic suffix, Kimi /coding,
|
|
# official OpenAI hosts → codex_responses, api.x.ai →
|
|
# codex_responses), then the provider's own declared transport.
|
|
api_mode = _fallback_api_mode(provider, base_url, effective_model)
|
|
|
|
# OpenCode base URLs end with /v1 for OpenAI-compatible models, but the
|
|
# Anthropic SDK prepends its own /v1/messages to the base_url. Normalize
|
|
# symmetrically: strip /v1 for anthropic_messages, re-append it for
|
|
# chat_completions / codex_responses (heals a stripped URL persisted to
|
|
# model.base_url by an earlier switch into an anthropic-routed model).
|
|
from hermes_cli.models import opencode_provider_family
|
|
if opencode_provider_family(provider) is not None:
|
|
from hermes_cli.models import normalize_opencode_base_url
|
|
|
|
base_url = normalize_opencode_base_url(provider, api_mode, base_url)
|
|
|
|
# Optional opt-in: route OpenAI/Codex turns through `codex app-server`.
|
|
# Inert when `model.openai_runtime` is unset or "auto".
|
|
api_mode = _maybe_apply_codex_app_server_runtime(
|
|
provider=provider, api_mode=api_mode, model_cfg=model_cfg
|
|
)
|
|
|
|
if provider == "lmstudio":
|
|
base_url = auth_mod._normalize_lmstudio_runtime_base_url(base_url)
|
|
|
|
return {
|
|
"provider": provider,
|
|
"api_mode": api_mode,
|
|
"base_url": base_url,
|
|
"api_key": api_key,
|
|
"source": getattr(entry, "source", "pool"),
|
|
"credential_pool": pool,
|
|
"requested_provider": requested_provider,
|
|
}
|
|
|
|
|
|
def resolve_requested_provider(requested: Optional[str] = None) -> str:
|
|
"""Resolve provider request from explicit arg, config, then env."""
|
|
if requested and requested.strip():
|
|
return requested.strip().lower()
|
|
|
|
model_cfg = _get_model_config()
|
|
cfg_provider = model_cfg.get("provider")
|
|
if isinstance(cfg_provider, str) and cfg_provider.strip():
|
|
return cfg_provider.strip().lower()
|
|
|
|
# Prefer the persisted config selection over any stale shell/.env
|
|
# provider override so chat uses the endpoint the user last saved.
|
|
env_provider = _getenv("HERMES_INFERENCE_PROVIDER", "").strip().lower()
|
|
if env_provider:
|
|
return env_provider
|
|
|
|
return "auto"
|
|
|
|
|
|
def _try_resolve_from_custom_pool(
|
|
base_url: str,
|
|
provider_label: str,
|
|
api_mode_override: Optional[str] = None,
|
|
provider_name: Optional[str] = None,
|
|
) -> Optional[Dict[str, Any]]:
|
|
"""Check if a credential pool exists for a custom endpoint and return a runtime dict if so."""
|
|
candidates: list[str] = []
|
|
seen = set()
|
|
|
|
def _add(key: Optional[str]) -> None:
|
|
normalized = str(key or "").strip().lower()
|
|
if normalized and normalized not in seen:
|
|
seen.add(normalized)
|
|
candidates.append(normalized)
|
|
|
|
try:
|
|
for key in custom_provider_pool_key_candidates(base_url, provider_name):
|
|
_add(key)
|
|
except Exception:
|
|
pass
|
|
if not candidates:
|
|
return None
|
|
|
|
for pool_key in candidates:
|
|
try:
|
|
pool = load_pool(pool_key)
|
|
if not pool.has_credentials():
|
|
continue
|
|
entry = pool.select()
|
|
if entry is None:
|
|
continue
|
|
pool_api_key = getattr(entry, "runtime_api_key", None) or getattr(entry, "access_token", "")
|
|
if not pool_api_key:
|
|
continue
|
|
if not has_usable_secret(pool_api_key) and _loopback_hostname(base_url_hostname(base_url)):
|
|
# Legacy configs commonly used short/placeholder keys ('123',
|
|
# 'm', ...) for local no-auth services like Ollama -- fine for
|
|
# the endpoint itself, but has_usable_secret's 4-char floor
|
|
# (added after these configs were written) now rejects them
|
|
# here with no migration path. Every OTHER resolution path in
|
|
# this file already substitutes "no-key-required" for a
|
|
# loopback endpoint with no usable secret (the config-based
|
|
# custom_providers fallback a few hundred lines below, and the
|
|
# "actual" provider's local-offline exemption further down) --
|
|
# this pool path was the one gap (issue #86864).
|
|
pool_api_key = "no-key-required"
|
|
return {
|
|
"provider": provider_label,
|
|
"api_mode": api_mode_override or _detect_api_mode_for_url(base_url) or "chat_completions",
|
|
"base_url": base_url,
|
|
"api_key": pool_api_key,
|
|
"source": f"pool:{pool_key}",
|
|
"credential_pool": pool,
|
|
}
|
|
except Exception:
|
|
continue
|
|
return None
|
|
|
|
|
|
def _filter_capabilities(value: Any) -> Dict[str, bool]:
|
|
"""Return the string-keyed boolean capabilities accepted at runtime."""
|
|
if not isinstance(value, dict):
|
|
return {}
|
|
return {
|
|
key: enabled
|
|
for key, enabled in value.items()
|
|
if isinstance(key, str) and isinstance(enabled, bool)
|
|
}
|
|
|
|
|
|
def _lift_model_capabilities(
|
|
entry: Dict[str, Any], model: Optional[str], result: Dict[str, Any]
|
|
) -> None:
|
|
"""Copy explicit boolean per-model capabilities into the runtime."""
|
|
capabilities = _filter_capabilities(entry.get("capabilities"))
|
|
models = entry.get("models")
|
|
model_config = models.get(model) if isinstance(models, dict) and model else None
|
|
if isinstance(model_config, dict):
|
|
capabilities.update(_filter_capabilities(model_config))
|
|
if capabilities:
|
|
result["capabilities"] = capabilities
|
|
|
|
|
|
def _lift_max_output_tokens(entry: Dict[str, Any], result: Dict[str, Any]) -> None:
|
|
"""Propagate a per-provider output cap onto the resolved runtime dict.
|
|
|
|
Accepts ``max_output_tokens`` or ``max_tokens`` on a ``custom_providers``
|
|
entry so a provider block can pin its own output limit. Gateway and CLI
|
|
map this onto ``AIAgent.max_tokens`` only when the top-level
|
|
``model.max_tokens`` isn't set, so the documented global key still wins.
|
|
"""
|
|
for _k in ("max_output_tokens", "max_tokens"):
|
|
_v = entry.get(_k)
|
|
if isinstance(_v, int) and _v > 0:
|
|
result["max_output_tokens"] = _v
|
|
return
|
|
|
|
|
|
def _lift_extra_headers(entry: Dict[str, Any], result: Dict[str, Any]) -> None:
|
|
"""Copy a validated ``extra_headers`` dict from a provider entry.
|
|
|
|
SECURITY: header values routinely carry credentials (Cloudflare Access
|
|
service tokens, proxy auth, custom bearer schemes). Never log them.
|
|
"""
|
|
extra_headers = normalize_extra_headers(entry.get("extra_headers"))
|
|
if extra_headers:
|
|
result["extra_headers"] = extra_headers
|
|
|
|
|
|
def _get_named_custom_provider(requested_provider: str) -> Optional[Dict[str, Any]]:
|
|
requested_norm = _normalize_custom_provider_name(requested_provider or "")
|
|
if not requested_norm:
|
|
return None
|
|
|
|
# Bare "custom" is normally an incomplete spec — the canonical form is
|
|
# "custom:<name>" — and is otherwise owned by the model.base_url "bare
|
|
# custom" trust path. BUT a user may literally name a ``providers:`` (or
|
|
# legacy ``custom_providers:``) entry "custom" (e.g. ``providers.custom``
|
|
# pointing at cliproxy). We used to return None here *before* scanning
|
|
# config, so such an entry was never matched and resolution fell through to
|
|
# the global default (Codex) — the cause of cron jobs with
|
|
# ``provider: "custom"`` failing with ``auth_unavailable: providers=codex``.
|
|
# Fall through to the config scan instead; if no entry is literally named
|
|
# "custom" it still returns None at the end, preserving the trust path.
|
|
|
|
# Raw names should only map to custom providers when they are not already
|
|
# valid built-in providers or aliases. Explicit menu keys like
|
|
# ``custom:local`` always target the saved custom provider. Bare "custom"
|
|
# is exempt from the shadow check — it is not a built-in to defer to.
|
|
if requested_norm == "auto":
|
|
return None
|
|
if requested_norm != "custom" and not requested_norm.startswith("custom:"):
|
|
try:
|
|
canonical = auth_mod.resolve_provider(requested_norm)
|
|
except AuthError:
|
|
pass
|
|
else:
|
|
# A user-declared ``custom_providers`` entry whose name matches
|
|
# only an *alias* (``kimi`` → built-in ``kimi-coding``) is the
|
|
# user's intended target — alias rewriting would otherwise hijack
|
|
# the request. We only defer to the built-in when the raw name is
|
|
# the canonical provider itself (``nous``, ``openrouter``, …) so
|
|
# accidentally shadowing a canonical provider still resolves to
|
|
# the built-in. See tests/hermes_cli/test_runtime_provider_resolution.py
|
|
# ``test_named_custom_provider_does_not_shadow_builtin_provider``.
|
|
if (canonical or "").strip().lower() == requested_norm:
|
|
return None
|
|
|
|
config = load_config()
|
|
|
|
# First check providers: dict (new-style user-defined providers)
|
|
providers = config.get("providers")
|
|
if isinstance(providers, dict):
|
|
from hermes_cli.config import is_provider_enabled
|
|
for ep_name, entry in providers.items():
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
# Skip providers the user explicitly disabled via
|
|
# ``providers.<name>.enabled: false``. They remain in config
|
|
# so re-enabling is a one-line edit, but the resolver pretends
|
|
# they're not configured.
|
|
if not is_provider_enabled(entry):
|
|
continue
|
|
# Resolve the API key from the env var name stored in key_env
|
|
key_env = str(
|
|
entry.get("key_env") or entry.get("api_key_env") or ""
|
|
).strip()
|
|
resolved_api_key = _getenv(key_env, "").strip() if key_env else ""
|
|
# Fall back to inline api_key when key_env is absent or unresolvable
|
|
if not resolved_api_key:
|
|
resolved_api_key = str(entry.get("api_key", "") or "").strip()
|
|
|
|
display_name = entry.get("name", "")
|
|
if requested_norm in custom_provider_aliases(
|
|
str(display_name or ep_name),
|
|
str(ep_name),
|
|
):
|
|
# Found match by provider key
|
|
base_url = entry.get("api") or entry.get("url") or entry.get("base_url") or ""
|
|
if base_url:
|
|
result: Dict[str, Any] = {
|
|
"name": entry.get("name", ep_name),
|
|
"base_url": base_url.strip(),
|
|
"api_key": resolved_api_key,
|
|
"model": entry.get("default_model", ""),
|
|
}
|
|
provider_key = str(ep_name or "").strip()
|
|
if provider_key:
|
|
result["provider_key"] = provider_key
|
|
if key_env:
|
|
result["key_env"] = key_env
|
|
extra_body = entry.get("extra_body")
|
|
if isinstance(extra_body, dict):
|
|
result["extra_body"] = dict(extra_body)
|
|
_lift_extra_headers(entry, result)
|
|
# Command that PRINTS a credential, for gateways issuing
|
|
# short-lived bearers instead of static keys. Propagated
|
|
# raw; wrapped in a per-request token provider at
|
|
# resolution.
|
|
key_cmd = str(entry.get("key_cmd", "") or "").strip()
|
|
if key_cmd:
|
|
result["key_cmd"] = key_cmd
|
|
# The v11→v12 migration writes the API mode under the new
|
|
# ``transport`` field, but hand-edited configs may still
|
|
# use the legacy ``api_mode`` spelling. Accept both —
|
|
# the runtime normaliser ``_normalize_custom_provider_entry``
|
|
# already does, so without this lift every migrated config
|
|
# silently downgrades codex_responses / anthropic_messages
|
|
# providers to chat_completions in the resolved runtime.
|
|
api_mode = _parse_api_mode(entry.get("api_mode") or entry.get("transport"))
|
|
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
|
|
return result
|
|
|
|
# Fall back to custom_providers: list (legacy format)
|
|
custom_providers = config.get("custom_providers")
|
|
if isinstance(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."
|
|
)
|
|
return None
|
|
|
|
custom_providers = get_compatible_custom_providers(config)
|
|
if not custom_providers:
|
|
return None
|
|
|
|
for entry in custom_providers:
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
name = entry.get("name")
|
|
base_url = entry.get("base_url")
|
|
if not isinstance(name, str) or not isinstance(base_url, str):
|
|
continue
|
|
provider_key = str(entry.get("provider_key", "") or "").strip()
|
|
if requested_norm not in custom_provider_aliases(name, provider_key):
|
|
continue
|
|
result = {
|
|
"name": name.strip(),
|
|
"base_url": base_url.strip(),
|
|
"api_key": str(entry.get("api_key", "") or "").strip(),
|
|
}
|
|
key_env = str(entry.get("key_env", "") or "").strip()
|
|
if key_env:
|
|
result["key_env"] = key_env
|
|
if provider_key:
|
|
result["provider_key"] = provider_key
|
|
extra_body = entry.get("extra_body")
|
|
if isinstance(extra_body, dict):
|
|
result["extra_body"] = dict(extra_body)
|
|
_lift_extra_headers(entry, result)
|
|
api_mode = _parse_api_mode(entry.get("api_mode"))
|
|
if api_mode:
|
|
result["api_mode"] = api_mode
|
|
model_name = str(entry.get("model", "") or "").strip()
|
|
if model_name:
|
|
result["model"] = model_name
|
|
_lift_max_output_tokens(entry, result)
|
|
capabilities = _filter_capabilities(entry.get("capabilities"))
|
|
if capabilities:
|
|
result["capabilities"] = capabilities
|
|
return result
|
|
|
|
return None
|
|
|
|
|
|
def has_named_custom_provider(requested_provider: str) -> bool:
|
|
"""Return True when config defines a custom provider matching the request.
|
|
|
|
Thin public wrapper around :func:`_get_named_custom_provider` so other
|
|
modules (e.g. the cronjob tool) can decide whether a provider name will
|
|
actually resolve to a configured ``providers:`` / ``custom_providers:``
|
|
entry — without reaching into a private helper or duplicating the scan.
|
|
"""
|
|
try:
|
|
return _get_named_custom_provider(requested_provider) is not None
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def find_custom_provider_identity(base_url: str) -> Optional[str]:
|
|
"""Map an endpoint URL back to its canonical ``custom:<name>`` menu key.
|
|
|
|
Returns the ``custom:<normalized-name>`` slug of the first ``providers:``
|
|
/ ``custom_providers:`` entry whose base_url matches, or ``None`` when no
|
|
entry owns the URL.
|
|
|
|
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. The
|
|
endpoint URL is the one durable fact that survives the round-trip, so
|
|
this reverse lookup lets persist/rebuild paths recover the entry identity
|
|
(and with it key_env/api_key/api_mode resolution via
|
|
:func:`_get_named_custom_provider`) instead of failing with
|
|
``auth_unavailable`` or silently rebuilding with placeholder credentials.
|
|
"""
|
|
target = _normalize_base_url_for_match(base_url)
|
|
if not target:
|
|
return None
|
|
try:
|
|
config = load_config()
|
|
except Exception:
|
|
return None
|
|
|
|
providers = config.get("providers")
|
|
if isinstance(providers, dict):
|
|
for ep_name, entry in providers.items():
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
entry_url = (
|
|
entry.get("api") or entry.get("url") or entry.get("base_url") or ""
|
|
)
|
|
if _normalize_base_url_for_match(entry_url) == target:
|
|
return custom_provider_slug(str(ep_name), str(ep_name))
|
|
|
|
try:
|
|
custom_providers = 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")
|
|
if not isinstance(name, str) or not name.strip():
|
|
continue
|
|
if _normalize_base_url_for_match(entry.get("base_url")) == target:
|
|
return custom_provider_slug(
|
|
name,
|
|
str(entry.get("provider_key", "") or ""),
|
|
)
|
|
|
|
return None
|
|
|
|
|
|
def find_custom_provider_identity_by_model(model: str) -> Optional[str]:
|
|
"""Map a model id back to the ``custom:<name>`` entry that serves it.
|
|
|
|
Returns the ``custom:<normalized-name>`` slug of the first ``providers:``
|
|
/ ``custom_providers:`` entry whose ``model`` / ``default_model`` matches,
|
|
or whose ``models`` catalog (dict or list shape) contains the id.
|
|
``None`` when no entry serves the model.
|
|
|
|
Companion to :func:`find_custom_provider_identity` (URL reverse-lookup)
|
|
for the persistence paths where no base_url survived the round-trip: the
|
|
session row always stores the model name, and a custom endpoint's model
|
|
ids (e.g. an in-house SFT checkpoint) virtually never collide with
|
|
catalog models on built-in providers, so the model is the last durable
|
|
fact that can recover the entry identity.
|
|
"""
|
|
target = str(model or "").strip().lower()
|
|
if not target:
|
|
return None
|
|
try:
|
|
config = load_config()
|
|
except Exception:
|
|
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
|
|
models = entry.get("models")
|
|
if isinstance(models, dict):
|
|
return any(
|
|
str(mid).strip().lower() == target for mid in models.keys()
|
|
)
|
|
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 False
|
|
|
|
providers = config.get("providers")
|
|
if isinstance(providers, dict):
|
|
for ep_name, entry in providers.items():
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
if _entry_serves_model(entry):
|
|
return custom_provider_slug(str(ep_name), str(ep_name))
|
|
|
|
try:
|
|
custom_providers = 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")
|
|
if not isinstance(name, str) or not name.strip():
|
|
continue
|
|
if _entry_serves_model(entry):
|
|
return custom_provider_slug(
|
|
name,
|
|
str(entry.get("provider_key", "") or ""),
|
|
)
|
|
|
|
return None
|
|
|
|
|
|
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.
|
|
|
|
The bare string ``"custom"`` is the *resolved billing class* shared by
|
|
every named ``providers:`` / ``custom_providers:`` entry — it is NOT a
|
|
routable provider identity (``resolve_runtime_provider("custom")`` falls
|
|
through to the OpenRouter default URL with no api_key, which surfaces to
|
|
the user as "No LLM provider configured").
|
|
|
|
Any code path that persists or restores a session's provider override
|
|
must run the resolved provider through this helper so a bare ``"custom"``
|
|
is upgraded back to its durable ``custom:<name>`` menu key. Three
|
|
recovery sources, in priority order:
|
|
|
|
1. ``base_url`` — reverse-lookup the entry that owns the endpoint URL
|
|
(the one fact that always survives the persistence round-trip when a
|
|
URL was recorded).
|
|
2. ``model`` — reverse-lookup the entry that serves the session's model
|
|
(``model``/``default_model``/``models`` catalog). The session row
|
|
always stores the model name, so when no base_url survived (the
|
|
recurring Desktop/TUI regression vector) the model is the last
|
|
session-scoped fact that can recover the entry — and unlike the
|
|
config fallback below it stays correct after the user points their
|
|
global default at a different provider.
|
|
3. ``config_provider`` — the active ``config.model.provider`` (or its
|
|
``provider``/``HERMES_INFERENCE_PROVIDER`` equivalent). When neither
|
|
a base_url nor a model recovered the entry, the configured provider
|
|
is the only durable identity left, so fall back to it when it names
|
|
a real entry.
|
|
|
|
Returns ``custom:<name>`` when a routable identity is recovered, else
|
|
``None`` (caller keeps whatever it had — bare ``"custom"`` only as a last
|
|
resort, e.g. a genuine ad-hoc endpoint with no config entry).
|
|
"""
|
|
# 1. Reverse-lookup by endpoint URL.
|
|
if base_url:
|
|
identity = find_custom_provider_identity(base_url)
|
|
if identity:
|
|
return identity
|
|
|
|
# 2. Reverse-lookup by the session's model name.
|
|
if model:
|
|
identity = find_custom_provider_identity_by_model(model)
|
|
if identity:
|
|
return identity
|
|
|
|
# 3. Fall back to the configured provider when it names a real entry.
|
|
candidate = str(config_provider or "").strip()
|
|
if not candidate:
|
|
try:
|
|
candidate = str(_get_model_config().get("provider") or "").strip()
|
|
except Exception:
|
|
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 return it when it actually resolves to a configured custom entry,
|
|
# so we never invent a `custom:<x>` that resolution can't honor.
|
|
try:
|
|
entry = _get_named_custom_provider(candidate)
|
|
if entry is not None:
|
|
# ``candidate`` matched, but it may be the entry's DISPLAY NAME —
|
|
# ``_get_named_custom_provider`` accepts either spelling. For a
|
|
# keyed ``providers:`` entry the display name is not the durable
|
|
# identity, so re-resolve through the endpoint the matched entry
|
|
# owns and return the same config-key slug every other path
|
|
# returns (7b5a18817). Without this, a display name that differs
|
|
# from its key heals to ``custom:<display-name>`` and stops
|
|
# matching the persisted identity.
|
|
identity = find_custom_provider_identity(str(entry.get("base_url") or ""))
|
|
if identity:
|
|
return identity
|
|
if candidate_norm.startswith("custom:"):
|
|
return candidate_norm
|
|
return f"custom:{candidate_norm}"
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def is_routable_provider(provider: Optional[str]) -> bool:
|
|
"""Whether a provider name currently resolves to a routable route.
|
|
|
|
Empty/None is vacuously routable: agent build falls back to the
|
|
configured default instead of failing. A name that resolves through
|
|
the full chain (built-in -> user ``providers:`` -> ``custom_providers:``
|
|
-> models.dev) is routable; anything else would fail agent init with
|
|
"Unknown provider '<name>'".
|
|
|
|
Session resume uses this to detect a stale/renamed/removed provider
|
|
persisted in an older session snapshot, so recovery can fall back to
|
|
the configured default or the model the user picked instead of letting
|
|
the agent build die.
|
|
"""
|
|
name = str(provider or "").strip()
|
|
if not name or name.lower() == "auto":
|
|
return True
|
|
if name.lower() == "custom":
|
|
# The bare string is the resolved billing class shared by every
|
|
# named custom entry — not a routable identity. restore paths must
|
|
# heal it (canonical_custom_identity) or fall back, never hand it
|
|
# straight to agent init.
|
|
return False
|
|
try:
|
|
from hermes_cli.providers import resolve_provider_full
|
|
|
|
config = load_config()
|
|
return (
|
|
resolve_provider_full(
|
|
name,
|
|
config.get("providers"),
|
|
get_compatible_custom_providers(config),
|
|
)
|
|
is not None
|
|
)
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _normalize_base_url_for_match(value) -> str:
|
|
return str(value or "").strip().rstrip("/").lower()
|
|
|
|
|
|
def _custom_provider_request_overrides(custom_provider: Dict[str, Any]) -> Dict[str, Any]:
|
|
extra_body = custom_provider.get("extra_body")
|
|
if not isinstance(extra_body, dict) or not extra_body:
|
|
return {}
|
|
return {"extra_body": dict(extra_body)}
|
|
|
|
|
|
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]]:
|
|
# Bare `provider="custom"` with an explicit base_url (e.g. propagated
|
|
# from a `model_aliases:` direct-alias resolution) — build a runtime
|
|
# directly so the alias's base_url actually takes effect.
|
|
#
|
|
# GitHub #27132: provider aliases that resolve to "custom" at runtime
|
|
# (ollama, vllm, llamacpp, …) are treated identically here, so a YAML
|
|
# `provider: ollama` with a LAN/WireGuard `base_url` doesn't silently
|
|
# fall through to OpenRouter.
|
|
requested_norm = (requested_provider or "").strip().lower()
|
|
|
|
# Managed llama.cpp runtime: a llamacpp-flavored alias with no explicit
|
|
# base_url resolves to the supervised server (or a detected external
|
|
# one) before the generic custom fallthrough. Explicit base_url always
|
|
# wins — a user pointing at a specific server means that server.
|
|
if requested_norm in ("llamacpp", "llama.cpp", "llama-cpp") and not explicit_base_url:
|
|
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 {
|
|
"provider": "custom",
|
|
"api_mode": "chat_completions",
|
|
"base_url": endpoint["base_url"],
|
|
"api_key": (explicit_api_key or "").strip()
|
|
or endpoint["api_key"] or "no-key-required",
|
|
"source": "local-runtime",
|
|
"requested_provider": requested_provider,
|
|
}
|
|
# No server to serve this model. Say so and stop — falling through
|
|
# to the generic custom path sends the request to whatever provider
|
|
# picks it up (OpenRouter with a placeholder key), and the user's
|
|
# "local server is off" surfaces as that provider's baffling
|
|
# "401 Invalid API key". The switch's own state picks the message:
|
|
# the user who turned the server off gets pointed at the switch,
|
|
# anyone else at the setup pane.
|
|
try:
|
|
from hermes_cli.config import load_config as _load_cfg
|
|
|
|
_lr_enabled = bool((_load_cfg().get("local_runtime") or {}).get("enabled"))
|
|
except Exception: # noqa: BLE001
|
|
_lr_enabled = False
|
|
if _lr_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."
|
|
)
|
|
|
|
if requested_norm and requested_norm != "custom":
|
|
try:
|
|
from hermes_cli.auth import resolve_provider as _resolve_provider
|
|
|
|
if _resolve_provider(requested_norm) == "custom":
|
|
requested_norm = "custom"
|
|
except Exception:
|
|
pass
|
|
if requested_norm == "custom" and explicit_base_url:
|
|
base_url = explicit_base_url.strip().rstrip("/")
|
|
# Check credential pool first — mirrors the named-custom-provider path
|
|
# so bare `provider: custom` with a configured custom_providers entry
|
|
# also gets its api_key from the pool instead of env var fallbacks.
|
|
pool_result = _try_resolve_from_custom_pool(base_url, "custom", None)
|
|
if pool_result:
|
|
pool_result["source"] = "direct-alias"
|
|
return pool_result
|
|
_da_is_openai_url = base_url_host_matches(base_url, "openai.com") or base_url_host_matches(base_url, "openai.azure.com")
|
|
_da_is_openrouter = base_url_host_matches(base_url, "openrouter.ai")
|
|
_da_is_ollama_url = base_url_host_matches(base_url, "ollama.com")
|
|
api_key_candidates = [
|
|
(explicit_api_key or "").strip(),
|
|
# Gate env key fallbacks on authoritative hosts (#28660)
|
|
# OLLAMA_API_KEY needs its own gate here: _host_derived_api_key
|
|
# deliberately skips it, expecting an explicit host-matched path
|
|
# like this one (GHSA-76xc-57q6-vm5m). Without it a `model_aliases:`
|
|
# entry pointing at Ollama Cloud resolved no key at all.
|
|
(_getenv("OLLAMA_API_KEY", "").strip() if _da_is_ollama_url else ""),
|
|
(_getenv("OPENAI_API_KEY", "").strip() if _da_is_openai_url else ""),
|
|
(_getenv("OPENROUTER_API_KEY", "").strip() if _da_is_openrouter else ""),
|
|
# Bonus (#28660): derive `<VENDOR>_API_KEY` from the host so users
|
|
# who set DEEPSEEK_API_KEY / GROQ_API_KEY / MISTRAL_API_KEY get the
|
|
# intuitive match without configuring `custom_providers` first.
|
|
_host_derived_api_key(base_url),
|
|
]
|
|
api_key = next(
|
|
(c for c in api_key_candidates if has_usable_secret(c)),
|
|
"",
|
|
) or "no-key-required"
|
|
return {
|
|
"provider": "custom",
|
|
"api_mode": _detect_api_mode_for_url(base_url) or "chat_completions",
|
|
"base_url": base_url,
|
|
"api_key": api_key,
|
|
"source": "direct-alias",
|
|
"requested_provider": requested_provider,
|
|
}
|
|
|
|
custom_provider = _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
|
|
|
|
# Check if a credential pool exists for this custom endpoint
|
|
pool_result = _try_resolve_from_custom_pool(
|
|
base_url,
|
|
"custom",
|
|
custom_provider.get("api_mode"),
|
|
provider_name=custom_provider.get("provider_key") or custom_provider.get("name"),
|
|
)
|
|
if pool_result:
|
|
# Propagate the model name even when using pooled credentials —
|
|
# the pool doesn't know about the custom_providers model field.
|
|
# An explicit ``target_model`` wins (same rule as the non-pool path).
|
|
model_name = target_model or custom_provider.get("model")
|
|
if model_name:
|
|
pool_result["model"] = model_name
|
|
_lift_model_capabilities(custom_provider, model_name, pool_result)
|
|
if isinstance(custom_provider.get("max_output_tokens"), int):
|
|
pool_result["max_output_tokens"] = custom_provider["max_output_tokens"]
|
|
request_overrides = _custom_provider_request_overrides(custom_provider)
|
|
if request_overrides:
|
|
pool_result["request_overrides"] = {
|
|
**dict(pool_result.get("request_overrides") or {}),
|
|
**request_overrides,
|
|
}
|
|
# Propagate extra_headers so custom-provider auth headers (e.g.
|
|
# Cloudflare Access service tokens) still apply with pooled
|
|
# credentials. NEVER log the values.
|
|
if custom_provider.get("extra_headers"):
|
|
pool_result["extra_headers"] = dict(custom_provider["extra_headers"])
|
|
return pool_result
|
|
|
|
_cp_is_openai_url = base_url_host_matches(base_url, "openai.com") or base_url_host_matches(base_url, "openai.azure.com")
|
|
_cp_is_openrouter = base_url_host_matches(base_url, "openrouter.ai")
|
|
api_key_candidates = [
|
|
(explicit_api_key or "").strip(),
|
|
str(custom_provider.get("api_key", "") or "").strip(),
|
|
_getenv(str(custom_provider.get("key_env", "") or "").strip(), "").strip(),
|
|
# Gate provider env keys on their authoritative hosts — sending
|
|
# OPENAI_API_KEY to a local-llm endpoint leaks credentials (#28660).
|
|
(_getenv("OPENAI_API_KEY", "").strip() if _cp_is_openai_url else ""),
|
|
(_getenv("OPENROUTER_API_KEY", "").strip() if _cp_is_openrouter else ""),
|
|
# Bonus (#28660): derive `<VENDOR>_API_KEY` from the host as a final
|
|
# fallback when key_env wasn't set explicitly.
|
|
_host_derived_api_key(base_url),
|
|
]
|
|
api_key = next((candidate for candidate in api_key_candidates if has_usable_secret(candidate)), "")
|
|
|
|
# A ``key_cmd`` credential is minted per request rather than resolved once:
|
|
# gateways that issue short-lived bearers would otherwise go stale
|
|
# mid-session and 401. Both wire clients already accept a callable api_key
|
|
# (the Entra ID contract) and invoke it per request. An explicit --api-key
|
|
# still wins — it is the one-off recovery escape hatch.
|
|
key_cmd = str(custom_provider.get("key_cmd", "") or "").strip()
|
|
if key_cmd and not has_usable_secret((explicit_api_key or "").strip()):
|
|
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"),
|
|
)
|
|
if token_provider is not None:
|
|
api_key = token_provider
|
|
|
|
result = {
|
|
"provider": "custom",
|
|
"api_mode": custom_provider.get("api_mode")
|
|
or _detect_api_mode_for_url(base_url)
|
|
or "chat_completions",
|
|
"base_url": base_url,
|
|
"api_key": api_key or "no-key-required",
|
|
"source": f"custom_provider:{custom_provider.get('name', requested_provider)}",
|
|
"requested_provider": requested_provider,
|
|
}
|
|
# Propagate the model name so callers can override self.model when the
|
|
# provider name differs from the actual model string the API expects.
|
|
# An explicit ``target_model`` wins over the provider's configured
|
|
# default (regression: auxiliary slots / background-review resolve a
|
|
# concrete model for a custom provider and must not silently fall back
|
|
# to ``default_model``).
|
|
if target_model:
|
|
result["model"] = target_model
|
|
elif custom_provider.get("model"):
|
|
result["model"] = custom_provider["model"]
|
|
_lift_model_capabilities(
|
|
custom_provider, result.get("model"), result
|
|
)
|
|
if isinstance(custom_provider.get("max_output_tokens"), int):
|
|
result["max_output_tokens"] = custom_provider["max_output_tokens"]
|
|
# Per-provider extra HTTP headers (proxies, gateways, custom auth).
|
|
# Values may carry credentials — NEVER log them.
|
|
if custom_provider.get("extra_headers"):
|
|
result["extra_headers"] = dict(custom_provider["extra_headers"])
|
|
request_overrides = _custom_provider_request_overrides(custom_provider)
|
|
if request_overrides:
|
|
result["request_overrides"] = request_overrides
|
|
|
|
# Custom providers in the OpenCode family (name extends opencode-go/zen,
|
|
# or base_url hosted on opencode.ai) serve models behind different API
|
|
# surfaces per model — a static api_mode 503s for /v1/responses-only
|
|
# models like grok-4.5 (#85589). Re-derive api_mode from the effective
|
|
# model and normalize the /v1 suffix, exactly like the built-in
|
|
# opencode-zen/go paths do.
|
|
from hermes_cli.models import opencode_provider_family
|
|
|
|
_oc_family = opencode_provider_family(requested_provider)
|
|
if _oc_family is None:
|
|
try:
|
|
from utils import base_url_hostname
|
|
|
|
if base_url_hostname(base_url).lower() == "opencode.ai":
|
|
_oc_family = (
|
|
"opencode-go" if "/zen/go" in base_url.lower() else "opencode-zen"
|
|
)
|
|
except Exception:
|
|
_oc_family = None
|
|
if _oc_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 _get_model_config().get("default")
|
|
or ""
|
|
).strip()
|
|
if _effective_model:
|
|
result["api_mode"] = opencode_model_api_mode(_oc_family, _effective_model)
|
|
result["base_url"] = normalize_opencode_base_url(
|
|
_oc_family, result["api_mode"], result["base_url"]
|
|
)
|
|
return result
|
|
|
|
|
|
def _resolve_openrouter_runtime(
|
|
*,
|
|
requested_provider: str,
|
|
explicit_api_key: Optional[str] = None,
|
|
explicit_base_url: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
model_cfg = _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 = ""
|
|
for k in ("api_key", "api"):
|
|
v = model_cfg.get(k)
|
|
if isinstance(v, str) and v.strip():
|
|
cfg_api_key = v.strip()
|
|
break
|
|
requested_norm = (requested_provider or "").strip().lower()
|
|
cfg_provider = cfg_provider.strip().lower()
|
|
# GitHub #27132: provider aliases that resolve to "custom" (ollama,
|
|
# vllm, llamacpp, …) follow the same base_url trust + routing rules
|
|
# as a bare `provider: custom`. Normalising here keeps every check
|
|
# below — `requested_norm == "custom"`, the trust check, the pool
|
|
# gate up the stack — alias-aware without duplicating the alias map.
|
|
if requested_norm and requested_norm != "custom":
|
|
try:
|
|
from hermes_cli.auth import resolve_provider as _resolve_provider
|
|
|
|
if _resolve_provider(requested_norm) == "custom":
|
|
requested_norm = "custom"
|
|
except Exception:
|
|
pass
|
|
|
|
env_openrouter_base_url = _getenv("OPENROUTER_BASE_URL", "").strip()
|
|
env_custom_base_url = _getenv("CUSTOM_BASE_URL", "").strip()
|
|
|
|
# Use config base_url when available and the provider context matches.
|
|
# OPENAI_BASE_URL env var is no longer consulted — config.yaml is
|
|
# the single source of truth for endpoint URLs.
|
|
use_config_base_url = False
|
|
if cfg_base_url.strip() and not explicit_base_url:
|
|
if requested_norm == "auto":
|
|
if not cfg_provider or cfg_provider == "auto":
|
|
use_config_base_url = True
|
|
elif requested_norm == "custom" and _config_base_url_trustworthy_for_bare_custom(
|
|
cfg_base_url, cfg_provider
|
|
):
|
|
use_config_base_url = True
|
|
|
|
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("/")
|
|
|
|
# Choose API key based on whether the resolved base_url targets OpenRouter.
|
|
# When hitting OpenRouter, prefer OPENROUTER_API_KEY (issue #289).
|
|
# When hitting a custom endpoint (e.g. Z.ai, local LLM), prefer
|
|
# OPENAI_API_KEY so the OpenRouter key doesn't leak to an unrelated
|
|
# provider (issues #420, #560).
|
|
_is_openrouter_url = base_url_host_matches(base_url, "openrouter.ai")
|
|
# Also treat explicitly-configured OpenRouter mirrors/proxies as OpenRouter
|
|
# for key selection — if the user set OPENROUTER_BASE_URL or requested
|
|
# provider=openrouter explicitly, OPENROUTER_API_KEY should still be used.
|
|
_is_openrouter_context = _is_openrouter_url or (
|
|
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:
|
|
api_key_candidates = [
|
|
explicit_api_key,
|
|
_getenv("OPENROUTER_API_KEY"),
|
|
_getenv("OPENAI_API_KEY"),
|
|
]
|
|
else:
|
|
# Custom endpoint: use api_key from config when using config base_url (#1760).
|
|
# When the endpoint is Ollama Cloud, check OLLAMA_API_KEY — it's
|
|
# the canonical env var for ollama.com authentication. Match on
|
|
# HOST, not substring — a custom base_url whose path contains
|
|
# "ollama.com" (e.g. http://127.0.0.1/ollama.com/v1) or whose
|
|
# hostname is a look-alike (ollama.com.attacker.test) must not
|
|
# receive the Ollama credential. See GHSA-76xc-57q6-vm5m.
|
|
_is_ollama_url = base_url_host_matches(base_url, "ollama.com")
|
|
_is_openai_url = base_url_host_matches(base_url, "openai.com")
|
|
_is_openai_azure = base_url_host_matches(base_url, "openai.azure.com")
|
|
# Gate each provider key on its own host — sending OPENAI_API_KEY or
|
|
# OPENROUTER_API_KEY to an unrelated custom endpoint (DeepSeek, Groq,
|
|
# Mistral, …) leaks credentials and causes 401s (issue #28660).
|
|
# Mirrors the OLLAMA_API_KEY host-gate added in GHSA-76xc-57q6-vm5m.
|
|
api_key_candidates = [
|
|
explicit_api_key,
|
|
(cfg_api_key if use_config_base_url else ""),
|
|
(_getenv("OLLAMA_API_KEY") if _is_ollama_url else ""),
|
|
(_getenv("OPENAI_API_KEY") if (_is_openai_url or _is_openai_azure) else ""),
|
|
(_getenv("OPENROUTER_API_KEY") if _is_openrouter_url else ""),
|
|
# Bonus (#28660): derive `<VENDOR>_API_KEY` from the host so users
|
|
# who set DEEPSEEK_API_KEY / GROQ_API_KEY / MISTRAL_API_KEY get the
|
|
# intuitive match. Helper returns "" for IPs/loopback and for env
|
|
# vars already handled by the explicit host-gated paths above.
|
|
_host_derived_api_key(base_url),
|
|
]
|
|
api_key = next(
|
|
(str(candidate or "").strip() for candidate in api_key_candidates if has_usable_secret(candidate)),
|
|
"",
|
|
)
|
|
|
|
source = "explicit" if (explicit_api_key or explicit_base_url) else "env/config"
|
|
|
|
# When "custom" was explicitly requested, preserve that as the provider
|
|
# name instead of silently relabeling to "openrouter" (#2562).
|
|
# Also provide a placeholder API key for local servers that don't require
|
|
# authentication — the OpenAI SDK requires a non-empty api_key string.
|
|
effective_provider = "custom" if requested_norm == "custom" else "openrouter"
|
|
|
|
# For custom endpoints, check if a credential pool exists
|
|
if effective_provider == "custom" and base_url:
|
|
# Pass requested_provider so pool lookup prefers name match over base_url,
|
|
# fixing credential mix-ups when multiple custom providers share a base_url.
|
|
pool_result = _try_resolve_from_custom_pool(
|
|
base_url, effective_provider, _parse_api_mode(model_cfg.get("api_mode")),
|
|
provider_name=requested_provider if requested_norm != "custom" else None,
|
|
)
|
|
if pool_result:
|
|
return pool_result
|
|
|
|
if effective_provider == "custom" and not api_key and not _is_openrouter_url:
|
|
api_key = "no-key-required"
|
|
|
|
return {
|
|
"provider": effective_provider,
|
|
"api_mode": _resolve_plain_custom_api_mode(model_cfg, base_url)
|
|
if effective_provider == "custom"
|
|
else _parse_api_mode(model_cfg.get("api_mode"))
|
|
or _detect_api_mode_for_url(base_url)
|
|
or "chat_completions",
|
|
"base_url": base_url,
|
|
"api_key": api_key,
|
|
"source": source,
|
|
}
|
|
|
|
|
|
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]:
|
|
"""Resolve an Azure Foundry runtime entry.
|
|
|
|
Reads ``model.base_url`` + ``model.api_mode`` from config.yaml (or
|
|
explicit overrides), pulls the API key from ``.env`` / env var, and
|
|
strips a trailing ``/v1`` for Anthropic-style endpoints because the
|
|
Anthropic SDK appends ``/v1/messages`` internally.
|
|
|
|
When ``model.auth_mode == "entra_id"`` (and the model is OpenAI-style),
|
|
the returned ``api_key`` is a zero-arg callable produced by
|
|
:func:`agent.azure_identity_adapter.build_token_provider` rather than
|
|
a string. Downstream code that constructs an OpenAI SDK client passes
|
|
this through unchanged (the SDK accepts ``Callable[[], str]`` for
|
|
``api_key`` and calls it before every request). Code paths that need
|
|
a string (logging, manual HTTP probes, header injection) must use the
|
|
helpers in ``agent.azure_identity_adapter``.
|
|
|
|
Raises :class:`AuthError` when required values are missing.
|
|
"""
|
|
explicit_api_key = str(explicit_api_key or "").strip()
|
|
explicit_base_url_clean = str(explicit_base_url or "").strip().rstrip("/")
|
|
|
|
cfg_provider = str(model_cfg.get("provider") or "").strip().lower()
|
|
cfg_base_url = ""
|
|
cfg_api_mode = "chat_completions"
|
|
cfg_auth_mode = "api_key"
|
|
cfg_entra: Dict[str, Any] = {}
|
|
if cfg_provider == "azure-foundry":
|
|
cfg_base_url = str(model_cfg.get("base_url") or "").strip().rstrip("/")
|
|
cfg_api_mode = _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"
|
|
_entra = model_cfg.get("entra")
|
|
if isinstance(_entra, dict):
|
|
cfg_entra = _entra
|
|
|
|
# Model-family inference: Azure Foundry deploys GPT-5.x / codex / o1-o4
|
|
# reasoning models as Responses-API-only. Calling /chat/completions
|
|
# against them returns 400 "The requested operation is unsupported."
|
|
# Upgrade api_mode when the model name matches, unless the user has
|
|
# explicitly chosen anthropic_messages (Anthropic-style endpoint).
|
|
effective_model = str(target_model or model_cfg.get("default") or "").strip()
|
|
if effective_model and cfg_api_mode != "anthropic_messages":
|
|
try:
|
|
from hermes_cli.models import azure_foundry_model_api_mode
|
|
|
|
inferred = azure_foundry_model_api_mode(effective_model)
|
|
except Exception:
|
|
inferred = None
|
|
if inferred:
|
|
cfg_api_mode = inferred
|
|
|
|
env_base_url = _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 AuthError(
|
|
"Azure Foundry requires a base URL. Set it via 'hermes model' or "
|
|
"the AZURE_FOUNDRY_BASE_URL environment variable."
|
|
)
|
|
|
|
# Anthropic SDK appends /v1/messages itself, so strip any trailing /v1
|
|
# we inherited from the configured base_url to avoid double-/v1 paths.
|
|
if cfg_api_mode == "anthropic_messages":
|
|
base_url = re.sub(r"/v1/?$", "", base_url)
|
|
|
|
# ── Entra ID (Microsoft Foundry recommended path) ──────────────────
|
|
#
|
|
# OpenAI-style endpoints use the OpenAI SDK's native callable
|
|
# ``api_key=`` contract — the SDK mints a fresh JWT per request
|
|
# automatically.
|
|
#
|
|
# Anthropic-style endpoints (Claude on Foundry) take the callable
|
|
# too: :func:`agent.anthropic_adapter.build_anthropic_client`
|
|
# detects the callable and constructs an ``httpx.Client`` with a
|
|
# request event hook that injects a fresh ``Authorization: Bearer``
|
|
# header per request (the Anthropic SDK does not accept callables
|
|
# natively). From the runtime resolver's perspective both modes
|
|
# are identical — return the callable api_key and let the
|
|
# downstream SDK wrapper handle the contract difference.
|
|
if cfg_auth_mode == "entra_id":
|
|
if explicit_api_key:
|
|
# User passed --api-key on the CLI while config says entra_id —
|
|
# honour the explicit string (escape hatch for one-off testing).
|
|
api_key: Any = explicit_api_key
|
|
source = "explicit"
|
|
auth_mode = "api_key"
|
|
else:
|
|
try:
|
|
from agent.azure_identity_adapter import (
|
|
EntraIdentityConfig,
|
|
SCOPE_AI_AZURE_DEFAULT,
|
|
build_token_provider,
|
|
)
|
|
except Exception as exc:
|
|
raise AuthError(
|
|
"Azure Foundry Entra ID auth requires the 'azure-identity' "
|
|
"package. Install it with: pip install azure-identity "
|
|
f"(import failed: {exc})"
|
|
) from exc
|
|
|
|
scope = (
|
|
str(cfg_entra.get("scope") or "").strip()
|
|
or SCOPE_AI_AZURE_DEFAULT
|
|
)
|
|
try:
|
|
entra_config = EntraIdentityConfig(
|
|
scope=scope,
|
|
)
|
|
token_provider = build_token_provider(config=entra_config)
|
|
except ImportError as exc:
|
|
raise AuthError(str(exc)) from exc
|
|
api_key = token_provider
|
|
source = "entra_id"
|
|
auth_mode = "entra_id"
|
|
|
|
clean_entra = {}
|
|
if auth_mode == "entra_id":
|
|
configured_scope = str(cfg_entra.get("scope") or "").strip()
|
|
if configured_scope:
|
|
clean_entra["scope"] = configured_scope
|
|
|
|
return {
|
|
"provider": "azure-foundry",
|
|
"api_mode": cfg_api_mode,
|
|
"base_url": base_url,
|
|
"api_key": api_key,
|
|
"auth_mode": auth_mode,
|
|
"entra": clean_entra,
|
|
"source": source,
|
|
"requested_provider": requested_provider,
|
|
}
|
|
|
|
# ── Static API key (legacy / default) ──────────────────────────────
|
|
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 = ""
|
|
if not api_key:
|
|
api_key = _getenv("AZURE_FOUNDRY_API_KEY", "").strip()
|
|
if not api_key:
|
|
raise AuthError(
|
|
"Azure Foundry requires an API key. Set AZURE_FOUNDRY_API_KEY in "
|
|
"~/.hermes/.env or run 'hermes model' to configure. To use "
|
|
"keyless Microsoft Entra ID auth instead, set "
|
|
"model.auth_mode: entra_id in config.yaml (or pick "
|
|
"'Microsoft Entra ID' in 'hermes model')."
|
|
)
|
|
|
|
source = "explicit" if (explicit_api_key or explicit_base_url) else "config"
|
|
return {
|
|
"provider": "azure-foundry",
|
|
"api_mode": cfg_api_mode,
|
|
"base_url": base_url,
|
|
"api_key": api_key,
|
|
"auth_mode": "api_key",
|
|
"source": source,
|
|
"requested_provider": requested_provider,
|
|
}
|
|
|
|
|
|
def _resolve_explicit_runtime(
|
|
*,
|
|
provider: str,
|
|
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,
|
|
) -> Optional[Dict[str, Any]]:
|
|
explicit_api_key = str(explicit_api_key or "").strip()
|
|
explicit_base_url = str(explicit_base_url or "").strip().rstrip("/")
|
|
if not explicit_api_key and not explicit_base_url:
|
|
return None
|
|
|
|
if provider == "anthropic":
|
|
cfg_provider = str(model_cfg.get("provider") or "").strip().lower()
|
|
cfg_base_url = ""
|
|
if cfg_provider == "anthropic":
|
|
cfg_base_url = str(model_cfg.get("base_url") or "").strip().rstrip("/")
|
|
if not _anthropic_base_url_override_ok(cfg_base_url):
|
|
cfg_base_url = ""
|
|
base_url = explicit_base_url or cfg_base_url or "https://api.anthropic.com"
|
|
api_key = explicit_api_key
|
|
if not api_key:
|
|
from agent.anthropic_adapter import resolve_anthropic_token
|
|
|
|
api_key = resolve_anthropic_token()
|
|
if not api_key:
|
|
raise AuthError(
|
|
"No Anthropic credentials found. Set ANTHROPIC_TOKEN or ANTHROPIC_API_KEY, "
|
|
"run 'claude setup-token', or authenticate with 'claude /login'."
|
|
)
|
|
return {
|
|
"provider": "anthropic",
|
|
"api_mode": "anthropic_messages",
|
|
"base_url": base_url,
|
|
"api_key": api_key,
|
|
"source": "explicit",
|
|
"requested_provider": requested_provider,
|
|
}
|
|
|
|
if provider == "openai-codex":
|
|
base_url = explicit_base_url or DEFAULT_CODEX_BASE_URL
|
|
api_key = explicit_api_key
|
|
last_refresh = None
|
|
if not api_key:
|
|
creds = resolve_codex_runtime_credentials()
|
|
api_key = creds.get("api_key", "")
|
|
last_refresh = creds.get("last_refresh")
|
|
if not explicit_base_url:
|
|
base_url = creds.get("base_url", "").rstrip("/") or base_url
|
|
return {
|
|
"provider": "openai-codex",
|
|
"api_mode": "codex_responses",
|
|
"base_url": base_url,
|
|
"api_key": api_key,
|
|
"source": "explicit",
|
|
"last_refresh": last_refresh,
|
|
"requested_provider": requested_provider,
|
|
}
|
|
|
|
if provider == "nous":
|
|
from hermes_cli.providers import nous_api_mode
|
|
|
|
state = auth_mod.get_provider_auth_state("nous") or {}
|
|
base_url = (
|
|
explicit_base_url
|
|
or _nous_inference_base_url_override()
|
|
or str(state.get("inference_base_url") or auth_mod.DEFAULT_NOUS_INFERENCE_URL).strip().rstrip("/")
|
|
)
|
|
# Only use the agent_key compatibility field for inference when it
|
|
# contains a NAS invoke JWT; raw OAuth access_token fallback is handled
|
|
# by resolve_nous_runtime_credentials().
|
|
api_key = explicit_api_key or (
|
|
str(state.get("agent_key") or "").strip()
|
|
if _agent_key_is_usable(
|
|
state,
|
|
max(60, env_int("HERMES_NOUS_MIN_KEY_TTL_SECONDS", 1800)),
|
|
)
|
|
else ""
|
|
)
|
|
expires_at = state.get("agent_key_expires_at") or state.get("expires_at")
|
|
if not api_key:
|
|
creds = resolve_nous_runtime_credentials(
|
|
timeout_seconds=float(_getenv("HERMES_NOUS_TIMEOUT_SECONDS", "15")),
|
|
)
|
|
api_key = creds.get("api_key", "")
|
|
expires_at = creds.get("expires_at")
|
|
if not explicit_base_url:
|
|
base_url = creds.get("base_url", "").rstrip("/") or base_url
|
|
return {
|
|
"provider": "nous",
|
|
"api_mode": nous_api_mode(target_model or model_cfg.get("default") or ""),
|
|
"base_url": base_url,
|
|
"api_key": api_key,
|
|
"source": "explicit",
|
|
"expires_at": expires_at,
|
|
"requested_provider": requested_provider,
|
|
}
|
|
|
|
# Azure Foundry: user-configured endpoint with selectable API mode
|
|
if provider == "azure-foundry":
|
|
return _resolve_azure_foundry_runtime(
|
|
requested_provider=requested_provider,
|
|
model_cfg=model_cfg,
|
|
explicit_api_key=explicit_api_key,
|
|
explicit_base_url=explicit_base_url,
|
|
)
|
|
|
|
pconfig = PROVIDER_REGISTRY.get(provider)
|
|
if pconfig and pconfig.auth_type == "api_key":
|
|
env_url = ""
|
|
if pconfig.base_url_env_var:
|
|
env_url = _getenv(pconfig.base_url_env_var, "").strip().rstrip("/")
|
|
|
|
base_url = explicit_base_url
|
|
if not base_url:
|
|
if provider in {"kimi-coding", "kimi-coding-cn"}:
|
|
creds = resolve_api_key_provider_credentials(provider)
|
|
base_url = creds.get("base_url", "").rstrip("/")
|
|
else:
|
|
base_url = env_url or pconfig.inference_base_url
|
|
|
|
if provider == "actual":
|
|
base_url = normalize_actual_base_url(base_url)
|
|
|
|
api_key = explicit_api_key
|
|
if not api_key:
|
|
creds = resolve_api_key_provider_credentials(provider)
|
|
api_key = creds.get("api_key", "")
|
|
if not base_url:
|
|
base_url = creds.get("base_url", "").rstrip("/")
|
|
if provider == "actual":
|
|
base_url = normalize_actual_base_url(base_url)
|
|
|
|
api_mode = "chat_completions"
|
|
if provider == "copilot":
|
|
api_mode = _copilot_runtime_api_mode(
|
|
model_cfg,
|
|
api_key,
|
|
target_model=target_model,
|
|
)
|
|
elif provider == "xai":
|
|
api_mode = "codex_responses"
|
|
elif provider == "actual":
|
|
api_mode = "codex_responses"
|
|
else:
|
|
configured_provider = str(model_cfg.get("provider") or "").strip().lower()
|
|
configured_mode = _parse_api_mode(model_cfg.get("api_mode"))
|
|
if configured_mode and _provider_supports_explicit_api_mode(provider, configured_provider):
|
|
api_mode = configured_mode
|
|
else:
|
|
# URL detection first, then the provider's declared transport
|
|
# (fixes regional OpenAI hosts and other non-chat overlays).
|
|
api_mode = _fallback_api_mode(
|
|
provider, base_url, target_model or model_cfg.get("default", "")
|
|
)
|
|
|
|
if provider == "actual" and not api_key and is_actual_local_base_url(base_url):
|
|
api_key = ACTUAL_LOCAL_NOAUTH_PLACEHOLDER
|
|
|
|
return {
|
|
"provider": provider,
|
|
"api_mode": api_mode,
|
|
"base_url": base_url.rstrip("/"),
|
|
"api_key": api_key,
|
|
"source": "explicit",
|
|
"requested_provider": requested_provider,
|
|
}
|
|
|
|
return None
|
|
|
|
|
|
def _is_external_process_provider(provider: str) -> bool:
|
|
"""Whether ``provider`` is declared as an external-process (CLI) provider.
|
|
|
|
Reads the CLI provider registry first (which now absorbs registered
|
|
ProviderProfiles, in-tree and out), then falls back to the profile registry
|
|
directly so the check works before the CLI registry has been extended.
|
|
"""
|
|
name = (provider or "").strip().lower()
|
|
if not name:
|
|
return False
|
|
try:
|
|
from hermes_cli.auth import PROVIDER_REGISTRY
|
|
|
|
pconfig = PROVIDER_REGISTRY.get(name)
|
|
if pconfig is not None:
|
|
return pconfig.auth_type == "external_process"
|
|
except Exception:
|
|
pass
|
|
try:
|
|
from providers import get_provider_profile
|
|
|
|
profile = get_provider_profile(name)
|
|
except Exception:
|
|
return False
|
|
return profile is not None and getattr(profile, "auth_type", "") == "external_process"
|
|
|
|
|
|
def resolve_runtime_provider(
|
|
*,
|
|
requested: Optional[str] = None,
|
|
explicit_api_key: Optional[str] = None,
|
|
explicit_base_url: Optional[str] = None,
|
|
target_model: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Resolve runtime provider credentials for agent execution.
|
|
|
|
target_model: Optional override for model_cfg.get("default") when
|
|
computing provider-specific api_mode (e.g. OpenCode Zen/Go where different
|
|
models route through different API surfaces). Callers performing an
|
|
explicit mid-session model switch should pass the new model here so
|
|
api_mode is derived from the model they are switching TO, not the stale
|
|
persisted default. Other callers can leave it None to preserve existing
|
|
behavior (api_mode derived from config).
|
|
"""
|
|
requested_provider = resolve_requested_provider(requested)
|
|
|
|
# Honour ``providers.<name>.enabled: false`` for BOTH user-defined
|
|
# custom providers and the built-in ones (openai / anthropic /
|
|
# openrouter / gemini / ...). The earlier ``_get_named_custom_provider``
|
|
# gate only covers custom blocks — built-in resolution paths
|
|
# (``resolve_provider`` + pool / explicit / generic runtime) walk
|
|
# their own short-circuits and would otherwise return stale config
|
|
# for a provider the user explicitly turned off.
|
|
#
|
|
# Fail fast with a typed error so the fallback chain can advance to
|
|
# the next provider instead of using a disabled one.
|
|
from hermes_cli.config import is_provider_enabled, load_config
|
|
_full_cfg = load_config()
|
|
_provs_cfg = _full_cfg.get("providers") if isinstance(_full_cfg, dict) else None
|
|
if isinstance(_provs_cfg, dict):
|
|
_block = _provs_cfg.get(requested_provider)
|
|
if isinstance(_block, dict) and not is_provider_enabled(_block):
|
|
raise ValueError(
|
|
f"provider {requested_provider!r} is disabled in config "
|
|
f"(providers.{requested_provider}.enabled: false)"
|
|
)
|
|
|
|
if requested_provider == "moa":
|
|
return {
|
|
"provider": "moa",
|
|
"api_mode": "chat_completions",
|
|
"base_url": "moa://local",
|
|
"api_key": "moa-virtual-provider",
|
|
"source": "moa-virtual-provider",
|
|
"requested_provider": requested_provider,
|
|
}
|
|
|
|
# Azure Anthropic short-circuit: when explicitly targeting an Azure endpoint
|
|
# with provider="anthropic", bypass _resolve_named_custom_runtime (which would
|
|
# return provider="custom" with chat_completions api_mode and no valid key).
|
|
# Instead, use the Azure key directly with anthropic_messages api_mode.
|
|
_eff_base = (explicit_base_url or "").strip()
|
|
if requested_provider == "anthropic" and base_url_host_matches(_eff_base, "azure.com"):
|
|
_azure_key = (
|
|
(explicit_api_key or "").strip()
|
|
or _getenv("AZURE_ANTHROPIC_KEY", "").strip()
|
|
or _getenv("ANTHROPIC_API_KEY", "").strip()
|
|
)
|
|
return {
|
|
"provider": "anthropic",
|
|
"api_mode": "anthropic_messages",
|
|
"base_url": _eff_base.rstrip("/"),
|
|
"api_key": _azure_key,
|
|
"source": "azure-explicit",
|
|
"requested_provider": requested_provider,
|
|
}
|
|
|
|
# Azure Foundry: user-configured endpoint with selectable API mode
|
|
# (OpenAI-style chat_completions or Anthropic-style anthropic_messages).
|
|
# Resolve before the custom-runtime / pool / generic paths so Azure
|
|
# config is always picked up from model.base_url + model.api_mode,
|
|
# regardless of whether the caller passed explicit_* args.
|
|
if requested_provider == "azure-foundry":
|
|
azure_runtime = _resolve_azure_foundry_runtime(
|
|
requested_provider=requested_provider,
|
|
model_cfg=_get_model_config(),
|
|
explicit_api_key=explicit_api_key,
|
|
explicit_base_url=explicit_base_url,
|
|
target_model=target_model,
|
|
)
|
|
return azure_runtime
|
|
|
|
# Vertex AI: OAuth2-token provider (Gemini via the OpenAI-compatible
|
|
# endpoint). Resolve BEFORE the custom-runtime / credential-pool / generic
|
|
# paths. The credential *path* (GOOGLE_APPLICATION_CREDENTIALS /
|
|
# VERTEX_CREDENTIALS_PATH) must never reach the credential pool or the
|
|
# generic api_key resolver — those would treat the file path as a static
|
|
# API key. Instead we mint a short-lived OAuth2 access token here and hand
|
|
# it to the standard OpenAI client as api_key, with base_url computed from
|
|
# the project ID + region. The token is re-minted per call (5-min refresh
|
|
# margin) by get_vertex_config(); mid-session expiry is additionally
|
|
# recovered on 401 by run_agent._try_refresh_vertex_client_credentials().
|
|
if requested_provider in ("vertex", "google-vertex", "vertex-ai", "gcp-vertex", "vertexai"):
|
|
from agent.vertex_adapter import get_vertex_config
|
|
|
|
token, base_url = get_vertex_config()
|
|
if not token or not base_url:
|
|
raise AuthError(
|
|
"Vertex AI credentials could not be resolved. Vertex uses "
|
|
"OAuth2 (not a static API key): provide a service-account JSON "
|
|
"via GOOGLE_APPLICATION_CREDENTIALS (or VERTEX_CREDENTIALS_PATH) "
|
|
"in ~/.hermes/.env, or run 'gcloud auth application-default "
|
|
"login' for ADC. Set the GCP project/region under vertex: in "
|
|
"config.yaml if they aren't embedded in the credentials. "
|
|
"Run `hermes setup` to install Vertex support."
|
|
)
|
|
return {
|
|
"provider": "vertex",
|
|
"api_mode": "chat_completions",
|
|
"base_url": base_url.rstrip("/"),
|
|
"api_key": token,
|
|
"source": "vertex-oauth",
|
|
"requested_provider": requested_provider,
|
|
}
|
|
|
|
custom_runtime = _resolve_named_custom_runtime(
|
|
requested_provider=requested_provider,
|
|
explicit_api_key=explicit_api_key,
|
|
explicit_base_url=explicit_base_url,
|
|
target_model=target_model,
|
|
)
|
|
if custom_runtime:
|
|
custom_runtime["requested_provider"] = requested_provider
|
|
return custom_runtime
|
|
|
|
# If provider is "auto" (or unset) but config.yaml has an explicit base_url
|
|
# pointing at a custom/local endpoint (e.g. Ollama at localhost:11434),
|
|
# route through the OpenAI-compatible resolver instead of letting
|
|
# resolve_provider() pick up an ANTHROPIC_API_KEY or OPENAI_API_KEY from
|
|
# the environment and send the request to a cloud API. Fixes #3846.
|
|
if not explicit_base_url and not explicit_api_key:
|
|
model_cfg = _get_model_config()
|
|
cfg_provider = str(model_cfg.get("provider") or "").strip().lower()
|
|
cfg_base_url = str(model_cfg.get("base_url") or "").strip()
|
|
if cfg_base_url and cfg_provider in ("auto", ""):
|
|
# Check that base_url isn't one of the well-known cloud API roots
|
|
# (OpenRouter, Anthropic, OpenAI). If it's something else (Ollama,
|
|
# LM Studio, vLLM, …) we honour it directly. The full detection
|
|
# logic lives in _resolve_openrouter_runtime; we just skip the
|
|
# resolve_provider() call so env-var credentials don't shadow it.
|
|
# Match on HOST, not substring, so a look-alike base_url
|
|
# (e.g. http://api.anthropic.com.attacker.test/v1, or one whose
|
|
# path merely contains "openai.com") cannot evade the bypass and
|
|
# leak a cloud credential. Mirrors the host-gating used for
|
|
# API-key selection in _resolve_openrouter_runtime.
|
|
_known_cloud_hosts = (
|
|
"openrouter.ai",
|
|
"anthropic.com",
|
|
"openai.com",
|
|
)
|
|
if not any(
|
|
base_url_host_matches(cfg_base_url, host)
|
|
for host in _known_cloud_hosts
|
|
):
|
|
runtime = _resolve_openrouter_runtime(
|
|
requested_provider=requested_provider,
|
|
explicit_api_key=explicit_api_key,
|
|
explicit_base_url=explicit_base_url,
|
|
)
|
|
runtime["requested_provider"] = requested_provider
|
|
return runtime
|
|
|
|
provider = resolve_provider(
|
|
requested_provider,
|
|
explicit_api_key=explicit_api_key,
|
|
explicit_base_url=explicit_base_url,
|
|
)
|
|
model_cfg = _get_model_config()
|
|
|
|
# OpenCode Zen free tier (*-free slugs, e.g. x-preview-f-free /
|
|
# "Ox Alpha"): served ANONYMOUSLY on the Zen relay ONLY. Any bearer the
|
|
# relay doesn't recognize is a 401 — and the Go relay doesn't serve the
|
|
# free tier at all ("Model x is not supported"), so a valid OpenCode GO
|
|
# subscription key still fails. Route free slugs through the keyless Zen
|
|
# runtime BEFORE the credential-pool / explicit / api_key paths so they
|
|
# work with any OpenCode credential state, including none.
|
|
from hermes_cli.models import (
|
|
opencode_provider_family as _oc_family_fn,
|
|
opencode_zen_free_runtime as _oc_free_runtime_fn,
|
|
)
|
|
if _oc_family_fn(provider) is not None:
|
|
_oc_model = str(
|
|
target_model or model_cfg.get("default") or model_cfg.get("model") or ""
|
|
).strip()
|
|
_free_runtime = _oc_free_runtime_fn(provider, _oc_model)
|
|
if _free_runtime is not None:
|
|
_free_runtime["requested_provider"] = requested_provider
|
|
return _free_runtime
|
|
|
|
explicit_runtime = _resolve_explicit_runtime(
|
|
provider=provider,
|
|
requested_provider=requested_provider,
|
|
model_cfg=model_cfg,
|
|
explicit_api_key=explicit_api_key,
|
|
explicit_base_url=explicit_base_url,
|
|
target_model=target_model,
|
|
)
|
|
if explicit_runtime:
|
|
return explicit_runtime
|
|
|
|
should_use_pool = provider != "openrouter"
|
|
if provider == "openrouter":
|
|
cfg_provider = str(model_cfg.get("provider") or "").strip().lower()
|
|
cfg_base_url = str(model_cfg.get("base_url") or "").strip()
|
|
env_openai_base_url = _getenv("OPENAI_BASE_URL", "").strip()
|
|
env_openrouter_base_url = _getenv("OPENROUTER_BASE_URL", "").strip()
|
|
has_custom_endpoint = bool(
|
|
explicit_base_url
|
|
or env_openai_base_url
|
|
or env_openrouter_base_url
|
|
)
|
|
if cfg_base_url and cfg_provider in {"auto", "custom"}:
|
|
has_custom_endpoint = True
|
|
has_runtime_override = bool(explicit_api_key or explicit_base_url)
|
|
should_use_pool = (
|
|
requested_provider in {"openrouter", "auto"}
|
|
and not has_custom_endpoint
|
|
and not has_runtime_override
|
|
)
|
|
|
|
try:
|
|
pool = load_pool(provider) if should_use_pool else None
|
|
except Exception:
|
|
pool = None
|
|
if pool and pool.has_credentials():
|
|
entry = pool.select()
|
|
pool_api_key = ""
|
|
if entry is not None:
|
|
pool_api_key = (
|
|
getattr(entry, "runtime_api_key", None)
|
|
or getattr(entry, "access_token", "")
|
|
)
|
|
# For Nous, the pool entry's runtime_api_key is the agent_key
|
|
# compatibility field. It must be an invoke JWT. The pool doesn't
|
|
# refresh it during selection (that would trigger network calls in
|
|
# non-runtime contexts like `hermes auth list`). If the key is
|
|
# expired/missing, refresh the selected pool entry before falling back
|
|
# to singleton auth resolution.
|
|
if provider == "nous" and entry is not None:
|
|
min_ttl = max(60, env_int("HERMES_NOUS_MIN_KEY_TTL_SECONDS", 1800))
|
|
nous_state = {
|
|
"agent_key": getattr(entry, "agent_key", None),
|
|
"agent_key_expires_at": getattr(entry, "agent_key_expires_at", None),
|
|
"scope": getattr(entry, "scope", None),
|
|
}
|
|
if not _agent_key_is_usable(nous_state, min_ttl):
|
|
logger.debug("Nous pool entry agent_key expired/missing, refreshing selected pool entry")
|
|
try:
|
|
refreshed = pool.try_refresh_current()
|
|
except Exception as exc:
|
|
logger.debug("Nous pool entry refresh failed: %s", exc)
|
|
refreshed = None
|
|
if refreshed is not None:
|
|
entry = refreshed
|
|
pool_api_key = (
|
|
getattr(entry, "runtime_api_key", None)
|
|
or getattr(entry, "access_token", "")
|
|
)
|
|
nous_state = {
|
|
"agent_key": getattr(entry, "agent_key", None),
|
|
"agent_key_expires_at": getattr(entry, "agent_key_expires_at", None),
|
|
"scope": getattr(entry, "scope", None),
|
|
}
|
|
if not pool_api_key or not _agent_key_is_usable(nous_state, min_ttl):
|
|
logger.debug("Nous pool entry agent_key still unavailable, falling through to runtime resolution")
|
|
pool_api_key = ""
|
|
if (
|
|
entry is not None
|
|
and pool_api_key
|
|
and credential_pool_matches_provider(
|
|
pool,
|
|
provider,
|
|
base_url=(
|
|
getattr(entry, "runtime_base_url", None)
|
|
or getattr(entry, "base_url", None)
|
|
or ""
|
|
),
|
|
)
|
|
):
|
|
return _resolve_runtime_from_pool_entry(
|
|
provider=provider,
|
|
entry=entry,
|
|
requested_provider=requested_provider,
|
|
model_cfg=model_cfg,
|
|
pool=pool,
|
|
target_model=target_model,
|
|
)
|
|
|
|
if provider == "nous":
|
|
try:
|
|
from hermes_cli.providers import nous_api_mode
|
|
|
|
creds = resolve_nous_runtime_credentials(
|
|
timeout_seconds=float(_getenv("HERMES_NOUS_TIMEOUT_SECONDS", "15")),
|
|
)
|
|
return {
|
|
"provider": "nous",
|
|
"api_mode": nous_api_mode(target_model or model_cfg.get("default") or ""),
|
|
"base_url": creds.get("base_url", "").rstrip("/"),
|
|
"api_key": creds.get("api_key", ""),
|
|
"source": creds.get("source", "portal"),
|
|
"expires_at": creds.get("expires_at"),
|
|
"requested_provider": requested_provider,
|
|
}
|
|
except AuthError:
|
|
if requested_provider != "auto":
|
|
raise
|
|
# Auto-detected Nous but credentials are stale/revoked —
|
|
# fall through to env-var providers (e.g. OpenRouter).
|
|
logger.info("Auto-detected Nous provider but credentials failed; "
|
|
"falling through to next provider.")
|
|
|
|
if provider == "openai-codex":
|
|
try:
|
|
creds = resolve_codex_runtime_credentials()
|
|
return {
|
|
"provider": "openai-codex",
|
|
"api_mode": "codex_responses",
|
|
"base_url": creds.get("base_url", "").rstrip("/"),
|
|
"api_key": creds.get("api_key", ""),
|
|
"source": creds.get("source", "hermes-auth-store"),
|
|
"last_refresh": creds.get("last_refresh"),
|
|
"requested_provider": requested_provider,
|
|
}
|
|
except AuthError:
|
|
if requested_provider != "auto":
|
|
raise
|
|
# Auto-detected Codex but credentials are stale/revoked —
|
|
# fall through to env-var providers (e.g. OpenRouter).
|
|
logger.info("Auto-detected Codex provider but credentials failed; "
|
|
"falling through to next provider.")
|
|
|
|
if provider == "xai-oauth":
|
|
try:
|
|
creds = resolve_xai_oauth_runtime_credentials()
|
|
return {
|
|
"provider": "xai-oauth",
|
|
"api_mode": "codex_responses",
|
|
"base_url": (creds.get("base_url") or "").rstrip("/") or DEFAULT_XAI_OAUTH_BASE_URL,
|
|
"api_key": creds.get("api_key", ""),
|
|
"source": creds.get("source", "hermes-auth-store"),
|
|
"last_refresh": creds.get("last_refresh"),
|
|
"requested_provider": requested_provider,
|
|
}
|
|
except AuthError:
|
|
if requested_provider != "auto":
|
|
raise
|
|
logger.info("Auto-detected xAI OAuth provider but credentials failed; "
|
|
"falling through to next provider.")
|
|
|
|
if provider == "qwen-oauth":
|
|
try:
|
|
creds = resolve_qwen_runtime_credentials()
|
|
return {
|
|
"provider": "qwen-oauth",
|
|
"api_mode": "chat_completions",
|
|
"base_url": creds.get("base_url", "").rstrip("/"),
|
|
"api_key": creds.get("api_key", ""),
|
|
"source": creds.get("source", "qwen-cli"),
|
|
"expires_at_ms": creds.get("expires_at_ms"),
|
|
"requested_provider": requested_provider,
|
|
}
|
|
except AuthError:
|
|
if requested_provider != "auto":
|
|
raise
|
|
logger.info("Qwen OAuth credentials failed; "
|
|
"falling through to next provider.")
|
|
|
|
if provider == "minimax-oauth":
|
|
pconfig = PROVIDER_REGISTRY.get(provider)
|
|
if pconfig and pconfig.auth_type == "oauth_minimax":
|
|
from hermes_cli.auth import resolve_minimax_oauth_runtime_credentials
|
|
creds = resolve_minimax_oauth_runtime_credentials()
|
|
return {
|
|
"provider": provider,
|
|
"api_mode": "anthropic_messages",
|
|
"base_url": creds["base_url"],
|
|
"api_key": creds["api_key"],
|
|
"source": creds.get("source", "oauth"),
|
|
"requested_provider": requested_provider,
|
|
}
|
|
|
|
# External-process providers (an agent CLI driven over stdio, e.g. ACP).
|
|
# Keyed on the registered provider's auth_type rather than on one name, so a
|
|
# provider shipped outside this tree lands on the same credential path.
|
|
if _is_external_process_provider(provider):
|
|
creds = resolve_external_process_provider_credentials(provider)
|
|
return {
|
|
"provider": provider,
|
|
"api_mode": "chat_completions",
|
|
"base_url": creds.get("base_url", "").rstrip("/"),
|
|
"api_key": creds.get("api_key", ""),
|
|
"command": creds.get("command", ""),
|
|
"args": list(creds.get("args") or []),
|
|
"source": creds.get("source", "process"),
|
|
"requested_provider": requested_provider,
|
|
}
|
|
|
|
# Anthropic (native Messages API)
|
|
if provider == "anthropic":
|
|
# Allow base URL override from config.yaml model.base_url, but only
|
|
# when the configured provider is anthropic — otherwise a non-Anthropic
|
|
# base_url (e.g. Codex endpoint) would leak into Anthropic requests.
|
|
cfg_provider = str(model_cfg.get("provider") or "").strip().lower()
|
|
cfg_base_url = ""
|
|
if cfg_provider == "anthropic":
|
|
cfg_base_url = (model_cfg.get("base_url") or "").strip().rstrip("/")
|
|
if not _anthropic_base_url_override_ok(cfg_base_url):
|
|
cfg_base_url = ""
|
|
base_url = cfg_base_url or "https://api.anthropic.com"
|
|
|
|
# For Microsoft Foundry endpoints, use ANTHROPIC_API_KEY directly —
|
|
# Claude Code OAuth tokens (sk-ant-oat01) are not accepted by Azure.
|
|
# Azure keys don't start with "sk-ant-" so resolve_anthropic_token()
|
|
# would find the Claude Code OAuth token first (priority 3) and return
|
|
# that instead, causing 401s. Detect Azure endpoints and use the env
|
|
# key directly to bypass the OAuth priority chain.
|
|
_is_azure_endpoint = base_url_host_matches(base_url, "azure.com") or (
|
|
cfg_base_url and base_url_host_matches(cfg_base_url, "azure.com")
|
|
)
|
|
if _is_azure_endpoint:
|
|
# Honor user-specified env var hints on the model config before
|
|
# falling back to the built-in AZURE_ANTHROPIC_KEY / ANTHROPIC_API_KEY
|
|
# chain. Accept both `key_env` (Hermes canonical — matches the
|
|
# custom_providers field name) and `api_key_env` (documented in the
|
|
# Azure Foundry guide and read by most Hermes-compatible importers).
|
|
# Matches the config.yaml examples in website/docs/guides/azure-foundry.md.
|
|
token = ""
|
|
for hint_key in ("key_env", "api_key_env"):
|
|
env_var = str(model_cfg.get(hint_key) or "").strip()
|
|
if env_var:
|
|
token = _getenv(env_var, "").strip()
|
|
if token:
|
|
break
|
|
# Next: an inline api_key on the model config (useful in multi-profile
|
|
# setups that want to avoid env-var juggling).
|
|
if not token:
|
|
token = str(model_cfg.get("api_key") or "").strip()
|
|
# Finally fall back to the historical fixed names.
|
|
if not token:
|
|
token = (
|
|
_getenv("AZURE_ANTHROPIC_KEY", "").strip()
|
|
or _getenv("ANTHROPIC_API_KEY", "").strip()
|
|
)
|
|
if not token:
|
|
raise AuthError(
|
|
"No Azure Anthropic API key found. Set AZURE_ANTHROPIC_KEY or "
|
|
"ANTHROPIC_API_KEY, or point key_env/api_key_env in your "
|
|
"config.yaml model section at a custom env var."
|
|
)
|
|
else:
|
|
from agent.anthropic_adapter import resolve_anthropic_token
|
|
token = resolve_anthropic_token()
|
|
if not token:
|
|
raise AuthError(
|
|
"No Anthropic credentials found. Set ANTHROPIC_TOKEN or ANTHROPIC_API_KEY, "
|
|
"run 'claude setup-token', or authenticate with 'claude /login'."
|
|
)
|
|
return {
|
|
"provider": "anthropic",
|
|
"api_mode": "anthropic_messages",
|
|
"base_url": base_url,
|
|
"api_key": token,
|
|
"source": "env",
|
|
"requested_provider": requested_provider,
|
|
}
|
|
|
|
# AWS Bedrock (native Converse API via boto3)
|
|
if provider == "bedrock":
|
|
from agent.bedrock_adapter import (
|
|
has_aws_credentials,
|
|
resolve_aws_auth_env_var,
|
|
resolve_bedrock_runtime_region,
|
|
is_anthropic_bedrock_model,
|
|
is_openai_bedrock_model,
|
|
bedrock_openai_base_url,
|
|
resolve_bedrock_bearer_token,
|
|
)
|
|
# When the user explicitly selected bedrock (not auto-detected),
|
|
# trust boto3's credential chain — it handles IMDS, ECS task roles,
|
|
# Lambda execution roles, SSO, and other implicit sources that our
|
|
# env-var check can't detect.
|
|
is_explicit = requested_provider in {"bedrock", "aws", "aws-bedrock", "amazon-bedrock", "amazon"}
|
|
if not is_explicit and not has_aws_credentials():
|
|
raise AuthError(
|
|
"No AWS credentials found for Bedrock. Configure one of:\n"
|
|
" - AWS_ACCESS_KEY_ID + AWS_SECRET_ACCESS_KEY\n"
|
|
" - AWS_PROFILE (for SSO / named profiles)\n"
|
|
" - IAM instance role (EC2, ECS, Lambda)\n"
|
|
"Or run 'aws configure' to set up credentials.",
|
|
code="no_aws_credentials",
|
|
)
|
|
# Read bedrock-specific config from config.yaml
|
|
_bedrock_cfg = load_config().get("bedrock", {})
|
|
# Region priority: config.yaml bedrock.region → env var → us-east-1.
|
|
# resolve_bedrock_runtime_region() is the canonical implementation of
|
|
# this priority; auxiliary resolution uses the same helper.
|
|
region = resolve_bedrock_runtime_region({"bedrock": _bedrock_cfg})
|
|
auth_source = resolve_aws_auth_env_var() or "aws-sdk-default-chain"
|
|
# Build guardrail config if configured
|
|
_gr = _bedrock_cfg.get("guardrail", {})
|
|
guardrail_config = None
|
|
if _gr.get("guardrail_identifier") and _gr.get("guardrail_version"):
|
|
guardrail_config = {
|
|
"guardrailIdentifier": _gr["guardrail_identifier"],
|
|
"guardrailVersion": _gr["guardrail_version"],
|
|
}
|
|
if _gr.get("stream_processing_mode"):
|
|
guardrail_config["streamProcessingMode"] = _gr["stream_processing_mode"]
|
|
if _gr.get("trace"):
|
|
guardrail_config["trace"] = _gr["trace"]
|
|
# Triple-path routing:
|
|
# - OpenAI GPT-5.5 on Bedrock uses Bedrock Mantle's OpenAI Responses
|
|
# endpoint (not Converse / bedrock-runtime).
|
|
# - Claude models use AnthropicBedrock SDK for prompt caching,
|
|
# thinking budgets, and adaptive thinking.
|
|
# - Other models use the native Converse API.
|
|
#
|
|
# Exception: Bearer Token auth (AWS_BEARER_TOKEN_BEDROCK) is NOT
|
|
# supported by the AnthropicBedrock SDK (it only does SigV4 signing —
|
|
# a bearer-only setup fails at runtime with "could not resolve
|
|
# credentials from session"). Route these users through the Converse
|
|
# API regardless of model. Ref: #28156.
|
|
_current_model = str(target_model or model_cfg.get("default") or "").strip()
|
|
_has_bearer_token = bool(os.environ.get("AWS_BEARER_TOKEN_BEDROCK", "").strip())
|
|
if is_openai_bedrock_model(_current_model):
|
|
bearer = resolve_bedrock_bearer_token()
|
|
runtime = {
|
|
"provider": "bedrock",
|
|
"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,
|
|
"region": region,
|
|
"model": _current_model,
|
|
"bedrock_openai": True,
|
|
"requested_provider": requested_provider,
|
|
}
|
|
elif is_anthropic_bedrock_model(_current_model) and not _has_bearer_token:
|
|
# Claude on Bedrock → AnthropicBedrock SDK → anthropic_messages path
|
|
runtime = {
|
|
"provider": "bedrock",
|
|
"api_mode": "anthropic_messages",
|
|
"base_url": f"https://bedrock-runtime.{region}.amazonaws.com",
|
|
"api_key": "aws-sdk",
|
|
"source": auth_source,
|
|
"region": region,
|
|
"bedrock_anthropic": True, # Signal to use AnthropicBedrock client
|
|
"requested_provider": requested_provider,
|
|
}
|
|
else:
|
|
# Non-Claude/OpenAI (Nova, DeepSeek, Llama, GPT-OSS, etc.) → Converse API
|
|
runtime = {
|
|
"provider": "bedrock",
|
|
"api_mode": "bedrock_converse",
|
|
"base_url": f"https://bedrock-runtime.{region}.amazonaws.com",
|
|
"api_key": "aws-sdk",
|
|
"source": auth_source,
|
|
"region": region,
|
|
"requested_provider": requested_provider,
|
|
}
|
|
if guardrail_config:
|
|
runtime["guardrail_config"] = guardrail_config
|
|
return runtime
|
|
|
|
# API-key providers (z.ai/GLM, Kimi, MiniMax, MiniMax-CN)
|
|
pconfig = PROVIDER_REGISTRY.get(provider)
|
|
if pconfig and pconfig.auth_type == "api_key":
|
|
creds = resolve_api_key_provider_credentials(provider)
|
|
# Actual Computer: a loopback base_url configured in model_cfg (not
|
|
# just env) selects the daemon's local offline API, which requires no
|
|
# auth. Inject the placeholder BEFORE the usable-secret gate below,
|
|
# mirroring the env-driven path inside the credential resolver.
|
|
if provider == "actual" and not has_usable_secret(creds.get("api_key")):
|
|
_cfg_provider = str(model_cfg.get("provider") or "").strip().lower()
|
|
_cfg_url = ""
|
|
if _cfg_provider == provider:
|
|
_cfg_url = (model_cfg.get("base_url") or "").strip().rstrip("/")
|
|
_effective_url = normalize_actual_base_url(
|
|
_cfg_url or creds.get("base_url", "").rstrip("/")
|
|
)
|
|
if is_actual_local_base_url(_effective_url):
|
|
creds = dict(creds)
|
|
creds["api_key"] = ACTUAL_LOCAL_NOAUTH_PLACEHOLDER
|
|
creds["source"] = creds.get("source") or "local-offline"
|
|
# An explicitly selected API-key provider is authoritative. Returning
|
|
# a runtime with an empty key defers failure until the first request and
|
|
# can make a later fallback look like a silent provider switch. Fail at
|
|
# resolution so callers surface the missing credential (or consult only
|
|
# an explicitly configured fallback chain). LM Studio's no-auth path
|
|
# supplies a non-empty placeholder in the credential resolver above.
|
|
if not has_usable_secret(creds.get("api_key")):
|
|
env_names = ", ".join(pconfig.api_key_env_vars)
|
|
hint = f" Set {env_names}." if env_names else ""
|
|
raise AuthError(
|
|
f"No usable credentials found for provider '{provider}'.{hint}",
|
|
provider=provider,
|
|
code="missing_api_key",
|
|
)
|
|
# Honour model.base_url from config.yaml when the configured provider
|
|
# matches this provider — mirrors the Anthropic path above. Without
|
|
# this, users who set model.base_url to e.g. api.minimaxi.com/anthropic
|
|
# (China endpoint) still get the hardcoded api.minimax.io default (#6039).
|
|
cfg_provider = str(model_cfg.get("provider") or "").strip().lower()
|
|
cfg_base_url = ""
|
|
if cfg_provider == provider:
|
|
cfg_base_url = (model_cfg.get("base_url") or "").strip().rstrip("/")
|
|
base_url = cfg_base_url or creds.get("base_url", "").rstrip("/")
|
|
if provider == "actual":
|
|
base_url = normalize_actual_base_url(base_url)
|
|
api_mode = "chat_completions"
|
|
if provider == "copilot":
|
|
api_mode = _copilot_runtime_api_mode(
|
|
model_cfg,
|
|
creds.get("api_key", ""),
|
|
target_model=target_model,
|
|
)
|
|
elif provider == "xai":
|
|
api_mode = "codex_responses"
|
|
elif provider == "actual":
|
|
api_mode = "codex_responses"
|
|
else:
|
|
configured_provider = str(model_cfg.get("provider") or "").strip().lower()
|
|
# Only honor persisted api_mode when it belongs to the same provider family.
|
|
configured_mode = _parse_api_mode(model_cfg.get("api_mode"))
|
|
from hermes_cli.models import opencode_provider_family
|
|
if opencode_provider_family(provider) is not None:
|
|
# opencode-zen/go must always re-derive api_mode from the
|
|
# target model (not the stale persisted api_mode), because
|
|
# the same provider serves both anthropic_messages
|
|
# (e.g. minimax-m2.7) and chat_completions (e.g.
|
|
# deepseek-v4-flash) and switching models via /model would
|
|
# otherwise carry the previous mode forward, stripping /v1
|
|
# from base_url for chat_completions models and 404'ing.
|
|
# Refs #16878.
|
|
from hermes_cli.models import opencode_model_api_mode
|
|
_effective = target_model or model_cfg.get("default", "")
|
|
api_mode = opencode_model_api_mode(provider, _effective)
|
|
elif configured_mode and _provider_supports_explicit_api_mode(provider, configured_provider):
|
|
api_mode = configured_mode
|
|
else:
|
|
# URL detection first (e.g. https://api.minimax.io/anthropic,
|
|
# official OpenAI hosts → codex_responses, api.x.ai →
|
|
# codex_responses), then the provider's declared transport.
|
|
api_mode = _fallback_api_mode(
|
|
provider, base_url, target_model or model_cfg.get("default", "")
|
|
)
|
|
# Normalize the /v1 suffix for OpenCode by API mode (see comment above).
|
|
from hermes_cli.models import opencode_provider_family
|
|
if opencode_provider_family(provider) is not None:
|
|
from hermes_cli.models import normalize_opencode_base_url
|
|
base_url = normalize_opencode_base_url(provider, api_mode, base_url)
|
|
if provider == "lmstudio":
|
|
base_url = auth_mod._normalize_lmstudio_runtime_base_url(base_url)
|
|
api_key = creds.get("api_key", "")
|
|
if provider == "actual" and not api_key and is_actual_local_base_url(base_url):
|
|
api_key = ACTUAL_LOCAL_NOAUTH_PLACEHOLDER
|
|
return {
|
|
"provider": provider,
|
|
"api_mode": api_mode,
|
|
"base_url": base_url,
|
|
"api_key": api_key,
|
|
"source": creds.get("source", "env"),
|
|
"requested_provider": requested_provider,
|
|
}
|
|
|
|
runtime = _resolve_openrouter_runtime(
|
|
requested_provider=requested_provider,
|
|
explicit_api_key=explicit_api_key,
|
|
explicit_base_url=explicit_base_url,
|
|
)
|
|
runtime["requested_provider"] = requested_provider
|
|
return runtime
|
|
|
|
|
|
def format_runtime_provider_error(error: Exception) -> str:
|
|
if isinstance(error, AuthError):
|
|
return format_auth_error(error)
|
|
return str(error)
|