Merge branch 'simp/r2-config' into simp/integration2
This commit is contained in:
Teknium
+28
-646
@@ -20,7 +20,6 @@ from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Dict, Any, Optional, List, Tuple, Set
|
||||
|
||||
from hermes_cli.route_identity import normalize_route_base_url
|
||||
from hermes_cli.secret_prompt import masked_secret_prompt
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -869,6 +868,34 @@ def _ensure_hermes_home_managed(home: Path):
|
||||
# =============================================================================
|
||||
|
||||
from hermes_cli.config_defaults import DEFAULT_CONFIG, OPTIONAL_ENV_VARS # noqa: F401
|
||||
from hermes_cli.config_providers import ( # noqa: F401 (re-exported; callers/tests use hermes_cli.config.<name>)
|
||||
_API_MODE_ALIASES,
|
||||
_CAMEL_ALIASES,
|
||||
_KNOWN_PROVIDER_KEYS,
|
||||
_PROVIDER_NORMALIZE_WARNED,
|
||||
_canonical_api_mode,
|
||||
_coerce_ssl_verify,
|
||||
_custom_provider_entry_to_provider_config,
|
||||
_entries_for_route,
|
||||
_normalize_custom_provider_entry,
|
||||
_normalize_provider_models,
|
||||
_pick_provider_base_url,
|
||||
_route_model_cfg,
|
||||
_warn_once_per_provider,
|
||||
apply_custom_provider_extra_headers_to_client_kwargs,
|
||||
apply_custom_provider_tls_to_client_kwargs,
|
||||
coerce_provider_id,
|
||||
find_provider_entry,
|
||||
get_compatible_custom_providers,
|
||||
get_custom_provider_context_length,
|
||||
get_custom_provider_extra_headers,
|
||||
get_custom_provider_model_capability,
|
||||
get_custom_provider_tls_settings,
|
||||
is_provider_enabled,
|
||||
normalize_extra_headers,
|
||||
providers_dict_to_custom_providers,
|
||||
stringify_provider_map,
|
||||
)
|
||||
|
||||
# =============================================================================
|
||||
# Config Migration System
|
||||
@@ -1226,631 +1253,6 @@ def get_missing_skill_config_vars() -> List[Dict[str, Any]]:
|
||||
return missing
|
||||
|
||||
|
||||
# ``_normalize_custom_provider_entry`` runs on every ``load_picker_context()``
|
||||
# call (i.e. per interactive picker/inventory request), so any warning it emits
|
||||
# fires repeatedly for the same static config. Deduplicate per (provider,
|
||||
# signature): on Windows a repeated-warning storm contends on
|
||||
# ``concurrent-log-handler``'s cross-process rotation lock and can peg a core /
|
||||
# stall the gateway/serve event loop. The cache lives for the process lifetime.
|
||||
_PROVIDER_NORMALIZE_WARNED: set = set()
|
||||
|
||||
|
||||
def _warn_once_per_provider(
|
||||
provider_key: str, signature: str, msg: str, *args: Any
|
||||
) -> None:
|
||||
"""Emit ``logger.warning(msg, *args)`` at most once per (provider, signature)."""
|
||||
dedup_key = (provider_key or "?", signature)
|
||||
if dedup_key in _PROVIDER_NORMALIZE_WARNED:
|
||||
return
|
||||
_PROVIDER_NORMALIZE_WARNED.add(dedup_key)
|
||||
logger.warning(msg, *args)
|
||||
|
||||
|
||||
_API_MODE_ALIASES = {
|
||||
# Values accepted by earlier releases (and natural spellings) mapped to
|
||||
# the canonical transport names consumed by agent_init. Before this map
|
||||
# existed, an unrecognized api_mode was silently ignored and the
|
||||
# transport fell through to hostname-based guessing, so a config that
|
||||
# said ``api_mode: openai`` (valid on older releases) could flip to
|
||||
# ``codex_responses`` after an update and break the provider (#66543
|
||||
# discussion; observed live against api.actual.inc).
|
||||
"openai": "chat_completions",
|
||||
"openai_chat": "chat_completions",
|
||||
"openai-chat": "chat_completions",
|
||||
"chat-completions": "chat_completions",
|
||||
"chatcompletions": "chat_completions",
|
||||
"responses": "codex_responses",
|
||||
"openai_responses": "codex_responses",
|
||||
"openai-responses": "codex_responses",
|
||||
"anthropic": "anthropic_messages",
|
||||
"anthropic-messages": "anthropic_messages",
|
||||
"messages": "anthropic_messages",
|
||||
"bedrock": "bedrock_converse",
|
||||
"bedrock-converse": "bedrock_converse",
|
||||
}
|
||||
|
||||
|
||||
def _canonical_api_mode(api_mode: str) -> str:
|
||||
"""Map legacy/alias ``api_mode`` spellings to canonical transport names.
|
||||
|
||||
Unknown values pass through unchanged (callers keep their existing fall-through behavior); known
|
||||
aliases are rewritten so downstream consumers (``agent_init``'s accepted-set check, runtime
|
||||
resolution) see a canonical name instead of silently discarding the user's intent.
|
||||
"""
|
||||
cleaned = api_mode.strip()
|
||||
return _API_MODE_ALIASES.get(cleaned.lower(), cleaned)
|
||||
|
||||
|
||||
def coerce_provider_id(value: Any) -> str:
|
||||
"""Provider identity fields are strings."""
|
||||
if value is None:
|
||||
return ""
|
||||
return str(value).strip()
|
||||
|
||||
|
||||
def stringify_provider_map(providers: Any) -> dict:
|
||||
"""Copy a ``providers:`` mapping so keys are strings.
|
||||
|
||||
Desktop Custom Endpoints store the name as the dict key; an unquoted YAML key ``2070:`` loads
|
||||
as int, so picker code calling ``ep_name.lower()`` crashes and CRUD lookups of ``"2070"`` miss.
|
||||
"""
|
||||
if not isinstance(providers, dict):
|
||||
return {}
|
||||
out: Dict[str, Any] = {}
|
||||
for stored, value in providers.items():
|
||||
key = coerce_provider_id(stored)
|
||||
if key:
|
||||
out[key] = value
|
||||
return out
|
||||
|
||||
|
||||
def find_provider_entry(providers: Any, key: Any) -> Tuple[Any, Optional[Dict[str, Any]]]:
|
||||
"""Return ``(stored_key, entry)`` matching *key* by string identity.
|
||||
|
||||
Prefer an exact string hit, then scan.
|
||||
"""
|
||||
if not isinstance(providers, dict):
|
||||
return None, None
|
||||
want = coerce_provider_id(key)
|
||||
if not want:
|
||||
return None, None
|
||||
exact = providers.get(want)
|
||||
if isinstance(exact, dict):
|
||||
return want, exact
|
||||
for stored, entry in providers.items():
|
||||
if coerce_provider_id(stored) == want and isinstance(entry, dict):
|
||||
return stored, entry
|
||||
return None, None
|
||||
|
||||
|
||||
# camelCase aliases commonly used in hand-written provider configs.
|
||||
_CAMEL_ALIASES: Dict[str, str] = {
|
||||
"apiKey": "api_key",
|
||||
"baseUrl": "base_url",
|
||||
"apiMode": "api_mode",
|
||||
"keyEnv": "key_env",
|
||||
"apiKeyEnv": "key_env", # alias — OpenClaw-compatible + docs variant
|
||||
"defaultModel": "default_model",
|
||||
"contextLength": "context_length",
|
||||
"rateLimitDelay": "rate_limit_delay",
|
||||
}
|
||||
_KNOWN_PROVIDER_KEYS = {
|
||||
# ``provider`` duplicates the ``providers.<name>`` mapping key and is unused
|
||||
# here, but Hermes' own config writer has historically emitted it into
|
||||
# provider entries. Accept it silently so self-written configs don't warn.
|
||||
"provider",
|
||||
"name", "api", "url", "base_url", "api_key", "key_env", "api_key_env", "key_cmd",
|
||||
"api_mode", "transport", "model", "default_model", "models", "models_discovered",
|
||||
"context_length", "rate_limit_delay", "request_timeout_seconds", "stale_timeout_seconds",
|
||||
"discover_models", "extra_body", "extra_headers", "capabilities", "ssl_ca_cert", "ssl_verify",
|
||||
}
|
||||
|
||||
|
||||
def _pick_provider_base_url(entry: Dict[str, Any], provider_key: str) -> str:
|
||||
"""First usable URL among ``base_url``/``url``/``api``, or "".
|
||||
|
||||
URLs containing unresolved placeholder tokens — ``${ENV_VAR}`` env-refs and bare ``{region}``
|
||||
templates — are accepted without validation: they are expanded at runtime, and a caller reaching
|
||||
this normalizer with raw config would otherwise see the provider silently dropped.
|
||||
"""
|
||||
from urllib.parse import urlparse
|
||||
|
||||
for url_key in ("base_url", "url", "api"):
|
||||
raw_url = entry.get(url_key)
|
||||
if not (isinstance(raw_url, str) and raw_url.strip()):
|
||||
continue
|
||||
candidate = raw_url.strip()
|
||||
if re.search(r"\{[^}]+\}", candidate):
|
||||
return candidate
|
||||
parsed = urlparse(candidate)
|
||||
if parsed.scheme and parsed.netloc:
|
||||
return candidate
|
||||
logger.warning(
|
||||
"providers.%s: '%s' value '%s' is not a valid URL "
|
||||
"(no scheme or host) — skipped",
|
||||
provider_key or "?", url_key, candidate,
|
||||
)
|
||||
return ""
|
||||
|
||||
|
||||
def _normalize_provider_models(models: Any) -> Tuple[Dict[str, Any], bool]:
|
||||
"""Normalize an entry's ``models`` to the dict shape downstream expects.
|
||||
|
||||
Returns ``(models_dict, discovered_flag)``. Older Hermes versions wrote an in-mapping
|
||||
``__discovered_model_catalog__`` sentinel (accepted on read, stripped so sentinel keys never
|
||||
surface as model IDs). Hand-edited/older configs may write a plain list of ids or ``[{id: ...}]``
|
||||
rows; both are converted so /model doesn't show the provider with (0) models.
|
||||
"""
|
||||
discovered = False
|
||||
if isinstance(models, dict) and models:
|
||||
# Shallow-copy: `models` may alias a cached config sub-dict, and the
|
||||
# normalized entry escapes into long-lived runtime state.
|
||||
models_copy = dict(models)
|
||||
if models_copy.pop("__discovered_model_catalog__", None) is True:
|
||||
discovered = True
|
||||
models_copy.pop("__explicit_model_allowlist__", None)
|
||||
return models_copy, discovered
|
||||
if isinstance(models, list) and models:
|
||||
normalized_models: Dict[str, Any] = {}
|
||||
for item in models:
|
||||
if isinstance(item, str) and item.strip():
|
||||
normalized_models[item.strip()] = {}
|
||||
continue
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
model_id = item.get("id")
|
||||
if not isinstance(model_id, str) or not model_id.strip():
|
||||
model_id = item.get("name")
|
||||
if not isinstance(model_id, str) or not model_id.strip():
|
||||
continue
|
||||
normalized_models[model_id.strip()] = {
|
||||
k: v for k, v in item.items() if k not in {"id", "name"}
|
||||
}
|
||||
return normalized_models, discovered
|
||||
return {}, discovered
|
||||
|
||||
|
||||
def _normalize_custom_provider_entry(
|
||||
entry: Any,
|
||||
*,
|
||||
provider_key: str = "",
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""Return a runtime-compatible custom provider entry or ``None``."""
|
||||
if not isinstance(entry, dict):
|
||||
return None
|
||||
|
||||
# Shallow-copy before alias normalization writes into the entry: callers
|
||||
# pass live sub-dicts from load_config_readonly()'s shared cache, and
|
||||
# mutating those violates the cache's no-mutation contract and leaks alias
|
||||
# keys back into config.yaml on a later save_config(load_config()).
|
||||
entry = dict(entry)
|
||||
provider_key = coerce_provider_id(provider_key)
|
||||
|
||||
# api_key_env is a documented snake_case alias for key_env (see
|
||||
# website/docs/guides/azure-foundry.md). Normalize it up front so the
|
||||
# rest of the normalizer treats it as the canonical field.
|
||||
if "api_key_env" in entry and "key_env" not in entry:
|
||||
entry["key_env"] = entry["api_key_env"]
|
||||
for camel, snake in _CAMEL_ALIASES.items():
|
||||
if camel in entry and snake not in entry:
|
||||
_warn_once_per_provider(
|
||||
provider_key, f"camel:{camel}",
|
||||
"providers.%s: camelCase key '%s' auto-mapped to '%s' "
|
||||
"(use snake_case to avoid this warning)",
|
||||
provider_key or "?", camel, snake,
|
||||
)
|
||||
entry[snake] = entry[camel]
|
||||
unknown = set(entry.keys()) - _KNOWN_PROVIDER_KEYS - set(_CAMEL_ALIASES.keys())
|
||||
if unknown:
|
||||
_warn_once_per_provider(
|
||||
provider_key, "unknown:" + ",".join(sorted(unknown)),
|
||||
"providers.%s: unknown config keys ignored: %s",
|
||||
provider_key or "?", ", ".join(sorted(unknown)),
|
||||
)
|
||||
|
||||
base_url = _pick_provider_base_url(entry, provider_key)
|
||||
if not base_url:
|
||||
return None
|
||||
|
||||
name = coerce_provider_id(entry.get("name")) or provider_key
|
||||
if not name:
|
||||
return None
|
||||
|
||||
normalized: Dict[str, Any] = {
|
||||
"name": name,
|
||||
"base_url": base_url,
|
||||
}
|
||||
|
||||
provider_key = provider_key.strip()
|
||||
if provider_key:
|
||||
normalized["provider_key"] = provider_key
|
||||
|
||||
def _stripped(*keys: str) -> str:
|
||||
val = None
|
||||
for k in keys:
|
||||
val = entry.get(k)
|
||||
if val:
|
||||
break
|
||||
return val.strip() if isinstance(val, str) else ""
|
||||
|
||||
if _stripped("api_key"):
|
||||
normalized["api_key"] = _stripped("api_key")
|
||||
|
||||
key_env = _stripped("key_env", "api_key_env")
|
||||
if key_env:
|
||||
normalized["key_env"] = key_env
|
||||
if entry.get("api_key_env") and not entry.get("key_env"):
|
||||
normalized["api_key_env"] = key_env
|
||||
|
||||
api_mode = _stripped("api_mode", "transport")
|
||||
if api_mode:
|
||||
normalized["api_mode"] = _canonical_api_mode(api_mode)
|
||||
|
||||
model_name = _stripped("model", "default_model")
|
||||
if model_name:
|
||||
normalized["model"] = model_name
|
||||
|
||||
# Entry-level marker: the ``models`` mapping was auto-discovered by Hermes
|
||||
# (``_save_discovered_models_to_config``), not hand-curated.
|
||||
models_dict, discovered = _normalize_provider_models(entry.get("models"))
|
||||
if models_dict:
|
||||
normalized["models"] = models_dict
|
||||
if entry.get("models_discovered") is True or discovered:
|
||||
normalized["models_discovered"] = True
|
||||
|
||||
capabilities = entry.get("capabilities")
|
||||
if isinstance(capabilities, dict):
|
||||
normalized_capabilities = {
|
||||
key: value
|
||||
for key, value in capabilities.items()
|
||||
if isinstance(key, str) and isinstance(value, bool)
|
||||
}
|
||||
if normalized_capabilities:
|
||||
normalized["capabilities"] = normalized_capabilities
|
||||
|
||||
context_length = entry.get("context_length")
|
||||
if isinstance(context_length, int) and context_length > 0:
|
||||
normalized["context_length"] = context_length
|
||||
|
||||
rate_limit_delay = entry.get("rate_limit_delay")
|
||||
if isinstance(rate_limit_delay, (int, float)) and rate_limit_delay >= 0:
|
||||
normalized["rate_limit_delay"] = rate_limit_delay
|
||||
|
||||
if isinstance(entry.get("discover_models"), bool):
|
||||
normalized["discover_models"] = entry["discover_models"]
|
||||
|
||||
if isinstance(entry.get("extra_body"), dict):
|
||||
normalized["extra_body"] = dict(entry["extra_body"])
|
||||
|
||||
# Per-provider extra HTTP headers (proxies, gateways, custom auth).
|
||||
# Values may carry credentials (e.g. CF-Access-Client-Secret) — never
|
||||
# log them anywhere downstream.
|
||||
normalized_headers = normalize_extra_headers(entry.get("extra_headers"))
|
||||
if normalized_headers:
|
||||
normalized["extra_headers"] = normalized_headers
|
||||
|
||||
ssl_ca_cert = entry.get("ssl_ca_cert")
|
||||
if isinstance(ssl_ca_cert, str) and ssl_ca_cert.strip():
|
||||
normalized["ssl_ca_cert"] = ssl_ca_cert.strip()
|
||||
|
||||
ssl_verify = entry.get("ssl_verify")
|
||||
if isinstance(ssl_verify, bool):
|
||||
normalized["ssl_verify"] = ssl_verify
|
||||
elif isinstance(ssl_verify, str) and ssl_verify.strip():
|
||||
normalized["ssl_verify"] = ssl_verify.strip()
|
||||
|
||||
return normalized
|
||||
|
||||
|
||||
def _custom_provider_entry_to_provider_config(
|
||||
entry: Any,
|
||||
*,
|
||||
provider_key: str = "",
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""Translate a legacy custom provider entry to the v12 providers shape."""
|
||||
normalized = _normalize_custom_provider_entry(
|
||||
dict(entry) if isinstance(entry, dict) else entry,
|
||||
provider_key=provider_key,
|
||||
)
|
||||
if normalized is None:
|
||||
return None
|
||||
|
||||
provider_entry: Dict[str, Any] = {"api": normalized["base_url"]}
|
||||
|
||||
for field in (
|
||||
"name", "api_key", "key_env", "models", "models_discovered", "context_length",
|
||||
"rate_limit_delay", "discover_models", "extra_body", "extra_headers",
|
||||
"ssl_ca_cert", "ssl_verify",
|
||||
):
|
||||
if field in normalized:
|
||||
provider_entry[field] = normalized[field]
|
||||
|
||||
if "model" in normalized:
|
||||
provider_entry["default_model"] = normalized["model"]
|
||||
if "api_mode" in normalized:
|
||||
provider_entry["transport"] = normalized["api_mode"]
|
||||
|
||||
return provider_entry
|
||||
|
||||
|
||||
def providers_dict_to_custom_providers(providers_dict: Any) -> List[Dict[str, Any]]:
|
||||
"""Normalize ``providers`` config entries into the legacy custom-provider shape."""
|
||||
if not isinstance(providers_dict, dict):
|
||||
return []
|
||||
|
||||
custom_providers: List[Dict[str, Any]] = []
|
||||
for key, entry in providers_dict.items():
|
||||
if isinstance(entry, dict) and not is_provider_enabled(entry):
|
||||
continue
|
||||
normalized = _normalize_custom_provider_entry(
|
||||
entry, provider_key=coerce_provider_id(key)
|
||||
)
|
||||
if normalized is not None:
|
||||
custom_providers.append(normalized)
|
||||
|
||||
return custom_providers
|
||||
|
||||
|
||||
def get_compatible_custom_providers(
|
||||
config: Optional[Dict[str, Any]] = None,
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""Return a deduplicated custom-provider view across legacy and v12+ config.
|
||||
|
||||
``custom_providers`` remains the on-disk legacy format, while ``providers`` is the newer keyed
|
||||
schema. Runtime and picker flows still need a single list-shaped view, but we should not
|
||||
materialise that compatibility layer back into config.yaml because it duplicates entries in UIs.
|
||||
"""
|
||||
if config is None:
|
||||
config = load_config()
|
||||
|
||||
compatible: List[Dict[str, Any]] = []
|
||||
seen_provider_keys: set = set()
|
||||
seen_name_url_pairs: set = set()
|
||||
|
||||
def _append_if_new(entry: Optional[Dict[str, Any]]) -> None:
|
||||
if entry is None:
|
||||
return
|
||||
provider_key = str(entry.get("provider_key", "") or "").strip().lower()
|
||||
name = str(entry.get("name", "") or "").strip().lower()
|
||||
base_url = str(entry.get("base_url", "") or "").strip().rstrip("/").lower()
|
||||
model = str(entry.get("model", "") or "").strip().lower()
|
||||
pair = (name, base_url, model)
|
||||
|
||||
if provider_key and provider_key in seen_provider_keys:
|
||||
return
|
||||
if name and base_url and pair in seen_name_url_pairs:
|
||||
return
|
||||
|
||||
compatible.append(entry)
|
||||
if provider_key:
|
||||
seen_provider_keys.add(provider_key)
|
||||
if name and base_url:
|
||||
seen_name_url_pairs.add(pair)
|
||||
|
||||
custom_providers = config.get("custom_providers")
|
||||
if custom_providers is not None:
|
||||
if not isinstance(custom_providers, list):
|
||||
return []
|
||||
for entry in custom_providers:
|
||||
_append_if_new(_normalize_custom_provider_entry(entry))
|
||||
|
||||
for entry in providers_dict_to_custom_providers(config.get("providers")):
|
||||
_append_if_new(entry)
|
||||
|
||||
return compatible
|
||||
|
||||
|
||||
def _entries_for_route(
|
||||
base_url: str,
|
||||
custom_providers: Optional[List[Dict[str, Any]]],
|
||||
config: Optional[Dict[str, Any]],
|
||||
):
|
||||
"""Yield custom-provider entries whose normalized route identity equals *base_url*.
|
||||
|
||||
Loads ``get_compatible_custom_providers(config)`` when *custom_providers* is None (failures →
|
||||
no entries). Yields nothing for an empty *base_url* or non-list input.
|
||||
"""
|
||||
if custom_providers is None:
|
||||
try:
|
||||
custom_providers = get_compatible_custom_providers(config)
|
||||
except Exception:
|
||||
custom_providers = []
|
||||
if not base_url or not isinstance(custom_providers, list):
|
||||
return
|
||||
target_url = normalize_route_base_url(base_url)
|
||||
if not target_url:
|
||||
return
|
||||
for entry in custom_providers:
|
||||
if not isinstance(entry, dict):
|
||||
continue
|
||||
entry_url = normalize_route_base_url(entry.get("base_url"))
|
||||
if entry_url and entry_url == target_url:
|
||||
yield entry
|
||||
|
||||
|
||||
def _route_model_cfg(entry: Dict[str, Any], model: str) -> Optional[Dict[str, Any]]:
|
||||
"""Return ``entry.models[model]`` when both are mappings, else None."""
|
||||
models = entry.get("models")
|
||||
if not isinstance(models, dict):
|
||||
return None
|
||||
model_cfg = models.get(model)
|
||||
return model_cfg if isinstance(model_cfg, dict) else None
|
||||
|
||||
|
||||
def _coerce_ssl_verify(value: Any) -> Optional[bool]:
|
||||
if value is None:
|
||||
return None
|
||||
if isinstance(value, bool):
|
||||
return value
|
||||
if isinstance(value, str):
|
||||
lowered = value.strip().lower()
|
||||
if lowered in {"false", "0", "no", "off"}:
|
||||
return False
|
||||
if lowered in {"true", "1", "yes", "on"}:
|
||||
return True
|
||||
return None
|
||||
|
||||
|
||||
def get_custom_provider_tls_settings(
|
||||
base_url: str,
|
||||
custom_providers: Optional[List[Dict[str, Any]]] = None,
|
||||
config: Optional[Dict[str, Any]] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""Return TLS settings from a matching ``custom_providers`` / ``providers`` entry."""
|
||||
for entry in _entries_for_route(base_url, custom_providers, config):
|
||||
out: Dict[str, Any] = {}
|
||||
ca = entry.get("ssl_ca_cert")
|
||||
if isinstance(ca, str) and ca.strip():
|
||||
out["ssl_ca_cert"] = ca.strip()
|
||||
verify = _coerce_ssl_verify(entry.get("ssl_verify"))
|
||||
if verify is not None:
|
||||
out["ssl_verify"] = verify
|
||||
return out
|
||||
return {}
|
||||
|
||||
|
||||
def apply_custom_provider_tls_to_client_kwargs(
|
||||
client_kwargs: Dict[str, Any],
|
||||
base_url: str,
|
||||
custom_providers: Optional[List[Dict[str, Any]]] = None,
|
||||
config: Optional[Dict[str, Any]] = None,
|
||||
) -> None:
|
||||
"""Attach per-provider TLS knobs to OpenAI client kwargs when matched."""
|
||||
tls = get_custom_provider_tls_settings(base_url, custom_providers, config)
|
||||
if tls.get("ssl_ca_cert"):
|
||||
client_kwargs["ssl_ca_cert"] = tls["ssl_ca_cert"]
|
||||
if "ssl_verify" in tls:
|
||||
client_kwargs["ssl_verify"] = tls["ssl_verify"]
|
||||
|
||||
|
||||
def normalize_extra_headers(extra_headers: Any) -> Dict[str, str]:
|
||||
"""Normalize a raw ``extra_headers`` value into a ``dict[str, str]``.
|
||||
|
||||
SECURITY: header values routinely carry credentials (Cloudflare Access service tokens, proxy
|
||||
auth, custom bearer schemes). Callers must never log the returned values.
|
||||
"""
|
||||
if not isinstance(extra_headers, dict) or not extra_headers:
|
||||
return {}
|
||||
return {str(k): str(v) for k, v in extra_headers.items() if v is not None}
|
||||
|
||||
|
||||
def get_custom_provider_extra_headers(
|
||||
base_url: str,
|
||||
custom_providers: Optional[List[Dict[str, Any]]] = None,
|
||||
config: Optional[Dict[str, Any]] = None,
|
||||
) -> Dict[str, str]:
|
||||
"""Return ``extra_headers`` from a matching ``providers`` / ``custom_providers`` entry.
|
||||
|
||||
Matches the entry whose normalized route identity equals *base_url*, mirroring
|
||||
:func:`get_custom_provider_tls_settings`, and returns its ``extra_headers`` dict, or ``{}`` when
|
||||
no entry matches or declares none.
|
||||
|
||||
SECURITY: header values routinely carry credentials (Cloudflare Access service tokens, proxy
|
||||
auth, custom bearer schemes). Callers must never log the returned values.
|
||||
"""
|
||||
for entry in _entries_for_route(base_url, custom_providers, config):
|
||||
headers = normalize_extra_headers(entry.get("extra_headers"))
|
||||
if headers:
|
||||
return headers
|
||||
return {}
|
||||
|
||||
|
||||
def apply_custom_provider_extra_headers_to_client_kwargs(
|
||||
client_kwargs: Dict[str, Any],
|
||||
base_url: str,
|
||||
custom_providers: Optional[List[Dict[str, Any]]] = None,
|
||||
config: Optional[Dict[str, Any]] = None,
|
||||
) -> None:
|
||||
"""Merge per-provider ``extra_headers`` onto OpenAI client ``default_headers``.
|
||||
|
||||
Provider-specific headers win over SDK/provider defaults already in ``client_kwargs`` (they
|
||||
are the most specific level). No-op when base_url matches no provider entry or none declares
|
||||
headers. SECURITY: values may carry credentials -- never log them.
|
||||
"""
|
||||
extra_headers = get_custom_provider_extra_headers(base_url, custom_providers, config)
|
||||
if not extra_headers:
|
||||
return
|
||||
merged = dict(client_kwargs.get("default_headers") or {})
|
||||
merged.update(extra_headers)
|
||||
client_kwargs["default_headers"] = merged
|
||||
|
||||
|
||||
def get_custom_provider_context_length(
|
||||
model: str,
|
||||
base_url: str,
|
||||
custom_providers: Optional[List[Dict[str, Any]]] = None,
|
||||
config: Optional[Dict[str, Any]] = None,
|
||||
) -> Optional[int]:
|
||||
"""Look up a per-model ``context_length`` override from ``custom_providers``.
|
||||
|
||||
Matches any entry whose normalized route identity equals ``base_url`` and returns
|
||||
``custom_providers[i].models.<model>.context_length`` if present and valid. Returns ``None``
|
||||
when no override applies.
|
||||
"""
|
||||
if not model or not base_url:
|
||||
return None
|
||||
if custom_providers is None:
|
||||
try:
|
||||
custom_providers = get_compatible_custom_providers(config)
|
||||
except Exception:
|
||||
if config is None:
|
||||
return None
|
||||
raw = config.get("custom_providers")
|
||||
custom_providers = raw if isinstance(raw, list) else []
|
||||
|
||||
for entry in _entries_for_route(base_url, custom_providers, config):
|
||||
model_cfg = _route_model_cfg(entry, model)
|
||||
if model_cfg is None:
|
||||
continue
|
||||
raw_ctx = model_cfg.get("context_length")
|
||||
if raw_ctx is None:
|
||||
continue
|
||||
try:
|
||||
ctx = int(raw_ctx)
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
if ctx > 0:
|
||||
return ctx
|
||||
return None
|
||||
|
||||
|
||||
def get_custom_provider_model_capability(
|
||||
model: str,
|
||||
base_url: str,
|
||||
capability: str,
|
||||
custom_providers: Optional[List[Dict[str, Any]]] = None,
|
||||
config: Optional[Dict[str, Any]] = None,
|
||||
) -> Optional[bool]:
|
||||
"""Return an explicit boolean capability for one custom-provider model.
|
||||
|
||||
Matching is scoped to the normalized route and exact runtime model id so aliases can declare
|
||||
capabilities without changing the id sent upstream. Missing or non-boolean declarations return
|
||||
``None``.
|
||||
"""
|
||||
if not model or not base_url or not capability:
|
||||
return None
|
||||
if custom_providers is None:
|
||||
try:
|
||||
if config is None:
|
||||
# Read-only path: this helper never mutates the entries it
|
||||
# scans, and get_compatible_custom_providers shallow-copies
|
||||
# each entry before normalizing, so the no-deepcopy cache is
|
||||
# safe here (~135us saved per call on the blank-stub paths).
|
||||
config = load_config_readonly()
|
||||
custom_providers = get_compatible_custom_providers(config)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
for entry in _entries_for_route(base_url, custom_providers, config):
|
||||
model_cfg = _route_model_cfg(entry, model)
|
||||
if model_cfg is None:
|
||||
continue
|
||||
value = model_cfg.get(capability)
|
||||
if isinstance(value, bool):
|
||||
return value
|
||||
return None
|
||||
|
||||
|
||||
def _coerce_config_version(value: Any) -> int:
|
||||
"""Return a safe integer config version, treating invalid values as legacy."""
|
||||
if isinstance(value, bool):
|
||||
@@ -2900,26 +2302,6 @@ def _normalize_max_turns_config(config: Dict[str, Any]) -> Dict[str, Any]:
|
||||
return config
|
||||
|
||||
|
||||
def is_provider_enabled(provider_cfg: Optional[Dict[str, Any]]) -> bool:
|
||||
"""Return whether a ``providers.<name>`` config block is enabled.
|
||||
|
||||
A provider is enabled by default. Only an explicit ``enabled: false`` in the block hides it from
|
||||
the model picker, ``/models`` listings, the runtime resolver and the doctor / status output.
|
||||
|
||||
Backward-compat: configs without the ``enabled`` key keep working as before — the default is
|
||||
``True``.
|
||||
"""
|
||||
if not isinstance(provider_cfg, dict):
|
||||
return True
|
||||
flag = provider_cfg.get("enabled", True)
|
||||
if isinstance(flag, bool):
|
||||
return flag
|
||||
# YAML can produce strings for "true"/"false" depending on quoting.
|
||||
if isinstance(flag, str):
|
||||
return flag.strip().lower() not in {"false", "0", "no", "off"}
|
||||
return bool(flag)
|
||||
|
||||
|
||||
# Sentinel for an unlimited turn budget. ``sys.maxsize`` survives the str→int
|
||||
# round-trip through the HERMES_MAX_ITERATIONS env bridge (gateway/run.py), works
|
||||
# in every ``<``/``>=``/``max - used`` comparison in agent/iteration_budget.py and
|
||||
|
||||
@@ -0,0 +1,668 @@
|
||||
"""Custom-provider entry normalization and per-route lookups (TLS, headers, context length, capabilities).
|
||||
|
||||
Split out of ``hermes_cli/config.py``; every name is re-imported there, so
|
||||
``hermes_cli.config.<name>`` keeps resolving (and monkeypatching) as before.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import re
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
from hermes_cli.route_identity import normalize_route_base_url
|
||||
|
||||
# Log-record parity with the origin module.
|
||||
logger = logging.getLogger("hermes_cli.config")
|
||||
|
||||
|
||||
# ``_normalize_custom_provider_entry`` runs on every ``load_picker_context()``
|
||||
# call (i.e. per interactive picker/inventory request), so any warning it emits
|
||||
# fires repeatedly for the same static config. Deduplicate per (provider,
|
||||
# signature): on Windows a repeated-warning storm contends on
|
||||
# ``concurrent-log-handler``'s cross-process rotation lock and can peg a core /
|
||||
# stall the gateway/serve event loop. The cache lives for the process lifetime.
|
||||
_PROVIDER_NORMALIZE_WARNED: set = set()
|
||||
|
||||
|
||||
def _warn_once_per_provider(
|
||||
provider_key: str, signature: str, msg: str, *args: Any
|
||||
) -> None:
|
||||
"""Emit ``logger.warning(msg, *args)`` at most once per (provider, signature)."""
|
||||
dedup_key = (provider_key or "?", signature)
|
||||
if dedup_key in _PROVIDER_NORMALIZE_WARNED:
|
||||
return
|
||||
_PROVIDER_NORMALIZE_WARNED.add(dedup_key)
|
||||
logger.warning(msg, *args)
|
||||
|
||||
|
||||
_API_MODE_ALIASES = {
|
||||
# Values accepted by earlier releases (and natural spellings) mapped to
|
||||
# the canonical transport names consumed by agent_init. Before this map
|
||||
# existed, an unrecognized api_mode was silently ignored and the
|
||||
# transport fell through to hostname-based guessing, so a config that
|
||||
# said ``api_mode: openai`` (valid on older releases) could flip to
|
||||
# ``codex_responses`` after an update and break the provider (#66543
|
||||
# discussion; observed live against api.actual.inc).
|
||||
"openai": "chat_completions",
|
||||
"openai_chat": "chat_completions",
|
||||
"openai-chat": "chat_completions",
|
||||
"chat-completions": "chat_completions",
|
||||
"chatcompletions": "chat_completions",
|
||||
"responses": "codex_responses",
|
||||
"openai_responses": "codex_responses",
|
||||
"openai-responses": "codex_responses",
|
||||
"anthropic": "anthropic_messages",
|
||||
"anthropic-messages": "anthropic_messages",
|
||||
"messages": "anthropic_messages",
|
||||
"bedrock": "bedrock_converse",
|
||||
"bedrock-converse": "bedrock_converse",
|
||||
}
|
||||
|
||||
|
||||
def _canonical_api_mode(api_mode: str) -> str:
|
||||
"""Map legacy/alias ``api_mode`` spellings to canonical transport names.
|
||||
|
||||
Unknown values pass through unchanged (callers keep their existing fall-through behavior); known
|
||||
aliases are rewritten so downstream consumers (``agent_init``'s accepted-set check, runtime
|
||||
resolution) see a canonical name instead of silently discarding the user's intent.
|
||||
"""
|
||||
cleaned = api_mode.strip()
|
||||
return _API_MODE_ALIASES.get(cleaned.lower(), cleaned)
|
||||
|
||||
|
||||
def coerce_provider_id(value: Any) -> str:
|
||||
"""Provider identity fields are strings."""
|
||||
if value is None:
|
||||
return ""
|
||||
return str(value).strip()
|
||||
|
||||
|
||||
def stringify_provider_map(providers: Any) -> dict:
|
||||
"""Copy a ``providers:`` mapping so keys are strings.
|
||||
|
||||
Desktop Custom Endpoints store the name as the dict key; an unquoted YAML key ``2070:`` loads
|
||||
as int, so picker code calling ``ep_name.lower()`` crashes and CRUD lookups of ``"2070"`` miss.
|
||||
"""
|
||||
if not isinstance(providers, dict):
|
||||
return {}
|
||||
out: Dict[str, Any] = {}
|
||||
for stored, value in providers.items():
|
||||
key = coerce_provider_id(stored)
|
||||
if key:
|
||||
out[key] = value
|
||||
return out
|
||||
|
||||
|
||||
def find_provider_entry(providers: Any, key: Any) -> Tuple[Any, Optional[Dict[str, Any]]]:
|
||||
"""Return ``(stored_key, entry)`` matching *key* by string identity.
|
||||
|
||||
Prefer an exact string hit, then scan.
|
||||
"""
|
||||
if not isinstance(providers, dict):
|
||||
return None, None
|
||||
want = coerce_provider_id(key)
|
||||
if not want:
|
||||
return None, None
|
||||
exact = providers.get(want)
|
||||
if isinstance(exact, dict):
|
||||
return want, exact
|
||||
for stored, entry in providers.items():
|
||||
if coerce_provider_id(stored) == want and isinstance(entry, dict):
|
||||
return stored, entry
|
||||
return None, None
|
||||
|
||||
|
||||
# camelCase aliases commonly used in hand-written provider configs.
|
||||
_CAMEL_ALIASES: Dict[str, str] = {
|
||||
"apiKey": "api_key",
|
||||
"baseUrl": "base_url",
|
||||
"apiMode": "api_mode",
|
||||
"keyEnv": "key_env",
|
||||
"apiKeyEnv": "key_env", # alias — OpenClaw-compatible + docs variant
|
||||
"defaultModel": "default_model",
|
||||
"contextLength": "context_length",
|
||||
"rateLimitDelay": "rate_limit_delay",
|
||||
}
|
||||
|
||||
|
||||
_KNOWN_PROVIDER_KEYS = {
|
||||
# ``provider`` duplicates the ``providers.<name>`` mapping key and is unused
|
||||
# here, but Hermes' own config writer has historically emitted it into
|
||||
# provider entries. Accept it silently so self-written configs don't warn.
|
||||
"provider",
|
||||
"name", "api", "url", "base_url", "api_key", "key_env", "api_key_env", "key_cmd",
|
||||
"api_mode", "transport", "model", "default_model", "models", "models_discovered",
|
||||
"context_length", "rate_limit_delay", "request_timeout_seconds", "stale_timeout_seconds",
|
||||
"discover_models", "extra_body", "extra_headers", "capabilities", "ssl_ca_cert", "ssl_verify",
|
||||
}
|
||||
|
||||
|
||||
def _pick_provider_base_url(entry: Dict[str, Any], provider_key: str) -> str:
|
||||
"""First usable URL among ``base_url``/``url``/``api``, or "".
|
||||
|
||||
URLs containing unresolved placeholder tokens — ``${ENV_VAR}`` env-refs and bare ``{region}``
|
||||
templates — are accepted without validation: they are expanded at runtime, and a caller reaching
|
||||
this normalizer with raw config would otherwise see the provider silently dropped.
|
||||
"""
|
||||
from urllib.parse import urlparse
|
||||
|
||||
for url_key in ("base_url", "url", "api"):
|
||||
raw_url = entry.get(url_key)
|
||||
if not (isinstance(raw_url, str) and raw_url.strip()):
|
||||
continue
|
||||
candidate = raw_url.strip()
|
||||
if re.search(r"\{[^}]+\}", candidate):
|
||||
return candidate
|
||||
parsed = urlparse(candidate)
|
||||
if parsed.scheme and parsed.netloc:
|
||||
return candidate
|
||||
logger.warning(
|
||||
"providers.%s: '%s' value '%s' is not a valid URL "
|
||||
"(no scheme or host) — skipped",
|
||||
provider_key or "?", url_key, candidate,
|
||||
)
|
||||
return ""
|
||||
|
||||
|
||||
def _normalize_provider_models(models: Any) -> Tuple[Dict[str, Any], bool]:
|
||||
"""Normalize an entry's ``models`` to the dict shape downstream expects.
|
||||
|
||||
Returns ``(models_dict, discovered_flag)``. Older Hermes versions wrote an in-mapping
|
||||
``__discovered_model_catalog__`` sentinel (accepted on read, stripped so sentinel keys never
|
||||
surface as model IDs). Hand-edited/older configs may write a plain list of ids or ``[{id: ...}]``
|
||||
rows; both are converted so /model doesn't show the provider with (0) models.
|
||||
"""
|
||||
discovered = False
|
||||
if isinstance(models, dict) and models:
|
||||
# Shallow-copy: `models` may alias a cached config sub-dict, and the
|
||||
# normalized entry escapes into long-lived runtime state.
|
||||
models_copy = dict(models)
|
||||
if models_copy.pop("__discovered_model_catalog__", None) is True:
|
||||
discovered = True
|
||||
models_copy.pop("__explicit_model_allowlist__", None)
|
||||
return models_copy, discovered
|
||||
if isinstance(models, list) and models:
|
||||
normalized_models: Dict[str, Any] = {}
|
||||
for item in models:
|
||||
if isinstance(item, str) and item.strip():
|
||||
normalized_models[item.strip()] = {}
|
||||
continue
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
model_id = item.get("id")
|
||||
if not isinstance(model_id, str) or not model_id.strip():
|
||||
model_id = item.get("name")
|
||||
if not isinstance(model_id, str) or not model_id.strip():
|
||||
continue
|
||||
normalized_models[model_id.strip()] = {
|
||||
k: v for k, v in item.items() if k not in {"id", "name"}
|
||||
}
|
||||
return normalized_models, discovered
|
||||
return {}, discovered
|
||||
|
||||
|
||||
def _normalize_custom_provider_entry(
|
||||
entry: Any,
|
||||
*,
|
||||
provider_key: str = "",
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""Return a runtime-compatible custom provider entry or ``None``."""
|
||||
from hermes_cli.config import normalize_extra_headers
|
||||
if not isinstance(entry, dict):
|
||||
return None
|
||||
|
||||
# Shallow-copy before alias normalization writes into the entry: callers
|
||||
# pass live sub-dicts from load_config_readonly()'s shared cache, and
|
||||
# mutating those violates the cache's no-mutation contract and leaks alias
|
||||
# keys back into config.yaml on a later save_config(load_config()).
|
||||
entry = dict(entry)
|
||||
provider_key = coerce_provider_id(provider_key)
|
||||
|
||||
# api_key_env is a documented snake_case alias for key_env (see
|
||||
# website/docs/guides/azure-foundry.md). Normalize it up front so the
|
||||
# rest of the normalizer treats it as the canonical field.
|
||||
if "api_key_env" in entry and "key_env" not in entry:
|
||||
entry["key_env"] = entry["api_key_env"]
|
||||
for camel, snake in _CAMEL_ALIASES.items():
|
||||
if camel in entry and snake not in entry:
|
||||
_warn_once_per_provider(
|
||||
provider_key, f"camel:{camel}",
|
||||
"providers.%s: camelCase key '%s' auto-mapped to '%s' "
|
||||
"(use snake_case to avoid this warning)",
|
||||
provider_key or "?", camel, snake,
|
||||
)
|
||||
entry[snake] = entry[camel]
|
||||
unknown = set(entry.keys()) - _KNOWN_PROVIDER_KEYS - set(_CAMEL_ALIASES.keys())
|
||||
if unknown:
|
||||
_warn_once_per_provider(
|
||||
provider_key, "unknown:" + ",".join(sorted(unknown)),
|
||||
"providers.%s: unknown config keys ignored: %s",
|
||||
provider_key or "?", ", ".join(sorted(unknown)),
|
||||
)
|
||||
|
||||
base_url = _pick_provider_base_url(entry, provider_key)
|
||||
if not base_url:
|
||||
return None
|
||||
|
||||
name = coerce_provider_id(entry.get("name")) or provider_key
|
||||
if not name:
|
||||
return None
|
||||
|
||||
normalized: Dict[str, Any] = {
|
||||
"name": name,
|
||||
"base_url": base_url,
|
||||
}
|
||||
|
||||
provider_key = provider_key.strip()
|
||||
if provider_key:
|
||||
normalized["provider_key"] = provider_key
|
||||
|
||||
def _stripped(*keys: str) -> str:
|
||||
val = None
|
||||
for k in keys:
|
||||
val = entry.get(k)
|
||||
if val:
|
||||
break
|
||||
return val.strip() if isinstance(val, str) else ""
|
||||
|
||||
if _stripped("api_key"):
|
||||
normalized["api_key"] = _stripped("api_key")
|
||||
|
||||
key_env = _stripped("key_env", "api_key_env")
|
||||
if key_env:
|
||||
normalized["key_env"] = key_env
|
||||
if entry.get("api_key_env") and not entry.get("key_env"):
|
||||
normalized["api_key_env"] = key_env
|
||||
|
||||
api_mode = _stripped("api_mode", "transport")
|
||||
if api_mode:
|
||||
normalized["api_mode"] = _canonical_api_mode(api_mode)
|
||||
|
||||
model_name = _stripped("model", "default_model")
|
||||
if model_name:
|
||||
normalized["model"] = model_name
|
||||
|
||||
# Entry-level marker: the ``models`` mapping was auto-discovered by Hermes
|
||||
# (``_save_discovered_models_to_config``), not hand-curated.
|
||||
models_dict, discovered = _normalize_provider_models(entry.get("models"))
|
||||
if models_dict:
|
||||
normalized["models"] = models_dict
|
||||
if entry.get("models_discovered") is True or discovered:
|
||||
normalized["models_discovered"] = True
|
||||
|
||||
capabilities = entry.get("capabilities")
|
||||
if isinstance(capabilities, dict):
|
||||
normalized_capabilities = {
|
||||
key: value
|
||||
for key, value in capabilities.items()
|
||||
if isinstance(key, str) and isinstance(value, bool)
|
||||
}
|
||||
if normalized_capabilities:
|
||||
normalized["capabilities"] = normalized_capabilities
|
||||
|
||||
context_length = entry.get("context_length")
|
||||
if isinstance(context_length, int) and context_length > 0:
|
||||
normalized["context_length"] = context_length
|
||||
|
||||
rate_limit_delay = entry.get("rate_limit_delay")
|
||||
if isinstance(rate_limit_delay, (int, float)) and rate_limit_delay >= 0:
|
||||
normalized["rate_limit_delay"] = rate_limit_delay
|
||||
|
||||
if isinstance(entry.get("discover_models"), bool):
|
||||
normalized["discover_models"] = entry["discover_models"]
|
||||
|
||||
if isinstance(entry.get("extra_body"), dict):
|
||||
normalized["extra_body"] = dict(entry["extra_body"])
|
||||
|
||||
# Per-provider extra HTTP headers (proxies, gateways, custom auth).
|
||||
# Values may carry credentials (e.g. CF-Access-Client-Secret) — never
|
||||
# log them anywhere downstream.
|
||||
normalized_headers = normalize_extra_headers(entry.get("extra_headers"))
|
||||
if normalized_headers:
|
||||
normalized["extra_headers"] = normalized_headers
|
||||
|
||||
ssl_ca_cert = entry.get("ssl_ca_cert")
|
||||
if isinstance(ssl_ca_cert, str) and ssl_ca_cert.strip():
|
||||
normalized["ssl_ca_cert"] = ssl_ca_cert.strip()
|
||||
|
||||
ssl_verify = entry.get("ssl_verify")
|
||||
if isinstance(ssl_verify, bool):
|
||||
normalized["ssl_verify"] = ssl_verify
|
||||
elif isinstance(ssl_verify, str) and ssl_verify.strip():
|
||||
normalized["ssl_verify"] = ssl_verify.strip()
|
||||
|
||||
return normalized
|
||||
|
||||
|
||||
def _custom_provider_entry_to_provider_config(
|
||||
entry: Any,
|
||||
*,
|
||||
provider_key: str = "",
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""Translate a legacy custom provider entry to the v12 providers shape."""
|
||||
from hermes_cli.config import _normalize_custom_provider_entry
|
||||
normalized = _normalize_custom_provider_entry(
|
||||
dict(entry) if isinstance(entry, dict) else entry,
|
||||
provider_key=provider_key,
|
||||
)
|
||||
if normalized is None:
|
||||
return None
|
||||
|
||||
provider_entry: Dict[str, Any] = {"api": normalized["base_url"]}
|
||||
|
||||
for field in (
|
||||
"name", "api_key", "key_env", "models", "models_discovered", "context_length",
|
||||
"rate_limit_delay", "discover_models", "extra_body", "extra_headers",
|
||||
"ssl_ca_cert", "ssl_verify",
|
||||
):
|
||||
if field in normalized:
|
||||
provider_entry[field] = normalized[field]
|
||||
|
||||
if "model" in normalized:
|
||||
provider_entry["default_model"] = normalized["model"]
|
||||
if "api_mode" in normalized:
|
||||
provider_entry["transport"] = normalized["api_mode"]
|
||||
|
||||
return provider_entry
|
||||
|
||||
|
||||
def providers_dict_to_custom_providers(providers_dict: Any) -> List[Dict[str, Any]]:
|
||||
"""Normalize ``providers`` config entries into the legacy custom-provider shape."""
|
||||
from hermes_cli.config import _normalize_custom_provider_entry
|
||||
if not isinstance(providers_dict, dict):
|
||||
return []
|
||||
|
||||
custom_providers: List[Dict[str, Any]] = []
|
||||
for key, entry in providers_dict.items():
|
||||
if isinstance(entry, dict) and not is_provider_enabled(entry):
|
||||
continue
|
||||
normalized = _normalize_custom_provider_entry(
|
||||
entry, provider_key=coerce_provider_id(key)
|
||||
)
|
||||
if normalized is not None:
|
||||
custom_providers.append(normalized)
|
||||
|
||||
return custom_providers
|
||||
|
||||
|
||||
def get_compatible_custom_providers(
|
||||
config: Optional[Dict[str, Any]] = None,
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""Return a deduplicated custom-provider view across legacy and v12+ config.
|
||||
|
||||
``custom_providers`` remains the on-disk legacy format, while ``providers`` is the newer keyed
|
||||
schema. Runtime and picker flows still need a single list-shaped view, but we should not
|
||||
materialise that compatibility layer back into config.yaml because it duplicates entries in UIs.
|
||||
"""
|
||||
from hermes_cli.config import _normalize_custom_provider_entry, load_config
|
||||
if config is None:
|
||||
config = load_config()
|
||||
|
||||
compatible: List[Dict[str, Any]] = []
|
||||
seen_provider_keys: set = set()
|
||||
seen_name_url_pairs: set = set()
|
||||
|
||||
def _append_if_new(entry: Optional[Dict[str, Any]]) -> None:
|
||||
if entry is None:
|
||||
return
|
||||
provider_key = str(entry.get("provider_key", "") or "").strip().lower()
|
||||
name = str(entry.get("name", "") or "").strip().lower()
|
||||
base_url = str(entry.get("base_url", "") or "").strip().rstrip("/").lower()
|
||||
model = str(entry.get("model", "") or "").strip().lower()
|
||||
pair = (name, base_url, model)
|
||||
|
||||
if provider_key and provider_key in seen_provider_keys:
|
||||
return
|
||||
if name and base_url and pair in seen_name_url_pairs:
|
||||
return
|
||||
|
||||
compatible.append(entry)
|
||||
if provider_key:
|
||||
seen_provider_keys.add(provider_key)
|
||||
if name and base_url:
|
||||
seen_name_url_pairs.add(pair)
|
||||
|
||||
custom_providers = config.get("custom_providers")
|
||||
if custom_providers is not None:
|
||||
if not isinstance(custom_providers, list):
|
||||
return []
|
||||
for entry in custom_providers:
|
||||
_append_if_new(_normalize_custom_provider_entry(entry))
|
||||
|
||||
for entry in providers_dict_to_custom_providers(config.get("providers")):
|
||||
_append_if_new(entry)
|
||||
|
||||
return compatible
|
||||
|
||||
|
||||
def _entries_for_route(
|
||||
base_url: str,
|
||||
custom_providers: Optional[List[Dict[str, Any]]],
|
||||
config: Optional[Dict[str, Any]],
|
||||
):
|
||||
"""Yield custom-provider entries whose normalized route identity equals *base_url*.
|
||||
|
||||
Loads ``get_compatible_custom_providers(config)`` when *custom_providers* is None (failures →
|
||||
no entries). Yields nothing for an empty *base_url* or non-list input.
|
||||
"""
|
||||
from hermes_cli.config import get_compatible_custom_providers
|
||||
if custom_providers is None:
|
||||
try:
|
||||
custom_providers = get_compatible_custom_providers(config)
|
||||
except Exception:
|
||||
custom_providers = []
|
||||
if not base_url or not isinstance(custom_providers, list):
|
||||
return
|
||||
target_url = normalize_route_base_url(base_url)
|
||||
if not target_url:
|
||||
return
|
||||
for entry in custom_providers:
|
||||
if not isinstance(entry, dict):
|
||||
continue
|
||||
entry_url = normalize_route_base_url(entry.get("base_url"))
|
||||
if entry_url and entry_url == target_url:
|
||||
yield entry
|
||||
|
||||
|
||||
def _route_model_cfg(entry: Dict[str, Any], model: str) -> Optional[Dict[str, Any]]:
|
||||
"""Return ``entry.models[model]`` when both are mappings, else None."""
|
||||
models = entry.get("models")
|
||||
if not isinstance(models, dict):
|
||||
return None
|
||||
model_cfg = models.get(model)
|
||||
return model_cfg if isinstance(model_cfg, dict) else None
|
||||
|
||||
|
||||
def _coerce_ssl_verify(value: Any) -> Optional[bool]:
|
||||
if value is None:
|
||||
return None
|
||||
if isinstance(value, bool):
|
||||
return value
|
||||
if isinstance(value, str):
|
||||
lowered = value.strip().lower()
|
||||
if lowered in {"false", "0", "no", "off"}:
|
||||
return False
|
||||
if lowered in {"true", "1", "yes", "on"}:
|
||||
return True
|
||||
return None
|
||||
|
||||
|
||||
def get_custom_provider_tls_settings(
|
||||
base_url: str,
|
||||
custom_providers: Optional[List[Dict[str, Any]]] = None,
|
||||
config: Optional[Dict[str, Any]] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""Return TLS settings from a matching ``custom_providers`` / ``providers`` entry."""
|
||||
for entry in _entries_for_route(base_url, custom_providers, config):
|
||||
out: Dict[str, Any] = {}
|
||||
ca = entry.get("ssl_ca_cert")
|
||||
if isinstance(ca, str) and ca.strip():
|
||||
out["ssl_ca_cert"] = ca.strip()
|
||||
verify = _coerce_ssl_verify(entry.get("ssl_verify"))
|
||||
if verify is not None:
|
||||
out["ssl_verify"] = verify
|
||||
return out
|
||||
return {}
|
||||
|
||||
|
||||
def apply_custom_provider_tls_to_client_kwargs(
|
||||
client_kwargs: Dict[str, Any],
|
||||
base_url: str,
|
||||
custom_providers: Optional[List[Dict[str, Any]]] = None,
|
||||
config: Optional[Dict[str, Any]] = None,
|
||||
) -> None:
|
||||
"""Attach per-provider TLS knobs to OpenAI client kwargs when matched."""
|
||||
tls = get_custom_provider_tls_settings(base_url, custom_providers, config)
|
||||
if tls.get("ssl_ca_cert"):
|
||||
client_kwargs["ssl_ca_cert"] = tls["ssl_ca_cert"]
|
||||
if "ssl_verify" in tls:
|
||||
client_kwargs["ssl_verify"] = tls["ssl_verify"]
|
||||
|
||||
|
||||
def normalize_extra_headers(extra_headers: Any) -> Dict[str, str]:
|
||||
"""Normalize a raw ``extra_headers`` value into a ``dict[str, str]``.
|
||||
|
||||
SECURITY: header values routinely carry credentials (Cloudflare Access service tokens, proxy
|
||||
auth, custom bearer schemes). Callers must never log the returned values.
|
||||
"""
|
||||
if not isinstance(extra_headers, dict) or not extra_headers:
|
||||
return {}
|
||||
return {str(k): str(v) for k, v in extra_headers.items() if v is not None}
|
||||
|
||||
|
||||
def get_custom_provider_extra_headers(
|
||||
base_url: str,
|
||||
custom_providers: Optional[List[Dict[str, Any]]] = None,
|
||||
config: Optional[Dict[str, Any]] = None,
|
||||
) -> Dict[str, str]:
|
||||
"""Return ``extra_headers`` from a matching ``providers`` / ``custom_providers`` entry.
|
||||
|
||||
Matches the entry whose normalized route identity equals *base_url*, mirroring
|
||||
:func:`get_custom_provider_tls_settings`, and returns its ``extra_headers`` dict, or ``{}`` when
|
||||
no entry matches or declares none.
|
||||
|
||||
SECURITY: header values routinely carry credentials (Cloudflare Access service tokens, proxy
|
||||
auth, custom bearer schemes). Callers must never log the returned values.
|
||||
"""
|
||||
from hermes_cli.config import normalize_extra_headers
|
||||
for entry in _entries_for_route(base_url, custom_providers, config):
|
||||
headers = normalize_extra_headers(entry.get("extra_headers"))
|
||||
if headers:
|
||||
return headers
|
||||
return {}
|
||||
|
||||
|
||||
def apply_custom_provider_extra_headers_to_client_kwargs(
|
||||
client_kwargs: Dict[str, Any],
|
||||
base_url: str,
|
||||
custom_providers: Optional[List[Dict[str, Any]]] = None,
|
||||
config: Optional[Dict[str, Any]] = None,
|
||||
) -> None:
|
||||
"""Merge per-provider ``extra_headers`` onto OpenAI client ``default_headers``.
|
||||
|
||||
Provider-specific headers win over SDK/provider defaults already in ``client_kwargs`` (they
|
||||
are the most specific level). No-op when base_url matches no provider entry or none declares
|
||||
headers. SECURITY: values may carry credentials -- never log them.
|
||||
"""
|
||||
extra_headers = get_custom_provider_extra_headers(base_url, custom_providers, config)
|
||||
if not extra_headers:
|
||||
return
|
||||
merged = dict(client_kwargs.get("default_headers") or {})
|
||||
merged.update(extra_headers)
|
||||
client_kwargs["default_headers"] = merged
|
||||
|
||||
|
||||
def get_custom_provider_context_length(
|
||||
model: str,
|
||||
base_url: str,
|
||||
custom_providers: Optional[List[Dict[str, Any]]] = None,
|
||||
config: Optional[Dict[str, Any]] = None,
|
||||
) -> Optional[int]:
|
||||
"""Look up a per-model ``context_length`` override from ``custom_providers``.
|
||||
|
||||
Matches any entry whose normalized route identity equals ``base_url`` and returns
|
||||
``custom_providers[i].models.<model>.context_length`` if present and valid. Returns ``None``
|
||||
when no override applies.
|
||||
"""
|
||||
from hermes_cli.config import get_compatible_custom_providers
|
||||
if not model or not base_url:
|
||||
return None
|
||||
if custom_providers is None:
|
||||
try:
|
||||
custom_providers = get_compatible_custom_providers(config)
|
||||
except Exception:
|
||||
if config is None:
|
||||
return None
|
||||
raw = config.get("custom_providers")
|
||||
custom_providers = raw if isinstance(raw, list) else []
|
||||
|
||||
for entry in _entries_for_route(base_url, custom_providers, config):
|
||||
model_cfg = _route_model_cfg(entry, model)
|
||||
if model_cfg is None:
|
||||
continue
|
||||
raw_ctx = model_cfg.get("context_length")
|
||||
if raw_ctx is None:
|
||||
continue
|
||||
try:
|
||||
ctx = int(raw_ctx)
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
if ctx > 0:
|
||||
return ctx
|
||||
return None
|
||||
|
||||
|
||||
def get_custom_provider_model_capability(
|
||||
model: str,
|
||||
base_url: str,
|
||||
capability: str,
|
||||
custom_providers: Optional[List[Dict[str, Any]]] = None,
|
||||
config: Optional[Dict[str, Any]] = None,
|
||||
) -> Optional[bool]:
|
||||
"""Return an explicit boolean capability for one custom-provider model.
|
||||
|
||||
Matching is scoped to the normalized route and exact runtime model id so aliases can declare
|
||||
capabilities without changing the id sent upstream. Missing or non-boolean declarations return
|
||||
``None``.
|
||||
"""
|
||||
from hermes_cli.config import get_compatible_custom_providers, load_config_readonly
|
||||
if not model or not base_url or not capability:
|
||||
return None
|
||||
if custom_providers is None:
|
||||
try:
|
||||
if config is None:
|
||||
# Read-only path: this helper never mutates the entries it
|
||||
# scans, and get_compatible_custom_providers shallow-copies
|
||||
# each entry before normalizing, so the no-deepcopy cache is
|
||||
# safe here (~135us saved per call on the blank-stub paths).
|
||||
config = load_config_readonly()
|
||||
custom_providers = get_compatible_custom_providers(config)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
for entry in _entries_for_route(base_url, custom_providers, config):
|
||||
model_cfg = _route_model_cfg(entry, model)
|
||||
if model_cfg is None:
|
||||
continue
|
||||
value = model_cfg.get(capability)
|
||||
if isinstance(value, bool):
|
||||
return value
|
||||
return None
|
||||
|
||||
|
||||
def is_provider_enabled(provider_cfg: Optional[Dict[str, Any]]) -> bool:
|
||||
"""Return whether a ``providers.<name>`` config block is enabled.
|
||||
|
||||
A provider is enabled by default. Only an explicit ``enabled: false`` in the block hides it from
|
||||
the model picker, ``/models`` listings, the runtime resolver and the doctor / status output.
|
||||
|
||||
Backward-compat: configs without the ``enabled`` key keep working as before — the default is
|
||||
``True``.
|
||||
"""
|
||||
if not isinstance(provider_cfg, dict):
|
||||
return True
|
||||
flag = provider_cfg.get("enabled", True)
|
||||
if isinstance(flag, bool):
|
||||
return flag
|
||||
# YAML can produce strings for "true"/"false" depending on quoting.
|
||||
if isinstance(flag, str):
|
||||
return flag.strip().lower() not in {"false", "0", "no", "off"}
|
||||
return bool(flag)
|
||||
Reference in New Issue
Block a user