"""Map Model Context Protocol servers onto the existing tool framework. **MCP is a distribution format, not a capability.** Nothing in the literature suggests a model calls tools better because they arrived over MCP; the model sees a name, a description and a JSON schema either way. What MCP changes is who writes the tool, how it reaches the operator's machine, and what an attacker can do with that. So this module is mostly about the second part. It does **not** build a parallel tool system. An MCP tool becomes an ordinary :class:`~distinct_tools.core.ToolSpec` in an ordinary :class:`~distinct_tools.core.Registry`, called through the ordinary per-job :class:`~distinct_tools.core.ToolBroker`. Every existing property survives unchanged: exact ``id@version`` references, the per-job allowlist snapshot, quotas, timeouts, byte caps and the fail-closed default. Four decisions carry the security argument, and each one exists because the obvious alternative is what the published MCP attacks target. 1. **Discovery happens at install time, never during a run.** ``tools/list`` is called once, by an operator who is installing something. A run never negotiates with a server. This is why ``ToolBroker.__init__`` can keep snapshotting the registry, and it is why lazy-loading tool registries, which the practitioner literature recommends for large catalogues, are the wrong pattern here and are deliberately not supported. 2. **Tool definitions are digest-pinned at approval.** A server that changes a tool's description or schema after the operator approved it produces a different digest and is refused. Tool poisoning by later mutation, the highest-ranked MCP client risk, is a rug-pull; pinning is the answer to a rug-pull. 3. **``required_hosts`` is declared by the operator, never read from the server.** A server that could declare its own network permissions would be granting them to itself. 4. **Descriptions are untrusted text that reaches the prompt.** They are bounded, stripped of control characters, and tagged with their origin so a harness can fence them the way it already fences tool *results*. **One capability the in-process path has and the stdio path deliberately does not.** A handler running in this process may be handed the calling run's :class:`~distinct_tools.core.ToolContext`, so a server holding per-run state can key it by run identity instead of asking the model for an identifier. Trusting the model with that identifier would make one run's notes reachable from another run by guessing a string, which is a disclosure rather than a bug in a prompt. A transport opts in by setting ``accepts_context``; :class:`InProcessMcpServer` does, :class:`StdioMcpTransport` does not. A ``ToolContext`` is a capability handle, and writing one down a pipe to a process this runtime cannot govern would advertise a boundary that is not there. **What this module cannot enforce, stated plainly.** It governs *declaration*, not *behaviour*, and that distinction is the whole of this paragraph. An MCP tool that declares a host is refused unless the operator separately approved egress for it, and the digest pin means a server cannot change what it declared after approval. Neither of those stops code from opening a socket. For an out-of-process (stdio) server the reason is obvious: the server performs its own I/O in its own process and nothing here is in the way. For an in-process server the reason is less obvious and was, until a red-team run found it, stated the other way round in this file. ``ToolContext.require_url`` is a check a cooperating handler calls, not a wall around one: a handler runs on an ordinary thread in the agent process with no import hook, no audit hook, no seccomp filter and no namespace. A handler that imports ``socket`` connects, and the trace records a clean success. Reproduced, with a spec declaring no hosts under a policy allowing none. So the honest statement is that in-process is preferred for a different reason than containment: it keeps the tool inside the same review, the same policy and the same digest pin as everything else, and it removes a process the runtime would have no visibility into at all. What actually contains either shape is the OS-native isolation layer around the worker, and ``distinct_agent/containment.py`` is candid about how much of that is in force on which platform. Installing a third-party in-process MCP server is a decision to run somebody else's code in this process, and this module makes that decision explicit and pinned rather than safe. """ from __future__ import annotations import hashlib import json import re import threading from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass, field from typing import Any, Protocol from .core import ( Registry, ToolCallError, ToolContext, ToolRef, ToolSpec, normalize_host, ) #: Namespace separator between a server slug and its tool name. Two servers #: may both publish ``search``; without a namespace one would shadow the other #: and the operator could not tell which they approved. NAMESPACE_SEPARATOR = "." _SLUG_RE = re.compile(r"^[a-z][a-z0-9]*(?:[_-][a-z0-9]+)*$") _MCP_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,95}$") MAX_TOOLS_PER_SERVER = 32 MAX_DESCRIPTION_CHARACTERS = 1_500 MAX_SCHEMA_CHARACTERS = 20_000 MAX_MESSAGE_BYTES = 1_048_576 PROTOCOL_VERSION = "2025-06-18" class McpError(RuntimeError): """An MCP server could not be used. Callers install nothing and continue.""" class McpDefinitionChanged(McpError): """A tool's definition no longer matches the digest the operator approved.""" def _clean_text(value: object, limit: int) -> str: text = "" if value is None else str(value) cleaned = "".join(character for character in text if ord(character) >= 32 or character == " ") return cleaned.strip()[:limit] def _canonical(value: object) -> str: """Stable rendering for wire messages and for wrapping handler output. ``default=str`` is a deliberate fallback here: a Python handler may return something unusual and describing it is better than failing the call. """ return json.dumps(value, ensure_ascii=False, allow_nan=False, sort_keys=True, separators=(",", ":"), default=str) def _canonical_strict(value: object) -> str: """Stable rendering for anything a digest is taken over. No ``default`` fallback, on purpose. A schema containing a value that is not plain JSON must be *refused*, not stringified: stringifying it would put ``"mappingproxy({...})"`` in the schema the model reads and would make the digest depend on a repr rather than on the definition. """ return json.dumps( value, ensure_ascii=False, allow_nan=False, sort_keys=True, separators=(",", ":") ) @dataclass(frozen=True) class McpToolDefinition: """One tool exactly as a server described it, before any of it is trusted.""" name: str description: str input_schema: Mapping[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: if not isinstance(self.name, str) or not _MCP_NAME_RE.fullmatch(self.name): raise McpError( f"MCP tool name {self.name!r} is not a short alphanumeric identifier" ) object.__setattr__(self, "description", _clean_text(self.description, MAX_DESCRIPTION_CHARACTERS)) if not self.description: # A tool the model cannot tell apart from another tool is worse # than no tool: selection error is the dominant small-model # failure and an empty description guarantees it. raise McpError(f"MCP tool {self.name!r} has no usable description") if not isinstance(self.input_schema, Mapping): raise McpError(f"MCP tool {self.name!r} has a malformed input schema") try: encoded = _canonical_strict(self.input_schema) except (TypeError, ValueError) as exc: raise McpError( f"MCP tool {self.name!r} has an input schema that is not plain JSON" ) from exc if len(encoded) > MAX_SCHEMA_CHARACTERS: raise McpError(f"MCP tool {self.name!r} has an oversized input schema") object.__setattr__(self, "input_schema", json.loads(encoded)) @property def digest(self) -> str: """Identity of everything the model will see about this tool. Name, description and schema, and nothing else. A server may change its implementation freely; it may not change what it told the operator it was, because that is what the operator approved. """ payload = _canonical_strict( { "name": self.name, "description": self.description, "input_schema": self.input_schema, } ) return hashlib.sha256(payload.encode("utf-8")).hexdigest() @classmethod def from_wire(cls, value: Mapping[str, Any]) -> McpToolDefinition: if not isinstance(value, Mapping): raise McpError("MCP tool entry is not an object") return cls( name=value.get("name", ""), description=value.get("description", "") or value.get("title", ""), input_schema=value.get("inputSchema") or {}, ) class McpTransport(Protocol): """The two operations distinct needs from an MCP server, and no more. Deliberately narrow. Resources, prompts, sampling and completion are not exposed: sampling in particular would let a server drive the model, which inverts the trust relationship this design exists to preserve. A transport may additionally set ``accepts_context = True``, which makes :func:`install_mcp_server` pass the calling run's ``ToolContext`` to ``call_tool`` as a keyword argument. It is opt-in rather than mandatory so that a transport written against the two methods above keeps working, and so that passing a capability handle is a decision each transport makes rather than something the framework does to it. """ def list_tools(self) -> Sequence[Mapping[str, Any]]: """Return raw tool entries. Called at install time only.""" def call_tool(self, name: str, arguments: Mapping[str, Any], *, timeout: float) -> Any: """Invoke one tool and return its already-decoded result payload.""" @dataclass(frozen=True) class McpServerManifest: """What the operator approved, recorded so it can be checked again later. ``pinned`` maps tool name to definition digest. An empty mapping means "trust on first install": the digests are recorded during :func:`install_mcp_server` and written back for the operator to persist. Once populated it is enforced. """ slug: str version: str = "1" allowed_hosts: frozenset[str] = frozenset() pinned: Mapping[str, str] = field(default_factory=dict) tool_names: tuple[str, ...] = () timeout_seconds: float = 15.0 def __post_init__(self) -> None: if not isinstance(self.slug, str) or not _SLUG_RE.fullmatch(self.slug): raise McpError( "server slug must be lowercase letters, digits, underscore or hyphen, " "starting with a letter" ) object.__setattr__(self, "allowed_hosts", frozenset(normalize_host(host) for host in self.allowed_hosts)) pinned = {} for name, digest in dict(self.pinned).items(): if not isinstance(digest, str) or len(digest) != 64 or not re.fullmatch(r"[0-9a-f]{64}", digest): raise McpError(f"pinned digest for {name!r} is not a sha256 hex digest") pinned[str(name)] = digest object.__setattr__(self, "pinned", dict(pinned)) object.__setattr__(self, "tool_names", tuple(str(name) for name in self.tool_names)) if not isinstance(self.timeout_seconds, int | float) or isinstance(self.timeout_seconds, bool): raise McpError("timeout_seconds must be a number") if not 0.1 <= float(self.timeout_seconds) <= 60.0: raise McpError("timeout_seconds must be between 0.1 and 60") def tool_id_for(self, name: str) -> str: return f"{self.slug}{NAMESPACE_SEPARATOR}{name.lower().replace('-', '_')}" @property def declares_network(self) -> bool: return bool(self.allowed_hosts) @dataclass(frozen=True) class InstalledMcpTool: """The result of installing one tool, for the operator's console.""" ref: ToolRef source_name: str digest: str newly_pinned: bool def _result_payload(raw: Any, tool_id: str) -> Any: """Turn an MCP call result into a plain JSON value, or raise. MCP returns ``content`` as a list of typed blocks. Text blocks are joined; anything else is described rather than dropped, because a tool silently returning nothing looks identical to a tool that worked. """ if not isinstance(raw, Mapping): return raw if raw.get("isError") is True: message = "" content = raw.get("content") if isinstance(content, Sequence) and not isinstance(content, str | bytes): for block in content: if isinstance(block, Mapping) and block.get("type") == "text": message = str(block.get("text", ""))[:500] break raise ToolCallError("tool_error", message or f"{tool_id} reported an error") if "structuredContent" in raw: return raw["structuredContent"] content = raw.get("content") if not isinstance(content, Sequence) or isinstance(content, str | bytes): return raw texts: list[str] = [] others: list[str] = [] for block in content: if not isinstance(block, Mapping): continue kind = str(block.get("type", "")) if kind == "text": texts.append(str(block.get("text", ""))) else: others.append(kind or "unknown") payload: dict[str, Any] = {"text": "\n".join(texts)} if others: payload["omitted_content_types"] = sorted(set(others)) return payload def _make_handler( *, transport: McpTransport, manifest: McpServerManifest, definition: McpToolDefinition, tool_id: str, ) -> Callable[[Mapping[str, Any], ToolContext], Any]: """Wrap one MCP tool as an ordinary broker handler. The broker already owns the deadline, the byte caps, the quota and the result validation, so this does exactly three things the broker cannot: it passes the remaining time down to the transport, it offers the run's context to a transport that has declared it can hold one, and it converts MCP's error shape into the framework's. """ accepts_context = bool(getattr(transport, "accepts_context", False)) def handler(arguments: Mapping[str, Any], context: ToolContext) -> Any: remaining = context.remaining_seconds() if remaining <= 0: raise ToolCallError("timeout", f"{tool_id} had no time left to run") budget = min(remaining, float(manifest.timeout_seconds)) try: if accepts_context: raw = transport.call_tool( definition.name, dict(arguments), timeout=budget, context=context ) else: raw = transport.call_tool(definition.name, dict(arguments), timeout=budget) except ToolCallError: raise except McpError as exc: raise ToolCallError("tool_error", str(exc)[:500]) from exc except Exception as exc: # contained: a server is not trusted to fail well raise ToolCallError( "tool_error", f"{tool_id} failed: {type(exc).__name__}" ) from exc return _result_payload(raw, tool_id) handler.__name__ = re.sub(r"[^0-9a-zA-Z_]", "_", tool_id) return handler def describe_definitions( definitions: Sequence[McpToolDefinition], manifest: McpServerManifest ) -> str: """Render what a server is asking to install, for an operator to read. Printed before approval, because "users approve tool invocations without inspecting them" is the documented root of the tool-poisoning problem and the only fix available here is to put the text in front of them. """ lines = [ f"MCP server '{manifest.slug}' version {manifest.version} offers " f"{len(definitions)} tool(s).", ] if manifest.declares_network: hosts = ", ".join(sorted(manifest.allowed_hosts)) lines.append( f" NETWORK: you have declared outbound access to {hosts} for this server. " "This breaks the runtime's no-outbound-request property." ) else: lines.append(" No outbound hosts declared for this server.") for definition in definitions: lines.append(f" {manifest.tool_id_for(definition.name)}@{manifest.version}") lines.append(f" from tool '{definition.name}', digest {definition.digest[:16]}") lines.append(f" description: {definition.description}") lines.append( " The description text above is supplied by the server and is read by the model. " "Treat it as you would any untrusted instruction." ) return "\n".join(lines) def install_mcp_server( registry: Registry, transport: McpTransport, manifest: McpServerManifest, ) -> tuple[InstalledMcpTool, ...]: """Read a server's tools once and register the approved ones. Called at install or enable time by the operator, never during a run. Raises rather than partially installing: a half-installed server would advertise a capability set nobody approved. """ if not isinstance(registry, Registry): raise TypeError("registry must be a Registry") try: raw_entries = list(transport.list_tools()) except McpError: raise except Exception as exc: raise McpError(f"could not list tools: {type(exc).__name__}") from exc if len(raw_entries) > MAX_TOOLS_PER_SERVER: raise McpError( f"server '{manifest.slug}' offers {len(raw_entries)} tools; the limit is " f"{MAX_TOOLS_PER_SERVER}. Large tool sets measurably reduce selection accuracy, " "so a server must be narrowed rather than accepted whole" ) definitions = [McpToolDefinition.from_wire(entry) for entry in raw_entries] seen: set[str] = set() for definition in definitions: if definition.name in seen: raise McpError(f"server '{manifest.slug}' lists {definition.name!r} twice") seen.add(definition.name) wanted = set(manifest.tool_names) if manifest.tool_names else {d.name for d in definitions} missing = sorted(wanted - seen) if missing: raise McpError( f"server '{manifest.slug}' does not offer approved tool(s): {', '.join(missing)}" ) # Verify every digest before registering any of them, so a changed # definition halts the whole install rather than leaving a mixture. chosen = [definition for definition in definitions if definition.name in wanted] for definition in chosen: expected = manifest.pinned.get(definition.name) if expected is not None and expected != definition.digest: raise McpDefinitionChanged( f"tool {definition.name!r} on server '{manifest.slug}' no longer matches the " "definition you approved. Its name, description or schema has changed. " "Review the new definition and re-approve it deliberately." ) installed: list[InstalledMcpTool] = [] for definition in chosen: tool_id = manifest.tool_id_for(definition.name) spec = ToolSpec( tool_id=tool_id, version=manifest.version, description=definition.description, input_schema=definition.input_schema, required_hosts=manifest.allowed_hosts, origin=f"mcp:{manifest.slug}", ) registry.register( spec, _make_handler( transport=transport, manifest=manifest, definition=definition, tool_id=tool_id, ), ) installed.append( InstalledMcpTool( ref=spec.ref, source_name=definition.name, digest=definition.digest, newly_pinned=definition.name not in manifest.pinned, ) ) return tuple(installed) def pin_manifest( manifest: McpServerManifest, installed: Sequence[InstalledMcpTool] ) -> McpServerManifest: """Return ``manifest`` with every installed digest recorded. The operator persists the result. Next install compares against it, so a server that changes a tool description is caught rather than trusted. """ pinned = dict(manifest.pinned) for item in installed: pinned[item.source_name] = item.digest return McpServerManifest( slug=manifest.slug, version=manifest.version, allowed_hosts=manifest.allowed_hosts, pinned=pinned, tool_names=tuple(item.source_name for item in installed), timeout_seconds=manifest.timeout_seconds, ) # -------------------------------------------------------------------------- # In-process transport: the default, and the only one with no isolation gap # -------------------------------------------------------------------------- class InProcessMcpServer: """An MCP server implemented as Python callables in the worker process. This is the shape the agent binary is built for: no daemon, no subprocess, no socket, nothing to leave running when the agent exits. Handlers run inside the broker's thread and deadline, and ``ToolContext.require_url`` genuinely governs them, which is not true of a subprocess server. """ #: Declares that this transport can hold the calling run's ``ToolContext``. #: Read by :func:`install_mcp_server`; see the module docstring for why the #: stdio transport does not set it. accepts_context = True def __init__(self) -> None: self._definitions: dict[str, McpToolDefinition] = {} self._handlers: dict[str, tuple[Callable[..., Any], bool]] = {} self._lock = threading.RLock() def add_tool( self, definition: McpToolDefinition, handler: Callable[[Mapping[str, Any]], Any], ) -> None: """Add a tool whose handler is a pure function of its arguments.""" self._add(definition, handler, wants_context=False) def add_context_tool( self, definition: McpToolDefinition, handler: Callable[[Mapping[str, Any], ToolContext], Any], ) -> None: """Add a tool whose handler also receives the calling run's context. The handler signature is exactly the framework's ``ToolHandler``, so a tool already written for the direct path can be published here without a shim, and the two paths cannot drift apart. A tool added this way refuses to run without a context rather than inventing one: its whole reason for wanting the context is that something in it is scoped to a run, and a fallback scope would be a shared scope. """ self._add(definition, handler, wants_context=True) def _add( self, definition: McpToolDefinition, handler: Callable[..., Any], *, wants_context: bool, ) -> None: if not callable(handler): raise TypeError("handler must be callable") with self._lock: if definition.name in self._definitions: raise McpError(f"tool {definition.name!r} is already defined") if len(self._definitions) >= MAX_TOOLS_PER_SERVER: raise McpError(f"an MCP server may define at most {MAX_TOOLS_PER_SERVER} tools") self._definitions[definition.name] = definition self._handlers[definition.name] = (handler, wants_context) def list_tools(self) -> Sequence[Mapping[str, Any]]: with self._lock: return [ { "name": definition.name, "description": definition.description, "inputSchema": definition.input_schema, } for definition in sorted(self._definitions.values(), key=lambda item: item.name) ] def call_tool( self, name: str, arguments: Mapping[str, Any], *, timeout: float, context: ToolContext | None = None, ) -> Any: with self._lock: entry = self._handlers.get(name) if entry is None: raise ToolCallError("tool_not_allowed", f"{name} is not defined on this server") handler, wants_context = entry if wants_context: if context is None: raise ToolCallError( "tool_error", f"{name} holds state that belongs to one run and was called without a run " "to belong to", ) value = handler(arguments, context) else: value = handler(arguments) if isinstance(value, Mapping) and ("content" in value or "isError" in value): return value return {"content": [{"type": "text", "text": _canonical(value)}]} # -------------------------------------------------------------------------- # Stdio transport: opt-in, and the isolation gap is real # -------------------------------------------------------------------------- SpawnCallable = Callable[[Sequence[str]], Any] class StdioMcpTransport: """Newline-delimited JSON-RPC over a child process's stdin and stdout. **This is the opt-in path and it has a gap the in-process path does not.** The child does its own I/O, so ``ToolContext.require_url`` cannot govern it and ``required_hosts`` is a record of consent rather than a control. Containing this child is the OS-native isolation layer's job: on Windows it must be assigned to the worker's Job Object before it is resumed, and on Linux it must be inside the same namespace and cgroup. A stdio MCP server spawned outside that boundary is an unsandboxed program running as the operator, and this class will not pretend otherwise. ``spawn`` is injected so the process-creation policy belongs to the caller (which is the component that owns the Job Object), and so the framing can be tested without a real server. """ def __init__( self, command: Sequence[str], *, spawn: SpawnCallable, client_name: str = "distinct-agent", handshake_timeout: float = 20.0, ) -> None: if not command or not all(isinstance(part, str) for part in command): raise McpError("command must be a non-empty sequence of strings") if not callable(spawn): raise McpError("spawn must be callable; it owns the isolation policy") self.command = tuple(command) self._spawn = spawn self._client_name = client_name self._handshake_timeout = handshake_timeout self._process: Any = None self._next_id = 0 self._lock = threading.Lock() def start(self) -> None: if self._process is not None: return process = self._spawn(self.command) if getattr(process, "stdin", None) is None or getattr(process, "stdout", None) is None: raise McpError("spawned MCP server has no stdin/stdout pipes") self._process = process self._request( "initialize", { "protocolVersion": PROTOCOL_VERSION, "capabilities": {}, "clientInfo": {"name": self._client_name, "version": "1"}, }, timeout=self._handshake_timeout, ) self._notify("notifications/initialized", {}) def close(self) -> None: process, self._process = self._process, None if process is None: return for closer in (getattr(process, "stdin", None), getattr(process, "stdout", None)): try: if closer is not None: closer.close() except OSError: pass terminate = getattr(process, "terminate", None) if callable(terminate): try: terminate() except OSError: pass def __enter__(self) -> StdioMcpTransport: self.start() return self def __exit__(self, *exc: object) -> None: self.close() def list_tools(self) -> Sequence[Mapping[str, Any]]: self.start() result = self._request("tools/list", {}, timeout=self._handshake_timeout) tools = result.get("tools") if isinstance(result, Mapping) else None if not isinstance(tools, Sequence) or isinstance(tools, str | bytes): raise McpError("tools/list did not return a list of tools") return [entry for entry in tools if isinstance(entry, Mapping)] def call_tool(self, name: str, arguments: Mapping[str, Any], *, timeout: float) -> Any: self.start() return self._request( "tools/call", {"name": name, "arguments": dict(arguments)}, timeout=timeout ) # -- framing ---------------------------------------------------------- def _write(self, message: Mapping[str, Any]) -> None: payload = _canonical(message) if len(payload.encode("utf-8")) > MAX_MESSAGE_BYTES: raise McpError("outbound MCP message is too large") try: self._process.stdin.write(payload + "\n") self._process.stdin.flush() except (OSError, ValueError, AttributeError) as exc: raise McpError("MCP server stdin is closed") from exc def _notify(self, method: str, params: Mapping[str, Any]) -> None: with self._lock: self._write({"jsonrpc": "2.0", "method": method, "params": dict(params)}) def _request(self, method: str, params: Mapping[str, Any], *, timeout: float) -> Any: with self._lock: self._next_id += 1 request_id = self._next_id self._write( { "jsonrpc": "2.0", "id": request_id, "method": method, "params": dict(params), } ) return self._read_response(request_id, timeout=timeout) def _read_response(self, request_id: int, *, timeout: float) -> Any: deadline_reads = 0 while True: deadline_reads += 1 if deadline_reads > 64: # A server that keeps sending notifications and never answers # is a hang; bound it rather than blocking the broker thread. raise McpError("MCP server sent too many messages without answering") try: line = self._process.stdout.readline() except (OSError, ValueError, AttributeError) as exc: raise McpError("MCP server stdout is closed") from exc if not line: raise McpError("MCP server closed the connection") if isinstance(line, bytes): line = line.decode("utf-8", errors="replace") if len(line) > MAX_MESSAGE_BYTES: raise McpError("MCP server sent an oversized message") stripped = line.strip() if not stripped: continue try: message = json.loads(stripped) except (ValueError, RecursionError) as exc: # RecursionError is a RuntimeError and CPython raises it on # deeply nested JSON, so a hostile or simply broken server # could raise straight past a handler catching only ValueError. raise McpError("MCP server sent malformed JSON") from exc if not isinstance(message, Mapping): raise McpError("MCP server sent a non-object message") if message.get("id") != request_id: # A notification or an answer to something else. Skip it. continue if "error" in message: error = message["error"] detail = "" if isinstance(error, Mapping): detail = _clean_text(error.get("message"), 300) raise ToolCallError("tool_error", detail or "MCP server returned an error") return message.get("result") def local_mcp_server() -> InProcessMcpServer: """An in-process MCP server exposing the local, no-egress tool set. Present so the MCP path is exercised by the same tools the direct path uses, rather than by a fixture that only exists in tests. Installing it produces ``localtools.calculate@1`` and friends, which are *different* refs from the directly registered ``calculate@1``: the same capability reached two ways, and an operator approves whichever one they want. """ from . import local as _local server = InProcessMcpServer() handlers = { "calculate": _local.calculate_handler, "calculate_date": _local.calculate_date_handler, "convert_units": _local.convert_units_handler, "extract_from_text": _local.extract_from_text_handler, "search_document": _local.search_document_handler, "verify_quote": _local.verify_quote_handler, } for spec in _local.LOCAL_SPECS: handler = handlers[spec.tool_id] # to_dict() thaws the deep-frozen schema back to ordinary dicts and # lists. Passing spec.input_schema directly would hand mapping proxies # to the JSON encoder, which is exactly what _canonical_strict refuses. published = spec.to_dict() server.add_tool( McpToolDefinition( name=spec.tool_id, description=published["description"], input_schema=published["input_schema"], ), _bind_local(handler), ) return server def _bind_local(handler: Callable[..., Any]) -> Callable[[Mapping[str, Any]], Any]: """Adapt a broker handler to the in-process MCP calling convention. The context passed here grants no hosts, which is correct: every local tool declares ``required_hosts=frozenset()`` and must not acquire network reach by travelling through the MCP path. """ def call(arguments: Mapping[str, Any]) -> Any: # The local handlers are pure and consult neither the deadline nor the # host list, but an empty host set is still the correct thing to hand # them: travelling through the MCP path must not grant network reach # that the direct path withholds. context = ToolContext(job_id="mcp-local", deadline=0.0, allowed_hosts=frozenset()) return handler(arguments, context) return call __all__ = [ "MAX_DESCRIPTION_CHARACTERS", "MAX_TOOLS_PER_SERVER", "NAMESPACE_SEPARATOR", "PROTOCOL_VERSION", "InProcessMcpServer", "InstalledMcpTool", "McpDefinitionChanged", "McpError", "McpServerManifest", "McpToolDefinition", "McpTransport", "StdioMcpTransport", "describe_definitions", "install_mcp_server", "local_mcp_server", "pin_manifest", ]