From f59fb2e2922905ab046503fa15d6a3a61cf76311 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 15:51:24 -0700 Subject: [PATCH] refactor(mcp): tighten registration candidates and sampling handler; drop dead _extract_tool_result_text --- tools/mcp_tool_registration.py | 294 ++++++++++++--------------------- tools/mcp_tool_sampling.py | 29 ++-- 2 files changed, 116 insertions(+), 207 deletions(-) diff --git a/tools/mcp_tool_registration.py b/tools/mcp_tool_registration.py index 3201fc5e6b..ecdfa24bb6 100644 --- a/tools/mcp_tool_registration.py +++ b/tools/mcp_tool_registration.py @@ -3,8 +3,8 @@ registry: include/exclude filtering, trust-tier metadata capture, utility-tool selection, name-collision resolution and the schema-cache write-through. Both entry points (``_register_server_tools`` for a live server, -``_register_from_cache_sync`` for a lazy cached manifest) build a list of -``_Candidate`` records and hand them to the single ``_register_candidates`` loop.""" +``_register_from_cache_sync`` for a lazy cached manifest) build ``_Candidate`` +records and feed the single ``_register_candidates`` loop.""" import logging from dataclasses import dataclass @@ -19,7 +19,6 @@ if TYPE_CHECKING: # pragma: no cover logger = logging.getLogger("tools.mcp_tool") _UTILITY_ORIGIN_PREFIX = "generated utility " - # Utility tool key -> handler factory; each takes (server_name, tool_timeout). _UTILITY_HANDLER_FACTORIES = { "list_resources": _make_list_resources_handler, @@ -30,37 +29,32 @@ _UTILITY_HANDLER_FACTORIES = { def _normalize_server_trust(value: Any) -> str: - """Normalize a config ``trust`` value: None -> ``full`` (backward-compatible - default); any unrecognized string -> ``untrusted``, so a misspelled tier - fails closed rather than silently disabling gating.""" + """Config ``trust`` -> tier. None -> ``full`` (backward-compatible default); + an unrecognized string -> ``untrusted`` so a misspelled tier fails closed.""" if value is None: return _core._TRUST_FULL text = str(value).strip().lower() if text in (_core._TRUST_FULL, _core._TRUST_UNTRUSTED): return text logger.warning( - "MCP trust: unrecognized trust value %r — treating as 'untrusted' " - "(valid values: full, untrusted)", value, + "MCP trust: unrecognized trust value %r — treating as 'untrusted' (valid values: full, untrusted)", value, ) return _core._TRUST_UNTRUSTED def _annotation_read_only_hint(mcp_tool: Any) -> bool: """True only when annotations (SDK object or schema-cache dict) carry - ``readOnlyHint is True``; anything else is False — unknown metadata means - the tool must be treated as write-capable.""" + ``readOnlyHint is True``; unknown metadata means write-capable.""" annotations = getattr(mcp_tool, "annotations", None) - if annotations is None: - return False if isinstance(annotations, dict): return annotations.get("readOnlyHint") is True return getattr(annotations, "readOnlyHint", None) is True def _record_tool_trust_metadata(server_name: str, config: dict, tools: List[Any]) -> None: - """Capture per-server trust and per-tool readOnlyHint at discovery (the + """Capture per-server trust and per-tool readOnlyHint at discovery — the security boundary: the call-time gate classifies from data we control, - never re-read server-supplied state).""" + never re-read server-supplied state.""" with _core._lock: _core._server_trust_levels[server_name] = _normalize_server_trust((config or {}).get("trust")) hints = _core._tool_read_only_hints.setdefault(server_name, {}) @@ -83,74 +77,59 @@ def _forget_mcp_tool_server(tool_name: str) -> None: def _select_utility_schemas(server_name: str, server: "MCPServerTask", config: dict) -> List[dict]: - """Select utility schemas based on config and server capabilities.""" + """Utility schemas allowed by config (``tools.resources``/``tools.prompts``) + and by the server's advertised capabilities. + + ``initialize_result.capabilities`` is the source of truth: its sub-objects are + non-None iff the server advertises that request family (the old + ``hasattr(server.session, ...)`` gate never filtered anything — ClientSession + defines all four methods). When no initialize_result was captured (test + fixtures, older paths) fall back to that legacy session-method check.""" tools_filter = config.get("tools") or {} - family_enabled = { - family: _parse_boolish(tools_filter.get(family), default=True) - for family in ("resources", "prompts") - } - # ``initialize_result.capabilities`` is the source of truth: its sub-objects - # are non-None iff the server advertises that request family. The old - # ``hasattr(server.session, ...)`` gate never filtered anything because - # ClientSession defines all four methods on the class. + enabled = {f: _parse_boolish(tools_filter.get(f), default=True) for f in ("resources", "prompts")} init_result = getattr(server, "initialize_result", None) - advertised_caps = getattr(init_result, "capabilities", None) if init_result is not None else None + advertised = getattr(init_result, "capabilities", None) if init_result is not None else None + + def _skip_reason(handler_key: str) -> Optional[str]: + family = _UTILITY_CAPABILITY_ATTRS[handler_key] + if not enabled[family]: + return f"{family} disabled" + if advertised is not None: + return None if getattr(advertised, family, None) is not None else f"server does not advertise '{family}' capability" + method = _UTILITY_CAPABILITY_METHODS[handler_key] + return None if hasattr(server.session, method) else f"session lacks {method}" selected: List[dict] = [] for entry in _build_utility_schemas(server_name): - handler_key = entry["handler_key"] - family = _UTILITY_CAPABILITY_ATTRS[handler_key] - if not family_enabled[family]: - logger.debug("MCP server '%s': skipping utility '%s' (%s disabled)", server_name, handler_key, family) - continue - if advertised_caps is not None: - if getattr(advertised_caps, family, None) is None: - logger.debug( - "MCP server '%s': skipping utility '%s' (server does not advertise '%s' capability)", - server_name, handler_key, family, - ) - continue - # Legacy fallback when initialize_result wasn't captured (test - # fixtures, older paths): register every stub the session can serve. - elif not hasattr(server.session, _UTILITY_CAPABILITY_METHODS[handler_key]): - logger.debug( - "MCP server '%s': skipping utility '%s' (session lacks %s)", - server_name, handler_key, _UTILITY_CAPABILITY_METHODS[handler_key], - ) + reason = _skip_reason(entry["handler_key"]) + if reason: + logger.debug("MCP server '%s': skipping utility '%s' (%s)", server_name, entry["handler_key"], reason) continue selected.append(entry) return selected def _existing_tool_names() -> List[str]: - """Return tool names for all currently connected servers.""" + """Tool names for all currently connected servers plus lazy (cache-registered) + servers, whose tools live only in the registry.""" names: List[str] = [] for _sname, server in _core._servers.items(): if hasattr(server, "_registered_tool_names"): names.extend(server._registered_tool_names) - continue - for mcp_tool in server._tools: - names.append(_core._convert_mcp_schema(server.name, mcp_tool)["name"]) - # Lazy servers registered from the schema cache have no MCPServerTask yet — - # their tools live only in the registry. + else: + names.extend(_core._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 + n for sname, tool_names in _core._lazy_server_tool_names.items() + if sname not in _core._servers for n in tool_names ) return names def _make_tool_filter(name: str, config: dict) -> Callable[[str], bool]: - """Build the include/exclude predicate for a server's tool names. - - Rules: ``tools.include`` is a whitelist, ``tools.exclude`` a blacklist; - entries may be exact names or fnmatch globs; include wins over exclude; - ``include: []`` is an explicit empty whitelist (register nothing); neither - set registers everything. - """ + """Include/exclude predicate for a server's tool names: ``tools.include`` is a + whitelist (``[]`` = register nothing), ``tools.exclude`` a blacklist; entries + are exact names or fnmatch globs; include wins over exclude.""" tools_filter = config.get("tools") or {} include_raw = tools_filter.get("include") include_set = _normalize_name_filter(include_raw, f"mcp_servers.{name}.tools.include") @@ -160,17 +139,14 @@ def _make_tool_filter(name: str, config: dict) -> Callable[[str], bool]: def _should_register(tool_name: str) -> bool: if include_active: return matches_name_filter(tool_name, include_set) - if exclude_set: - return not matches_name_filter(tool_name, exclude_set) - return True + return not (exclude_set and matches_name_filter(tool_name, exclude_set)) return _should_register class _CachedMCPTool: - """Minimal stand-in for MCP Tool objects loaded from the schema cache. - Missing/non-dict ``annotations`` (older cache files) fail closed to - write-capable via ``_annotation_read_only_hint``.""" + """Stand-in for MCP Tool objects loaded from the schema cache. Missing or + non-dict ``annotations`` (older cache files) fail closed to write-capable.""" __slots__ = ("name", "description", "inputSchema", "annotations") @@ -182,26 +158,19 @@ class _CachedMCPTool: @classmethod def from_cache_dicts(cls, raws: Iterable[Any]) -> List["_CachedMCPTool"]: - """Cached tool rows -> stand-ins; rows that are not dicts or lack a name are dropped.""" - tools = [] + """Cached rows -> stand-ins; rows that are not dicts or lack a name are dropped.""" + out = [] for raw in raws: - if not isinstance(raw, dict) or not raw.get("name"): - continue - schema = raw.get("inputSchema") - tools.append(cls( - raw["name"], - raw.get("description") or "", - schema if isinstance(schema, dict) else {}, - raw.get("annotations"), - )) - return tools + if isinstance(raw, dict) and raw.get("name"): + schema = raw.get("inputSchema") + out.append(cls(raw["name"], raw.get("description") or "", schema if isinstance(schema, dict) else {}, raw.get("annotations"))) + return out @dataclass class _Candidate: - """One registry registration attempt for a server: a native tool or a - generated utility. ``origin`` is the human-readable provenance used in - collision diagnostics.""" + """One registration attempt: a native tool or a generated utility. + ``origin`` is the provenance text used in collision diagnostics.""" registry_name: str origin: str @@ -212,62 +181,46 @@ class _Candidate: def is_utility(self) -> bool: return self.origin.startswith(_UTILITY_ORIGIN_PREFIX) - @property - def description(self) -> str: - return self.schema.get("description") or "" - def _tool_candidates(name: str, tools: Iterable[Any], should_register: Callable[[str], bool], tool_timeout) -> List[_Candidate]: - """Native tools (live SDK objects or ``_CachedMCPTool``) -> candidates. - The description scan runs on BOTH paths: the cache file is user-writable JSON.""" - candidates: List[_Candidate] = [] - for mcp_tool in tools: - if not should_register(mcp_tool.name): - logger.debug("MCP server '%s': skipping tool '%s' (filtered by config)", name, mcp_tool.name) + """Native tools (live SDK objects or ``_CachedMCPTool``) -> candidates. The + injection scan runs on BOTH paths: the cache file is user-writable JSON.""" + out: List[_Candidate] = [] + for t in tools: + 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, mcp_tool.name, mcp_tool.description or "") - schema = _core._convert_mcp_schema(name, mcp_tool) - candidates.append(_Candidate( - schema["name"], f"tool {mcp_tool.name!r}", schema, - _core._make_tool_handler(name, mcp_tool.name, tool_timeout), - )) - return candidates + _core._scan_mcp_description(name, t.name, t.description or "") + schema = _core._convert_mcp_schema(name, t) + out.append(_Candidate(schema["name"], f"tool {t.name!r}", schema, _core._make_tool_handler(name, t.name, tool_timeout))) + return out def _utility_candidates(name: str, entries: Iterable[Any], tool_timeout) -> List[_Candidate]: - """``{schema, handler_key}`` rows (live selection or cache) -> candidates; - malformed rows are dropped.""" - candidates: List[_Candidate] = [] + """``{schema, handler_key}`` rows (live selection or cache) -> candidates; malformed rows dropped.""" + out: List[_Candidate] = [] for raw in entries: if not isinstance(raw, dict): continue - schema, handler_key = raw.get("schema"), raw.get("handler_key") - if not isinstance(schema, dict) or handler_key not in _UTILITY_HANDLER_FACTORIES or not schema.get("name"): - continue - candidates.append(_Candidate( - schema["name"], f"{_UTILITY_ORIGIN_PREFIX}{handler_key!r}", schema, - _UTILITY_HANDLER_FACTORIES[handler_key](name, tool_timeout), - )) - return candidates + schema, key = raw.get("schema"), raw.get("handler_key") + if isinstance(schema, dict) and key in _UTILITY_HANDLER_FACTORIES and schema.get("name"): + out.append(_Candidate(schema["name"], f"{_UTILITY_ORIGIN_PREFIX}{key!r}", schema, _UTILITY_HANDLER_FACTORIES[key](name, tool_timeout))) + return out def _resolve_name_collisions(name: str, candidates: List[_Candidate]) -> List[_Candidate]: """Preflight registry-name collisions among one server's candidates. - Exact duplicate rows (same name + origin) are dropped silently; a generated + Exact duplicates (same name + origin) are dropped silently; a generated utility that normalizes onto a server-native tool's name is shadowed (the native tool wins); any other multi-origin collision is ambiguous and every - colliding entry is skipped (fail closed). Returns the survivors in order. - """ + colliding entry is skipped (fail closed). Returns the survivors in order.""" unique: List[_Candidate] = [] seen: set[tuple[str, str]] = set() origins_by_name: Dict[str, set[str]] = {} for c in candidates: if (c.registry_name, c.origin) in seen: - logger.debug( - "MCP server '%s': duplicate registration candidate %s for '%s'; keeping one", - name, c.origin, c.registry_name, - ) + logger.debug("MCP server '%s': duplicate registration candidate %s for '%s'; keeping one", name, c.origin, c.registry_name) continue seen.add((c.registry_name, c.origin)) unique.append(c) @@ -283,57 +236,44 @@ def _resolve_name_collisions(name: str, candidates: List[_Candidate]) -> List[_C if len(native_origins) == 1 and utility_origins: shadowed.update((registry_name, o) for o in utility_origins) logger.info( - "MCP server '%s': generated utility %s normalizes onto " - "server-native %s — keeping the native tool and dropping the " - "utility (the utility only applies when the server has no such " - "tool of its own)", + "MCP server '%s': generated utility %s normalizes onto server-native %s — keeping the " + "native tool and dropping the utility (the utility only applies when the server has no " + "such tool of its own)", name, ", ".join(utility_origins), native_origins[0], ) continue ambiguous[registry_name] = sorted(origins) - for registry_name, origins in sorted(ambiguous.items()): logger.error( - "MCP server '%s': name normalization collision for '%s' from %s; " - "skipping every colliding entry instead of choosing an arbitrary handler", + "MCP server '%s': name normalization collision for '%s' from %s; skipping every colliding " + "entry instead of choosing an arbitrary handler", name, registry_name, ", ".join(origins), ) - return [ - c for c in unique - if c.registry_name not in ambiguous and (c.registry_name, c.origin) not in shadowed - ] + return [c for c in unique if c.registry_name not in ambiguous and (c.registry_name, c.origin) not in shadowed] def _log_foreign_owner(name: str, c: _Candidate, existing_toolset: str, lazy: bool) -> None: - """Diagnostics for a candidate whose registry name is already owned by - another toolset (skipped to preserve the existing owner).""" + """Diagnostics for a name already owned by another toolset (skipped to preserve the owner).""" if lazy: if not c.is_utility: - logger.warning( - "MCP server '%s' (lazy): cached tool '%s' collides with toolset '%s' — skipping", - name, c.registry_name, existing_toolset, - ) + logger.warning("MCP server '%s' (lazy): cached tool '%s' collides with toolset '%s' — skipping", name, c.registry_name, existing_toolset) elif existing_toolset.startswith("mcp-"): logger.error( - "MCP server '%s': %s normalizes to '%s', already owned by MCP toolset '%s' " - "— skipping to preserve the existing owner", + "MCP server '%s': %s normalizes to '%s', already owned by MCP toolset '%s' — skipping to preserve the existing owner", name, c.origin, c.registry_name, existing_toolset, ) else: logger.warning( - "MCP server '%s': %s (→ '%s') collides with built-in tool in toolset '%s' " - "— skipping to preserve built-in", + "MCP server '%s': %s (→ '%s') collides with built-in tool in toolset '%s' — skipping to preserve built-in", name, c.origin, c.registry_name, existing_toolset, ) -def _register_candidates( - name: str, candidates: List[_Candidate], *, check_fn: Callable, scope: Callable[[], Optional[str]], lazy: bool, -) -> List[str]: +def _register_candidates(name: str, candidates: List[_Candidate], *, check_fn: Callable, scope: Callable[[], Optional[str]], lazy: bool) -> List[str]: """Register candidates under toolset ``mcp-{name}``; returns the names that - actually landed. The ownership pre-check is advisory only — multiple - servers connect in parallel, so ``ToolRegistry.register()`` is the atomic - ownership gate and its verdict is re-read after every call.""" + landed. The ownership pre-check is advisory only — servers connect in + parallel, so ``ToolRegistry.register()`` is the atomic ownership gate and + its verdict is re-read after every call.""" from tools.registry import registry toolset_name = f"mcp-{name}" @@ -344,26 +284,18 @@ def _register_candidates( _log_foreign_owner(name, c, existing_toolset, lazy) continue registry.register( - name=c.registry_name, - toolset=toolset_name, - schema=c.schema, - handler=c.handler, - check_fn=check_fn, - is_async=False, - description=c.description, - scope=scope(), + 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: if not lazy: logger.error( - "MCP server '%s': registration of %s as '%s' was rejected by " - "the registry; skipping provenance/count updates", + "MCP server '%s': registration of %s as '%s' was rejected by the registry; skipping provenance/count updates", name, c.origin, c.registry_name, ) continue _core._track_mcp_tool_server(c.registry_name, name) registered.append(c.registry_name) - if registered: registry.register_toolset_alias(name, toolset_name) return registered @@ -376,53 +308,42 @@ def _write_schema_cache(name: str, server: "MCPServerTask", config: dict, should from tools.mcp_schema_cache import config_fingerprint, write_cache_entry tools_payload = [] - for mcp_tool in server._tools: - if not should_register(mcp_tool.name): - continue - schema_obj = getattr(mcp_tool, "inputSchema", None) - tools_payload.append({ - "name": mcp_tool.name, - "description": mcp_tool.description or "", - "inputSchema": schema_obj if isinstance(schema_obj, dict) else {}, - # Persisted so the lazy path trust-gates identically next startup. - "annotations": {"readOnlyHint": _annotation_read_only_hint(mcp_tool)}, - }) - utility_payload = [ - {"schema": entry["schema"], "handler_key": entry["handler_key"]} - for entry in _select_utility_schemas(name, server, config) - ] + for t in server._tools: + if should_register(t.name): + schema_obj = getattr(t, "inputSchema", None) + tools_payload.append({ + "name": t.name, + "description": t.description or "", + "inputSchema": schema_obj if isinstance(schema_obj, dict) else {}, + # Persisted so the lazy path trust-gates identically next startup. + "annotations": {"readOnlyHint": _annotation_read_only_hint(t)}, + }) + utility_payload = [{"schema": e["schema"], "handler_key": e["handler_key"]} for e in _select_utility_schemas(name, server, config)] cache_meta = getattr(server, "_list_cache_meta", None) or {} write_cache_entry( - name, - config_fingerprint(config), - tools=tools_payload, - utility_tools=utility_payload, - ttl_ms=cache_meta.get("ttl_ms"), - cache_scope=cache_meta.get("cache_scope"), + name, config_fingerprint(config), tools=tools_payload, utility_tools=utility_payload, + ttl_ms=cache_meta.get("ttl_ms"), cache_scope=cache_meta.get("cache_scope"), ) except Exception as exc: logger.debug("MCP schema cache write failed for '%s': %s", name, exc) def _register_server_tools(name: str, server: "MCPServerTask", config: dict) -> List[str]: - """Register an already-connected server's tools (plus utility tools) into - the registry; used by initial discovery and list_changed refresh. + """Register an already-connected server's tools (plus utility tools); used by + initial discovery and list_changed refresh. Returns the registered names. Toolset resolution for ``mcp-{server}`` / raw-name aliases derives from the live registry rather than mutating ``toolsets.TOOLSETS``. Lossy name normalization can map distinct raw names (``read-file``/``read_file``) to - one registry name; such collisions fail closed — every ambiguous entry is - skipped. Generated utilities share the namespace and join the same - preflight. Returns the registered prefixed names. - """ + one registry name; such collisions fail closed. Generated utilities share + the namespace and join the same preflight.""" should_register = _make_tool_filter(name, config) - check_fn = _make_check_fn(name) _record_tool_trust_metadata(name, config, server._tools) candidates = _tool_candidates(name, server._tools, should_register, server.tool_timeout) candidates += _utility_candidates(name, _select_utility_schemas(name, server, config), server.tool_timeout) registered = _register_candidates( name, _resolve_name_collisions(name, candidates), - check_fn=check_fn, scope=lambda: _core._server_registry_scope(name), lazy=False, + check_fn=_make_check_fn(name), scope=lambda: _core._server_registry_scope(name), lazy=False, ) if registered: _write_schema_cache(name, server, config, should_register) @@ -431,21 +352,18 @@ def _register_server_tools(name: str, server: "MCPServerTask", config: dict) -> def _register_from_cache_sync(name: str, config: dict, entry: dict) -> List[str]: """Lazy startup: register a server's tools from a cached manifest with no - child process. The first real call routes through + child process; the first real call goes through ``_get_connected_server_for_call`` -> ``_ensure_lazy_server_connected``. Trust metadata is recorded first so the call-time gate is identical whether the server was spawned live or registered from cache.""" from tools.mcp_schema_cache import config_fingerprint, tools_from_cache_entry, utility_tools_from_cache_entry tool_timeout = _resolve_tool_timeout(config) - check_fn = _make_check_fn(name) cached_tools = _CachedMCPTool.from_cache_dicts(tools_from_cache_entry(entry)) _record_tool_trust_metadata(name, config, cached_tools) candidates = _tool_candidates(name, cached_tools, _make_tool_filter(name, config), tool_timeout) candidates += _utility_candidates(name, utility_tools_from_cache_entry(entry), tool_timeout) - registered = _register_candidates( - name, candidates, check_fn=check_fn, scope=_core._mcp_registry_scope, lazy=True, - ) + registered = _register_candidates(name, candidates, check_fn=_make_check_fn(name), scope=_core._mcp_registry_scope, lazy=True) if registered: with _core._lock: _core._lazy_server_configs[name] = dict(config) diff --git a/tools/mcp_tool_sampling.py b/tools/mcp_tool_sampling.py index 48813e3968..91c3492e22 100644 --- a/tools/mcp_tool_sampling.py +++ b/tools/mcp_tool_sampling.py @@ -105,15 +105,14 @@ def _response_total_tokens(response, default): class SamplingHandler: - """Handles sampling/createMessage requests for one MCP server. + """Handles sampling/createMessage requests for one MCP server; passed to + ``ClientSession`` as ``sampling_callback``. All state (rate-limit + timestamps, metrics, tool-loop counter) is per instance. Runs on the MCP + background loop; the sync LLM call is offloaded via ``asyncio.to_thread``. Deprecated upstream (MCP 2026-07-28, SEP-2577, 12-month window): stays fully functional because handshake-era servers still issue it, but do NOT grow new capability here — modern servers use MRTR, handled by the SDK session layer. - - Callable; passed to ``ClientSession`` as ``sampling_callback``. All state - (rate-limit timestamps, metrics, tool-loop counter) is per instance. Runs on - the MCP background loop; the sync LLM call is offloaded via ``asyncio.to_thread``. """ _STOP_REASON_MAP = {"stop": "endTurn", "length": "maxTokens", "tool_calls": "toolUse"} @@ -135,8 +134,7 @@ class SamplingHandler: def _check_rate_limit(self) -> bool: """Sliding-window (60s) limiter; True if the request is allowed.""" now = time.time() - window = now - 60 - self._rate_timestamps[:] = [t for t in self._rate_timestamps if t > window] + self._rate_timestamps[:] = [t for t in self._rate_timestamps if t > now - 60] if len(self._rate_timestamps) >= self.max_rpm: return False self._rate_timestamps.append(now) @@ -151,11 +149,6 @@ class SamplingHandler: return hint.name return None - @staticmethod - def _extract_tool_result_text(block) -> str: - """Extract text from a ToolResultContent block.""" - return _tool_result_text(block) - def _convert_messages(self, params) -> List[dict]: """Convert MCP SamplingMessages to OpenAI format (``content_as_list`` when the SDK provides it; per-block duck-typed dispatch).""" @@ -327,13 +320,11 @@ def _format_elicitation_schema_summary(schema: dict, server_name: str) -> str: class ElicitationHandler: - """Handles ``elicitation/create`` requests for one MCP server. - - Callable; passed to ``ClientSession`` as ``elicitation_callback``. Form-mode - requests route through Hermes' approval system (CLI, TUI, Telegram, ...); - URL-mode is declined as unsupported. Fail-closed: any timeout, exception, - or unexpected state returns decline/cancel, never a silent accept. - """ + """Handles ``elicitation/create`` requests for one MCP server; passed to + ``ClientSession`` as ``elicitation_callback``. Form-mode requests route + through Hermes' approval system (CLI, TUI, Telegram, ...); URL-mode is + declined as unsupported. Fail-closed: any timeout, exception or unexpected + state returns decline/cancel, never a silent accept.""" # asyncio-side safety net over the approval's own input() timeout so the # MCP loop never blocks indefinitely if the inner timeout is bypassed.