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

feat(mcp): expose profile-scoped cached connection health

Browse Source
This commit is contained in:
Joey
2026-09-04 04:15:39 +02:00
committed by Teknium
parent b167e81750
commit a6699d60f4
6 changed files with 479 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/tools/test_mcp_tool.py
+180 -1
View File
@@ -4,9 +4,11 @@ All tests use mocks -- no real MCP servers or subprocesses are started.
"""
import asyncio
import copy
import json
import logging
import os
import pickle
import sys
import threading
import time
@@ -272,6 +274,19 @@ class TestMCPStatus:
import tools.mcp_tool as mcp_tool
from tools import mcp_tool_config as _mcp_config
from tools import mcp_tool_discovery as _mcp_discovery
from tools import mcp_tool_errors as _mcp_errors
from tools.mcp_oauth import OAuthNonInteractiveError
assert _mcp_errors._connect_failure_reason(RuntimeError("network")) == "check_failed"
assert _mcp_errors._connect_failure_reason(
OAuthNonInteractiveError("browser required")
) == "auth_required"
tagged = _mcp_errors._connect_error_text(
"Connection closed", OAuthNonInteractiveError("browser required")
)
for cloned in (copy.copy(tagged), copy.deepcopy(tagged), pickle.loads(pickle.dumps(tagged))):
assert cloned == "Connection closed"
assert getattr(cloned, "reason") == "auth_required"
monkeypatch.setattr(
_mcp_config, "_load_mcp_config",
@@ -284,13 +299,17 @@ class TestMCPStatus:
)
with mcp_tool._lock:
saved_servers = dict(mcp_tool._servers)
saved_scopes = dict(mcp_tool._server_scope_keys)
saved_connecting = set(mcp_tool._server_connecting)
saved_errors = dict(mcp_tool._server_connect_errors)
mcp_tool._servers.clear()
mcp_tool._server_scope_keys.clear()
mcp_tool._server_connecting.clear()
mcp_tool._server_connect_errors.clear()
mcp_tool._server_connecting.add("connecting")
mcp_tool._server_connect_errors["failed"] = "Connection closed"
mcp_tool._server_connect_errors["failed"] = _mcp_errors._connect_error_text(
"Connection closed", OAuthNonInteractiveError("browser required")
)
try:
statuses = {
@@ -301,6 +320,8 @@ class TestMCPStatus:
with mcp_tool._lock:
mcp_tool._servers.clear()
mcp_tool._servers.update(saved_servers)
mcp_tool._server_scope_keys.clear()
mcp_tool._server_scope_keys.update(saved_scopes)
mcp_tool._server_connecting.clear()
mcp_tool._server_connecting.update(saved_connecting)
mcp_tool._server_connect_errors.clear()
@@ -310,11 +331,122 @@ class TestMCPStatus:
assert statuses["configured"]["connected"] is False
assert statuses["configured"]["disabled"] is False
assert statuses["connecting"]["status"] == "connecting"
assert statuses["connecting"]["reason"] == "stale"
assert statuses["failed"]["status"] == "failed"
assert statuses["failed"]["error"] == "Connection closed"
assert statuses["failed"]["reason"] == "auth_required"
assert statuses["disabled"]["status"] == "disabled"
assert statuses["disabled"]["disabled"] is True
def test_status_ignores_a_runtime_owned_by_another_profile(self, monkeypatch):
import tools.mcp_tool as mcp_tool
from tools import mcp_tool_config as _mcp_config
from tools import mcp_tool_discovery as _mcp_discovery
monkeypatch.setattr(
_mcp_config, "_load_mcp_config",
lambda: {"shared": {"command": "shared-mcp"}},
)
monkeypatch.setattr(mcp_tool, "_mcp_registry_scope", lambda: "profile:work")
foreign_server = MagicMock(spec=mcp_tool.MCPServerTask)
foreign_server.session = object()
foreign_server._registered_tool_names = ["secret_tool"]
foreign_server._tools = []
foreign_server._sampling = None
with mcp_tool._lock:
saved_servers = dict(mcp_tool._servers)
saved_scopes = dict(mcp_tool._server_scope_keys)
mcp_tool._servers["shared"] = foreign_server
mcp_tool._server_scope_keys["shared"] = "profile:other"
try:
[status] = _mcp_discovery.get_mcp_status()
with mcp_tool._lock:
mcp_tool._server_scope_keys.pop("shared", None)
[unscoped_status] = _mcp_discovery.get_mcp_status()
finally:
with mcp_tool._lock:
mcp_tool._servers.clear()
mcp_tool._servers.update(saved_servers)
mcp_tool._server_scope_keys.clear()
mcp_tool._server_scope_keys.update(saved_scopes)
assert status["status"] == "configured"
assert status["reason"] == "stale"
assert status["tools"] == 0
assert unscoped_status["status"] == "configured"
assert unscoped_status["reason"] == "stale"
def test_scoped_shutdown_clears_only_its_connection_status(self, monkeypatch):
import tools.mcp_tool as mcp_tool
from tools import mcp_tool_lifecycle, mcp_tool_loop, mcp_tool_discovery
monkeypatch.setattr(mcp_tool_loop, "_stop_mcp_loop", lambda **_kwargs: None)
with mcp_tool._lock:
saved_servers = dict(mcp_tool._servers)
saved_scopes = dict(mcp_tool._server_scope_keys)
saved_connecting = set(mcp_tool._server_connecting)
saved_errors = dict(mcp_tool._server_connect_errors)
saved_retry_after = dict(mcp_tool._server_connect_retry_after)
saved_failures = dict(mcp_tool._server_connect_failures)
mcp_tool._servers.clear()
mcp_tool._server_scope_keys.clear()
mcp_tool._server_scope_keys.update({
"work-connecting": "profile:work",
"work-failed": "profile:work",
"other-failed": "profile:other",
})
mcp_tool._server_connecting.clear()
mcp_tool._server_connecting.add("work-connecting")
mcp_tool._server_connect_errors.clear()
mcp_tool._server_connect_errors.update({
"work-failed": "work error",
"other-failed": "other error",
})
mcp_tool._server_connect_retry_after.clear()
mcp_tool._server_connect_retry_after.update({
"work-failed": 1.0,
"other-failed": 2.0,
})
mcp_tool._server_connect_failures.clear()
mcp_tool._server_connect_failures.update({
"work-failed": 1,
"other-failed": 2,
})
try:
mcp_tool_lifecycle.shutdown_mcp_servers(scope="profile:work")
with mcp_tool._lock:
assert mcp_tool._server_connecting == set()
assert mcp_tool._server_connect_errors == {
"other-failed": "other error"
}
assert mcp_tool._server_scope_keys == {
"other-failed": "profile:other"
}
assert mcp_tool._server_connect_retry_after == {
"other-failed": 2.0
}
assert mcp_tool._server_connect_failures == {
"other-failed": 2
}
finally:
with mcp_tool._lock:
mcp_tool._servers.clear()
mcp_tool._servers.update(saved_servers)
mcp_tool._server_scope_keys.clear()
mcp_tool._server_scope_keys.update(saved_scopes)
mcp_tool._server_connecting.clear()
mcp_tool._server_connecting.update(saved_connecting)
mcp_tool._server_connect_errors.clear()
mcp_tool._server_connect_errors.update(saved_errors)
mcp_tool._server_connect_retry_after.clear()
mcp_tool._server_connect_retry_after.update(saved_retry_after)
mcp_tool._server_connect_failures.clear()
mcp_tool._server_connect_failures.update(saved_failures)
class TestLifecycleConfig:
def test_get_lifecycle_seconds_accepts_top_level_and_nested_values(self):
@@ -2688,6 +2820,53 @@ class TestSanitizeMcpNameComponent:
class TestRegisterMcpServers:
def test_records_profile_scope_before_connect(self):
import tools.mcp_tool as mcp_tool
from tools import mcp_tool_lifecycle, mcp_tool_loop, mcp_tool_discovery
seen = []
async def fake_register(name, _cfg):
with mcp_tool._lock:
seen.append((
name in mcp_tool._server_connecting,
mcp_tool._server_scope_keys.get(name),
))
raise RuntimeError("expected connect failure")
with mcp_tool._lock:
saved_servers = dict(mcp_tool._servers)
saved_scopes = dict(mcp_tool._server_scope_keys)
saved_connecting = set(mcp_tool._server_connecting)
saved_errors = dict(mcp_tool._server_connect_errors)
mcp_tool._servers.clear()
mcp_tool._server_scope_keys.clear()
mcp_tool._server_connecting.clear()
mcp_tool._server_connect_errors.clear()
try:
with patch("tools.mcp_tool._MCP_AVAILABLE", True), \
patch("tools.mcp_tool_config._filter_suspicious_mcp_servers", side_effect=lambda value: value), \
patch("tools.mcp_tool_discovery._connect_cooldown_active", return_value=False), \
patch("tools.mcp_tool._mcp_registry_scope", return_value="profile:work"), \
patch("tools.mcp_tool_discovery._discover_and_register_server", side_effect=fake_register):
mcp_tool_discovery.register_mcp_servers({"scoped": {"command": "test"}})
assert seen == [(True, "profile:work")]
with mcp_tool._lock:
assert mcp_tool._server_connect_errors["scoped"].reason == "check_failed"
assert mcp_tool._server_scope_keys["scoped"] == "profile:work"
finally:
with mcp_tool._lock:
mcp_tool._servers.clear()
mcp_tool._servers.update(saved_servers)
mcp_tool._server_scope_keys.clear()
mcp_tool._server_scope_keys.update(saved_scopes)
mcp_tool._server_connecting.clear()
mcp_tool._server_connecting.update(saved_connecting)
mcp_tool._server_connect_errors.clear()
mcp_tool._server_connect_errors.update(saved_errors)
"""Verify the new register_mcp_servers() public API."""
def test_mcp_not_available_returns_empty(self):
tests/tui_gateway/test_mcp_profile_rpcs.py
+171
View File
@@ -13,6 +13,7 @@ from __future__ import annotations
import os
from pathlib import Path
from types import SimpleNamespace
import pytest
@@ -110,6 +111,176 @@ def test_list_reflects_the_scoped_profile(hermes_root):
assert work_server["command"] == "svc-a-bin"
def test_status_is_profile_scoped_and_credential_safe(hermes_root):
_result(
_call(
"mcp.servers.add",
{"profile": "work", "name": "svc-a", "config": {"command": "svc-a-bin"}},
)
)
_result(
_call(
"mcp.servers.add",
{"profile": "other", "name": "svc-b", "config": {"command": "svc-b-bin"}},
)
)
payload = _result(_call("mcp.servers.status", {"profile": "work"}))
assert payload["checked_at"] > 0
assert payload["servers"] == [
{
"name": "svc-a",
"transport": "stdio",
"tools": 0,
"connected": False,
"disabled": False,
"status": "configured",
"reason": "stale",
}
]
assert "error" not in str(payload)
def test_status_redacts_internal_failures(hermes_root, monkeypatch):
from tools import mcp_tool_discovery
def fail(*args, **kwargs):
raise RuntimeError("token=must-not-leak")
monkeypatch.setattr(mcp_tool_discovery, "get_mcp_status", fail)
response = _call("mcp.servers.status", {"profile": "work"})
assert response["error"]["message"] == "MCP status unavailable"
assert "must-not-leak" not in str(response)
def test_status_omits_raw_server_errors(hermes_root, monkeypatch):
from tools import mcp_tool_discovery
monkeypatch.setattr(
mcp_tool_discovery,
"get_mcp_status",
lambda configured, include_runtime: [{
"name": "svc",
"transport": "stdio",
"tools": 0,
"connected": False,
"disabled": False,
"status": "failed",
"reason": "auth_required",
"error": "token=must-not-leak",
}],
)
payload = _result(_call("mcp.servers.status"))
assert payload["servers"][0]["reason"] == "auth_required"
assert "error" not in payload["servers"][0]
assert "must-not-leak" not in str(payload)
def test_status_does_not_run_the_runtime_config_loader(hermes_root, monkeypatch):
from tools import mcp_tool_config
_result(
_call(
"mcp.servers.add",
{"profile": "work", "name": "svc-a", "config": {"command": "svc-a-bin"}},
)
)
def forbidden():
raise AssertionError("runtime config loader executed")
monkeypatch.setattr(mcp_tool_config, "_load_mcp_config", forbidden)
payload = _result(_call("mcp.servers.status", {"profile": "work"}))
assert [server["name"] for server in payload["servers"]] == ["svc-a"]
def test_status_does_not_mix_launch_runtime_into_another_profile(hermes_root):
import tools.mcp_tool as mcp_tool
_result(
_call(
"mcp.servers.add",
{"profile": "work", "name": "shared", "config": {"command": "work-bin"}},
)
)
launch_server = SimpleNamespace(
session=object(),
_registered_tool_names=["launch_secret_tool"],
_tools=[],
_sampling=None,
)
with mcp_tool._lock:
saved_servers = dict(mcp_tool._servers)
saved_scopes = dict(mcp_tool._server_scope_keys)
mcp_tool._servers["shared"] = launch_server
mcp_tool._server_scope_keys.pop("shared", None)
try:
payload = _result(_call("mcp.servers.status", {"profile": "work"}))
finally:
with mcp_tool._lock:
mcp_tool._servers.clear()
mcp_tool._servers.update(saved_servers)
mcp_tool._server_scope_keys.clear()
mcp_tool._server_scope_keys.update(saved_scopes)
assert payload["servers"][0]["status"] == "configured"
assert payload["servers"][0]["tools"] == 0
def test_status_includes_named_profile_runtime_in_multiplex(hermes_root):
from agent.secret_scope import is_multiplex_active, set_multiplex_active
from hermes_constants import (
hermes_home_key,
reset_hermes_home_override,
set_hermes_home_override,
)
import tools.mcp_tool as mcp_tool
_result(
_call(
"mcp.servers.add",
{"profile": "work", "name": "shared", "config": {"command": "work-bin"}},
)
)
work_token = set_hermes_home_override(hermes_root / "profiles" / "work")
try:
work_scope = hermes_home_key()
finally:
reset_hermes_home_override(work_token)
work_server = SimpleNamespace(
session=object(),
_registered_tool_names=["work_tool"],
_tools=[],
_sampling=None,
)
previous_multiplex = is_multiplex_active()
with mcp_tool._lock:
saved_servers = dict(mcp_tool._servers)
saved_scopes = dict(mcp_tool._server_scope_keys)
mcp_tool._servers["shared"] = work_server # type: ignore[assignment]
mcp_tool._server_scope_keys["shared"] = work_scope
set_multiplex_active(True)
try:
payload = _result(_call("mcp.servers.status", {"profile": "work"}))
finally:
set_multiplex_active(previous_multiplex)
with mcp_tool._lock:
mcp_tool._servers.clear()
mcp_tool._servers.update(saved_servers)
mcp_tool._server_scope_keys.clear()
mcp_tool._server_scope_keys.update(saved_scopes)
assert payload["servers"][0]["status"] == "connected"
assert payload["servers"][0]["tools"] == 1
def test_set_api_key_writes_env_and_header_to_right_profile(hermes_root):
root = hermes_root
_result(
tools/mcp_tool_discovery.py
+46 -11
View File
@@ -112,7 +112,7 @@ def _note_connect_failure(name: str, exc: BaseException) -> str:
message = _errors._format_connect_error(exc)
with _core._lock:
_core._server_connecting.discard(name)
_core._server_connect_errors[name] = message
_core._server_connect_errors[name] = _errors._connect_error_text(message, exc)
_record_connect_failure(name)
return message
@@ -245,7 +245,9 @@ def _select_new_servers(servers: Dict[str, dict]) -> Dict[str, dict]:
stale_cached = [_core._servers[k] for k in servers
if k in _core._servers and getattr(_core._servers[k], "session", None) is None]
_core._server_connecting.update(new_servers)
current_scope = _core._mcp_registry_scope()
for srv_name in new_servers:
_core._server_scope_keys[srv_name] = current_scope
_core._server_connect_errors.pop(srv_name, None)
# Track which servers opt-in to parallel tool calls (idempotent).
for srv_name, srv_cfg in servers.items():
@@ -327,7 +329,9 @@ def _run_discovery_pass(new_servers: Dict[str, dict]) -> None:
"connecting set: %s", how, len(stale), ", ".join(stale))
_core._server_connecting.difference_update(stale)
for _sn in stale:
_core._server_connect_errors.setdefault(_sn, f"Connection attempt {how} during discovery")
message = f"Connection attempt {how} during discovery"
_core._server_connect_errors.setdefault(
_sn, _errors._MCPConnectErrorText(message, "check_failed"))
raise
finally:
if _was_interrupted:
@@ -453,16 +457,36 @@ def is_mcp_tool_parallel_safe(tool_name: str) -> bool:
return bool(server_name and server_name in _core._parallel_safe_servers)
def get_mcp_status() -> List[dict]:
"""Per-server status dicts for banner/TUI: name, transport, tools, connected, disabled,
status (connected / disabled / connecting / failed / configured) and error for failed."""
configured = _config._load_mcp_config()
def get_mcp_status(
configured: Optional[Dict[str, dict]] = None,
*,
include_runtime: bool = True,
) -> List[dict]:
"""Per-server cached status without initiating connections."""
configured = _config._load_mcp_config() if configured is None else dict(configured)
if not configured:
return []
current_scope = _core._mcp_registry_scope()
with _core._lock:
active_servers = dict(_core._servers)
connecting = set(_core._server_connecting)
connect_errors = dict(_core._server_connect_errors)
scope_keys = dict(_core._server_scope_keys) if include_runtime else {}
def visible_in_current_scope(name: str) -> bool:
if name in scope_keys:
return scope_keys[name] == current_scope
return current_scope is None
active_servers = {
name: server for name, server in _core._servers.items()
if include_runtime and visible_in_current_scope(name)
}
connecting = {
name for name in _core._server_connecting
if include_runtime and visible_in_current_scope(name)
}
connect_errors = {
name: error for name, error in _core._server_connect_errors.items()
if include_runtime and visible_in_current_scope(name)
}
result: List[dict] = []
for name, cfg in configured.items():
@@ -471,8 +495,13 @@ def get_mcp_status() -> List[dict]:
live = server is not None and server.session is not None
status = ("connected" if live else "disabled" if not enabled else "connecting" if name in connecting
else "failed" if name in connect_errors else "configured")
reason = {
"connected": "healthy", "disabled": "not_configured",
"connecting": "stale", "configured": "stale",
}.get(status, "check_failed")
entry = {"name": name, "transport": cfg.get("transport", "http") if "url" in cfg else "stdio",
"tools": 0, "connected": False, "disabled": status == "disabled", "status": status}
"tools": 0, "connected": False, "disabled": status == "disabled",
"status": status, "reason": reason}
if live:
entry["connected"] = True
entry["tools"] = (len(server._registered_tool_names) if hasattr(server, "_registered_tool_names")
@@ -480,7 +509,13 @@ def get_mcp_status() -> List[dict]:
if server._sampling:
entry["sampling"] = dict(server._sampling.metrics)
elif status == "failed":
entry["error"] = connect_errors[name]
stored_error = connect_errors[name]
failure = getattr(server, "_error", None) if server is not None else None
failure_reason = getattr(stored_error, "reason", None)
if failure_reason is None and isinstance(failure, BaseException):
failure_reason = _errors._connect_failure_reason(failure)
entry["reason"] = failure_reason or "check_failed"
entry["error"] = str(stored_error)
result.append(entry)
return result
tools/mcp_tool_errors.py
+23
View File
@@ -283,6 +283,29 @@ def _is_auth_error(exc: BaseException) -> bool:
return getattr(exc.response, "status_code", None) == 401 if isinstance(exc, http_types) else True
def _connect_failure_reason(exc: BaseException) -> str:
"""Reduce an MCP connect exception to a credential-safe health reason."""
return "auth_required" if _is_auth_error(_unwrap_exception_group(exc)) else "check_failed"
class _MCPConnectErrorText(str):
"""Backward-compatible error text carrying a credential-safe reason."""
reason: str
def __new__(cls, message: str, reason: str):
value = super().__new__(cls, message)
value.reason = reason
return value
def __reduce__(self):
return type(self), (str(self), self.reason)
def _connect_error_text(message: str, exc: BaseException) -> str:
return _MCPConnectErrorText(message, _connect_failure_reason(exc))
# Lower-cased substrings meaning the transport session expired / was GC'd (OAuth token still valid).
# Substrings (lower-cased match) that indicate the MCP server rejected the request because its server-side
# transport session expired / was garbage-collected. See #13383.
tools/mcp_tool_lifecycle.py
+26 -5
View File
@@ -84,11 +84,16 @@ def _filter_mcp_children(pids: set) -> set:
return kept
def _clear_connect_cooldowns() -> None:
def _clear_connect_cooldowns(names=None) -> None:
"""Drop connect-retry cooldowns: a restart must re-attempt every server immediately, not
honour a stale per-server backoff. Caller holds ``_core._lock``."""
_core._server_connect_retry_after.clear()
_core._server_connect_failures.clear()
if names is None:
_core._server_connect_retry_after.clear()
_core._server_connect_failures.clear()
else:
for name in names:
_core._server_connect_retry_after.pop(name, None)
_core._server_connect_failures.pop(name, None)
def shutdown_mcp_servers(*, scope: Optional[str] = None):
@@ -100,6 +105,19 @@ def shutdown_mcp_servers(*, scope: Optional[str] = None):
with _core._lock:
selected = [name for name in _core._servers if scope is None or _core._server_scope_keys.get(name) == scope]
servers_snapshot = [_core._servers[name] for name in selected]
selected_status = (
set(_core._servers) | set(_core._server_scope_keys)
| set(_core._server_connecting) | set(_core._server_connect_errors)
if scope is None else {
name for name, owner in _core._server_scope_keys.items() if owner == scope
}
)
def clear_selected_status():
_core._server_connecting.difference_update(selected_status)
for name in selected_status:
_core._server_connect_errors.pop(name, None)
_core._server_scope_keys.pop(name, None)
# Fast path: nothing to shut down. The connect-cooldown maps can still be populated here — a server that
# failed to connect is never recorded in ``_servers`` (that is the very premise of the #50394 cooldown),
@@ -115,7 +133,8 @@ def shutdown_mcp_servers(*, scope: Optional[str] = None):
for name in selected:
_core._servers.pop(name, None)
_core._server_scope_keys.pop(name, None)
_clear_connect_cooldowns()
clear_selected_status()
_clear_connect_cooldowns(None if scope is None else selected_status)
with _core._lock:
loop = _core._mcp_loop
@@ -132,7 +151,9 @@ def shutdown_mcp_servers(*, scope: Optional[str] = None):
# (a server that failed to connect is never in ``_servers`` — the most likely state for
# stale backoff entries), no connect-cooldown state may survive shutdown.
with _core._lock:
_clear_connect_cooldowns()
if not servers_snapshot:
clear_selected_status()
_clear_connect_cooldowns(None if scope is None else selected_status)
_loop._stop_mcp_loop(only_if_idle=scope is not None)
tui_gateway/methods_tools.py
+33
View File
@@ -1171,6 +1171,39 @@ def _(rid, params: dict) -> dict:
return _ok(rid, {"servers": [_mcp_summarize_server(name, cfg) for name, cfg in sorted(servers.items())]})
@method("mcp.servers.status")
def _(rid, params: dict) -> dict:
"""Return credential-safe cached MCP runtime state without connecting."""
hc = _tools_mod("hermes_constants")
runtime_home_key = hc.hermes_home_key()
token = None
try:
if profile := _str_arg(params, "profile"):
profile_dir = _tools_mod("hermes_cli.profiles").get_profile_dir(profile)
if not profile_dir or not profile_dir.is_dir():
return _err(rid, 4064, f"profile '{profile}' not found")
token = hc.set_hermes_home_override(str(profile_dir))
import time
from agent.secret_scope import is_multiplex_active
configured = _tools_mod("hermes_cli.mcp_config")._get_mcp_servers()
get_status = _tools_mod("tools.mcp_tool_discovery").get_mcp_status
safe_fields = ("name", "transport", "tools", "connected", "disabled", "status", "reason")
servers = get_status(
configured,
include_runtime=is_multiplex_active() or hc.hermes_home_key() == runtime_home_key,
)
return _ok(rid, {
"servers": [{key: entry[key] for key in safe_fields if key in entry} for entry in servers],
"checked_at": int(time.time() * 1000),
})
except Exception:
return _err(rid, 5024, "MCP status unavailable")
finally:
_mcp_reset_profile(token)
@_mcp_rpc("add")
def _(rid, params: dict) -> dict:
"""Add ``name`` from ``preset`` (catalog id) and/or ``config`` (url/command/args/env/headers/auth/