From f4d4831e70bf4cb53d0dab8c1de59e8a74a106f2 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Thu, 3 Sep 2026 13:28:47 -0700 Subject: [PATCH] =?UTF-8?q?simplify(compat):=20tools/mcp=5Ftool=20facade?= =?UTF-8?q?=20=E2=80=94=20drop=2013=20re-export=20blocks=20(127=20names)?= =?UTF-8?q?=20+=20shutil=20re-import;=20siblings=20read=20sibling-defined?= =?UTF-8?q?=20names=20directly=20(=5Fcore=20kept=20for=20facade=20state)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tools/mcp_tool.py | 57 ++--------------------- tools/mcp_tool_agent.py | 2 +- tools/mcp_tool_config.py | 12 ++--- tools/mcp_tool_content.py | 6 +-- tools/mcp_tool_discovery.py | 84 ++++++++++++++++++---------------- tools/mcp_tool_errors.py | 2 +- tools/mcp_tool_handlers.py | 29 ++++++------ tools/mcp_tool_health.py | 3 +- tools/mcp_tool_lifecycle.py | 5 +- tools/mcp_tool_loop.py | 19 ++++---- tools/mcp_tool_registration.py | 12 +++-- tools/mcp_tool_server_run.py | 37 ++++++++------- tools/mcp_tool_transport.py | 17 ++++--- 13 files changed, 129 insertions(+), 156 deletions(-) diff --git a/tools/mcp_tool.py b/tools/mcp_tool.py index f02ae3b156..2e0c3c2551 100644 --- a/tools/mcp_tool.py +++ b/tools/mcp_tool.py @@ -6,8 +6,8 @@ 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 it back through ``tools.mcp_tool`` at call -time and every name they define is re-exported here so ``patch("tools.mcp_tool.X")`` works.""" +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 @@ -16,7 +16,6 @@ import importlib.util import inspect import logging import os -import shutil # noqa: F401 — tests patch ``tools.mcp_tool.shutil.which`` import sys import threading import time @@ -24,58 +23,12 @@ from typing import Any, Callable, Dict, List, Optional, Set logger = logging.getLogger(__name__) -from tools.mcp_tool_common import ( # noqa: F401 - _DEFAULT_TOOL_TIMEOUT, _env_ref_name, _exc_str, _get_lifecycle_seconds, _jittered, - _parse_boolish, _resolve_tool_timeout, _safe_numeric, _sanitize_error, mcp_field) -from tools.mcp_tool_schema import ( # noqa: F401 - MCP_TOOL_NAME_PREFIX, _build_utility_schemas, _convert_mcp_schema, _normalize_mcp_input_schema, - _scan_mcp_description, matches_name_filter, mcp_prefixed_tool_name, sanitize_mcp_name_component) -from tools.mcp_tool_content import ( # noqa: F401 - _MCP_HARD_RESULT_CAP_CHARS, _MCP_RESOURCE_MAX_B64_CHARS, _MCP_RESOURCE_MAX_BYTES, - _cache_mcp_audio_block, _cache_mcp_image_block, _is_reserved_mcp_meta_key, - _mcp_image_extension_for_mime_type, _mcp_resource_filename, _render_mcp_dropped_block_notice, - _render_mcp_resource_block, _truncate_mcp_text_result) -from tools.mcp_tool_errors import ( # noqa: F401 - InvalidMcpUrlError, NonMcpEndpointError, _EXC_TRAVERSAL_MAX_NODES, - _JSONRPC_UNSUPPORTED_PROTOCOL_VERSION, _classify_mcp_failure, _format_connect_error, - _handshake_rejected_as_modern, _is_auth_error, _is_method_not_found_error, - _is_session_expired_error, _make_redirect_header_stripper, _resolve_client_cert, - _resolve_identity_header, _unwrap_exception_group, _validate_remote_mcp_url) -from tools.mcp_tool_config import ( # noqa: F401 - _ENV_VAR_PATTERN, _build_safe_env, _filter_suspicious_mcp_servers, _get_mcp_stderr_log, - _interpolate_env_vars, _load_mcp_config, _resolve_stdio_command, _warn_hidden_whitespace, - _npx_bin_candidates, _npx_cached_bin, _whitespace_warned, _workspace_folder, _write_stderr_log_header) -from tools.mcp_tool_sampling import ( # noqa: F401 - ElicitationHandler, SamplingHandler, _format_elicitation_schema_summary) -from tools.mcp_tool_handlers import ( # noqa: F401 - _handle_auth_error_and_retry, _handle_session_expired_and_retry, _make_check_fn, - _make_get_prompt_handler, _make_list_prompts_handler, _make_list_resources_handler, - _make_read_resource_handler, _make_tool_handler) -from tools.mcp_tool_registration import ( # noqa: F401 - _annotation_read_only_hint, _existing_tool_names, _forget_mcp_tool_server, - _normalize_server_trust, _register_from_cache_sync, _register_server_tools, - _select_utility_schemas, _track_mcp_tool_server) -from tools.mcp_tool_lifecycle import ( # noqa: F401 - _drain_and_stop_mcp_loop, _drain_mcp_loop_tasks, _filter_mcp_children, - _kill_orphaned_mcp_children, _orphan_stdio_pid_servers, _orphan_stdio_pids, - _snapshot_child_pids, _stdio_pgids, _stdio_pids, _stop_mcp_loop_if_idle, shutdown_mcp_servers) -from tools.mcp_tool_agent import ( # noqa: F401 - _reinject_post_build_tools, persist_agent_tool_names, refresh_agent_mcp_tools, - reprobe_tool_availability, restore_agent_tool_prefix) +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 -from tools.mcp_tool_loop import ( # noqa: F401 - _running_loop, _LockCookie, _acquire_lock_on_fh, _try_acquire_mcp_discovery_lock, - _mcp_loop_exception_handler, _wrap_with_home_override, _wrap_with_dashboard_oauth_flow, - _run_on_mcp_loop, _signal_reconnect, reconnect_mcp_server, _wait_for_server_session_ready, - _signal_reconnect_and_wait, _ensure_mcp_loop, _stop_mcp_loop) -from tools.mcp_tool_discovery import ( # noqa: F401 - _record_connect_failure, _clear_connect_failure, _connect_cooldown_active, _connect_server, - _request_lazy_reconnect, _resolve_server_lazy, _ensure_lazy_server_connected, - _get_connected_server_for_call, _discover_and_register_server, register_mcp_servers, - discover_mcp_tools, is_mcp_tool_parallel_safe, get_mcp_status, probe_mcp_server_tools, - has_registered_mcp_tools, get_registered_mcp_server_names) # Wall-clock bound on the fail-open OSV malware preflight before a stdio spawn; just ABOVE diff --git a/tools/mcp_tool_agent.py b/tools/mcp_tool_agent.py index 2795d647e9..c7f924c4dc 100644 --- a/tools/mcp_tool_agent.py +++ b/tools/mcp_tool_agent.py @@ -104,7 +104,7 @@ def refresh_agent_mcp_tools( new_defs = list(get_tool_definitions(enabled_toolsets=enabled, disabled_toolsets=disabled, quiet_mode=quiet_mode) or []) new_names = {_def_name(t) for t in new_defs} # Post-build families re-appended on LOCALS only; live attributes untouched until publish. - staged_engine_names = _core._reinject_post_build_tools(agent, new_defs, new_names) + staged_engine_names = _reinject_post_build_tools(agent, new_defs, new_names) # Registry membership is read OUTSIDE ``_agent_tools_lock``: taking ``registry._lock`` # under the tools lock would be the first nesting of the two. prefix_registered: Optional[set] = None diff --git a/tools/mcp_tool_config.py b/tools/mcp_tool_config.py index 756aba02ad..cdf80c4b73 100644 --- a/tools/mcp_tool_config.py +++ b/tools/mcp_tool_config.py @@ -11,7 +11,7 @@ import sys import threading from datetime import datetime from typing import Any, Dict, List, Optional, Set, Tuple -from tools.mcp_tool_common import _env_ref_name, _prepend_path, _core +from tools.mcp_tool_common import _env_ref_name, _prepend_path logger = logging.getLogger("tools.mcp_tool") @@ -45,7 +45,7 @@ def _get_mcp_stderr_log() -> Any: def _write_stderr_log_header(server_name: str) -> None: """Session marker so operators can find each server's output in the shared log (per-line prefixes would need a pipe + reader thread).""" - fh = _core._get_mcp_stderr_log() + fh = _get_mcp_stderr_log() try: fh.write(f"\n===== [{datetime.now():%Y-%m-%d %H:%M:%S}] starting MCP server '{server_name}' =====\n") fh.flush() @@ -81,13 +81,13 @@ def _workspace_folder() -> str: def _workspace_basename() -> str: - root = _core._workspace_folder() + root = _workspace_folder() return os.path.basename(root.rstrip("/\\")) or root # Cursor's case-sensitive context vars -> resolver. _CONTEXT_VAR_RESOLVERS = { - "userHome": lambda: os.path.expanduser("~"), "workspaceFolder": lambda: _core._workspace_folder(), + "userHome": lambda: os.path.expanduser("~"), "workspaceFolder": lambda: _workspace_folder(), "workspaceFolderBasename": _workspace_basename, "pathSeparator": lambda: os.sep, "/": lambda: os.sep} @@ -300,7 +300,7 @@ def _portable_mcp_servers(safe_servers: Dict[str, dict]) -> None: from hermes_cli.plugins import discover_plugins, get_plugin_manager discover_plugins() portable = get_plugin_manager().get_portable_mcp_servers() - for name, cfg in _core._filter_suspicious_mcp_servers(portable).items(): + for name, cfg in _filter_suspicious_mcp_servers(portable).items(): if name in safe_servers: logger.warning("Portable MCP server '%s' conflicts with native config; skipping", name) else: @@ -323,7 +323,7 @@ def _load_mcp_config() -> Dict[str, dict]: except Exception: pass safe_servers: Dict[str, dict] = {} - for name, cfg in _core._filter_suspicious_mcp_servers(servers if isinstance(servers, dict) else {}).items(): + for name, cfg in _filter_suspicious_mcp_servers(servers if isinstance(servers, dict) else {}).items(): interpolated = _interpolate_env_vars(cfg) if isinstance(interpolated, dict): _warn_hidden_whitespace(name, interpolated) diff --git a/tools/mcp_tool_content.py b/tools/mcp_tool_content.py index 4b7c303400..e83f9dfa80 100644 --- a/tools/mcp_tool_content.py +++ b/tools/mcp_tool_content.py @@ -7,7 +7,7 @@ import logging import mimetypes from typing import Any, Dict, Optional, Tuple from tools.ansi_strip import strip_unicode_tags -from tools.mcp_tool_common import mcp_field, _core +from tools.mcp_tool_common import mcp_field from tools.mcp_tool_schema import mcp_prefixed_tool_name logger = logging.getLogger("tools.mcp_tool") @@ -83,14 +83,14 @@ def _decode_block_b64(data, what: str, label: str, *, cap_what: Optional[str] = """Base64-decode one block payload: ``(bytes, "")`` or ``(None, inline_marker)``. With ``cap_what`` the payload is rejected on b64 length BEFORE decoding and on decoded size after. Decode failures warn and return ``decode_fail`` ("" = drop the block).""" - if cap_what and len(data) > _core._MCP_RESOURCE_MAX_B64_CHARS: + if cap_what and len(data) > _MCP_RESOURCE_MAX_B64_CHARS: return None, f"[MCP {cap_what} too large to cache: ~{len(data) * 3 // 4} bytes{cap_suffix}]" try: raw_bytes = base64.b64decode(data) except (TypeError, ValueError) as exc: logger.warning("MCP %s decode failed (%s): %s", what, label, exc) return None, decode_fail - if cap_what and len(raw_bytes) > _core._MCP_RESOURCE_MAX_BYTES: + if cap_what and len(raw_bytes) > _MCP_RESOURCE_MAX_BYTES: return None, f"[MCP {cap_what} too large to cache: {len(raw_bytes)} bytes{cap_suffix}]" return raw_bytes, "" diff --git a/tools/mcp_tool_discovery.py b/tools/mcp_tool_discovery.py index caf352f102..3fd427fd51 100644 --- a/tools/mcp_tool_discovery.py +++ b/tools/mcp_tool_discovery.py @@ -9,7 +9,13 @@ import asyncio import logging import time from typing import Dict, List, Optional, Tuple -from tools.mcp_tool_common import _core +from tools.mcp_tool_common import _core, _parse_boolish +from tools import mcp_tool_config as _config +from tools import mcp_tool_errors as _errors +from tools import mcp_tool_lifecycle as _lifecycle +from tools import mcp_tool_loop as _loop +from tools import mcp_tool_registration as _registration +from tools.mcp_tool_schema import MCP_TOOL_NAME_PREFIX logger = logging.getLogger("tools.mcp_tool") @@ -35,7 +41,7 @@ def _connect_cooldown_active(server_name: str) -> bool: def _enabled(cfg: dict) -> bool: - return _core._parse_boolish(cfg.get("enabled", True), default=True) + return _parse_boolish(cfg.get("enabled", True), default=True) async def _connect_server(name: str, config: dict) -> _core.MCPServerTask: @@ -67,7 +73,7 @@ async def _connect_server(name: str, config: dict) -> _core.MCPServerTask: def _request_lazy_reconnect(server_name: str, server: _core.MCPServerTask) -> bool: """Wake a recycled stdio server and wait briefly for a fresh session.""" - loop = _core._running_loop() if server._is_recycled_stdio() else None + loop = _loop._running_loop() if server._is_recycled_stdio() else None if loop is None: return False @@ -86,7 +92,7 @@ def _request_lazy_reconnect(server_name: str, server: _core.MCPServerTask) -> bo return False try: - return bool(_core._run_on_mcp_loop(_await_ready, timeout=_core._RECYCLED_RECONNECT_TIMEOUT)) + return bool(_loop._run_on_mcp_loop(_await_ready, timeout=_core._RECYCLED_RECONNECT_TIMEOUT)) except Exception as exc: logger.warning("MCP server '%s': lazy reconnect after stdio recycle failed: %s", server_name, exc) return False @@ -98,16 +104,16 @@ def _resolve_server_lazy(name: str, config: dict) -> bool: Gated per-server by ``mcp_servers..lazy`` in config (default OFF), following the same per-server key pattern as ``idle_timeout_seconds``. Design from #56832 (Vansh5632). """ - return _core._parse_boolish(config.get("lazy", False), default=False) + return _parse_boolish(config.get("lazy", False), default=False) def _note_connect_failure(name: str, exc: BaseException) -> str: """Record a failed connect (under ``_lock``): error text for status, cooldown stamp.""" - message = _core._format_connect_error(exc) + message = _errors._format_connect_error(exc) with _core._lock: _core._server_connecting.discard(name) _core._server_connect_errors[name] = message - _core._record_connect_failure(name) + _record_connect_failure(name) return message @@ -116,7 +122,7 @@ def _note_connect_success(name: str) -> None: with _core._lock: _core._server_connecting.discard(name) _core._server_connect_errors.pop(name, None) - _core._clear_connect_failure(name) + _clear_connect_failure(name) def _adopt_server(name: str, server: _core.MCPServerTask) -> None: @@ -138,16 +144,16 @@ def _ensure_lazy_server_connected(server_name: str) -> bool: if server is not None and server.session is not None: return True config = _core._lazy_server_configs.get(server_name) - if (not config or _core._connect_cooldown_active(server_name) + if (not config or _connect_cooldown_active(server_name) or server_name in _core._server_connecting): return False _core._server_connecting.add(server_name) _core._server_connect_errors.pop(server_name, None) logger.info("MCP server '%s': lazy start on first use", server_name) - _core._ensure_mcp_loop() + _loop._ensure_mcp_loop() connect_timeout = config.get("connect_timeout", _core._DEFAULT_CONNECT_TIMEOUT) try: - _core._run_on_mcp_loop(lambda: _core._discover_and_register_server(server_name, config), + _loop._run_on_mcp_loop(lambda: _discover_and_register_server(server_name, config), timeout=float(connect_timeout) + 30.0) except BaseException as exc: logger.warning("Lazy MCP connect failed for '%s': %s", server_name, _note_connect_failure(server_name, exc)) @@ -165,7 +171,7 @@ def _ensure_lazy_server_connected(server_name: str) -> bool: from tools.registry import registry for tool_name in phantom_names: registry.deregister(tool_name, scope=_core._server_registry_scope(server_name)) - _core._forget_mcp_tool_server(tool_name) + _registration._forget_mcp_tool_server(tool_name) logger.info("MCP server '%s': deregistered %d phantom cached tool(s) not served live (stale schema-cache " "fingerprint %s): %s", server_name, len(phantom_names), stale_fingerprint, ", ".join(phantom_names)) return server is not None and server.session is not None @@ -182,9 +188,9 @@ def _get_connected_server_for_call(server_name: str) -> Optional[_core.MCPServer server = _core._servers.get(server_name) is_lazy = server_name in _core._lazy_server_configs if is_lazy and (server is None or server.session is None): - _core._ensure_lazy_server_connected(server_name) + _ensure_lazy_server_connected(server_name) elif server is not None and server.session is None and server._is_recycled_stdio(): - _core._request_lazy_reconnect(server_name, server) + _request_lazy_reconnect(server_name, server) else: return server with _core._lock: @@ -197,7 +203,7 @@ async def _discover_and_register_server(name: str, config: dict) -> List[str]: claimed: List[_core.MCPServerTask] = [] claim_token = _core._connect_server_claim.set(claimed.append) try: - server = await asyncio.wait_for(_core._connect_server(name, config), + server = await asyncio.wait_for(_connect_server(name, config), timeout=config.get("connect_timeout", _core._DEFAULT_CONNECT_TIMEOUT)) except BaseException: server = claimed[0] if claimed else None @@ -216,7 +222,7 @@ async def _discover_and_register_server(name: str, config: dict) -> List[str]: _core._server_connecting.discard(name) _core._server_connect_errors.pop(name, None) _adopt_server(name, server) - registered_names = _core._register_server_tools(name, server, config) + registered_names = _registration._register_server_tools(name, server, config) server._registered_tool_names = list(registered_names) logger.info("MCP server '%s' (%s): registered %d tool(s): %s", name, "HTTP" if "url" in config else "stdio", len(registered_names), ", ".join(registered_names)) @@ -235,7 +241,7 @@ def _select_new_servers(servers: Dict[str, dict]) -> Dict[str, dict]: new_servers = { k: v for k, v in servers.items() if k not in _core._servers and k not in connecting and k not in _core._lazy_server_configs - and _enabled(v) and not _core._connect_cooldown_active(k)} + and _enabled(v) and not _connect_cooldown_active(k)} stale_cached = [_core._servers[k] for k in servers if k in _core._servers and getattr(_core._servers[k], "session", None) is None] _core._server_connecting.update(new_servers) @@ -243,12 +249,12 @@ def _select_new_servers(servers: Dict[str, dict]) -> Dict[str, dict]: _core._server_connect_errors.pop(srv_name, None) # Track which servers opt-in to parallel tool calls (idempotent). for srv_name, srv_cfg in servers.items(): - if _core._parse_boolish(srv_cfg.get("supports_parallel_tool_calls", False), default=False): + if _parse_boolish(srv_cfg.get("supports_parallel_tool_calls", False), default=False): _core._parallel_safe_servers.add(srv_name) else: _core._parallel_safe_servers.discard(srv_name) for srv in stale_cached: - _core._signal_reconnect(srv) + _loop._signal_reconnect(srv) return new_servers @@ -266,7 +272,7 @@ def _register_lazy_from_cache(new_servers: Dict[str, dict]) -> Tuple[Dict[str, d except Exception: # pragma: no cover - cache module missing return eager_servers, 0, 0 for name, cfg in new_servers.items(): - if not _core._resolve_server_lazy(name, cfg): + if not _resolve_server_lazy(name, cfg): continue entry = get_cached_entry(name, config_fingerprint(cfg)) if not entry: @@ -274,7 +280,7 @@ def _register_lazy_from_cache(new_servers: Dict[str, dict]) -> Tuple[Dict[str, d with _core._lock: _core._server_connecting.discard(name) try: - names = _core._register_from_cache_sync(name, cfg, entry) + names = _registration._register_from_cache_sync(name, cfg, entry) except Exception as exc: logger.warning("Failed lazy MCP registration for '%s': %s", name, exc) with _core._lock: @@ -289,7 +295,7 @@ def _register_lazy_from_cache(new_servers: Dict[str, dict]) -> Tuple[Dict[str, d async def _discover_all(new_servers: Dict[str, dict]) -> None: """Connect every candidate concurrently; record per-server outcome.""" results = await asyncio.gather( - *(_core._discover_and_register_server(name, cfg) for name, cfg in new_servers.items()), + *(_discover_and_register_server(name, cfg) for name, cfg in new_servers.items()), return_exceptions=True) for name, result in zip(new_servers, results): if isinstance(result, BaseException): @@ -310,7 +316,7 @@ def _run_discovery_pass(new_servers: Dict[str, dict]) -> None: if _was_interrupted: _set_interrupt(False) try: - _core._run_on_mcp_loop(lambda: _discover_all(new_servers), timeout=120) + _loop._run_on_mcp_loop(lambda: _discover_all(new_servers), timeout=120) except (TimeoutError, InterruptedError) as _e: # Stranded _server_connecting entries would block future reconnects. how = "timed out" if isinstance(_e, TimeoutError) else "interrupted" @@ -353,35 +359,35 @@ def register_mcp_servers(servers: Dict[str, dict]) -> List[str]: if not _core._ensure_mcp_sdk(): logger.debug("MCP SDK not available -- skipping explicit MCP registration") return [] - servers = _core._filter_suspicious_mcp_servers(servers) + servers = _config._filter_suspicious_mcp_servers(servers) if not servers: logger.debug("No explicit MCP servers provided") return [] new_servers = _select_new_servers(servers) if not new_servers: - return _core._existing_tool_names() + return _registration._existing_tool_names() new_servers, lazy_registered, lazy_server_count = _register_lazy_from_cache(new_servers) if not new_servers: if lazy_registered: logger.info("MCP: registered %d lazy tool(s) from schema cache (no processes spawned)", lazy_registered) - return _core._existing_tool_names() - _core._ensure_mcp_loop() + return _registration._existing_tool_names() + _loop._ensure_mcp_loop() _run_discovery_pass(new_servers) _log_summary("MCP: registered", new_servers, lazy_tools=lazy_registered, lazy_servers=lazy_server_count) - return _core._existing_tool_names() + return _registration._existing_tool_names() def _acquire_discovery_lock_with_retry(): """Cross-process guard: a lock loser waits for the holder then discovers itself; unavailable locking or an expired wait runs unguarded (fail-soft). None / _LOCK_UNAVAILABLE = unguarded.""" - cookie = _core._try_acquire_mcp_discovery_lock() + cookie = _loop._try_acquire_mcp_discovery_lock() if cookie is not None: return cookie logger.debug("Another process holds MCP discovery lock -- retrying with backoff") for _ in range(_core._MCP_DISCOVERY_LOCK_MAX_RETRIES): time.sleep(_core._MCP_DISCOVERY_LOCK_RETRY_DELAY_S) - cookie = _core._try_acquire_mcp_discovery_lock() + cookie = _loop._try_acquire_mcp_discovery_lock() if cookie is not None: break # Cross-process discovery guard (#62771). A lock loser waits for the holder, then performs its own @@ -403,7 +409,7 @@ def discover_mcp_tools(allowed_mcp_names: Optional[List[str]] = None) -> List[st list simply don't match); ``None`` spawns every configured server. Used by ``hermes -z -t `` to skip cold-starting servers the caller doesn't need (10-60s each); it only affects which servers start, not which names ``-t`` validation can see.""" - servers = _core._load_mcp_config() + servers = _config._load_mcp_config() if not servers: logger.debug("No MCP servers configured") return [] @@ -428,7 +434,7 @@ def discover_mcp_tools(allowed_mcp_names: Optional[List[str]] = None) -> List[st connecting = set(_core._server_connecting) new_server_names = [name for name, cfg in servers.items() if name not in _core._servers and name not in connecting and _enabled(cfg)] - tool_names = _core.register_mcp_servers(servers) + tool_names = register_mcp_servers(servers) if new_server_names: _log_summary(" MCP:", new_server_names) return tool_names @@ -440,7 +446,7 @@ def discover_mcp_tools(allowed_mcp_names: Optional[List[str]] = None) -> List[st def is_mcp_tool_parallel_safe(tool_name: str) -> bool: """True when the tool's server opted into ``supports_parallel_tool_calls`` (provenance captured at registration, never the ambiguous ``mcp__{server}__{tool}`` shape).""" - if not tool_name.startswith(_core.MCP_TOOL_NAME_PREFIX): + if not tool_name.startswith(MCP_TOOL_NAME_PREFIX): return False with _core._lock: server_name = _core._mcp_tool_server_names.get(tool_name) @@ -450,7 +456,7 @@ def is_mcp_tool_parallel_safe(tool_name: str) -> bool: def get_mcp_status() -> List[dict]: """Per-server status dicts for banner/TUI: name, transport, tools, connected, disabled, status (connected / disabled / connecting / failed / configured) and error for failed.""" - configured = _core._load_mcp_config() + configured = _config._load_mcp_config() if not configured: return [] with _core._lock: @@ -484,15 +490,15 @@ def probe_mcp_server_tools() -> Dict[str, List[tuple]]: registered and failed servers are omitted.""" if not _core._ensure_mcp_sdk(): return {} - enabled = {k: v for k, v in (_core._load_mcp_config() or {}).items() if _enabled(v)} + enabled = {k: v for k, v in (_config._load_mcp_config() or {}).items() if _enabled(v)} if not enabled: return {} - _core._ensure_mcp_loop() + _loop._ensure_mcp_loop() result: Dict[str, List[tuple]] = {} probed_servers: List[_core.MCPServerTask] = [] async def _probe_all(): - coros = [asyncio.wait_for(_core._connect_server(name, cfg), + coros = [asyncio.wait_for(_connect_server(name, cfg), timeout=cfg.get("connect_timeout", _core._DEFAULT_CONNECT_TIMEOUT)) for name, cfg in enabled.items()] outcomes = await asyncio.gather(*coros, return_exceptions=True) @@ -505,11 +511,11 @@ def probe_mcp_server_tools() -> Dict[str, List[tuple]]: await asyncio.gather(*(s.shutdown() for s in probed_servers), return_exceptions=True) try: - _core._run_on_mcp_loop(_probe_all, timeout=120) + _loop._run_on_mcp_loop(_probe_all, timeout=120) except Exception as exc: logger.debug("MCP probe failed: %s", exc) finally: - _core._stop_mcp_loop_if_idle() + _lifecycle._stop_mcp_loop_if_idle() return result diff --git a/tools/mcp_tool_errors.py b/tools/mcp_tool_errors.py index e01327c078..fc0b5eca1b 100644 --- a/tools/mcp_tool_errors.py +++ b/tools/mcp_tool_errors.py @@ -87,7 +87,7 @@ def _classify_mcp_failure(exc: BaseException) -> str: """``'permanent'`` (``run()`` parks instead of burning the retry ladder: auth 401/403, NonMcpEndpointError, InvalidMcpUrlError, missing stdio command) or ``'transient'`` (backoff retry).""" root = _unwrap_exception_group(exc) - permanent = (_core._is_auth_error(root) + permanent = (_is_auth_error(root) or isinstance(root, (NonMcpEndpointError, InvalidMcpUrlError, FileNotFoundError)) or (isinstance(root, OSError) and getattr(root, "errno", None) == errno.ENOENT) # 401/403 HTTPStatusError that _is_auth_error's type-gate missed (auth types not importable here) diff --git a/tools/mcp_tool_handlers.py b/tools/mcp_tool_handlers.py index 24c2eb3fc6..9d9c4a360a 100644 --- a/tools/mcp_tool_handlers.py +++ b/tools/mcp_tool_handlers.py @@ -13,11 +13,12 @@ from typing import Any, Callable, Dict, List, Optional, Tuple from tools.registry import tool_error from tools.ansi_strip import strip_unicode_tags from tools.mcp_tool_common import _exc_str, _sanitize_error, mcp_field, _core +from tools import mcp_tool_loop as _loop from tools.mcp_tool_content import ( _MCP_HARD_RESULT_CAP_CHARS, _cache_mcp_audio_block, _cache_mcp_image_block, _render_mcp_dropped_block_notice, _render_mcp_resource_block, _strip_reserved_meta_keys, _truncate_mcp_text_result) -from tools.mcp_tool_errors import _is_session_expired_error +from tools.mcp_tool_errors import _is_auth_error, _is_session_expired_error logger = logging.getLogger("tools.mcp_tool") _MISSING = object() @@ -41,7 +42,7 @@ def _trust_gate_check(server_name: str, tool_name: str) -> Optional[str]: or _core._tool_read_only_hints.get(server_name, {}).get(tool_name) is True): return None try: # lazy: tools.approval routes the prompt to whichever surface owns the session - from tools.approval import request_elicitation_consent + from tools.approval_prompt import request_elicitation_consent answer = request_elicitation_consent( f"MCP tool '{tool_name}' on UNTRUSTED server '{server_name}' wants to run. This tool is write-capable " f"(no readOnlyHint=true annotation) and may modify external state.", @@ -76,13 +77,14 @@ def _acquire_call_server(server_name: str, tool_timeout: float): """``(server, None)`` when a call may be dispatched, else ``(None, error)``. No session: a reconnect may be completing, so wait briefly before a breaker strike; still down -> ask the server task to rebuild (probing a dead transport would re-arm the breaker forever).""" + from tools import mcp_tool_discovery as _discovery # lazy: discovery -> registration -> handlers cycle not_connected = tool_error(f"MCP server '{server_name}' is not connected") - server = _core._get_connected_server_for_call(server_name) + server = _discovery._get_connected_server_for_call(server_name) wait = min(5.0, float(tool_timeout or 5.0)) - if server and (server.session or _core._wait_for_server_session_ready(server, timeout=wait)): + if server and (server.session or _loop._wait_for_server_session_ready(server, timeout=wait)): return server, None _core._bump_server_error(server_name) - if server and _core._signal_reconnect(server): + if server and _loop._signal_reconnect(server): return None, tool_error(f"MCP server '{server_name}' transport is down; reconnect requested. Do NOT retry this " f"tool immediately — give it a few seconds to come back.") return None, not_connected @@ -139,11 +141,11 @@ def _handle_auth_error_and_retry(server_name: str, exc: BaseException, retry_cal """OAuth recovery + one retry; None when *exc* is not an auth error. ``handle_401`` decides viability; if viable, signal a reconnect (fresh credentials), wait ready, retry once. Any failure returns the structured ``needs_reauth`` error so the model stops refreshing.""" - if not _core._is_auth_error(exc): + if not _is_auth_error(exc): return None from tools.mcp_oauth_manager import get_manager try: - recovered = _core._run_on_mcp_loop(lambda: get_manager().handle_401(server_name, None), timeout=10) + recovered = _loop._run_on_mcp_loop(lambda: get_manager().handle_401(server_name, None), timeout=10) except Exception as rec_exc: logger.warning("MCP OAuth '%s': recovery attempt failed: %s", server_name, rec_exc) recovered = False @@ -151,7 +153,7 @@ def _handle_auth_error_and_retry(server_name: str, exc: BaseException, retry_cal srv = _lookup_reconnectable_server(server_name) # Recovery + reconnect is independent evidence of viability: close the breaker here, not only on # retry success (else a failing retry pins it open forever). - if srv is not None and _core._signal_reconnect_and_wait( + if srv is not None and _loop._signal_reconnect_and_wait( server_name, srv, op_description=f"{op_description} after OAuth recovery", timeout=15): _core._reset_server_error(server_name) result = _retry_once(server_name, retry_call, op_description, "auth recovery") @@ -175,7 +177,7 @@ def _handle_session_expired_and_retry(server_name: str, exc: BaseException, retr return None logger.info("MCP server '%s': %s failed with session-expired error (%s); signalling transport reconnect " "and retrying once.", server_name, op_description, exc) - if not _core._signal_reconnect_and_wait(server_name, srv, op_description=op_description, timeout=15): + if not _loop._signal_reconnect_and_wait(server_name, srv, op_description=op_description, timeout=15): logger.warning("MCP server '%s': reconnect did not ready within 15s after session-expired error; " "falling through to error response.", server_name) return None @@ -204,10 +206,10 @@ def _handle_stdio_child_exited_and_retry(server_name: str, exc: Exception, retry logger.info("MCP server '%s': %s found the stdio subprocess dead (%s); respawning and retrying once.", server_name, op_description, exc) if _mcp_loop_running(): - reconnected = _core._signal_reconnect_and_wait( + reconnected = _loop._signal_reconnect_and_wait( server_name, srv, op_description=op_description, timeout=_core._STDIO_RESPAWN_WAIT_SEC) else: # No MCP loop to wait on (non-async adapters, tests): still request the respawn. - _core._signal_reconnect(srv) + _loop._signal_reconnect(srv) if not reconnected: return _strike(server_name, _STDIO_NO_RESPAWN_MSG.format(s=server_name, t=_core._STDIO_RESPAWN_WAIT_SEC)) try: @@ -234,7 +236,7 @@ def _dispatch(server_name: str, server: Any, op: str, call, tool_timeout: float, server.mark_tool_call() def call_once(): - return _core._run_on_mcp_loop(call, timeout=tool_timeout) + return _loop._run_on_mcp_loop(call, timeout=tool_timeout) try: result = call_once() @@ -454,7 +456,8 @@ def _make_utility_handler(op: str, log_label: str, rpc, render, required: Option payload, ``required`` validated before any transport work.""" def _factory(server_name: str, tool_timeout: float): def _handler(args: dict, **kwargs) -> str: - server = _core._get_connected_server_for_call(server_name) + from tools import mcp_tool_discovery as _discovery # lazy: import cycle + server = _discovery._get_connected_server_for_call(server_name) if not server or not server.session: return tool_error(f"MCP server '{server_name}' is not connected") if required and not args.get(required): diff --git a/tools/mcp_tool_health.py b/tools/mcp_tool_health.py index 69f42c352b..1b328cfeda 100644 --- a/tools/mcp_tool_health.py +++ b/tools/mcp_tool_health.py @@ -11,6 +11,7 @@ from tools.mcp_tool_errors import _is_method_not_found_error, _unwrap_exception_ from tools.mcp_tool_schema import mcp_prefixed_tool_name from tools.mcp_tool_registration import _forget_mcp_tool_server from tools.mcp_tool_common import _core +from tools import mcp_tool_registration as _registration logger = logging.getLogger("tools.mcp_tool") @@ -145,7 +146,7 @@ class MCPServerHealthMixin: # Re-register; a raw name can become ambiguous after normalization without changing # its normalized name, so also drop old entries the final registration no longer owns. self._tools = new_mcp_tools - registered_names = _core._register_server_tools(self.name, self, self._config) + registered_names = _registration._register_server_tools(self.name, self, self._config) self._deregister_owned(old_tool_names - set(registered_names)) self._registered_tool_names = registered_names new_tool_names = set(registered_names) diff --git a/tools/mcp_tool_lifecycle.py b/tools/mcp_tool_lifecycle.py index 7cd0defbd1..a1377368d3 100644 --- a/tools/mcp_tool_lifecycle.py +++ b/tools/mcp_tool_lifecycle.py @@ -7,6 +7,7 @@ import os import time from typing import Dict, Optional from tools.mcp_tool_common import _core +from tools import mcp_tool_loop as _loop logger = logging.getLogger("tools.mcp_tool") @@ -132,7 +133,7 @@ def shutdown_mcp_servers(*, scope: Optional[str] = None): # stale backoff entries), no connect-cooldown state may survive shutdown. with _core._lock: _clear_connect_cooldowns() - _core._stop_mcp_loop(only_if_idle=scope is not None) + _loop._stop_mcp_loop(only_if_idle=scope is not None) def _take_reapable_pids(include_active: bool, server_name: Optional[str]) -> tuple[Dict[int, str], Dict[int, int]]: @@ -219,7 +220,7 @@ def _stop_mcp_loop_if_idle() -> bool: """Stop the MCP loop only when no registered server still owns it. Probe paths create temporary MCPServerTasks not placed in ``_servers``; they may clean up an idle loop but must not tear down the process-global loop under live agent tools.""" - return _core._stop_mcp_loop(only_if_idle=True) + return _loop._stop_mcp_loop(only_if_idle=True) async def _drain_mcp_loop_tasks(*, timeout: Optional[float] = None) -> None: diff --git a/tools/mcp_tool_loop.py b/tools/mcp_tool_loop.py index 677e95caff..3337b278f5 100644 --- a/tools/mcp_tool_loop.py +++ b/tools/mcp_tool_loop.py @@ -15,6 +15,7 @@ import threading import time from typing import Any, Coroutine, Optional from tools.mcp_tool_common import _core +from tools import mcp_tool_lifecycle as _lifecycle logger = logging.getLogger("tools.mcp_tool") @@ -75,12 +76,12 @@ def _try_acquire_mcp_discovery_lock() -> Any: except Exception: return _core._LOCK_UNAVAILABLE try: - acquired = _core._acquire_lock_on_fh(fh) + acquired = _acquire_lock_on_fh(fh) except Exception: fh.close() return _core._LOCK_UNAVAILABLE if acquired: - return _core._LockCookie(fh) + return _LockCookie(fh) fh.close() return None @@ -151,7 +152,7 @@ def _run_on_mcp_loop(coro_or_factory, timeout: float = 30): raise RuntimeError("MCP event loop is not running") # run_coroutine_threadsafe copies the LOOP thread's context, so a per-request profile scope # would vanish here; re-establish it inside the task's own context. - coro = _core._wrap_with_dashboard_oauth_flow(_core._wrap_with_home_override( + coro = _wrap_with_dashboard_oauth_flow(_wrap_with_home_override( coro_or_factory() if callable(coro_or_factory) else coro_or_factory)) future = safe_schedule_threadsafe(coro, loop, logger=logger, log_message="MCP scheduling failed") if future is None: @@ -195,7 +196,7 @@ def reconnect_mcp_server(server_name: str) -> bool: """Ask a currently-live MCP server to rebuild after external re-auth.""" with _core._lock: server = _core._servers.get(server_name) - return server is not None and _core._signal_reconnect(server) + return server is not None and _signal_reconnect(server) def _wait_for_server_session_ready(srv: Any, *, old_session: Any = None, timeout: float = 15.0) -> bool: @@ -234,7 +235,7 @@ def _signal_reconnect_and_wait(server_name: str, srv: Any, *, op_description: st old_session = getattr(srv, "session", None) logger.info("MCP server '%s': %s requesting transport reconnect", server_name, op_description) loop.call_soon_threadsafe(_request_reconnect) - return _core._wait_for_server_session_ready(srv, old_session=old_session, timeout=timeout) + return _wait_for_server_session_ready(srv, old_session=old_session, timeout=timeout) def _ensure_mcp_loop(): @@ -245,7 +246,7 @@ def _ensure_mcp_loop(): if _origin._mcp_loop is not None and _origin._mcp_loop.is_running(): return loop = _origin._mcp_loop = asyncio.new_event_loop() - loop.set_exception_handler(_core._mcp_loop_exception_handler) + loop.set_exception_handler(_mcp_loop_exception_handler) _origin._mcp_thread = threading.Thread(target=loop.run_forever, name="mcp-event-loop", daemon=True) _origin._mcp_thread.start() @@ -271,7 +272,7 @@ def _stop_mcp_loop(*, only_if_idle: bool = False) -> bool: from agent.async_utils import safe_schedule_threadsafe future = safe_schedule_threadsafe( - _core._drain_and_stop_mcp_loop(), loop, logger=logger, + _lifecycle._drain_and_stop_mcp_loop(), loop, logger=logger, log_message="MCP loop drain: failed to schedule", log_level=logging.WARNING) if future is not None: try: @@ -282,7 +283,7 @@ def _stop_mcp_loop(*, only_if_idle: bool = False) -> bool: logger.warning("Error draining MCP loop tasks: %s", exc) elif not loop.is_closed(): try: - loop.run_until_complete(_core._drain_mcp_loop_tasks(timeout=_core._MCP_LOOP_DRAIN_TIMEOUT)) + loop.run_until_complete(_lifecycle._drain_mcp_loop_tasks(timeout=_core._MCP_LOOP_DRAIN_TIMEOUT)) except BaseException as exc: logger.warning("Error draining stopped MCP loop tasks: %s", exc) if future is None and loop.is_running(): # drain-and-stop wasn't scheduled: stop it ourselves @@ -296,5 +297,5 @@ def _stop_mcp_loop(*, only_if_idle: bool = False) -> bool: except Exception as exc: logger.warning("Unable to close MCP event loop cleanly: %s", exc) # The loop is gone, so no session can be in flight: reap active too. - _core._kill_orphaned_mcp_children(include_active=True) + _lifecycle._kill_orphaned_mcp_children(include_active=True) return True diff --git a/tools/mcp_tool_registration.py b/tools/mcp_tool_registration.py index a9f4a9af67..10b65b5238 100644 --- a/tools/mcp_tool_registration.py +++ b/tools/mcp_tool_registration.py @@ -8,6 +8,8 @@ from dataclasses import dataclass from types import SimpleNamespace from typing import TYPE_CHECKING, Any, Callable, Dict, Iterable, List, Optional from tools.mcp_tool_common import _parse_boolish, _core, _resolve_tool_timeout, mcp_field +from tools import mcp_tool_handlers as _handlers +from tools import mcp_tool_schema as _schema from tools.mcp_tool_handlers import ( _make_check_fn, _make_get_prompt_handler, _make_list_prompts_handler, _make_list_resources_handler, _make_read_resource_handler) @@ -100,7 +102,7 @@ def _existing_tool_names() -> List[str]: names: List[str] = [] for server in _core._servers.values(): names.extend(server._registered_tool_names if hasattr(server, "_registered_tool_names") - else (_core._convert_mcp_schema(server.name, t)["name"] for t in server._tools)) + else (_schema._convert_mcp_schema(server.name, t)["name"] for t in server._tools)) with _core._lock: names.extend(n for sname, tool_names in _core._lazy_server_tool_names.items() if sname not in _core._servers for n in tool_names) @@ -157,9 +159,9 @@ def _tool_candidates(name: str, tools: Iterable[Any], should_register: Callable[ if not should_register(t.name): logger.debug("MCP server '%s': skipping tool '%s' (filtered by config)", name, t.name) continue - _core._scan_mcp_description(name, t.name, t.description or "") - schema = _core._convert_mcp_schema(name, t) - handler = _core._make_tool_handler(name, t.name, tool_timeout) + _schema._scan_mcp_description(name, t.name, t.description or "") + schema = _schema._convert_mcp_schema(name, t) + handler = _handlers._make_tool_handler(name, t.name, tool_timeout) out.append(_Candidate(schema["name"], f"tool {t.name!r}", schema, handler)) return out @@ -244,7 +246,7 @@ def _register_candidates(name: str, candidates: List[_Candidate], *, check_fn: C name=c.registry_name, toolset=toolset_name, schema=c.schema, handler=c.handler, check_fn=check_fn, is_async=False, description=c.schema.get("description") or "", scope=scope()) if registry.get_toolset_for_tool(c.registry_name) == toolset_name: - _core._track_mcp_tool_server(c.registry_name, name) + _track_mcp_tool_server(c.registry_name, name) registered.append(c.registry_name) elif not lazy: logger.error("MCP server '%s': registration of %s as '%s' was rejected by the registry; " diff --git a/tools/mcp_tool_server_run.py b/tools/mcp_tool_server_run.py index 9c2f0fd3cf..5cb8fe17cf 100644 --- a/tools/mcp_tool_server_run.py +++ b/tools/mcp_tool_server_run.py @@ -8,7 +8,10 @@ import logging import time from dataclasses import dataclass from typing import Optional -from tools.mcp_tool_common import _core +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") @@ -86,7 +89,7 @@ class MCPServerRunMixin: async with self._rpc_lock: await self._keepalive_probe() except Exception as exc: - root = _core._unwrap_exception_group(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}") @@ -146,19 +149,19 @@ class MCPServerRunMixin: 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 = _core._resolve_tool_timeout(config) + self.tool_timeout = _resolve_tool_timeout(config) self._auth_type = (config.get("auth") or "").lower().strip() - self._idle_timeout_seconds = _core._get_lifecycle_seconds(config, "idle_timeout_seconds") - self._max_lifetime_seconds = _core._get_lifecycle_seconds(config, "max_lifetime_seconds") + 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 = (_core.SamplingHandler(self.name, sampling_config) + 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 = (_core.ElicitationHandler(self.name, elicitation_config, owner=self) + self._elicitation = (_sampling.ElicitationHandler(self.name, elicitation_config, owner=self) 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 " @@ -166,7 +169,7 @@ class MCPServerRunMixin: if not self._is_http(): return True try: - _core._validate_remote_mcp_url(self.name, config.get("url")) + _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). @@ -175,8 +178,8 @@ class MCPServerRunMixin: await self._preflight_content_type( config["url"], headers=dict(config.get("headers") or {}), ssl_verify=config.get("ssl_verify", True), - client_cert=_core._resolve_client_cert(self.name, config)) - except (_core.InvalidMcpUrlError, _core.NonMcpEndpointError) as exc: + client_cert=_errors._resolve_client_cert(self.name, config)) + except (_errors.InvalidMcpUrlError, _errors.NonMcpEndpointError) as exc: logger.warning("%s", exc) self._publish_error(exc) # fail fast and non-retryably return False @@ -280,14 +283,14 @@ class MCPServerRunMixin: return True async def _backoff_sleep(self, budget: "_RetryBudget") -> None: - await asyncio.sleep(_core._jittered(budget.backoff)) + 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 = _core._unwrap_exception_group(exc) - failure_class = _core._classify_mcp_failure(root) + 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) @@ -327,7 +330,7 @@ class MCPServerRunMixin: # 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 _core._is_auth_error(root) + f"`hermes mcp login {self.name}`" if _errors._is_auth_error(root) else "connection with a permanent error, parking without retries") logger.warning("MCP server '%s' failed initial %s (state: connecting → parked): %s: %s", self.name, detail, type(root).__name__, root) @@ -351,7 +354,7 @@ class MCPServerRunMixin: 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 _core._is_auth_error(root) and self._session_proven and not self._permanent_grace_used: + 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( @@ -359,7 +362,7 @@ class MCPServerRunMixin: "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(_core._jittered(1.0)) + await asyncio.sleep(_jittered(1.0)) return not self._shutdown_event.is_set() # Deterministic failure on a working server: park now. logger.warning( @@ -412,5 +415,5 @@ class MCPServerRunMixin: from tools.registry import registry for tool_name in list(getattr(self, "_registered_tool_names", [])): registry.deregister(tool_name, scope=_core._server_registry_scope(self.name)) - _core._forget_mcp_tool_server(tool_name) + _registration._forget_mcp_tool_server(tool_name) self._registered_tool_names = [] diff --git a/tools/mcp_tool_transport.py b/tools/mcp_tool_transport.py index cd856527a8..1324179788 100644 --- a/tools/mcp_tool_transport.py +++ b/tools/mcp_tool_transport.py @@ -10,6 +10,9 @@ from typing import Dict, Optional, Set from tools.mcp_tool_errors import NonMcpEndpointError, _apply_identity_header, _handshake_rejected_as_modern, _make_redirect_header_stripper, _resolve_client_cert from tools.mcp_tool_lifecycle import _filter_mcp_children, _orphan_stdio_pid_servers, _orphan_stdio_pids, _stdio_pgids, _stdio_pids from tools.mcp_tool_common import _core +from tools import mcp_tool_config as _config +from tools import mcp_tool_lifecycle as _lifecycle +from tools import mcp_tool_registration as _registration logger = logging.getLogger("tools.mcp_tool") @@ -210,7 +213,7 @@ class MCPServerTransportMixin: command = config.get("command") if not command: raise ValueError(f"MCP server '{self.name}' has no 'command' in config") - command, safe_env = _core._resolve_stdio_command(command, _core._build_safe_env(config.get("env"))) + command, safe_env = _config._resolve_stdio_command(command, _config._build_safe_env(config.get("env"))) # OSV malware preflight, then the cached-npx swap (ordering enforced there). command, args = await _core._preflight_stdio_command(self.name, command, config.get("args", [])) server_params = _core.StdioServerParameters( @@ -219,8 +222,8 @@ class MCPServerTransportMixin: encoding_error_handler="replace") # Reap orphans of prior attempts first (else retries pile up zombie pairs); unscoped on purpose; # off-loop because the reaper blocks up to 2s. - await asyncio.to_thread(_core._kill_orphaned_mcp_children) - pids_before = _core._snapshot_child_pids() # so the new child can be identified after spawn + await asyncio.to_thread(_lifecycle._kill_orphaned_mcp_children) + pids_before = _lifecycle._snapshot_child_pids() # so the new child can be identified after spawn # Reap any orphaned subprocesses from prior failed connection attempts before spawning a new one. # Without this, each retry in the run() reconnect loop spawns a fresh process pair while the # previous failed pair lingers — leading to rapid zombie accumulation (see #57355, #57228). The @@ -230,13 +233,13 @@ class MCPServerTransportMixin: # otherwise stall the shared MCP event loop. new_pids: set = set() # Subprocess stderr goes to ~/.hermes/logs/mcp-stderr.log so banners can't corrupt the TUI. - _core._write_stderr_log_header(self.name) + _config._write_stderr_log_header(self.name) try: - errlog = _core._get_mcp_stderr_log() + errlog = _config._get_mcp_stderr_log() async with _core.stdio_client(server_params, errlog=errlog) as (read_stream, write_stream): # New PIDs for force-kill cleanup, minus non-MCP children (slash_worker, LSP) racing # into the window: they share the TUI's pgid — leaking them would killpg() the TUI. - new_pids = _filter_mcp_children(_core._snapshot_child_pids() - pids_before) + new_pids = _filter_mcp_children(_lifecycle._snapshot_child_pids() - pids_before) if new_pids: self._track_spawned_children(new_pids) self._stdio_child_pids = set(new_pids) # so in-flight calls fail fast when the child dies @@ -449,7 +452,7 @@ class MCPServerTransportMixin: owned = _core._servers.get(self.name) is self if not owned and not self._ready.is_set(): return - self._registered_tool_names = _core._register_server_tools(self.name, self, self._config) + self._registered_tool_names = _registration._register_server_tools(self.name, self, self._config) with _core._lock: # a retained initial-failure server that just published tools has recovered if _core._servers.get(self.name) is self: _core._server_connect_errors.pop(self.name, None)