#!/usr/bin/env python3 """Discover account-visible provider models without leaking credentials.""" from __future__ import annotations import json import os import select import shutil import subprocess import time from typing import Any from urllib.error import HTTPError, URLError from urllib.parse import urlencode from urllib.request import Request, urlopen from provider_model_catalog import Catalog, model_records, unique_models def codex_cli_authenticated() -> bool: """Return whether the installed Codex CLI has a usable local login.""" codex = shutil.which("codex") if not codex: return False try: status = subprocess.run( [codex, "login", "status"], capture_output=True, check=False, text=True, timeout=10, ) except (OSError, subprocess.SubprocessError): return False detail = f"{status.stdout}\n{status.stderr}".lower() return status.returncode == 0 and ( "logged in" in detail or "authenticated" in detail ) def _codex_app_server_records(codex: str) -> tuple[list[str], dict[str, dict[str, Any]]] | None: """Read the authenticated app-server ``model/list`` response when available.""" try: process = subprocess.Popen( [codex, "app-server"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=False, ) except OSError: return None if process.stdin is None or process.stdout is None: process.terminate() return None buffer = b"" def send(message: dict[str, Any]) -> None: assert process.stdin is not None process.stdin.write(json.dumps(message, separators=(",", ":")).encode() + b"\n") process.stdin.flush() def response(request_id: int, deadline: float) -> dict[str, Any] | None: nonlocal buffer assert process.stdout is not None while time.monotonic() < deadline: readable, _, _ = select.select([process.stdout.fileno()], [], [], 0.25) if not readable: continue chunk = os.read(process.stdout.fileno(), 65536) if not chunk: return None buffer += chunk while b"\n" in buffer: line, buffer = buffer.split(b"\n", 1) try: document = json.loads(line) except (TypeError, ValueError, json.JSONDecodeError): continue if document.get("id") == request_id and isinstance(document.get("result"), dict): return document["result"] return None try: send({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"clientInfo": {"name": "hermes-catalog", "version": "1"}, "capabilities": {}}}) if response(1, time.monotonic() + 12) is None: return None send({"jsonrpc": "2.0", "method": "initialized", "params": {}}) values: list[Any] = [] cursor: str | None = None request_id = 2 while True: params: dict[str, Any] = {"includeHidden": False, "limit": 100} if cursor: params["cursor"] = cursor send({"jsonrpc": "2.0", "id": request_id, "method": "model/list", "params": params}) result = response(request_id, time.monotonic() + 12) if result is None: return None page = result.get("models", result.get("data", [])) if not isinstance(page, list): return None values.extend(page) cursor = result.get("nextCursor") or result.get("next_cursor") if not isinstance(cursor, str) or not cursor: return model_records(values) request_id += 1 except (OSError, ValueError, json.JSONDecodeError): pass finally: process.terminate() try: process.wait(timeout=2) except subprocess.TimeoutExpired: process.kill() return None def _anthropic_api_records() -> tuple[list[str], dict[str, dict[str, Any]]] | None: """List current Anthropic account models, including documented capabilities.""" token = os.environ.get("ANTHROPIC_API_KEY", "").strip() if not token: return None request = Request( "https://api.anthropic.com/v1/models?" + urlencode({"limit": 100}), headers={"x-api-key": token, "anthropic-version": "2023-06-01"}, ) try: with urlopen(request, timeout=15) as response: document = json.load(response) except (HTTPError, URLError, OSError, TimeoutError, ValueError): return None values = document.get("data", []) if isinstance(document, dict) else [] return model_records(values if isinstance(values, list) else []) def discover_codex_models() -> Catalog: """Use the authenticated Codex endpoint; catalogs are status-only fallback.""" token = "" try: from hermes_cli.auth import resolve_codex_runtime_credentials credentials = resolve_codex_runtime_credentials(refresh_if_expiring=True) or {} token = str(credentials.get("api_key") or "").strip() except Exception: token = "" api_records: tuple[list[str], dict[str, dict[str, Any]]] | None = None if token: try: from hermes_cli.codex_models import _fetch_models_from_api api_records = model_records(_fetch_models_from_api(token)) except Exception: api_records = None if api_records is not None: live, live_metadata = api_records return Catalog("openai-codex", live, True, True, "connected-api", live_metadata) codex = shutil.which("codex") if codex: app_records = _codex_app_server_records(codex) if app_records is not None: app_models, app_metadata = app_records return Catalog("openai-codex", app_models, True, True, "connected-app-server", app_metadata) try: from hermes_cli.models import provider_model_ids known, known_metadata = model_records( provider_model_ids("openai-codex", force_refresh=True) ) except Exception: known, known_metadata = [], {} # The owner agent deliberately uses the local Codex app-server and the # Codex CLI's ChatGPT login. That credential is not exported into Hermes' # bearer-token store, so API discovery above may be empty even though the # runtime is fully authenticated. Treat a successful CLI status check as a # connected, degraded catalog: routing may use Codex while retaining the # last-known working model names until app-server discovery is available. if codex_cli_authenticated(): return Catalog( "openai-codex", known, False, True, "connected-cli", known_metadata, ) state = "degraded" if token else "not-configured" return Catalog("openai-codex", known, False, bool(token), state, known_metadata) def discover_claude_models() -> Catalog: """Discover subscription models first, retaining API discovery as fallback.""" try: from claude_model_discovery import discover_claude_subscription_models subscription = discover_claude_subscription_models() except Exception: subscription = None if subscription is not None and subscription.live: records = [model.as_dict() for model in subscription.models] models, metadata = model_records(records) return Catalog( "anthropic", models, True, subscription.connected, subscription.state, metadata, ) api_records = _anthropic_api_records() if api_records is not None: live_models, live_metadata = api_records return Catalog("anthropic", live_models, True, True, "connected-api", live_metadata) try: from hermes_cli.models import provider_model_ids known, known_metadata = model_records( provider_model_ids("anthropic", force_refresh=True) ) except Exception: known, known_metadata = [], {} if subscription is not None: return Catalog( "anthropic", known, False, subscription.connected, subscription.state, known_metadata, ) return Catalog("anthropic", known, False, False, "not-configured", known_metadata)