#!/usr/bin/env python3 """MCP (Model Context Protocol) client: connects to the ``mcp_servers`` configured in ~/.hermes/config.yaml (stdio, Streamable HTTP or SSE), discovers their tools and registers them into the hermes tool registry. The ``mcp`` package is optional (no-op without it). One background event loop (``_mcp_loop``) in a daemon thread runs each server as a long-lived Task (``MCPServerTask``) so the transport's anyio cancel scopes enter and exit in one Task; every ``_servers``/loop mutation holds ``_lock``. This module keeps the SDK loader, ``MCPServerTask`` and all shared state; the ``mcp_tool_*`` siblings read that state back through ``tools.mcp_tool`` at call time (``_core``) and are imported directly by their callers.""" import asyncio import contextvars import importlib import importlib.util import inspect import logging import os import sys import threading import time from typing import Any, Callable, Dict, List, Optional, Set logger = logging.getLogger(__name__) from tools.mcp_tool_common import _DEFAULT_TOOL_TIMEOUT, mcp_field from tools.mcp_tool_config import _get_mcp_stderr_log, _npx_cached_bin from tools.mcp_tool_sampling import ElicitationHandler, SamplingHandler from tools.mcp_tool_transport import MCPServerTransportMixin from tools.mcp_tool_server_run import MCPServerRunMixin from tools.mcp_tool_health import MCPServerHealthMixin # Wall-clock bound on the fail-open OSV malware preflight before a stdio spawn; just ABOVE # osv_check._TIMEOUT (10s) so it only bites when a stalled SSL handshake defeats that. _OSV_MALWARE_CHECK_TIMEOUT_S = 12.0 async def _preflight_stdio_command(server_name: str, command: str, args: list) -> tuple[str, list]: """OSV malware preflight (off-loop, wall-clock bound, fail-open on timeout), THEN the cached-npx swap. The preflight must see the REAL command/args: anything that rewrites argv to a wrapper or resolved binary has to happen after it, or the check silently inspects the wrapper and becomes a no-op (``_infer_ecosystem`` keys off the command basename being npx/uvx/pipx).""" from tools.osv_check import check_package_for_malware try: malware_error = await asyncio.wait_for( asyncio.to_thread(check_package_for_malware, command, args), timeout=_OSV_MALWARE_CHECK_TIMEOUT_S) except asyncio.TimeoutError: logger.warning("MCP server '%s': OSV malware preflight timed out after %.0fs " "(network slow/unreachable) — proceeding without the check.", server_name, _OSV_MALWARE_CHECK_TIMEOUT_S) malware_error = None if malware_error: raise ValueError(f"MCP server '{server_name}': {malware_error}") # npx resolves the package and then FORKS, staying resident as the real server's parent for # nothing (~48 MB per server, measured). Hermes already supervises the child (shared death # supervisor), so a cached package is spawned directly; a cache miss leaves npx untouched. if os.path.basename(command).lower().startswith("npx"): cached = _npx_cached_bin(args) if cached: direct_command, direct_args = cached logger.debug("MCP server '%s': using cached npx binary %s (skipping the " "resident `npm exec` parent)", server_name, direct_command) command, args = direct_command, direct_args return command, args # ---- Optional MCP SDK: availability probe now, symbol import on first use ---- _MCP_AVAILABLE = _MCP_HTTP_AVAILABLE = _MCP_NEW_HTTP = _MCP_LEGACY_HTTP = False _MCP_SAMPLING_TYPES = _MCP_NOTIFICATION_TYPES = _MCP_ELICITATION_TYPES = False _MCP_MESSAGE_HANDLER_SUPPORTED = _MCP_LOGGING_CALLBACK_SUPPORTED = False sse_client = None # Fallback for SDKs without LATEST_PROTOCOL_VERSION (Streamable HTTP arrived with 2025-03-26). LATEST_PROTOCOL_VERSION = "2025-03-26" # Newest revision ``ClientSession.initialize()`` speaks; from 2026-07-28 the handshake is a # per-request envelope so this can be OLDER than LATEST_PROTOCOL_VERSION, and the # MCP-Protocol-Version header must be seeded from THIS one. LATEST_HANDSHAKE_VERSION = LATEST_PROTOCOL_VERSION # Importing ``mcp`` costs ~260ms, so it is deferred to first use (_ensure_mcp_sdk); availability # is decided now via find_spec so every ``if not _MCP_AVAILABLE`` gate / patch / skipif holds. try: _MCP_AVAILABLE = importlib.util.find_spec("mcp") is not None except Exception: _MCP_AVAILABLE = False if not _MCP_AVAILABLE: logger.debug("mcp package not installed -- MCP tool support disabled") ClientSession: Any = None _MCP_SDK_IMPORT_ATTEMPTED = False _MCP_SDK_IMPORT_LOCK = threading.Lock() # Optional SDK type families (module, names, debug message when absent), bound in this order to # _MCP_SAMPLING_TYPES / _MCP_ELICITATION_TYPES / _MCP_NOTIFICATION_TYPES; an older SDK only # loses that feature, not MCP. _OPTIONAL_TYPE_FAMILIES = ( ("mcp.types", ("CreateMessageResult", "CreateMessageResultWithTools", "ErrorData", "SamplingCapability", "SamplingToolsCapability", "TextContent", "ToolUseContent"), "MCP sampling types not available -- sampling disabled"), ("mcp.types", ("ElicitRequestParams", "ElicitResult"), "MCP elicitation types not available -- elicitation disabled"), ("mcp.types", ("ServerNotification", "ToolListChangedNotification", "PromptListChangedNotification", "ResourceListChangedNotification"), "MCP notification types not available -- dynamic tool discovery disabled"), ) # Bound by _ensure_mcp_sdk(); module __getattr__ (PEP 562) imports the SDK on first external # access so mock.patch("tools.mcp_tool.stdio_client") sees a real original, never clobbered. _MCP_SDK_LAZY_SYMBOLS = frozenset( {"StdioServerParameters", "stdio_client", "streamablehttp_client", "streamable_http_client"} | {n for _mod, names, _msg in _OPTIONAL_TYPE_FAMILIES for n in names}) def __getattr__(name: str): if name in _MCP_SDK_LAZY_SYMBOLS: _ensure_mcp_sdk() try: return globals()[name] except KeyError: pass # SDK missing or symbol absent on this SDK build raise AttributeError(f"module {__name__!r} has no attribute {name!r}") def _import_sdk_names(module: str, names: tuple, missing_msg: Optional[str] = None) -> bool: """Bind ``names`` from SDK ``module`` into this module's globals; False (nothing bound, optional debug line) when this SDK build lacks the module or any of the names.""" try: mod = importlib.import_module(module) values = {n: getattr(mod, n) for n in names} except (ImportError, AttributeError): if missing_msg: logger.debug(missing_msg) return False globals().update(values) return True def _ensure_mcp_sdk() -> bool: """Import the optional ``mcp`` SDK on first use; return availability. Idempotent and thread-safe; honors a test-patched ``_MCP_AVAILABLE=False`` (no import) and pre-installed mocks (``ClientSession`` already set means no re-import).""" global _MCP_SDK_IMPORT_ATTEMPTED, _MCP_AVAILABLE, _MCP_HTTP_AVAILABLE, _MCP_NEW_HTTP, _MCP_LEGACY_HTTP global _MCP_SAMPLING_TYPES, _MCP_NOTIFICATION_TYPES, _MCP_ELICITATION_TYPES, sse_client global _MCP_MESSAGE_HANDLER_SUPPORTED, _MCP_LOGGING_CALLBACK_SUPPORTED, LATEST_HANDSHAKE_VERSION global _JSONRPC_METHOD_NOT_FOUND if not _MCP_AVAILABLE: return False if _MCP_SDK_IMPORT_ATTEMPTED or ClientSession is not None: return _MCP_AVAILABLE with _MCP_SDK_IMPORT_LOCK: if _MCP_SDK_IMPORT_ATTEMPTED or ClientSession is not None: return _MCP_AVAILABLE if (_import_sdk_names("mcp", ("ClientSession", "StdioServerParameters")) and _import_sdk_names("mcp.client.stdio", ("stdio_client",))): _MCP_AVAILABLE = True # mcp >= 1.24 ships streamable_http_client; 2.0 dropped the deprecated # streamablehttp_client alias. Either one gives HTTP. _MCP_NEW_HTTP = _import_sdk_names("mcp.client.streamable_http", ("streamable_http_client",)) _MCP_LEGACY_HTTP = _import_sdk_names("mcp.client.streamable_http", ("streamablehttp_client",)) _MCP_HTTP_AVAILABLE = _MCP_NEW_HTTP or _MCP_LEGACY_HTTP _import_sdk_names("mcp.types", ("LATEST_PROTOCOL_VERSION",), "mcp.types.LATEST_PROTOCOL_VERSION not available -- using fallback protocol version") if not _import_sdk_names("mcp.client.session", ("LATEST_HANDSHAKE_VERSION",)): # Pre-2.x SDKs: newest revision IS the handshake revision. LATEST_HANDSHAKE_VERSION = LATEST_PROTOCOL_VERSION if not _import_sdk_names("mcp.client.sse", ("sse_client",), "mcp.client.sse.sse_client not available -- SSE transport disabled"): sse_client = None _MCP_SAMPLING_TYPES, _MCP_ELICITATION_TYPES, _MCP_NOTIFICATION_TYPES = [ _import_sdk_names(*family) for family in _OPTIONAL_TYPE_FAMILIES] else: logger.debug("mcp package not installed -- MCP tool support disabled") if _MCP_AVAILABLE: try: _JSONRPC_METHOD_NOT_FOUND = importlib.import_module("mcp.types").METHOD_NOT_FOUND except Exception: # pragma: no cover — SDK without the constant pass _MCP_MESSAGE_HANDLER_SUPPORTED = _client_session_accepts("message_handler") if _MCP_AVAILABLE and not _MCP_MESSAGE_HANDLER_SUPPORTED: logger.debug("MCP SDK does not support message_handler -- dynamic tool discovery disabled") _MCP_LOGGING_CALLBACK_SUPPORTED = _client_session_accepts("logging_callback") _MCP_SDK_IMPORT_ATTEMPTED = True return _MCP_AVAILABLE _SDK_HTTPX_MOD = None def sdk_httpx(): """The httpx module the *installed* MCP SDK is built against (mcp 2.0 moved to ``httpx2``). Every object crossing the SDK boundary (AsyncClient, OAuth Request, exception classes) must come from the module the SDK itself imports or it fails at the transport layer. Resolved from the SDK's transport module, else the newest present; ``None`` if neither imports.""" global _SDK_HTTPX_MOD if _SDK_HTTPX_MOD is not None: return _SDK_HTTPX_MOD try: from mcp.client import streamable_http as _transport _SDK_HTTPX_MOD = getattr(_transport, "httpx2", None) or getattr(_transport, "httpx", None) except ImportError: _SDK_HTTPX_MOD = None for fallback in ("httpx2", "httpx"): if _SDK_HTTPX_MOD is not None: break try: _SDK_HTTPX_MOD = importlib.import_module(fallback) except ImportError: pass return _SDK_HTTPX_MOD def _client_session_accepts(kwarg: str) -> bool: """Whether this SDK's ``ClientSession.__init__`` takes ``kwarg`` (older SDKs lack ``message_handler`` and ``logging_callback``).""" if not _MCP_AVAILABLE: return False try: return kwarg in inspect.signature(ClientSession).parameters except (TypeError, ValueError): return False # MCP logging levels (RFC 5424 syslog severities) -> Python logging levels. # Port of anomalyco/opencode#34529's serverLog mapping. _MCP_LOG_LEVEL_MAP = { "debug": logging.DEBUG, "info": logging.INFO, "notice": logging.INFO, "warning": logging.WARNING, "error": logging.ERROR, "critical": logging.ERROR, "alert": logging.ERROR, "emergency": logging.ERROR} # ---- Reconnect / keepalive tuning ---- _DEFAULT_CONNECT_TIMEOUT = 60 # seconds for initial connection per server _MAX_RECONNECT_RETRIES = 5 _MAX_INITIAL_CONNECT_RETRIES = 3 # retries for the very first connection attempt _MAX_BACKOFF_SECONDS = 60 _RECYCLED_RECONNECT_TIMEOUT = 15.0 # Parked servers (tools deregistered) self-probe on this cadence: nothing else can revive them. _PARKED_RETRY_INTERVAL = 300 # Bounded wait for a respawned stdio child when a call finds it dead (gateway restarts kill # every MCP child); bounded so a broken server still parks via run()'s rapid-drop budget. _STDIO_RESPAWN_WAIT_SEC = 15.0 # Remote clients MUST ping faster than the server's idle-session TTL (short-TTL servers need a # smaller configured ``keepalive_interval``); stdio only opts in explicitly because local pipes # have no remote session TTL. The floor stops a tiny interval busy-looping. _DEFAULT_KEEPALIVE_INTERVAL, _MIN_KEEPALIVE_INTERVAL = 180, 5 # One bounded cancellation cycle at final shutdown so resistant tasks cannot hang exit. _MCP_LOOP_DRAIN_TIMEOUT = 3.0 # JSON-RPC 2.0 "method not found" (server without optional ``ping``); _ensure_mcp_sdk() # overrides it from mcp.types once loaded. _JSONRPC_METHOD_NOT_FOUND = -32601 # nextCursor pagination cap so a forever-cursor cannot spin discovery (50 pages = thousands). _MCP_LIST_MAX_PAGES = 50 async def _paginate_full_list(list_method, items_attr: str, server_name: str, cache_meta_out: Optional[dict] = None): """Drain a paginated ``list_*`` call by following ``nextCursor``; ``cache_meta_out`` gets the first page's SEP-2549 hints. Callers must hold the server's ``_rpc_lock``.""" items: list = [] cursor = None for _ in range(_MCP_LIST_MAX_PAGES): if not cursor: result = await list_method() else: # mcp 2.0 takes params=PaginatedRequestParams, 1.x takes cursor=. # Inspect before awaiting: an internal TypeError is not a signature mismatch. import inspect try: signature = inspect.signature(list_method) except (TypeError, ValueError): accepts_params = True # Opaque callables use the current SDK convention. else: accepts_params = any( p.kind == inspect.Parameter.VAR_KEYWORD or (p.name == "params" and p.kind != inspect.Parameter.POSITIONAL_ONLY) for p in signature.parameters.values() ) if accepts_params: import mcp.types as _types # late: keeps the SDK import lazy _params_cls = getattr(_types, "PaginatedRequestParams", None) if _params_cls is not None: result = await list_method(params=_params_cls(cursor=cursor)) else: result = await list_method(cursor=cursor) else: result = await list_method(cursor=cursor) if cache_meta_out is not None and not items: for key, snake, camel in (("ttl_ms", "ttl_ms", "ttlMs"), ("cache_scope", "cache_scope", "cacheScope")): hint = mcp_field(result, snake, camel) if hint is not None: cache_meta_out[key] = hint items.extend(getattr(result, items_attr, None) or []) cursor = mcp_field(result, "next_cursor", "nextCursor") # Cursor is an opaque string; anything else (incl. mocks) = last page. if not isinstance(cursor, str) or not cursor: break else: logger.warning("MCP server '%s': %s pagination exceeded %d pages; truncating at %d items", server_name, items_attr, _MCP_LIST_MAX_PAGES, len(items)) return items # ---- Server task -- each MCP server lives in one long-lived asyncio Task ---- class MCPServerTask(MCPServerRunMixin, MCPServerTransportMixin, MCPServerHealthMixin): """One MCP server connection in one long-lived asyncio Task (the transport's anyio cancel scopes must enter/exit in the same Task). Run state machine, transport bring-up and keepalive/liveness live in the three mixins.""" __slots__ = ( "name", "session", "tool_timeout", "_task", "_ready", "_shutdown_event", "_reconnect_event", "_tools", "_error", "_config", "_sampling", "_elicitation", "_registered_tool_names", "_auth_type", "_refresh_lock", "_rpc_lock", "_pending_refresh_tasks", "_pending_call_context", "_lifecycle_started_at", "_last_tool_call_at", "_idle_timeout_seconds", "_max_lifetime_seconds", "_recycled_reason", "initialize_result", "_ping_unsupported", "_list_cache_meta", "_reconnect_retries", "_session_proven", "_was_parked", "_inflight_tasks", "_reconnecting", "_suspect_reason", "_teardown_race", "_permanent_grace_used", "_stdio_child_pids", "_ever_connected", "_sse_fallback", "_park_reason", "_last_park_line", "_resolved_identity") def __init__(self, name: str): self.name = name self.session: Optional[Any] = None self.tool_timeout: float = _DEFAULT_TOOL_TIMEOUT self._task: Optional[asyncio.Task] = None self._ready = asyncio.Event() self._shutdown_event = asyncio.Event() # Set -> _run_http/_run_stdio exit cleanly and run() re-enters the transport. self._reconnect_event = asyncio.Event() self._tools: list = [] self._registered_tool_names: list[str] = [] self._config: dict = {} self._error: Optional[Exception] = None self._sampling: Optional[SamplingHandler] = None self._elicitation: Optional[ElicitationHandler] = None self._reconnect_retries: int = 0 # Rapid-drop budget: a session is UNPROVEN until it survives a keepalive interval or a # successful call; only a proven session clears the budget, so a post-handshake flapper # still parks. # Rapid-drop budget (#62212): a freshly (re)established session is UNPROVEN until it demonstrates # real health — it survived at least one full keepalive interval (keepalive success path) or served # at least one successful tool call. Only a proven session clears the reconnect budget; a transport # that flaps right after the handshake keeps getting charged and still reaches the park instead of # hot-cycling respawns forever. self._session_proven: bool = False # Never cleared (unlike _ready): separates first-connect from reconnect failures. self._ever_connected: bool = False # Latched when the Streamable HTTP -> SSE fallback connects: reconnects reuse SSE directly. self._sse_fallback: bool = False # Status/URL/body of the last HTTP rejection the Streamable HTTP client saw; names the real # cause when the SDK reports only ``Server returned an error response``. self._http_rejection: dict = {} # True from park until proven healthy again; logs the revival once. self._was_parked: bool = False # Why the server is parked (the revival_reason handed to _park), None once healthy again. # Lets cron preflight tell a network-blip park (recovering) from a permanent-error park # (revoked credentials, dead endpoint) that must not run the job tool-less forever. self._park_reason: Optional[str] = None self._last_park_line: Optional[str] = None # last park WARNING text; identical re-parks log at DEBUG # Digest of the resolved inputs the transport last connected with; a cross-profile adopter # must resolve the same digest in its own scope. None until the transport publishes it. self._resolved_identity: Optional[str] = None # In-flight RPC tasks so a deliberate teardown fails them fast; _reconnecting is True # during that teardown so _track_inflight_rpc turns the cancel into a retryable error. # In-flight RPC bookkeeping (#48069 salvage): user-visible requests registered while running so a # reconnect/shutdown teardown can fail them fast instead of orphaning them on a dying transport. self._inflight_tasks: set = set() self._reconnecting: bool = False # Latched by races (teardown-vs-keepalive, auth-lock corruption); ensure_healthy() # verifies before the next call. # See #77765, #81051, #84132. self._suspect_reason: Optional[str] = None # Teardown that failed in-flight calls => next reconnect is RACE RECOVERY, not a # budget charge. self._teardown_race: bool = False # One-time grace: auth/permanent failure on a PROVEN session gets one suspect+reconnect # cycle before parking. self._permanent_grace_used: bool = False # Children of the current stdio transport: in-flight calls fail FAST when one dies. # PIDs of the stdio subprocess spawned for the current transport (captured in _run_stdio). Used to # fail in-flight calls FAST when the child dies instead of waiting out the full tool timeout # (#81995). self._stdio_child_pids: Set[int] = set() self._auth_type: str = "" self._refresh_lock = asyncio.Lock() # A stdio session is one JSON-RPC stream (a concurrent list_tools can wedge a tool # call): serialize client-initiated RPCs per server (HTTP too, for ordering). self._rpc_lock = asyncio.Lock() self._pending_refresh_tasks: set[asyncio.Task] = set() # contextvars snapshot inside session.call_tool(): the SDK runs elicitation/create on a # task that does not inherit HERMES_SESSION_PLATFORM, so the callback replays this. self._pending_call_context: Optional[contextvars.Context] = None self._lifecycle_started_at = self._last_tool_call_at = time.monotonic() self._idle_timeout_seconds = self._max_lifetime_seconds = self._recycled_reason = None # Handshake InitializeResult: the server's REAL advertised capabilities. # Captures the ``InitializeResult`` returned by ``await session.initialize()`` so downstream code # can inspect the server's real advertised capabilities (``.capabilities.resources``, # ``.capabilities.prompts``) instead of assuming every ``ClientSession`` method attribute # corresponds to a supported server method. See #18051. self.initialize_result: Optional[Any] = None # SEP-2549 cache hints from the last tools/list (ttl_ms, cache_scope). self._list_cache_meta: dict = {} # Latched when ``ping`` returns -32601; keepalives then use list_tools. Reset per connect. self._ping_unsupported: bool = False # Content types a real Streamable-HTTP endpoint may return on the initial POST/GET; # anything else on a 2xx means the URL is not an MCP endpoint. _MCP_CONTENT_TYPES = ("application/json", "text/event-stream") # ---- Module-level state (every mutation under ``_lock``) ---- # # Every ledger below is keyed by the CONNECTION KEY from ``tools.mcp_tool_scope``: the bare # server name outside a multiplexer, ``(owner_scope, name)`` under one. Two profiles naming the # same server with their own credentials are two connections; a name-keyed ledger let the first # profile's connection shadow the second's (never connected, silently tool-less — #106005). _servers: Dict[Any, MCPServerTask] = {} # Profile registry scope per live connection (None outside multiplex) so a multiplexed # /reload-mcp tears down only its own profile's servers. _server_scope_keys: Dict[Any, Optional[str]] = {} # Registry scopes that have adopted a live server connection. The owning scope above remains # authoritative for connection teardown; this set preserves visibility for shared connections. _server_tool_scopes: Dict[Any, set] = {} _server_connecting: set = set() _server_connect_errors: Dict[Any, str] = {} # adopter scope -> server names whose shared connection an owner's scoped shutdown tore down; # drained by the next discovery pass so the adopter is re-registered (see mcp_tool_lifecycle). _orphaned_adopters: Dict[str, set] = {} # Lazy startup: servers registered from the schema cache without connecting; popped on # first real connection. # Keyed by connection key; entries are popped once a real connection is established on first use. See #56832. _lazy_server_configs: Dict[Any, dict] = {} _lazy_server_fingerprints: Dict[Any, str] = {} _lazy_server_tool_names: Dict[Any, List[str]] = {} # Task-local claim around ``_connect_server``: discovery retains a recoverable parked task # while standalone probes never publish failed servers into module-global ownership. _connect_server_claim: contextvars.ContextVar[Optional[Callable[[MCPServerTask], None]]] = ( contextvars.ContextVar("mcp_connect_server_claim", default=None)) # Per-server connect cooldown: a server that fails to spawn never reaches ``_servers``, so # without it every ``discover_mcp_tools()`` (one per worker session) would respawn it — a # restart storm whose unreaped children destabilise healthy servers. Exponential-backoff # deadline honoured by ``register_mcp_servers``; cleared on success. # Connection-retry cooldown (per-server isolation against restart storms). A single stdio MCP server that # fails to spawn (bad PATH, ``exec: not found``, crash-on-start) is never recorded in ``_servers`` -- # ``start()`` raises and ``_discover_and_register_server`` aborts before the ``_servers[name] = server`` # line. Without a cooldown, EVERY subsequent ``discover_mcp_tools()`` (one per agent worker session, i.e. # every few seconds) sees the server as "not connected" and re-spawns it from scratch. That is the restart # storm in #50394: the failing server is re-attempted on the shared MCP event loop on every worker session, # the subprocesses pile up unreaped, and the churn destabilises the healthy co-located servers (their tools # intermittently surface as "Unknown tool"). Fix: after a failed connection attempt, stamp a monotonic # ``retry_after`` deadline with exponential backoff. ``register_mcp_servers`` skips a server whose cooldown # has not elapsed, so a chronically failing server is retried on a backoff schedule instead of on every # worker session -- isolating it from the rest of the bridge. A successful connection clears the state. _server_connect_retry_after: Dict[Any, float] = {} # connection key -> monotonic deadline _server_connect_failures: Dict[Any, int] = {} # connection key -> consecutive failures _CONNECT_RETRY_BASE_BACKOFF_SEC, _CONNECT_RETRY_MAX_BACKOFF_SEC = 30.0, 600.0 # Per-server circuit breaker: closed -> open (calls short-circuit until the cooldown) -> # half-open (next call probes). Mutate only via _bump_server_error / _reset_server_error. # After _CIRCUIT_BREAKER_THRESHOLD consecutive failures, the handler returns a "server unreachable" message # that tells the model to stop retrying, preventing the 90-iteration burn loop described in #10447. State # machine: closed — error count below threshold; all calls go through. open — threshold reached; # calls short-circuit until the cooldown elapses. half-open — cooldown elapsed; the next call is a probe # that actually hits the session. Probe success → closed. Probe failure → reopens (cooldown re-armed). # ``_server_breaker_opened_at`` records the monotonic timestamp when the breaker most recently transitioned # into the open state. Use the ``_bump_server_error`` / ``_reset_server_error`` helpers to mutate this state # — they keep the count and timestamp in sync. _server_error_counts: Dict[Any, int] = {} _server_breaker_opened_at: Dict[Any, float] = {} # True while every strike in the current streak was the tool's own error payload (server reachable, # call rejected); picks the open-breaker wording, since "unreachable" was false for that case (#11113). _server_errors_all_application: Dict[Any, bool] = {} _CIRCUIT_BREAKER_THRESHOLD, _CIRCUIT_BREAKER_COOLDOWN_SEC = 3, 60.0 # Trust-tier gating (``trust: full | untrusted``): on an untrusted server every write-capable # call (discovery-time ``readOnlyHint`` not exactly True; malformed fails closed) needs approval # before the RPC fires. A lying readOnlyHint can only skip approval for calls the operator was # already warned about, never widen access. Missing trust = full; unrecognized = untrusted (a # typo must never disable the gate). Classified at CALL time from DISCOVERY data: no schema # mutation, prompt cache intact. ``_server_trust_levels`` is keyed by the CONSUMING profile's own # key (its policy for the name, even when it adopted another profile's connection); # ``_tool_read_only_hints`` by the connection key (the server's own tool annotations). _server_trust_levels: Dict[Any, str] = {} _tool_read_only_hints: Dict[Any, Dict[str, bool]] = {} _TRUST_FULL, _TRUST_UNTRUSTED = "full", "untrusted" def _bump_server_error(server_name: str, *, application: bool = False) -> None: """Count a failure; at the threshold (re)stamp the breaker-open time. Keyed by the calling scope's connection so one profile's failing server never opens another profile's breaker. *application*: the call completed and the payload was an error (transport is fine).""" from tools.mcp_tool_scope import _resolve_server_key key = _resolve_server_key(server_name) n = _server_error_counts.get(key, 0) + 1 _server_error_counts[key] = n _server_errors_all_application[key] = application and (n == 1 or _server_errors_all_application.get(key, False)) if n >= _CIRCUIT_BREAKER_THRESHOLD: _server_breaker_opened_at[key] = time.monotonic() def _reset_server_error(server_name: str) -> None: """Close the breaker on any unambiguous success signal.""" from tools.mcp_tool_scope import _resolve_server_key key = _resolve_server_key(server_name) _server_error_counts[key] = 0 _server_breaker_opened_at.pop(key, None) _server_errors_all_application.pop(key, None) # Servers opted into parallel tool calls, keyed by the consuming profile's own key (``foo-bar``/ # ``foo_bar`` sanitize alike but must not share policy; neither do two profiles' same-named servers). _parallel_safe_servers: set = set() # registry tool name -> raw server name (the generated name is lossy; never re-parse it). _mcp_tool_server_names: Dict[str, str] = {} # Dedicated event loop in a background daemon thread; _lock guards the loop handles, _servers, # the status maps and the PID ledgers. _mcp_loop: Optional[asyncio.AbstractEventLoop] = None _mcp_thread: Optional[threading.Thread] = None _lock = threading.Lock() # ---- Shared parent-death supervisor (state lives HERE: tests rebind ``_death_supervisor``) ---- # If this process dies without running its cleanup path (kill -9, OOM, crash, force-quit), stdio # MCP children reparent to init and run forever; macOS has no PR_SET_PDEATHSIG, so something has # to outlive us and reap them. ONE supervisor process serves all stdio servers and is told which # process groups to reap over a pipe; it detects our death as EOF on that pipe (exact, instant) # rather than polling getppid(). Replaced the per-server watchdog wrapper (~10 MB resident per # server, plus a signal-forwarding layer because wrapping put the server in a different session # from the pgid tracked for killpg). See tools/mcp_death_supervisor.py. POSIX-only, matching the # killpg-based orphan cleanup below. _death_supervisor = None # Optional[subprocess.Popen] _death_supervisor_lock = threading.Lock() # Groups the supervisor is reaping on our behalf; replayed verbatim on respawn so a respawn never # silently drops coverage for servers that are still running. _supervised_pgids: set = set() def _spawn_death_supervisor(): """Start the shared supervisor, or None if it cannot be started.""" import subprocess supervisor = os.path.join(os.path.dirname(os.path.abspath(__file__)), "mcp_death_supervisor.py") try: # start_new_session=True is load-bearing: shutdown paths killpg this process's own group, # which would kill the supervisor before it could reap anything. return subprocess.Popen( [sys.executable, supervisor, "--parent-pgid", str(os.getpgid(0))], stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=_get_mcp_stderr_log(), start_new_session=True, close_fds=True, text=True) except Exception: # Never let supervisor bookkeeping block a real MCP connection: graceful shutdown paths # still reap normally; only the ungraceful-exit safety net is lost. logger.debug("Could not start the MCP parent-death supervisor", exc_info=True) return None def _prune_dead_supervised_pgids() -> set: """Forget supervised groups with no members left; return what went. Caller holds ``_death_supervisor_lock``. Signal 0 is a pure existence probe (cannot terminate anything). It narrows, but cannot close, the window where a dead group's pgid is recycled before we notice (residual-risk note in ``tools/mcp_death_supervisor.py``).""" killpg = getattr(os, "killpg", None) if killpg is None: # windows-footgun: ok - POSIX-only, guarded return set() stale = set() for pgid in list(_supervised_pgids): try: killpg(pgid, 0) except ProcessLookupError: stale.add(pgid) except (PermissionError, OSError): # Exists but not ours to signal, or the probe failed: keep it — dropping coverage on # an ambiguous answer is the more expensive mistake. pass _supervised_pgids.difference_update(stale) return stale def _update_death_supervisor(verb: str, pgids) -> None: """Register or unregister process groups (``verb`` is ``"register"``/``"unregister"``) with the shared supervisor. Failures are swallowed: losing the safety net must never fail a live MCP session.""" if os.name != "posix": return wanted = {int(pgid) for pgid in pgids} if not wanted: return global _death_supervisor with _death_supervisor_lock: if verb == "register": _supervised_pgids.update(wanted) else: _supervised_pgids.difference_update(wanted) # A registration outlives the server only while some member survives (e.g. an orphaned # grandchild teardown failed to kill, deliberately kept registered). Once that group is # empty its pgid can be recycled by a stranger, so prune here too — the orphan sweep # unregisters what it reaps but is not guaranteed to run in a given process. stale = _prune_dead_supervised_pgids() proc = _death_supervisor if proc is None or proc.poll() is not None: if not _supervised_pgids: # Nothing left to cover: nothing to tell and nothing to respawn for. Keyed on # the SET, not the verb: after a broken-pipe write dropped the supervisor with # groups still registered, an unregister must still rebuild coverage for the # survivors. return # See #93517. proc = _spawn_death_supervisor() _death_supervisor = proc if proc is None: return # A fresh supervisor knows nothing: replay live coverage (already reflects this # call's mutation and the prune, so pruned groups never reach the replacement). payload = "".join(f"register {pgid}\n" for pgid in _supervised_pgids) else: payload = "".join(f"{verb} {pgid}\n" for pgid in wanted) payload += "".join(f"unregister {pgid}\n" for pgid in stale) try: proc.stdin.write(payload) proc.stdin.flush() except (BrokenPipeError, ValueError, OSError): # It exited between poll() and write(). Drop it so the next call respawns and replays # from ``_supervised_pgids`` (the set, not the pipe, is the record of what needs reaping). _death_supervisor = None return if not _supervised_pgids: # Nothing left to reap: release the supervisor rather than keep a ~15 MB process and a # pipe resident for the life of a gateway. Closing our write end is the same EOF parent # death sends; with an empty set it exits. The next register respawns and replays. try: proc.stdin.close() except (BrokenPipeError, ValueError, OSError): pass # Reap it, or the exited supervisor stays a zombie until the next Popen in this process. try: proc.wait(timeout=5) except Exception: # noqa: BLE001 - timeout or already gone; either way we drop it pass _death_supervisor = None def _mcp_registry_scope() -> Optional[str]: """Registry scope for MCP registrations: a profile overlay when this process serves profiles, else None. Under ``gateway.multiplex_profiles`` every turn runs scoped; a process that serves a routed profile through the HERMES_HOME override (dashboard/desktop backend, per-profile cron ticker) is a multiplexer too, even with the flag off — keying its connections by the bare name would hand one profile's credentialed connection to every other served profile (#111151). Single-profile processes (no override, or an override naming their own home) keep bare names.""" from agent.secret_scope import serves_routed_profile if not serves_routed_profile(): return None from tools.registry import registry return registry.current_scope_key() def _server_registry_scope(key) -> Optional[str]: """Scope owning the connection under *key*'s tools: the one captured at adoption (teardown runs on the MCP loop without the discovering profile's context), else the current one.""" if key in _server_scope_keys: return _server_scope_keys[key] from tools.mcp_tool_scope import _key_scope return _key_scope(key) or _mcp_registry_scope() def _server_visible_in_scope(key, scope: Optional[str]) -> bool: """Whether the live connection under *key* is visible from ``scope`` without changing its teardown owner.""" if scope is None: return True return (_server_scope_keys.get(key) == scope or scope in _server_tool_scopes.get(key, ())) # Cross-process discovery guard: advisory file lock so gateway + CLI + TUI don't all discover. # See issue #62771. _LOCK_UNAVAILABLE: Any = object() # sentinel: locking broken/unavailable _MCP_DISCOVERY_LOCK_PATH: Optional[str] = None # resolved lazily # A discovery pass (bounded gathers, one 120 s budget per wave) may legitimately # run past the 120 s a scatter completes in. The waiter must outlast the worst # legitimate holder (pass ceiling + slack), or it fails over at 120 s, discovers # unguarded beside a still-connecting holder, and spawns duplicate stdio trees. # See #117373: the concurrency cap made the old 120 s waiter budget stale. _MCP_DISCOVERY_PASS_MAX_SEC = 300 # overall pass ceiling _MCP_DISCOVERY_LOCK_RETRY_DELAY_S = 0.5 # Waiter budget (max_retries * delay) must outlast the pass ceiling: 320 s > 300 s. _MCP_DISCOVERY_LOCK_MAX_RETRIES = int( _MCP_DISCOVERY_PASS_MAX_SEC / _MCP_DISCOVERY_LOCK_RETRY_DELAY_S) + 20