"""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: # Under the owner's FRESH scope: the run task's copied one is the connect-time # snapshot, so a header rendered before the profile's secret source answered # would re-render to the same literal ${VAR} on every probe (#119092). from tools import mcp_tool_config as _config from tools.mcp_tool_discovery import _install_owner_secret_scope scope_token = await _install_owner_secret_scope() try: config = _config._require_rendered_remote(self.name, self._refresh_remote_config(config)) finally: if scope_token is not None: from agent.secret_scope import reset_secret_scope reset_secret_scope(scope_token) 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 = []