"""Helpers for model-provider plugins (``plugins/model-providers/<name>/``).The Hermes agent discovers these via ``providers.list_providers()`` and exposesthem in the CLI model picker. WebUI must mirror that registry instead ofrelying only on the static ``_PROVIDER_DISPLAY`` / ``_PROVIDER_MODELS`` tables.Bundled agent profiles (gemini, nous, custom, …) also live in``list_providers()``. WebUI already handles those via static tables anddedicated code paths — only *plugin-only* slugs (e.g. user-installed yandex)should take the plugin discovery path."""from __future__ import annotationsimport loggingimport threadingfrom typing import Anylogger = logging.getLogger(__name__)_PROFILES_LOCK = threading.Lock()_PROFILES_BY_NAME: dict[str, Any] | None = None_WEBUI_STATIC_PROVIDER_IDS: frozenset[str] | None = Nonedef _webui_static_provider_ids() -> frozenset[str]: """Provider slugs already owned by WebUI static tables / special cases.""" global _WEBUI_STATIC_PROVIDER_IDS if _WEBUI_STATIC_PROVIDER_IDS is not None: return _WEBUI_STATIC_PROVIDER_IDS try: from api.config import _PROVIDER_DISPLAY, _PROVIDER_MODELS static = ( frozenset(_PROVIDER_DISPLAY.keys()) | frozenset(_PROVIDER_MODELS.keys()) | frozenset({"custom"}) ) except Exception: static = frozenset({"custom"}) _WEBUI_STATIC_PROVIDER_IDS = static return staticdef _load_profiles_by_name() -> dict[str, Any]: try: from providers import list_providers except Exception: logger.debug("providers package unavailable for plugin discovery", exc_info=True) return {} result: dict[str, Any] = {} try: for profile in list_providers(): name = str(getattr(profile, "name", "") or "").strip().lower() if name: result[name] = profile except Exception: logger.debug("Failed to enumerate model-provider plugins", exc_info=True) return {} return resultdef plugin_model_provider_profiles() -> dict[str, Any]: """Return registered model-provider profiles keyed by canonical slug.""" global _PROFILES_BY_NAME cached = _PROFILES_BY_NAME if cached is not None: return cached with _PROFILES_LOCK: if _PROFILES_BY_NAME is None: _PROFILES_BY_NAME = _load_profiles_by_name() return _PROFILES_BY_NAMEdef invalidate_plugin_model_provider_cache() -> None: """Clear cached plugin discovery (e.g. after config reload).""" global _PROFILES_BY_NAME with _PROFILES_LOCK: _PROFILES_BY_NAME = Nonedef plugin_model_provider_ids() -> frozenset[str]: """Slugs from ``list_providers()`` that are not already WebUI-static.""" static = _webui_static_provider_ids() return frozenset( pid for pid in plugin_model_provider_profiles().keys() if pid not in static )def plugin_model_provider_display_name(provider_id: str) -> str | None: profile = plugin_model_provider_profiles().get((provider_id or "").strip().lower()) if profile is None: return None return str(getattr(profile, "display_name", "") or getattr(profile, "name", "") or "").strip() or Nonedef plugin_model_provider_api_key_env_var(provider_id: str) -> str | None: """Return the primary API-key env var for a plugin provider, if any.""" profile = plugin_model_provider_profiles().get((provider_id or "").strip().lower()) if profile is None: return None env_vars = getattr(profile, "env_vars", ()) or () for var in env_vars: upper = str(var).upper() if upper.endswith("_BASE_URL") or upper.endswith("_URL"): continue if upper.endswith("_FOLDER_ID"): continue return str(var) return Nonedef effective_provider_env_var(provider_id: str, static_map: dict[str, str]) -> str | None: pid = (provider_id or "").strip().lower() if not pid: return None if pid in static_map: return static_map[pid] if not is_plugin_model_provider(pid): return None return plugin_model_provider_api_key_env_var(pid)def effective_provider_display_name(provider_id: str, static_map: dict[str, str]) -> str: pid = (provider_id or "").strip().lower() if pid in static_map: return static_map[pid] if is_plugin_model_provider(pid): plugin_name = plugin_model_provider_display_name(pid) if plugin_name: return plugin_name return pid.replace("-", " ").title()def is_plugin_model_provider(provider_id: str) -> bool: """True for plugin-only providers (not already in WebUI static tables).""" pid = (provider_id or "").strip().lower() if not pid or pid in _webui_static_provider_ids(): return False return pid in plugin_model_provider_profiles()
"""Helpers for model-provider plugins (``plugins/model-providers/<name>/``).
The Hermes agent discovers these via ``providers.list_providers()`` and exposes
them in the CLI model picker. WebUI must mirror that registry instead of
relying only on the static ``_PROVIDER_DISPLAY`` / ``_PROVIDER_MODELS`` tables.
Bundled agent profiles (gemini, nous, custom, …) also live in
``list_providers()``. WebUI already handles those via static tables and
dedicated code paths — only *plugin-only* slugs (e.g. user-installed yandex)
should take the plugin discovery path.
"""
from __future__ import annotations
import logging
import threading
from typing import Any
logger = logging.getLogger(__name__)
_PROFILES_LOCK = threading.Lock()
_PROFILES_BY_NAME: dict[str, Any] | None = None
_WEBUI_STATIC_PROVIDER_IDS: frozenset[str] | None = None
def _webui_static_provider_ids() -> frozenset[str]:
"""Provider slugs already owned by WebUI static tables / special cases."""
global _WEBUI_STATIC_PROVIDER_IDS
if _WEBUI_STATIC_PROVIDER_IDS is not None:
return _WEBUI_STATIC_PROVIDER_IDS
try:
from api.config import _PROVIDER_DISPLAY, _PROVIDER_MODELS
static = (
frozenset(_PROVIDER_DISPLAY.keys())
| frozenset(_PROVIDER_MODELS.keys())
| frozenset({"custom"})
)
except Exception:
static = frozenset({"custom"})
_WEBUI_STATIC_PROVIDER_IDS = static
return static
def _load_profiles_by_name() -> dict[str, Any]:
try:
from providers import list_providers
except Exception:
logger.debug("providers package unavailable for plugin discovery", exc_info=True)
return {}
result: dict[str, Any] = {}
try:
for profile in list_providers():
name = str(getattr(profile, "name", "") or "").strip().lower()
if name:
result[name] = profile
except Exception:
logger.debug("Failed to enumerate model-provider plugins", exc_info=True)
return {}
return result
def plugin_model_provider_profiles() -> dict[str, Any]:
"""Return registered model-provider profiles keyed by canonical slug."""
global _PROFILES_BY_NAME
cached = _PROFILES_BY_NAME
if cached is not None:
return cached
with _PROFILES_LOCK:
if _PROFILES_BY_NAME is None:
_PROFILES_BY_NAME = _load_profiles_by_name()
return _PROFILES_BY_NAME
def invalidate_plugin_model_provider_cache() -> None:
"""Clear cached plugin discovery (e.g. after config reload)."""
global _PROFILES_BY_NAME
with _PROFILES_LOCK:
_PROFILES_BY_NAME = None
def plugin_model_provider_ids() -> frozenset[str]:
"""Slugs from ``list_providers()`` that are not already WebUI-static."""
static = _webui_static_provider_ids()
return frozenset(
pid for pid in plugin_model_provider_profiles().keys() if pid not in static
)
def plugin_model_provider_display_name(provider_id: str) -> str | None:
profile = plugin_model_provider_profiles().get((provider_id or "").strip().lower())
if profile is None:
return None
return str(getattr(profile, "display_name", "") or getattr(profile, "name", "") or "").strip() or None
def plugin_model_provider_api_key_env_var(provider_id: str) -> str | None:
"""Return the primary API-key env var for a plugin provider, if any."""
profile = plugin_model_provider_profiles().get((provider_id or "").strip().lower())
if profile is None:
return None
env_vars = getattr(profile, "env_vars", ()) or ()
for var in env_vars:
upper = str(var).upper()
if upper.endswith("_BASE_URL") or upper.endswith("_URL"):
continue
if upper.endswith("_FOLDER_ID"):
continue
return str(var)
return None
def effective_provider_env_var(provider_id: str, static_map: dict[str, str]) -> str | None:
pid = (provider_id or "").strip().lower()
if not pid:
return None
if pid in static_map:
return static_map[pid]
if not is_plugin_model_provider(pid):
return None
return plugin_model_provider_api_key_env_var(pid)
def effective_provider_display_name(provider_id: str, static_map: dict[str, str]) -> str:
pid = (provider_id or "").strip().lower()
if pid in static_map:
return static_map[pid]
if is_plugin_model_provider(pid):
plugin_name = plugin_model_provider_display_name(pid)
if plugin_name:
return plugin_name
return pid.replace("-", " ").title()
def is_plugin_model_provider(provider_id: str) -> bool:
"""True for plugin-only providers (not already in WebUI static tables)."""
pid = (provider_id or "").strip().lower()
if not pid or pid in _webui_static_provider_ids():
return False
return pid in plugin_model_provider_profiles()