6d8d565a7c
The connecting loop hashed the resolved inputs, then the transport resolved them again; a runtime-file rotation between the two reads published endpoint A's digest alongside a session opened with B. The transport now resolves once per attempt (_http_endpoint / _stdio_launch), connects with that value and publishes its digest; the adopter recomputes through the same resolvers. The loop clears the digest per attempt. (cherry picked from commit f543a4d3aadcdf7ec10093ebb99e14a740d43494)
559 lines
32 KiB
Python
559 lines
32 KiB
Python
"""Lifecycle of :class:`tools.mcp_tool.MCPServerTask`: the long-lived ``run`` state machine
|
|
(connect -> serve -> reconnect/park/recycle), keepalive-driven lifecycle waits, start/shutdown
|
|
and tool deregistration. Origin state and patchable helpers are read through ``_core`` so
|
|
``mock.patch("tools.mcp_tool.X")`` keeps working."""
|
|
|
|
import asyncio
|
|
import logging
|
|
import time
|
|
from dataclasses import dataclass
|
|
from typing import Optional
|
|
from tools.mcp_tool_common import _core, _get_lifecycle_seconds, _jittered, _resolve_tool_timeout
|
|
from tools import mcp_tool_errors as _errors
|
|
from tools import mcp_tool_registration as _registration
|
|
from tools import mcp_tool_sampling as _sampling
|
|
|
|
logger = logging.getLogger("tools.mcp_tool")
|
|
|
|
|
|
@dataclass
|
|
class _RetryBudget:
|
|
"""Per-run() retry counters (``_reconnect_retries`` stays on the task: handlers/tests read it)."""
|
|
|
|
initial_retries: int = 0
|
|
backoff: float = 1.0
|
|
|
|
|
|
class MCPServerRunMixin:
|
|
"""Methods of :class:`tools.mcp_tool.MCPServerTask` (mixed in; relies on its attributes)."""
|
|
|
|
@staticmethod
|
|
async def _cancel_waiters(*tasks: asyncio.Task) -> None:
|
|
for t in tasks:
|
|
if not t.done():
|
|
t.cancel()
|
|
try:
|
|
await t
|
|
except (asyncio.CancelledError, Exception):
|
|
pass
|
|
|
|
def _event_waiters(self) -> tuple:
|
|
"""Fresh ``(shutdown, reconnect)`` wait tasks; cancel them via ``_cancel_waiters``."""
|
|
return (asyncio.ensure_future(self._shutdown_event.wait()),
|
|
asyncio.ensure_future(self._reconnect_event.wait()))
|
|
|
|
def _recycle_if_due(self) -> bool:
|
|
"""Latch a stdio idle/lifetime recycle when its deadline has passed."""
|
|
recycle_reason = self._stdio_recycle_reason()
|
|
if recycle_reason is None:
|
|
return False
|
|
self._mark_stdio_recycled(recycle_reason)
|
|
return True
|
|
|
|
async def _wait_for_rpc_idle(self) -> None:
|
|
"""Wake the lifecycle loop when the active RPC releases its lock."""
|
|
async with self._rpc_lock:
|
|
pass
|
|
|
|
async def _wait_for_lifecycle_event(self) -> str:
|
|
"""Serve until a lifecycle event: ``"shutdown"`` (exits run), ``"reconnect"`` (session torn
|
|
down, transport re-entered; event cleared first) or ``"recycle"`` (stdio idle/lifetime
|
|
limit; restarts lazily on next call). Shutdown wins a tie. Remote transports run a
|
|
keepalive (``ping``, with a ``list_tools`` fallback for servers lacking the optional ping
|
|
utility — see :meth:`_keepalive_probe`) every ``keepalive_interval`` (which must stay
|
|
below the server's session TTL) so idle TCP/session state never goes stale (#17003);
|
|
stdio does so only when explicitly configured. A keepalive failure triggers a reconnect.
|
|
"""
|
|
is_http = self._is_http()
|
|
configured = self._config.get("keepalive_interval")
|
|
keepalive_interval = None
|
|
if is_http or configured is not None:
|
|
keepalive_interval = max(
|
|
_core._MIN_KEEPALIVE_INTERVAL,
|
|
float(_core._DEFAULT_KEEPALIVE_INTERVAL if configured is None else configured))
|
|
# No keepalive, but an unproven stdio session must still get its chance to prove
|
|
# itself: it counts as proven only once a FULL default interval has elapsed — not on
|
|
# the first timeout wake, which a shorter recycle deadline may cause (see below).
|
|
proof_at = None
|
|
if keepalive_interval is None and not self._session_proven:
|
|
proof_at = time.monotonic() + _core._DEFAULT_KEEPALIVE_INTERVAL
|
|
shutdown_task, reconnect_task = self._event_waiters()
|
|
rpc_idle_task = None
|
|
waiters = [shutdown_task, reconnect_task]
|
|
try:
|
|
while True:
|
|
if self._recycle_if_due():
|
|
return "recycle"
|
|
timeout = keepalive_interval
|
|
if timeout is None and not self._session_proven and proof_at is not None:
|
|
timeout = max(0.0, proof_at - time.monotonic())
|
|
recycle_deadline = self._next_stdio_recycle_deadline()
|
|
if recycle_deadline is not None:
|
|
recycle_timeout = max(0.0, recycle_deadline - time.monotonic())
|
|
timeout = recycle_timeout if timeout is None else min(timeout, recycle_timeout)
|
|
elif not is_http and self._rpc_lock.locked():
|
|
# Recycle deadlines are intentionally hidden while an RPC is active. Without
|
|
# a default stdio keepalive timeout, lock release must wake this loop so the
|
|
# now-visible deadline is evaluated instead of waiting forever. (For a stdio
|
|
# server with no limits the wake is a harmless extra iteration.)
|
|
rpc_idle_task = rpc_idle_task or asyncio.ensure_future(self._wait_for_rpc_idle())
|
|
waiters = [t for t in (shutdown_task, reconnect_task, rpc_idle_task) if t is not None]
|
|
done, _pending = await asyncio.wait(
|
|
waiters, timeout=timeout, return_when=asyncio.FIRST_COMPLETED)
|
|
if shutdown_task in done or reconnect_task in done:
|
|
break
|
|
if rpc_idle_task in done:
|
|
rpc_idle_task = None
|
|
continue
|
|
if self._recycle_if_due():
|
|
return "recycle"
|
|
if keepalive_interval is None:
|
|
# Stdio without a keepalive: idling a full default interval with the child
|
|
# still alive is the proof of health a successful ping gives remote
|
|
# transports — clear the rapid-drop budget without pinging (#62212). An
|
|
# earlier wake (a hidden-then-missed recycle deadline) is not that proof.
|
|
if (not self._session_proven and proof_at is not None
|
|
and time.monotonic() >= proof_at):
|
|
if self._stdio_children_dead():
|
|
# A dead child never fires a lifecycle event, so continuing here
|
|
# would spin on zero-timeout waits forever (#115483). No mark_suspect:
|
|
# the reconnect rebuilds the transport and the new child's handshake is
|
|
# the health check.
|
|
logger.warning("MCP server '%s' stdio child exited before the session proved "
|
|
"healthy; triggering reconnect (state: connected → degraded)",
|
|
self.name)
|
|
break
|
|
self._mark_session_proven()
|
|
continue
|
|
# Timeout: probe for a stale session — NEVER while an RPC is in flight (a
|
|
# concurrent ping can wedge the stdio stream; a busy server is alive anyway).
|
|
# Timeout — no lifecycle event fired. See #48069.
|
|
if self.session:
|
|
if self._rpc_lock.locked() or any(not t.done() for t in self._inflight_tasks):
|
|
continue
|
|
try:
|
|
async with self._rpc_lock:
|
|
await self._keepalive_probe()
|
|
except Exception as exc:
|
|
root = _errors._unwrap_exception_group(exc)
|
|
logger.warning("MCP server '%s' keepalive failed, triggering reconnect (state: connected → "
|
|
"degraded): %s: %s", self.name, type(root).__name__, root)
|
|
self.mark_suspect(f"keepalive failed: {type(root).__name__}: {root}")
|
|
self._reconnect_event.set()
|
|
break
|
|
# Survived a full keepalive interval: real proof of health.
|
|
# Clear the rapid-drop budget (#62212).
|
|
self._mark_session_proven()
|
|
finally:
|
|
await self._cancel_waiters(*waiters)
|
|
if self._shutdown_event.is_set():
|
|
self._fail_inflight_calls("shutdown")
|
|
return "shutdown"
|
|
# Deliberate teardown: fail in-flight RPCs NOW instead of riding out the tool timeout.
|
|
# See #48069, #81995.
|
|
self._fail_inflight_calls("reconnect")
|
|
self._reconnect_event.clear()
|
|
return "reconnect"
|
|
|
|
async def _wait_for_reconnect_or_shutdown(self, timeout: Optional[float] = None) -> str:
|
|
"""Parked wait: ``"shutdown"``, ``"reconnect"`` (explicit request; event cleared first) or
|
|
``"self-probe"`` (``timeout`` elapsed with neither). Shutdown wins a tie."""
|
|
shutdown_task, reconnect_task = self._event_waiters()
|
|
try:
|
|
await asyncio.wait({shutdown_task, reconnect_task}, return_when=asyncio.FIRST_COMPLETED, timeout=timeout)
|
|
finally:
|
|
await self._cancel_waiters(shutdown_task, reconnect_task)
|
|
if self._shutdown_event.is_set():
|
|
return "shutdown"
|
|
if not self._reconnect_event.is_set():
|
|
return "self-probe"
|
|
self._reconnect_event.clear()
|
|
return "reconnect"
|
|
|
|
def _log_park(self, msg: str, *args) -> None:
|
|
"""Park chatter control (#115713): re-parking a server that never revived is not a state
|
|
transition — ``hermes mcp list`` already surfaces the parked state, so one identical
|
|
WARNING per self-probe carries no new information. The first park (and the revived line
|
|
in ``_mark_session_proven``) stays a WARNING; an identical repeat while still parked is
|
|
demoted to DEBUG so a long-lived gateway's error log is not flooded (10k+ identical
|
|
lines/month). A park for a DIFFERENT reason (auth error after connection refused) is new
|
|
information and warns again."""
|
|
line = msg % args if args else msg
|
|
if self._was_parked and line == self._last_park_line:
|
|
logger.debug(msg, *args)
|
|
else:
|
|
self._last_park_line = line
|
|
logger.warning(msg, *args)
|
|
|
|
async def _park(self, revival_reason: str) -> bool:
|
|
"""Drop this server's tools and wait for a reconnect request; True when shutdown came instead.
|
|
The run task must NOT exit (it is the only ``_reconnect_event`` listener, so returning
|
|
leaves the server unrevivable). With tools deregistered no call can reach the breaker
|
|
probe, so the wait is TIMED (one self-probe per ``_PARKED_RETRY_INTERVAL``); an explicit
|
|
``_reconnect_event.set()`` wakes it immediately."""
|
|
# Do NOT return — exiting the task orphans the server: nothing would ever listen for
|
|
# _reconnect_event again and the server would be permanently wedged for the life of the process
|
|
# (#16788). Instead, drop the phantom tools from the registry and park. Because parking deregisters
|
|
# the tools, no tool call can reach the circuit-breaker half-open probe or _signal_reconnect — so
|
|
# the park is a TIMED wait: every _PARKED_RETRY_INTERVAL we wake and attempt one reconnect ourselves
|
|
# (#57129). An explicit _reconnect_event.set() (OAuth recovery, manual /mcp refresh) still wakes us
|
|
# immediately.
|
|
self._was_parked = True
|
|
self._park_reason = revival_reason
|
|
self._deregister_tools()
|
|
self._reconnect_event.clear()
|
|
paused = False
|
|
while True:
|
|
outcome = await self._wait_for_reconnect_or_shutdown(
|
|
timeout=_core._PARKED_RETRY_INTERVAL)
|
|
if outcome == "shutdown":
|
|
return True
|
|
# A disabled or deleted config entry must stop the self-probe here: the probe
|
|
# rebuilds the transport from the config captured at start, so it would re-run
|
|
# OAuth setup for a server the user turned off, every interval, for the life of
|
|
# the process (background loops that do not run the gateway reconcile tick
|
|
# never learn the entry changed). An explicit reconnect request — manual
|
|
# refresh or `hermes mcp login` — still revives immediately regardless of the
|
|
# config gate; only the unattended probe honours it. Announce the pause once:
|
|
# a line per skipped wake would be the very flood this gate exists to stop.
|
|
if outcome == "self-probe" and not self._still_configured_enabled():
|
|
(logger.debug if paused else logger.info)(
|
|
"MCP server '%s': parked entry is disabled or gone from mcp_servers; pausing the "
|
|
"self-probe until it is re-enabled", self.name)
|
|
paused = True
|
|
continue
|
|
# Nobody asked for this revival: a self-probe must never open a browser OAuth flow. The
|
|
# OAuth provider runs inside THIS task (the SDK's auth flow sits in the transport), so a
|
|
# task-local ContextVar reaches it; it stays set for the task's life — every later
|
|
# revival of a once-parked server is unattended too. Left interactive, an expired
|
|
# refresh token opened a new authorize tab every _PARKED_RETRY_INTERVAL, all night.
|
|
if outcome == "self-probe":
|
|
from tools.mcp_oauth import _oauth_interactive_enabled
|
|
|
|
_oauth_interactive_enabled.set(False)
|
|
logger.debug(
|
|
"MCP server '%s': attempting revival %s (%s); rebuilding transport.",
|
|
self.name,
|
|
revival_reason,
|
|
outcome,
|
|
)
|
|
return False
|
|
|
|
def _still_configured_enabled(self) -> bool:
|
|
"""Whether ``mcp_servers`` on disk still wants this server connected: entry present
|
|
and ``enabled`` not false. Config read is cached on the file signature, so a parked
|
|
task polling this every ``_PARKED_RETRY_INTERVAL`` stays cheap. Fail-open on a
|
|
config-read error: a broken config must not wedge a healthy server's revival."""
|
|
try:
|
|
from tools import mcp_tool_config as _config
|
|
entry = (_config._load_mcp_config() or {}).get(self.name)
|
|
if entry is None:
|
|
return False
|
|
from tools.mcp_tool_common import mcp_server_enabled
|
|
return mcp_server_enabled(entry)
|
|
except Exception:
|
|
return True
|
|
|
|
async def _prepare_run(self, config: dict) -> bool:
|
|
"""Bind config, build sampling/elicitation handlers, validate HTTP. False when the server
|
|
must not start (bad remote URL / non-MCP endpoint: fail fast with ``_error`` set and
|
|
``_ready`` fired instead of burning the reconnect ladder inside the SDK's httpx layer)."""
|
|
self._config = config
|
|
self.tool_timeout = _resolve_tool_timeout(config)
|
|
self._auth_type = (config.get("auth") or "").lower().strip()
|
|
self._idle_timeout_seconds = _get_lifecycle_seconds(config, "idle_timeout_seconds")
|
|
self._max_lifetime_seconds = _get_lifecycle_seconds(config, "max_lifetime_seconds")
|
|
# The _MCP_*_TYPES flags are False until the lazy SDK import runs.
|
|
_core._ensure_mcp_sdk()
|
|
sampling_config = config.get("sampling", {})
|
|
self._sampling = (_sampling.SamplingHandler(self.name, sampling_config)
|
|
if sampling_config.get("enabled", True) and _core._MCP_SAMPLING_TYPES else None)
|
|
# elicitation/create lets a server ask for structured input mid-call; the handler
|
|
# routes it through Hermes' approval system.
|
|
elicitation_config = config.get("elicitation", {})
|
|
self._elicitation = (_sampling.ElicitationHandler(self.name, elicitation_config,
|
|
call_context=lambda: self._pending_call_context)
|
|
if elicitation_config.get("enabled", True) and _core._MCP_ELICITATION_TYPES else None)
|
|
if "url" in config and "command" in config:
|
|
logger.warning("MCP server '%s' has both 'url' and 'command' in config. Using HTTP transport "
|
|
"('url'). Remove 'command' to silence this warning.", self.name)
|
|
if not self._is_http():
|
|
return True
|
|
try:
|
|
_errors._validate_remote_mcp_url(self.name, config.get("url"))
|
|
# Content-type preflight (Streamable HTTP only; SSE serves text/event-stream): a
|
|
# web-app root returns HTML and would hang the SDK for connect_timeout. Skipped once
|
|
# _ready was ever set and for OAuth servers (a token-less probe sees HTML/401).
|
|
from tools.mcp_liveness import liveness_for
|
|
if (config.get("transport") != "sse" and not config.get("skip_preflight")
|
|
and liveness_for(self.name).kind != "server_json"
|
|
and not self._ready.is_set() and self._auth_type != "oauth"):
|
|
await self._preflight_content_type(
|
|
config["url"], headers=dict(config.get("headers") or {}),
|
|
ssl_verify=config.get("ssl_verify", True),
|
|
client_cert=_errors._resolve_client_cert(self.name, config),
|
|
strict_redirect_headers=bool(config.get("strict_redirect_headers")))
|
|
except (_errors.InvalidMcpUrlError, _errors.NonMcpEndpointError) as exc:
|
|
logger.warning("%s", exc)
|
|
self._publish_error(exc) # fail fast and non-retryably
|
|
return False
|
|
return True
|
|
|
|
def _publish_error(self, exc: BaseException) -> None:
|
|
"""Hand *exc* to the waiting ``start()``."""
|
|
self._error = exc
|
|
self._ready.set()
|
|
|
|
_REMOTE_REBIND_KEYS = ("url", "auth", "oauth", "headers", "transport")
|
|
|
|
def _refresh_remote_config(self, config: dict) -> dict:
|
|
"""Before rebuilding a remote transport, re-read this server's definition from config.yaml and
|
|
adopt it when the endpoint or its auth changed (#113907). ``run()`` otherwise keeps the dict it
|
|
was started with, so a ``url`` edited while the process runs (dashboard/Desktop re-auth, a
|
|
catalog migration) left the loop probing the OLD URL — and that stale-URL provider evicted the
|
|
fresh one the dashboard had just authorised for the new URL, dropping its tokens/DCR client."""
|
|
if not self._is_http():
|
|
return config
|
|
from tools import mcp_tool_config as _config
|
|
fresh = (_config._load_mcp_config() or {}).get(self.name)
|
|
if not isinstance(fresh, dict) or "url" not in fresh or all(
|
|
fresh.get(k) == config.get(k) for k in self._REMOTE_REBIND_KEYS):
|
|
return config
|
|
logger.info("MCP server '%s': definition changed in config.yaml (%s -> %s); rebuilding with the new one",
|
|
self.name, config.get("url"), fresh.get("url"))
|
|
self._config = fresh
|
|
self._auth_type = (fresh.get("auth") or "").lower().strip()
|
|
self._sse_fallback = False # latched for the old endpoint
|
|
return fresh
|
|
|
|
async def run(self, config: dict):
|
|
"""Long-lived: connecting -> connected -> (degraded -> parked -> revived)*. Unproven drops
|
|
and transport errors charge a rapid-drop budget with jittered backoff; exhausting it (or
|
|
a permanent error) parks via :meth:`_park` rather than exiting, so the server stays
|
|
revivable. Branch helpers return True to keep looping, False to exit."""
|
|
if not await self._prepare_run(config):
|
|
return
|
|
self._reconnect_retries = 0
|
|
budget = _RetryBudget()
|
|
rebuild = False
|
|
while True:
|
|
try:
|
|
if rebuild:
|
|
config = self._refresh_remote_config(config)
|
|
rebuild = True
|
|
run_transport = self._run_http if self._is_http() else self._run_stdio
|
|
self._resolved_identity = None # the transport publishes the inputs it connects with
|
|
if not await self._on_clean_return(await run_transport(config), budget):
|
|
break
|
|
except asyncio.CancelledError:
|
|
# Not a connection failure: re-raise so shutdown()'s ``await self._task`` completes.
|
|
# Task was cancelled (shutdown, gateway restart, explicit task.cancel()). Don't treat this
|
|
# as a connection failure — CancelledError inherits from BaseException (not Exception) in
|
|
# Python 3.11+, so the broad ``except Exception`` below would NOT catch it; we'd silently
|
|
# exit the reconnect loop and the MCP server would stay dead until Hermes is fully
|
|
# restarted. See #9930.
|
|
self.session = None
|
|
raise
|
|
except Exception as exc:
|
|
self.session = None
|
|
if not await self._on_transport_error(exc, budget):
|
|
break
|
|
finally:
|
|
self.session = None
|
|
# Stale PIDs must never fast-fail the NEXT transport's calls.
|
|
self._stdio_child_pids = set()
|
|
|
|
async def _on_clean_return(self, lifecycle_reason: str, budget: "_RetryBudget") -> bool:
|
|
"""Clean transport return: shutdown, stdio recycle, or a requested rebuild (not a failure
|
|
for the retry counters)."""
|
|
if self._shutdown_event.is_set():
|
|
return False
|
|
if lifecycle_reason == "recycle":
|
|
logger.info("MCP server '%s': stdio session recycled after %s; waiting for lazy reconnect",
|
|
self.name, self._recycled_reason)
|
|
self.session = None
|
|
# Dormant until a lazy call wakes it (untimed: nothing to self-probe).
|
|
return await self._wait_for_reconnect_or_shutdown() != "shutdown"
|
|
# Per-cycle chatter stays DEBUG; WARNINGs mark state transitions.
|
|
logger.debug("MCP server '%s': reconnecting (OAuth recovery or manual refresh)", self.name)
|
|
# A clean return is NOT proof of health (a flapper handshakes fine, then drops). Only a
|
|
# PROVEN session clears the budget; a teardown race is recovery, never a park charge.
|
|
# A clean transport return means a session was established and then asked to rebuild (auth recovery
|
|
# / manual refresh / keepalive failure / transport TaskGroup drop). That alone is NOT proof of
|
|
# health: a flapping transport handshakes fine and drops moments later, and resetting the budget
|
|
# here let such servers respawn forever (#62212 — 6212 spawns in 63h). Only clear the
|
|
# consecutive-failure budget once the session PROVED healthy — survived >=1 full keepalive interval
|
|
# or served >=1 successful tool call (_mark_session_proven).
|
|
if self._teardown_race and not self._session_proven:
|
|
logger.info("MCP server '%s': reconnect after teardown race (in-flight calls were failed); "
|
|
"not charging the rapid-drop budget", self.name)
|
|
self._teardown_race, budget.backoff = False, 1.0
|
|
elif self._session_proven:
|
|
self._reconnect_retries, budget.backoff = 0, 1.0
|
|
else:
|
|
self._reconnect_retries += 1
|
|
if self._reconnect_retries > _core._MAX_RECONNECT_RETRIES:
|
|
self._log_park(
|
|
"MCP server '%s': %d consecutive reconnects without a healthy session (rapid-drop budget "
|
|
"exhausted), parking; will self-probe every %ds until it recovers (state: degraded → parked)",
|
|
self.name, _core._MAX_RECONNECT_RETRIES, _core._PARKED_RETRY_INTERVAL)
|
|
if not await self._park_and_rearm("from parked state", budget):
|
|
return False
|
|
# Clear readiness too: a stale _ready lets handler recovery mistake old for fresh.
|
|
self._ready.clear()
|
|
self.session = None
|
|
return True
|
|
|
|
async def _park_and_rearm(self, revival_reason: str, budget: "_RetryBudget") -> bool:
|
|
"""Park; on revival leave ONE probe per wake so a still-dead server re-parks instead of
|
|
burning 5 rapid retries. False on shutdown."""
|
|
if await self._park(revival_reason):
|
|
return False
|
|
self._reconnect_retries, budget.backoff = _core._MAX_RECONNECT_RETRIES, 1.0
|
|
return True
|
|
|
|
async def _park_initial_failure(self, exc: Exception, revival_reason: str, budget: "_RetryBudget") -> bool:
|
|
"""Publish ``exc`` to ``start()``, park, and on revival reset every counter. False on shutdown."""
|
|
self._publish_error(exc)
|
|
if await self._park(revival_reason):
|
|
return False
|
|
budget.initial_retries = self._reconnect_retries = 0
|
|
budget.backoff = 1.0
|
|
self._error = None
|
|
self._ready.clear()
|
|
return True
|
|
|
|
async def _backoff_sleep(self, budget: "_RetryBudget") -> None:
|
|
await asyncio.sleep(_jittered(budget.backoff))
|
|
budget.backoff = min(budget.backoff * 2, _core._MAX_BACKOFF_SECONDS)
|
|
|
|
async def _on_transport_error(self, exc: Exception, budget: "_RetryBudget") -> bool:
|
|
"""Transport raised: classify, then run the initial-connect or reconnect ladder. False = exit."""
|
|
# Unwrap anyio TaskGroup wrappers: the group's str() hides the root cause.
|
|
root = _errors._unwrap_exception_group(exc)
|
|
failure_class = _errors._classify_mcp_failure(root)
|
|
if self._is_recycled_stdio():
|
|
logger.warning("MCP server '%s': lazy reconnect after stdio recycle failed, marking unavailable "
|
|
"while retrying: %s: %s", self.name, type(root).__name__, root)
|
|
self._recycled_reason = None
|
|
# Initial-connect ladder (a startup blip must not kill the server); gated on
|
|
# _ever_connected, not _ready (which clears every reconnect cycle).
|
|
# If this is the first connection attempt, retry with backoff before giving up. Gated on
|
|
# ``_ever_connected`` rather than ``_ready`` — ``_ready`` is cleared on every reconnect cycle (see
|
|
# below), so a server that already registered tools once and then dropped would otherwise be
|
|
# misclassified as never having connected and re-enter this initial-connect ladder (#94654).
|
|
# ``_ever_connected`` itself is set once and never cleared. (Ported from Kilo Code's MCP resilience
|
|
# fix.)
|
|
if not self._ever_connected:
|
|
return await self._on_initial_connect_error(exc, root, failure_class, budget)
|
|
if self._shutdown_event.is_set():
|
|
logger.debug("MCP server '%s' disconnected during shutdown: %s: %s",
|
|
self.name, type(root).__name__, root)
|
|
return False
|
|
if failure_class == "permanent":
|
|
return await self._on_permanent_error(root, budget)
|
|
self._reconnect_retries += 1
|
|
if self._reconnect_retries > _core._MAX_RECONNECT_RETRIES:
|
|
self._log_park(
|
|
"MCP server '%s' failed after %d reconnection attempts, parking; will self-probe every %ds "
|
|
"until it recovers (state: degraded → parked): %s: %s",
|
|
self.name, _core._MAX_RECONNECT_RETRIES, _core._PARKED_RETRY_INTERVAL, type(root).__name__, root)
|
|
return await self._park_and_rearm("from parked state", budget)
|
|
logger.debug("MCP server '%s' connection lost (attempt %d/%d), reconnecting in %.0fs: %s: %s",
|
|
self.name, self._reconnect_retries, _core._MAX_RECONNECT_RETRIES, budget.backoff,
|
|
type(root).__name__, root)
|
|
await self._backoff_sleep(budget)
|
|
return not self._shutdown_event.is_set()
|
|
|
|
async def _on_initial_connect_error(self, exc: Exception, root: BaseException,
|
|
failure_class: str, budget: "_RetryBudget") -> bool:
|
|
if failure_class == "permanent":
|
|
# Deterministic failure (bad command, non-MCP URL, 401/403): park at once; auth
|
|
# failures park (not return) so the task can pick up fresh tokens later.
|
|
detail = (f"authentication, parking until credentials change; re-authenticate with "
|
|
f"`hermes mcp login {self.name}`" if _errors._is_auth_error(root)
|
|
else "connection with a permanent error, parking without retries")
|
|
self._log_park("MCP server '%s' failed initial %s (state: connecting → parked): %s: %s",
|
|
self.name, detail, type(root).__name__, root)
|
|
return await self._park_initial_failure(exc, "after permanent initial failure", budget)
|
|
budget.initial_retries += 1
|
|
if budget.initial_retries > _core._MAX_INITIAL_CONNECT_RETRIES:
|
|
self._log_park(
|
|
"MCP server '%s' failed initial connection after %d attempts, parking until a reconnect is "
|
|
"requested (state: connecting → parked): %s: %s",
|
|
self.name, _core._MAX_INITIAL_CONNECT_RETRIES, type(root).__name__, root)
|
|
return await self._park_initial_failure(exc, "after initial connection failures", budget)
|
|
logger.debug(
|
|
"MCP server '%s' initial connection failed (attempt %d/%d), retrying in %.0fs: %s: %s",
|
|
self.name, budget.initial_retries, _core._MAX_INITIAL_CONNECT_RETRIES, budget.backoff,
|
|
type(root).__name__, root)
|
|
await self._backoff_sleep(budget)
|
|
if self._shutdown_event.is_set():
|
|
self._publish_error(exc)
|
|
return not self._shutdown_event.is_set()
|
|
|
|
async def _on_permanent_error(self, root: BaseException, budget: "_RetryBudget") -> bool:
|
|
# Auth failure on a PROVEN session is often a raced-teardown OAuth lock, not revoked
|
|
# credentials: grant ONE suspect+reconnect cycle first.
|
|
if _errors._is_auth_error(root) and self._session_proven and not self._permanent_grace_used:
|
|
self._permanent_grace_used = True
|
|
self.mark_suspect(f"auth error on proven session: {root}")
|
|
logger.warning(
|
|
"MCP server '%s': auth error on a previously healthy session — marking suspect and forcing "
|
|
"one reconnect instead of parking (state: connected → suspect): %s: %s",
|
|
self.name, type(root).__name__, root)
|
|
self._reconnect_retries, budget.backoff = 0, 1.0
|
|
await asyncio.sleep(_jittered(1.0))
|
|
return not self._shutdown_event.is_set()
|
|
# Deterministic failure on a working server: park now.
|
|
self._log_park(
|
|
"MCP server '%s' hit a permanent error, parking without retries; will self-probe every %ds "
|
|
"(state: connected → parked): %s: %s", self.name, _core._PARKED_RETRY_INTERVAL, type(root).__name__, root)
|
|
return await self._park_and_rearm("from parked state (permanent error)", budget)
|
|
|
|
async def start(self, config: dict):
|
|
"""Create the background Task and wait until ready (or failed)."""
|
|
self._task = asyncio.ensure_future(self.run(config))
|
|
try:
|
|
await self._ready.wait()
|
|
except asyncio.CancelledError:
|
|
# The caller's connect timeout cancels *this* coroutine; the ensure_future'd run()
|
|
# task would otherwise keep running detached on a hung transport with no owner.
|
|
# Propagate so the transport context managers unwind and release child / FDs.
|
|
if self._task and not self._task.done():
|
|
self._task.cancel()
|
|
raise
|
|
if self._error:
|
|
raise self._error
|
|
|
|
async def shutdown(self):
|
|
"""Signal the Task to exit and wait for clean resource teardown."""
|
|
self._shutdown_event.set()
|
|
# Also set reconnect: closes any race where _wait_for_lifecycle_event misses the
|
|
# shutdown flag after returning "reconnect".
|
|
self._reconnect_event.set()
|
|
if self._task and not self._task.done():
|
|
try:
|
|
await asyncio.wait_for(self._task, timeout=10)
|
|
except asyncio.TimeoutError:
|
|
logger.warning("MCP server '%s' shutdown timed out, cancelling task", self.name)
|
|
self._task.cancel()
|
|
try:
|
|
await self._task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
if self._pending_refresh_tasks:
|
|
for task in list(self._pending_refresh_tasks):
|
|
task.cancel()
|
|
await asyncio.gather(*self._pending_refresh_tasks, return_exceptions=True)
|
|
self._pending_refresh_tasks.clear()
|
|
self._deregister_tools()
|
|
self.session = None
|
|
|
|
def _deregister_tools(self) -> None:
|
|
"""Drop this server's tools from the registry (idempotent); on shutdown AND budget
|
|
exhaustion, so a dead server never leaves phantom tools in the prompt."""
|
|
for tool_name in list(getattr(self, "_registered_tool_names", [])):
|
|
_registration._deregister_mcp_tool_all_scopes(self, tool_name)
|
|
self._registered_tool_names = []
|