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

refactor(doctor): extract API connectivity probes to doctor_connectivity.py; unify result rows

Browse Source
This commit is contained in:
Teknium
2026-09-02 15:47:50 -07:00
parent 5b475406d4
commit 29d24bfbc0
2 changed files with 410 additions and 549 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/doctor.py
+16 -549
View File
@@ -33,10 +33,13 @@ _env_path = get_env_path()
load_hermes_dotenv(hermes_home=_env_path.parent, project_env=PROJECT_ROOT / ".env")
from hermes_cli.colors import Colors, color
from hermes_cli.models import _HERMES_USER_AGENT
from hermes_cli.doctor_connectivity import ( # noqa: F401 (re-exported; tests import from hermes_cli.doctor)
_build_apikey_providers_list,
_has_healthy_oauth_fallback_for_apikey_provider,
build_probes,
run_probes,
)
from hermes_cli.vercel_auth import describe_vercel_auth
from hermes_constants import OPENROUTER_MODELS_URL
from utils import base_url_host_matches
_PROVIDER_ENV_HINTS = (
@@ -369,30 +372,6 @@ def _apply_doctor_tool_availability_overrides(available: list[str], unavailable:
return updated_available, updated_unavailable
def _has_healthy_oauth_fallback_for_apikey_provider(provider_label: str) -> bool:
"""Return True when a direct API-key probe failure is non-blocking.
Some provider families support both a direct API-key path and a separate
OAuth runtime path. When the OAuth path is already healthy, doctor should
still show a failed API-key connectivity row, but it should not promote
that direct-key problem into the final blocking summary.
"""
normalized = (provider_label or "").strip().lower()
if normalized == "minimax":
try:
from hermes_cli.auth import get_minimax_oauth_auth_status
return bool((get_minimax_oauth_auth_status() or {}).get("logged_in"))
except Exception:
return False
if normalized == "xai":
try:
from hermes_cli.auth import get_xai_oauth_auth_status
return bool((get_xai_oauth_auth_status() or {}).get("logged_in"))
except Exception:
return False
return False
def check_ok(text: str, detail: str = ""):
print(f" {color('✓', Colors.GREEN)} {text}" + (f" {color(detail, Colors.DIM)}" if detail else ""))
@@ -952,102 +931,6 @@ def _check_gateway_service_linger(issues: list[str]) -> None:
check_warn("Could not verify systemd linger", f"({linger_detail})")
_APIKEY_PROVIDERS_CACHE: list | None = None
def _build_apikey_providers_list() -> list:
"""Build the API-key provider health-check list once and cache it.
Tuple format: (name, env_vars, default_url, base_env, supports_models_endpoint)
Base list augmented with any ProviderProfile with auth_type="api_key" not
already present — adding plugins/model-providers/<name>/ is sufficient to get into doctor.
"""
_static = [
("Z.AI / GLM", ("GLM_API_KEY", "ZAI_API_KEY", "Z_AI_API_KEY"), "https://api.z.ai/api/paas/v4/models", "GLM_BASE_URL", True),
("Kimi / Moonshot", ("KIMI_API_KEY",), "https://api.moonshot.ai/v1/models", "KIMI_BASE_URL", True),
("StepFun Step Plan", ("STEPFUN_API_KEY",), "https://api.stepfun.ai/step_plan/v1/models", "STEPFUN_BASE_URL", True),
("Kimi / Moonshot (China)", ("KIMI_CN_API_KEY",), "https://api.moonshot.cn/v1/models", None, True),
("Arcee AI", ("ARCEEAI_API_KEY",), "https://api.arcee.ai/api/v1/models", "ARCEE_BASE_URL", True),
("GMI Cloud", ("GMI_API_KEY",), "https://api.gmi-serving.com/v1/models", "GMI_BASE_URL", True),
("DeepSeek", ("DEEPSEEK_API_KEY",), "https://api.deepseek.com/v1/models", "DEEPSEEK_BASE_URL", True),
("Hugging Face", ("HF_TOKEN",), "https://router.huggingface.co/v1/models", "HF_BASE_URL", True),
("NVIDIA NIM", ("NVIDIA_API_KEY",), "https://integrate.api.nvidia.com/v1/models", "NVIDIA_BASE_URL", True),
("Alibaba/DashScope", ("DASHSCOPE_API_KEY",), "https://dashscope-intl.aliyuncs.com/compatible-mode/v1/models", "DASHSCOPE_BASE_URL", True),
# MiniMax global: /v1 endpoint supports /models.
("MiniMax", ("MINIMAX_API_KEY",), "https://api.minimax.io/v1/models", "MINIMAX_BASE_URL", True),
# MiniMax CN: /v1 endpoint does NOT support /models (returns 404).
("MiniMax (China)", ("MINIMAX_CN_API_KEY",), "https://api.minimaxi.com/v1/models", "MINIMAX_CN_BASE_URL", False),
("Vercel AI Gateway", ("AI_GATEWAY_API_KEY",), "https://ai-gateway.vercel.sh/v1/models", "AI_GATEWAY_BASE_URL", True),
("Kilo Code", ("KILOCODE_API_KEY",), "https://api.kilo.ai/api/gateway/models", "KILOCODE_BASE_URL", True),
("OpenCode Zen", ("OPENCODE_ZEN_API_KEY",), "https://opencode.ai/zen/v1/models", "OPENCODE_ZEN_BASE_URL", True),
# OpenCode Go has no shared /models endpoint; skip the health check.
("OpenCode Go", ("OPENCODE_GO_API_KEY",), None, "OPENCODE_GO_BASE_URL", False),
]
_known_names = {t[0] for t in _static}
# Also index by profile canonical name so profiles without display_name
# don't create duplicate entries for providers already in the static list.
_known_canonical: set[str] = set()
_name_to_canonical = {
"Z.AI / GLM": "zai", "Kimi / Moonshot": "kimi-coding",
"StepFun Step Plan": "stepfun", "Kimi / Moonshot (China)": "kimi-coding-cn",
"Arcee AI": "arcee", "GMI Cloud": "gmi", "DeepSeek": "deepseek",
"Hugging Face": "huggingface", "NVIDIA NIM": "nvidia",
"Alibaba/DashScope": "alibaba", "MiniMax": "minimax",
"MiniMax (China)": "minimax-cn", "Vercel AI Gateway": "ai-gateway",
"Kilo Code": "kilocode", "OpenCode Zen": "opencode-zen",
"OpenCode Go": "opencode-go",
}
for _label, _canonical in _name_to_canonical.items():
_known_canonical.add(_canonical)
# Providers that already have a dedicated health check above the generic
# API-key loop (with custom headers/auth). Skip their pluggable profiles
# here so the generic Bearer-auth loop doesn't run a duplicate, broken
# check (e.g. Anthropic native API requires x-api-key, not Bearer).
_dedicated_canonical = {"anthropic", "openrouter", "bedrock"}
_known_canonical.update(_dedicated_canonical)
try:
from providers import list_providers
from providers.base import ProviderProfile as _PP
try:
from hermes_cli.providers import normalize_provider as _normalize_provider
except Exception: # pragma: no cover - normalization is best-effort
def _normalize_provider(_name: str) -> str:
return (_name or "").strip().lower()
for _pp in list_providers():
if not isinstance(_pp, _PP) or _pp.auth_type != "api_key" or not _pp.env_vars:
continue
_label = _pp.display_name or _pp.name
if _label in _known_names or _pp.name in _known_canonical:
continue
_candidates = {_normalize_provider(_pp.name)}
for _alias in (_pp.aliases or ()):
_candidates.add(_normalize_provider(_alias))
if _candidates & _dedicated_canonical:
continue
# Separate API-key vars from base-URL override vars — the health-check
# loop sends the first found value as Authorization: Bearer, so a URL
# string must never be picked.
_key_vars = tuple(
v for v in _pp.env_vars
if not v.endswith("_BASE_URL") and not v.endswith("_URL")
)
_base_var = next(
(v for v in _pp.env_vars if v.endswith("_BASE_URL") or v.endswith("_URL")),
None,
)
if not _key_vars:
continue
_models_url = (
(_pp.models_url or (_pp.base_url.rstrip("/") + "/models"))
if _pp.base_url else None
)
_hc = getattr(_pp, "supports_health_check", True)
_static.append((_label, _key_vars, _models_url, _base_var, _hc))
except Exception:
pass
return _static
def managed_scope_check() -> None:
"""Report the active managed scope (resolved dir + pinned key counts).
@@ -2822,434 +2705,18 @@ def _check_npm_audit(should_fix: bool) -> Finding:
def _check_api_connectivity(should_fix: bool) -> Finding:
"""Parallel HTTP/SDK probes for every configured provider; results printed in submission order."""
f = Finding()
issues = f.issues
# Refactor: every connectivity probe below is HTTP-bound and fully
# independent. Running them in series spent ~5s wall on a typical
# workstation (2s of that was boto3's IMDS lookup for AWS credentials,
# which times out unless you're actually on EC2). Threading them with
# a small executor pool collapses the section to roughly the slowest
# single probe — about 2s — without changing the output format.
#
# Each ``_probe_*`` helper is a pure function: takes its inputs,
# makes one HTTP/SDK call, returns a ``_ConnectivityResult`` carrying
# the line(s) to print and any issue strings to append. No globals,
# no shared mutable state, no printing inside the workers.
import concurrent.futures as _futures
from collections import namedtuple as _namedtuple
_ConnectivityResult = _namedtuple(
"_ConnectivityResult", ["label", "lines", "issues"]
)
_probes: list = [] # list of (label, callable) submitted in display order
def _probe_openrouter() -> _ConnectivityResult:
key = os.getenv("OPENROUTER_API_KEY")
if not key:
return _ConnectivityResult(
"OpenRouter API",
[(color("⚠", Colors.YELLOW), "OpenRouter API",
color("(not configured)", Colors.DIM))],
[],
)
try:
import httpx
r = httpx.get(
OPENROUTER_MODELS_URL,
headers={"Authorization": f"Bearer {key}"},
timeout=10,
)
if r.status_code == 200:
return _ConnectivityResult(
"OpenRouter API",
[(color("✓", Colors.GREEN), "OpenRouter API", "")],
[],
)
if r.status_code == 401:
return _ConnectivityResult(
"OpenRouter API",
[(color("✗", Colors.RED), "OpenRouter API",
color("(invalid API key)", Colors.DIM))],
["Check OPENROUTER_API_KEY in .env"],
)
if r.status_code == 402:
return _ConnectivityResult(
"OpenRouter API",
[(color("✗", Colors.RED), "OpenRouter API",
color("(out of credits — payment required)", Colors.DIM))],
["OpenRouter account has insufficient credits. "
"Fix: run 'hermes config set model.provider <provider>' "
"to switch providers, or fund your OpenRouter account "
"at https://openrouter.ai/settings/credits"],
)
if r.status_code == 429:
return _ConnectivityResult(
"OpenRouter API",
[(color("✗", Colors.RED), "OpenRouter API",
color("(rate limited)", Colors.DIM))],
["OpenRouter rate limit hit — consider switching to "
"a different provider or waiting"],
)
return _ConnectivityResult(
"OpenRouter API",
[(color("✗", Colors.RED), "OpenRouter API",
color(f"(HTTP {r.status_code})", Colors.DIM))],
[],
)
except Exception as e:
return _ConnectivityResult(
"OpenRouter API",
[(color("✗", Colors.RED), "OpenRouter API",
color(f"({e})", Colors.DIM))],
["Check network connectivity"],
)
def _probe_anthropic() -> _ConnectivityResult:
from hermes_cli.auth import get_anthropic_key
key = get_anthropic_key()
if not key:
return _ConnectivityResult("Anthropic API", [], [])
try:
import httpx
from agent.anthropic_adapter import (
_is_oauth_token,
_COMMON_BETAS,
_OAUTH_ONLY_BETAS,
_CONTEXT_1M_BETA,
)
headers = {"anthropic-version": "2023-06-01"}
is_oauth = _is_oauth_token(key)
if is_oauth:
headers["Authorization"] = f"Bearer {key}"
headers["anthropic-beta"] = ",".join(_COMMON_BETAS + _OAUTH_ONLY_BETAS)
else:
headers["x-api-key"] = key
r = httpx.get(
"https://api.anthropic.com/v1/models",
headers=headers, timeout=10,
)
# Reactive recovery: OAuth subscriptions without 1M context reject the
# request with 400 "long context beta is not yet available for this
# subscription". Retry once with that beta stripped so the doctor
# check doesn't falsely report Anthropic as unreachable.
if (
is_oauth
and r.status_code == 400
and "long context beta" in r.text.lower()
and "not yet available" in r.text.lower()
):
headers["anthropic-beta"] = ",".join(
[b for b in _COMMON_BETAS if b != _CONTEXT_1M_BETA]
+ list(_OAUTH_ONLY_BETAS)
)
r = httpx.get(
"https://api.anthropic.com/v1/models",
headers=headers, timeout=10,
)
if r.status_code == 200:
return _ConnectivityResult(
"Anthropic API",
[(color("✓", Colors.GREEN), "Anthropic API", "")],
[],
)
if r.status_code == 401:
return _ConnectivityResult(
"Anthropic API",
[(color("✗", Colors.RED), "Anthropic API",
color("(invalid API key)", Colors.DIM))],
[],
)
return _ConnectivityResult(
"Anthropic API",
[(color("⚠", Colors.YELLOW), "Anthropic API",
color("(couldn't verify)", Colors.DIM))],
[],
)
except Exception as e:
return _ConnectivityResult(
"Anthropic API",
[(color("⚠", Colors.YELLOW), "Anthropic API",
color(f"({e})", Colors.DIM))],
[],
)
def _probe_apikey_provider(pname, env_vars, default_url, base_env,
supports_health_check) -> _ConnectivityResult:
key = ""
for ev in env_vars:
key = os.getenv(ev, "")
if key:
break
if not key:
return _ConnectivityResult(pname, [], [])
label = pname.ljust(20)
if not supports_health_check:
return _ConnectivityResult(
pname,
[(color("✓", Colors.GREEN), label,
color("(key configured)", Colors.DIM))],
[],
)
try:
import httpx
base = os.getenv(base_env, "") if base_env else ""
# Auto-detect Kimi Code keys (sk-kimi-) → api.kimi.com/coding/v1
# (OpenAI-compat surface, which exposes /models for health check).
if not base and key.startswith("sk-kimi-"):
base = "https://api.kimi.com/coding/v1"
# Anthropic-compat endpoints (/anthropic, api.kimi.com/coding
# with no /v1) don't support /models. Rewrite to OpenAI-compat
# /v1 surface for health checks.
if base and base.rstrip("/").endswith("/anthropic"):
from agent.auxiliary_client import _to_openai_base_url
base = _to_openai_base_url(base)
if base_url_host_matches(base, "api.kimi.com") and base.rstrip("/").endswith("/coding"):
base = base.rstrip("/") + "/v1"
url = (base.rstrip("/") + "/models") if base else default_url
headers = {
"Authorization": f"Bearer {key}",
"User-Agent": _HERMES_USER_AGENT,
}
if base_url_host_matches(base, "api.kimi.com"):
headers["User-Agent"] = "claude-code/0.1.0"
# Google's Generative Language API (generativelanguage.googleapis.com)
# rejects ``Authorization: Bearer <api-key>`` with 401
# ``ACCESS_TOKEN_TYPE_UNSUPPORTED`` — that header is reserved for
# OAuth 2 access tokens, not plain API keys. Plain keys use
# ``x-goog-api-key`` (or ``?key=``). Without this, a perfectly valid
# GOOGLE_API_KEY/GEMINI_API_KEY always shows red in ``hermes doctor``.
if url and base_url_host_matches(url, "generativelanguage.googleapis.com"):
headers.pop("Authorization", None)
headers["x-goog-api-key"] = key
r = httpx.get(url, headers=headers, timeout=10)
if (
pname == "Alibaba/DashScope"
and not base
and r.status_code == 401
):
r = httpx.get(
"https://dashscope.aliyuncs.com/compatible-mode/v1/models",
headers=headers, timeout=10,
)
if r.status_code == 200:
return _ConnectivityResult(
pname,
[(color("✓", Colors.GREEN), label, "")],
[],
)
if r.status_code == 401:
return _ConnectivityResult(
pname,
[(color("✗", Colors.RED), label,
color("(invalid API key)", Colors.DIM))],
[f"Check {env_vars[0]} in .env"],
)
return _ConnectivityResult(
pname,
[(color("⚠", Colors.YELLOW), label,
color(f"(HTTP {r.status_code})", Colors.DIM))],
[],
)
except Exception as e:
return _ConnectivityResult(
pname,
[(color("⚠", Colors.YELLOW), label,
color(f"({e})", Colors.DIM))],
[],
)
def _probe_bedrock() -> _ConnectivityResult:
try:
from agent.bedrock_adapter import (
has_aws_credentials,
resolve_aws_auth_env_var,
resolve_bedrock_region,
)
except ImportError:
return _ConnectivityResult("AWS Bedrock", [], [])
if not has_aws_credentials():
return _ConnectivityResult("AWS Bedrock", [], [])
auth_var = resolve_aws_auth_env_var()
region = resolve_bedrock_region()
label = "AWS Bedrock".ljust(20)
try:
import boto3
from botocore.config import Config as _BotoConfig
# Trim retries on the actual Bedrock API call so a transient
# failure doesn't pad the doctor run by 30+ seconds.
cfg = _BotoConfig(
connect_timeout=5,
read_timeout=10,
retries={"max_attempts": 1},
)
client = boto3.client("bedrock", region_name=region, config=cfg)
resp = client.list_foundation_models()
n = len(resp.get("modelSummaries", []))
return _ConnectivityResult(
"AWS Bedrock",
[(color("✓", Colors.GREEN), label,
color(f"({auth_var}, {region}, {n} models)", Colors.DIM))],
[],
)
except ImportError:
return _ConnectivityResult(
"AWS Bedrock",
[(color("⚠", Colors.YELLOW), label,
color(f"(boto3 not installed — {sys.executable} -m pip install boto3)",
Colors.DIM))],
[f"Install boto3 for Bedrock: {sys.executable} -m pip install boto3"],
)
except Exception as e:
err_name = type(e).__name__
return _ConnectivityResult(
"AWS Bedrock",
[(color("⚠", Colors.YELLOW), label,
color(f"({err_name}: {e})", Colors.DIM))],
[f"AWS Bedrock: {err_name} — check IAM permissions for "
f"bedrock:ListFoundationModels"],
)
def _probe_azure_entra() -> _ConnectivityResult:
"""Probe Azure Foundry Entra ID auth, parallel to ``_probe_bedrock``.
Skipped unless the active config has ``model.provider:
azure-foundry`` AND ``model.auth_mode: entra_id`` — we don't probe
the token-service / CLI chain for users on plain API-key Azure.
Bounded by a 10s timeout (via
:func:`agent.azure_identity_adapter.describe_active_credential`)
so a slow token service can't pad the doctor run.
"""
label = "Azure Foundry (Entra ID)".ljust(28)
try:
from hermes_cli.config import load_config
cfg = load_config()
model_cfg = cfg.get("model") if isinstance(cfg, dict) else {}
if not isinstance(model_cfg, dict):
return _ConnectivityResult("Azure Foundry (Entra ID)", [], [])
cfg_provider = str(model_cfg.get("provider") or "").strip().lower()
auth_mode = str(model_cfg.get("auth_mode") or "").strip().lower()
if cfg_provider != "azure-foundry" or auth_mode != "entra_id":
return _ConnectivityResult("Azure Foundry (Entra ID)", [], [])
except Exception:
return _ConnectivityResult("Azure Foundry (Entra ID)", [], [])
try:
from agent.azure_identity_adapter import (
EntraIdentityConfig,
SCOPE_AI_AZURE_DEFAULT,
describe_active_credential,
has_azure_identity_installed,
)
except Exception as exc:
return _ConnectivityResult(
"Azure Foundry (Entra ID)",
[(color("⚠", Colors.YELLOW), label,
color(f"(adapter import failed: {exc})", Colors.DIM))],
[f"Azure Foundry adapter import failed: {exc}"],
)
if not has_azure_identity_installed():
return _ConnectivityResult(
"Azure Foundry (Entra ID)",
[(color("⚠", Colors.YELLOW), label,
color("(azure-identity not installed)", Colors.DIM))],
[f"Install azure-identity: {sys.executable} -m pip install azure-identity"],
)
entra_cfg = model_cfg.get("entra") or {}
if not isinstance(entra_cfg, dict):
entra_cfg = {}
scope = (
str(entra_cfg.get("scope") or "").strip()
or SCOPE_AI_AZURE_DEFAULT
)
config = EntraIdentityConfig(
scope=scope,
)
info = describe_active_credential(config=config, timeout_seconds=10.0)
if info.get("ok"):
env_sources = info.get("env_sources") or []
tag = ", ".join(env_sources) if env_sources else "default credential chain"
return _ConnectivityResult(
"Azure Foundry (Entra ID)",
[(color("✓", Colors.GREEN), label,
color(f"({tag}, scope={scope})", Colors.DIM))],
[],
)
err = info.get("error") or "credential chain exhausted"
hint = info.get("hint") or (
"Run `az login`, set AZURE_TENANT_ID/AZURE_CLIENT_ID/"
"AZURE_CLIENT_SECRET, or attach a managed identity to this VM."
)
return _ConnectivityResult(
"Azure Foundry (Entra ID)",
[(color("⚠", Colors.YELLOW), label,
color(f"({err})", Colors.DIM))],
[f"Azure Foundry Entra: {err}. {hint}"],
)
# Build the probe submission list in display order
_probes.append(("OpenRouter API", _probe_openrouter))
_probes.append(("Anthropic API", _probe_anthropic))
global _APIKEY_PROVIDERS_CACHE
if _APIKEY_PROVIDERS_CACHE is None:
_APIKEY_PROVIDERS_CACHE = _build_apikey_providers_list()
for _entry in _APIKEY_PROVIDERS_CACHE:
_pname, _env_vars, _default_url, _base_env, _supports = _entry
# Capture loop vars by binding default args — without this, all closures
# would share the final iteration's values and every probe would hit
# the last provider's URL.
_probes.append((_pname, lambda p=_pname, e=_env_vars, u=_default_url,
b=_base_env, s=_supports:
_probe_apikey_provider(p, e, u, b, s)))
_probes.append(("AWS Bedrock", _probe_bedrock))
_probes.append(("Azure Foundry (Entra ID)", _probe_azure_entra))
# Print a single status line so users see something happening, then
# fan out. ``\r`` clears it once the first real result line lands.
print(f" {color(f'Running {len(_probes)} connectivity checks in parallel…', Colors.DIM)}",
probes = build_probes()
# Single status line so users see something happening; ``\r`` clears it
# once the first real result line lands.
print(f" {color(f'Running {len(probes)} connectivity checks in parallel…', Colors.DIM)}",
end="", flush=True)
# Disable boto3's EC2 instance-metadata-service probe for the duration
# of the parallel block. boto's default credential chain tries
# 169.254.169.254 with a multi-second timeout when we're not on EC2,
# which dominated the section's wall time before this fix
# (~2s on a developer laptop, even with the rest parallelized).
# Set on the parent thread before submitting work so the env-var
# mutation never races with another worker. has_aws_credentials() in
# the bedrock probe already gates on real env-var creds, so IMDS is
# never the legitimate source for `hermes doctor`.
_imds_prev = os.environ.get("AWS_EC2_METADATA_DISABLED")
os.environ["AWS_EC2_METADATA_DISABLED"] = "true"
try:
# 8 workers is plenty — each probe is a single HTTP call plus a TLS
# handshake. More than that wastes thread-startup cost and risks
# noisy output if anything ever printed from inside a worker.
with _futures.ThreadPoolExecutor(max_workers=8,
thread_name_prefix="doctor-probe") as _ex:
_futures_in_order = [_ex.submit(_fn) for _, _fn in _probes]
_results = [_f.result() for _f in _futures_in_order]
finally:
if _imds_prev is None:
os.environ.pop("AWS_EC2_METADATA_DISABLED", None)
else:
os.environ["AWS_EC2_METADATA_DISABLED"] = _imds_prev
# Clear the "Running …" line and print all results in submission order.
results = run_probes(probes)
print("\r" + " " * 70 + "\r", end="")
for _r in _results:
for _glyph, _label, _detail in _r.lines:
if _detail:
print(f" {_glyph} {_label} {_detail}")
else:
print(f" {_glyph} {_label}")
_issues_to_add = list(_r.issues)
if _issues_to_add and _has_healthy_oauth_fallback_for_apikey_provider(_r.label):
_issues_to_add = []
for _issue in _issues_to_add:
issues.append(_issue)
for r in results:
for glyph, label, detail in r.lines:
print(f" {glyph} {label} {detail}" if detail else f" {glyph} {label}")
if r.issues and not _has_healthy_oauth_fallback_for_apikey_provider(r.label):
f.issues.extend(r.issues)
return f
hermes_cli/doctor_connectivity.py
+394
View File
@@ -0,0 +1,394 @@
"""API connectivity probes for ``hermes doctor`` (split out of ``doctor.py``).
Every probe is a pure function: takes its inputs, makes one HTTP/SDK call and
returns a ``ProbeResult`` carrying the row(s) to print and issue strings to
append. No printing inside workers — the caller prints in submission order.
"""
from __future__ import annotations
import concurrent.futures
import os
import sys
from typing import NamedTuple
from hermes_cli.colors import Colors, color
from hermes_cli.models import _HERMES_USER_AGENT
from hermes_constants import OPENROUTER_MODELS_URL
from utils import base_url_host_matches
_APIKEY_PROVIDERS_CACHE: list | None = None
class ProbeResult(NamedTuple):
label: str
lines: list # [(glyph, label, detail)]
issues: list
_GLYPH = {"ok": ("✓", Colors.GREEN), "warn": ("⚠", Colors.YELLOW), "fail": ("✗", Colors.RED)}
def _row(name: str, status: str, detail: str = "", issues: list | None = None, label: str | None = None) -> ProbeResult:
glyph, col = _GLYPH[status]
return ProbeResult(
name,
[(color(glyph, col), label if label is not None else name, color(detail, Colors.DIM) if detail else "")],
list(issues or []),
)
def _skip(name: str) -> ProbeResult:
return ProbeResult(name, [], [])
def _has_healthy_oauth_fallback_for_apikey_provider(provider_label: str) -> bool:
"""True when a failed direct API-key probe is non-blocking because the same
provider family's OAuth runtime path is already healthy: the failed row is
still shown, but not promoted into the final blocking summary."""
normalized = (provider_label or "").strip().lower()
getter = {"minimax": "get_minimax_oauth_auth_status", "xai": "get_xai_oauth_auth_status"}.get(normalized)
if not getter:
return False
try:
from hermes_cli import auth
return bool((getattr(auth, getter)() or {}).get("logged_in"))
except Exception:
return False
def _build_apikey_providers_list() -> list:
"""Build the API-key provider health-check list once and cache it.
Tuple format: (name, env_vars, default_url, base_env, supports_models_endpoint)
Base list augmented with any ProviderProfile with auth_type="api_key" not
already present — adding plugins/model-providers/<name>/ is sufficient to get into doctor.
"""
_static = [
("Z.AI / GLM", ("GLM_API_KEY", "ZAI_API_KEY", "Z_AI_API_KEY"), "https://api.z.ai/api/paas/v4/models", "GLM_BASE_URL", True),
("Kimi / Moonshot", ("KIMI_API_KEY",), "https://api.moonshot.ai/v1/models", "KIMI_BASE_URL", True),
("StepFun Step Plan", ("STEPFUN_API_KEY",), "https://api.stepfun.ai/step_plan/v1/models", "STEPFUN_BASE_URL", True),
("Kimi / Moonshot (China)", ("KIMI_CN_API_KEY",), "https://api.moonshot.cn/v1/models", None, True),
("Arcee AI", ("ARCEEAI_API_KEY",), "https://api.arcee.ai/api/v1/models", "ARCEE_BASE_URL", True),
("GMI Cloud", ("GMI_API_KEY",), "https://api.gmi-serving.com/v1/models", "GMI_BASE_URL", True),
("DeepSeek", ("DEEPSEEK_API_KEY",), "https://api.deepseek.com/v1/models", "DEEPSEEK_BASE_URL", True),
("Hugging Face", ("HF_TOKEN",), "https://router.huggingface.co/v1/models", "HF_BASE_URL", True),
("NVIDIA NIM", ("NVIDIA_API_KEY",), "https://integrate.api.nvidia.com/v1/models", "NVIDIA_BASE_URL", True),
("Alibaba/DashScope", ("DASHSCOPE_API_KEY",), "https://dashscope-intl.aliyuncs.com/compatible-mode/v1/models", "DASHSCOPE_BASE_URL", True),
# MiniMax global: /v1 endpoint supports /models.
("MiniMax", ("MINIMAX_API_KEY",), "https://api.minimax.io/v1/models", "MINIMAX_BASE_URL", True),
# MiniMax CN: /v1 endpoint does NOT support /models (returns 404).
("MiniMax (China)", ("MINIMAX_CN_API_KEY",), "https://api.minimaxi.com/v1/models", "MINIMAX_CN_BASE_URL", False),
("Vercel AI Gateway", ("AI_GATEWAY_API_KEY",), "https://ai-gateway.vercel.sh/v1/models", "AI_GATEWAY_BASE_URL", True),
("Kilo Code", ("KILOCODE_API_KEY",), "https://api.kilo.ai/api/gateway/models", "KILOCODE_BASE_URL", True),
("OpenCode Zen", ("OPENCODE_ZEN_API_KEY",), "https://opencode.ai/zen/v1/models", "OPENCODE_ZEN_BASE_URL", True),
# OpenCode Go has no shared /models endpoint; skip the health check.
("OpenCode Go", ("OPENCODE_GO_API_KEY",), None, "OPENCODE_GO_BASE_URL", False),
]
_known_names = {t[0] for t in _static}
# Canonical profile names of the static rows, so profiles without a
# display_name don't duplicate providers already listed above.
_known_canonical = {
"zai", "kimi-coding", "stepfun", "kimi-coding-cn", "arcee", "gmi", "deepseek",
"huggingface", "nvidia", "alibaba", "minimax", "minimax-cn", "ai-gateway",
"kilocode", "opencode-zen", "opencode-go",
}
# Providers with a dedicated health check (custom headers/auth). Skip their
# pluggable profiles so the generic Bearer-auth loop doesn't run a duplicate,
# broken check (e.g. Anthropic native API requires x-api-key, not Bearer).
_dedicated_canonical = {"anthropic", "openrouter", "bedrock"}
_known_canonical.update(_dedicated_canonical)
try:
from providers import list_providers
from providers.base import ProviderProfile as _PP
try:
from hermes_cli.providers import normalize_provider as _normalize_provider
except Exception: # pragma: no cover - normalization is best-effort
def _normalize_provider(_name: str) -> str:
return (_name or "").strip().lower()
for _pp in list_providers():
if not isinstance(_pp, _PP) or _pp.auth_type != "api_key" or not _pp.env_vars:
continue
_label = _pp.display_name or _pp.name
if _label in _known_names or _pp.name in _known_canonical:
continue
_candidates = {_normalize_provider(_pp.name)}
for _alias in (_pp.aliases or ()):
_candidates.add(_normalize_provider(_alias))
if _candidates & _dedicated_canonical:
continue
# Separate API-key vars from base-URL override vars — the health-check
# loop sends the first found value as Authorization: Bearer, so a URL
# string must never be picked.
_key_vars = tuple(
v for v in _pp.env_vars
if not v.endswith("_BASE_URL") and not v.endswith("_URL")
)
_base_var = next(
(v for v in _pp.env_vars if v.endswith("_BASE_URL") or v.endswith("_URL")),
None,
)
if not _key_vars:
continue
_models_url = (
(_pp.models_url or (_pp.base_url.rstrip("/") + "/models"))
if _pp.base_url else None
)
_hc = getattr(_pp, "supports_health_check", True)
_static.append((_label, _key_vars, _models_url, _base_var, _hc))
except Exception:
pass
return _static
def _probe_openrouter() -> ProbeResult:
name = "OpenRouter API"
key = os.getenv("OPENROUTER_API_KEY")
if not key:
return _row(name, "warn", "(not configured)")
try:
import httpx
r = httpx.get(OPENROUTER_MODELS_URL, headers={"Authorization": f"Bearer {key}"}, timeout=10)
except Exception as e:
return _row(name, "fail", f"({e})", ["Check network connectivity"])
if r.status_code == 200:
return _row(name, "ok")
if r.status_code == 401:
return _row(name, "fail", "(invalid API key)", ["Check OPENROUTER_API_KEY in .env"])
if r.status_code == 402:
return _row(name, "fail", "(out of credits — payment required)", [
"OpenRouter account has insufficient credits. "
"Fix: run 'hermes config set model.provider <provider>' "
"to switch providers, or fund your OpenRouter account "
"at https://openrouter.ai/settings/credits"])
if r.status_code == 429:
return _row(name, "fail", "(rate limited)", [
"OpenRouter rate limit hit — consider switching to a different provider or waiting"])
return _row(name, "fail", f"(HTTP {r.status_code})")
def _probe_anthropic() -> ProbeResult:
name = "Anthropic API"
from hermes_cli.auth import get_anthropic_key
key = get_anthropic_key()
if not key:
return _skip(name)
try:
import httpx
from agent.anthropic_adapter import (
_is_oauth_token,
_COMMON_BETAS,
_OAUTH_ONLY_BETAS,
_CONTEXT_1M_BETA,
)
headers = {"anthropic-version": "2023-06-01"}
is_oauth = _is_oauth_token(key)
if is_oauth:
headers["Authorization"] = f"Bearer {key}"
headers["anthropic-beta"] = ",".join(_COMMON_BETAS + _OAUTH_ONLY_BETAS)
else:
headers["x-api-key"] = key
url = "https://api.anthropic.com/v1/models"
r = httpx.get(url, headers=headers, timeout=10)
# Reactive recovery: OAuth subscriptions without 1M context reject the
# request with 400 "long context beta is not yet available for this
# subscription". Retry once with that beta stripped so the doctor
# check doesn't falsely report Anthropic as unreachable.
if (
is_oauth
and r.status_code == 400
and "long context beta" in r.text.lower()
and "not yet available" in r.text.lower()
):
headers["anthropic-beta"] = ",".join(
[b for b in _COMMON_BETAS if b != _CONTEXT_1M_BETA] + list(_OAUTH_ONLY_BETAS)
)
r = httpx.get(url, headers=headers, timeout=10)
except Exception as e:
return _row(name, "warn", f"({e})")
if r.status_code == 200:
return _row(name, "ok")
if r.status_code == 401:
return _row(name, "fail", "(invalid API key)")
return _row(name, "warn", "(couldn't verify)")
def _probe_apikey_provider(pname, env_vars, default_url, base_env, supports_health_check) -> ProbeResult:
key = ""
for ev in env_vars:
key = os.getenv(ev, "")
if key:
break
if not key:
return _skip(pname)
label = pname.ljust(20)
if not supports_health_check:
return _row(pname, "ok", "(key configured)", label=label)
try:
import httpx
base = os.getenv(base_env, "") if base_env else ""
# Auto-detect Kimi Code keys (sk-kimi-) → api.kimi.com/coding/v1
# (OpenAI-compat surface, which exposes /models for health check).
if not base and key.startswith("sk-kimi-"):
base = "https://api.kimi.com/coding/v1"
# Anthropic-compat endpoints (/anthropic, api.kimi.com/coding
# with no /v1) don't support /models. Rewrite to OpenAI-compat
# /v1 surface for health checks.
if base and base.rstrip("/").endswith("/anthropic"):
from agent.auxiliary_client import _to_openai_base_url
base = _to_openai_base_url(base)
if base_url_host_matches(base, "api.kimi.com") and base.rstrip("/").endswith("/coding"):
base = base.rstrip("/") + "/v1"
url = (base.rstrip("/") + "/models") if base else default_url
headers = {
"Authorization": f"Bearer {key}",
"User-Agent": _HERMES_USER_AGENT,
}
if base_url_host_matches(base, "api.kimi.com"):
headers["User-Agent"] = "claude-code/0.1.0"
# Google's Generative Language API rejects ``Authorization: Bearer
# <api-key>`` with 401 ``ACCESS_TOKEN_TYPE_UNSUPPORTED`` — that header is
# reserved for OAuth 2 access tokens; plain keys use ``x-goog-api-key``.
if url and base_url_host_matches(url, "generativelanguage.googleapis.com"):
headers.pop("Authorization", None)
headers["x-goog-api-key"] = key
r = httpx.get(url, headers=headers, timeout=10)
if pname == "Alibaba/DashScope" and not base and r.status_code == 401:
r = httpx.get("https://dashscope.aliyuncs.com/compatible-mode/v1/models", headers=headers, timeout=10)
except Exception as e:
return _row(pname, "warn", f"({e})", label=label)
if r.status_code == 200:
return _row(pname, "ok", label=label)
if r.status_code == 401:
return _row(pname, "fail", "(invalid API key)", [f"Check {env_vars[0]} in .env"], label=label)
return _row(pname, "warn", f"(HTTP {r.status_code})", label=label)
def _probe_bedrock() -> ProbeResult:
name = "AWS Bedrock"
try:
from agent.bedrock_adapter import (
has_aws_credentials,
resolve_aws_auth_env_var,
resolve_bedrock_region,
)
except ImportError:
return _skip(name)
if not has_aws_credentials():
return _skip(name)
auth_var = resolve_aws_auth_env_var()
region = resolve_bedrock_region()
label = name.ljust(20)
try:
import boto3
from botocore.config import Config as _BotoConfig
# Trim retries on the actual Bedrock API call so a transient
# failure doesn't pad the doctor run by 30+ seconds.
cfg = _BotoConfig(connect_timeout=5, read_timeout=10, retries={"max_attempts": 1})
client = boto3.client("bedrock", region_name=region, config=cfg)
resp = client.list_foundation_models()
n = len(resp.get("modelSummaries", []))
return _row(name, "ok", f"({auth_var}, {region}, {n} models)", label=label)
except ImportError:
pip = f"{sys.executable} -m pip install boto3"
return _row(name, "warn", f"(boto3 not installed — {pip})", [f"Install boto3 for Bedrock: {pip}"], label=label)
except Exception as e:
err_name = type(e).__name__
return _row(name, "warn", f"({err_name}: {e})",
[f"AWS Bedrock: {err_name} — check IAM permissions for bedrock:ListFoundationModels"], label=label)
def _probe_azure_entra() -> ProbeResult:
"""Probe Azure Foundry Entra ID auth, parallel to ``_probe_bedrock``.
Skipped unless the active config has ``model.provider: azure-foundry`` AND
``model.auth_mode: entra_id`` — we don't probe the token-service / CLI
chain for users on plain API-key Azure. Bounded by a 10s timeout (via
:func:`agent.azure_identity_adapter.describe_active_credential`).
"""
name = "Azure Foundry (Entra ID)"
label = name.ljust(28)
try:
from hermes_cli.config import load_config
cfg = load_config()
model_cfg = cfg.get("model") if isinstance(cfg, dict) else {}
if not isinstance(model_cfg, dict):
return _skip(name)
cfg_provider = str(model_cfg.get("provider") or "").strip().lower()
auth_mode = str(model_cfg.get("auth_mode") or "").strip().lower()
if cfg_provider != "azure-foundry" or auth_mode != "entra_id":
return _skip(name)
except Exception:
return _skip(name)
try:
from agent.azure_identity_adapter import (
EntraIdentityConfig,
SCOPE_AI_AZURE_DEFAULT,
describe_active_credential,
has_azure_identity_installed,
)
except Exception as exc:
return _row(name, "warn", f"(adapter import failed: {exc})", [f"Azure Foundry adapter import failed: {exc}"], label=label)
if not has_azure_identity_installed():
return _row(name, "warn", "(azure-identity not installed)",
[f"Install azure-identity: {sys.executable} -m pip install azure-identity"], label=label)
entra_cfg = model_cfg.get("entra") or {}
if not isinstance(entra_cfg, dict):
entra_cfg = {}
scope = str(entra_cfg.get("scope") or "").strip() or SCOPE_AI_AZURE_DEFAULT
info = describe_active_credential(config=EntraIdentityConfig(scope=scope), timeout_seconds=10.0)
if info.get("ok"):
env_sources = info.get("env_sources") or []
tag = ", ".join(env_sources) if env_sources else "default credential chain"
return _row(name, "ok", f"({tag}, scope={scope})", label=label)
err = info.get("error") or "credential chain exhausted"
hint = info.get("hint") or (
"Run `az login`, set AZURE_TENANT_ID/AZURE_CLIENT_ID/"
"AZURE_CLIENT_SECRET, or attach a managed identity to this VM."
)
return _row(name, "warn", f"({err})", [f"Azure Foundry Entra: {err}. {hint}"], label=label)
def build_probes() -> list:
"""(label, callable) pairs in display order."""
global _APIKEY_PROVIDERS_CACHE
if _APIKEY_PROVIDERS_CACHE is None:
_APIKEY_PROVIDERS_CACHE = _build_apikey_providers_list()
probes = [("OpenRouter API", _probe_openrouter), ("Anthropic API", _probe_anthropic)]
for _pname, _env_vars, _default_url, _base_env, _supports in _APIKEY_PROVIDERS_CACHE:
# Bind loop vars via default args so every closure keeps its own provider.
probes.append((_pname, lambda p=_pname, e=_env_vars, u=_default_url, b=_base_env, s=_supports:
_probe_apikey_provider(p, e, u, b, s)))
probes.append(("AWS Bedrock", _probe_bedrock))
probes.append(("Azure Foundry (Entra ID)", _probe_azure_entra))
return probes
def run_probes(probes: list) -> list:
"""Run every probe in a thread pool; results in submission order.
Probes are independent HTTP calls, so running them in series cost ~5s wall
(2s of it boto3's IMDS lookup). Parallel collapses the section to roughly
the slowest single probe without changing the output.
"""
# Disable boto3's EC2 instance-metadata probe for the parallel block: the
# default credential chain tries 169.254.169.254 with a multi-second
# timeout off-EC2. Set on the parent thread before submitting so the env
# mutation never races a worker; has_aws_credentials() already gates on
# real env creds, so IMDS is never the legitimate source for doctor.
_imds_prev = os.environ.get("AWS_EC2_METADATA_DISABLED")
os.environ["AWS_EC2_METADATA_DISABLED"] = "true"
try:
# 8 workers is plenty — each probe is one HTTP call plus a TLS handshake.
with concurrent.futures.ThreadPoolExecutor(max_workers=8, thread_name_prefix="doctor-probe") as ex:
futures = [ex.submit(fn) for _, fn in probes]
return [f.result() for f in futures]
finally:
if _imds_prev is None:
os.environ.pop("AWS_EC2_METADATA_DISABLED", None)
else:
os.environ["AWS_EC2_METADATA_DISABLED"] = _imds_prev