"""Door A ingest — the MCP connector, as a TRANSPORT inside the ``http`` source family (§4). ``shared/ingest-spec.md`` §4: "An MCP-based connector is an extension of this family and MUST honour the same extraction, materialization, and gate contracts." Commons settled this on 2026-08-01 — MCP is **not** a fourth source family. This module is that extension, and nothing more: it adapts an MCP tool call into the shape the shared library's transport seam already expects, so every contract downstream of the transport is the EXISTING one. **Where the transport discriminator lives, and why.** The shared library validates manifests fail-fast and rejects unknown keys (``llm_ingestion_okf.manifest._require_keys``): an ``http`` source admits exactly ``{type, id, base_url}`` plus optional ``credential_ref``. We consume that library PULL-ONLY at a pinned ``v0.3.2``, so introducing ``server_ref``/``tool`` as manifest fields would mean a spec amendment plus a library release. It buys nothing: the library's own URL join composes ``base_url`` + ``/`` + ``query``, so base_url: "mcp+stdio://" query: "" reproduces exactly the two-part structure a dedicated ``mcp`` family would have carried — with zero schema change. The spec's MUST sits on the CONTRACTS (extraction, materialization, gates), not on field names; naming was explicitly left to the consumer. **What we inherit for free by staying inside the family** — and would have had to write, and could have forgotten, as a fourth family: - the §8 network gate (an MCP source is a non-local transport; ``allow_network`` is a run argument, never a manifest field, and it refuses BEFORE any transport call); - the §8 ``max_rows`` cap, enforced as an ERROR rather than a silent truncation; - §5 verbatim fenced-body rendering, including the code-fence-marker refusal; - the §7 ``generated: true`` + ``ingest_manifest`` provenance stamp; - the §4 rule that credentials resolve at run time from the environment, never from the manifest. **MAF-free (D7-portable).** MCP is an open protocol; ``agent_framework`` is one framework's runtime. This module may import the former and never the latter, so the sibling Claude-SDK implementation can consume the identical seam. Guarded by ``tests/test_ingest_mcp_loadbearing.py::test_ingest_mcp_is_maf_free``. Note this module is deliberately NOT imported by ``portfolio_optimiser.ingest``: that module is AST-guarded against importing ``mcp`` at all (``tests/test_ingest_loadbearing.py::test_ingest_module_is_maf_free_and_context_layer_pure``). The MCP transport is opt-in at the CALL SITE — ``materialize(..., http_get=mcp_get(...))`` — which is also why an ``http`` manifest can never quietly acquire an MCP transport. """ from __future__ import annotations import os from collections.abc import Callable from typing import Any from llm_ingestion_okf.connectors import HttpGet from portfolio_optimiser.ingest import IngestError #: The transport token in ``base_url``. One source of truth: the parser accepts this and nothing #: else, so manifest-authoring docs and the parser cannot drift apart. MCP_SCHEME = "mcp+stdio" #: An MCP tool invocation reduced to what the transport seam needs: given the server reference, #: the tool name and an optional run-time-resolved credential, return the response body verbatim. #: This is the extension point — ``stdio_call_tool`` is the real implementation, and tests inject #: a canned one so the suite runs with no subprocess and no socket. McpCallTool = Callable[[str, str, "str | None"], str] def parse_mcp_url(url: str) -> tuple[str, str]: """Split a joined MCP URL back into ``(server_ref, tool)``, fail-fast on anything else. The library hands the transport ``base_url`` + ``/`` + ``query``; this recovers the two parts. Parsing is deliberately STRING-based rather than ``urllib.parse.urlsplit``: ``urlsplit`` lowercases the host, and ``server_ref`` names an environment variable, whose name is case-sensitive. Going through ``urlsplit`` would turn ``PORTFOLIO_DOCS_MCP`` into ``portfolio_docs_mcp`` and fail the lookup for a reason invisible at the call site. Refusing a URL this transport does not own is a GATE, not a label: it stops an MCP transport from quietly serving a plain ``https://`` manifest, which would leave the bundle's provenance claiming a transport that was never used. """ prefix = f"{MCP_SCHEME}://" if not url.startswith(prefix): raise IngestError( f"MCP transport received a URL it does not own: {url!r} " f"(expected a {prefix}… base_url; spec §4 — the manifest declares its transport)", code="mcp_scheme_mismatch", ) remainder = url[len(prefix) :] server_ref, separator, tool = remainder.partition("/") if not separator or not server_ref or not tool: raise IngestError( f"MCP URL must be {prefix}/, got {url!r} " "(server_ref names an environment variable holding the server command; " "tool is the extraction's `query`)", code="mcp_url_malformed", ) return server_ref, tool def mcp_get(call_tool: McpCallTool) -> HttpGet: """Adapt an MCP tool call into the library's ``HttpGet`` transport seam. Returns a ``(url, credential) -> str`` callable — the exact shape the library injects as ``http_get`` — which parses the URL, refuses one it does not own, and delegates the fetch to ``call_tool``. The credential arrives already resolved from the environment by the library and is passed out-of-band; it is never joined into the URL, never logged, never stamped. """ def get(url: str, credential: str | None) -> str: server_ref, tool = parse_mcp_url(url) return call_tool(server_ref, tool, credential) return get def stdio_call_tool( *, args: tuple[str, ...] = (), timeout_seconds: float = 30.0, ) -> McpCallTool: """The real stdio transport: launch the MCP server named by ``server_ref`` and call its tool. ``server_ref`` is the NAME of an environment variable holding the server command — mirroring the ``sql`` family's ``connection_ref`` — so the manifest carries a reference, never an executable path, and stays versionable and shareable. A missing or empty variable is fail-fast BEFORE any process is spawned. The credential, when present, is passed to the server as the ``MCP_CREDENTIAL`` environment variable rather than as a tool argument: tool arguments are echoed in MCP progress and result payloads and would land in logs. Kept thin and separate from :func:`mcp_get` on purpose — the CONTRACT tests (``tests/test_ingest_mcp_loadbearing.py``) inject a canned ``McpCallTool``, so they spawn no subprocess. The stdio path itself is covered separately by ``tests/test_ingest_golden_mcp.py``, which runs it against a REAL server subprocess (a local process costs no model tokens, so the cost discipline that governs this repo is untouched). A server consumed here must expose a NULL-ARGUMENT tool: the URL carries both coordinates and the tool is invoked with an empty argument dict. ``datasource.build_mcp_server`` does not qualify — its ``retrieve_cost_docs(query)`` has a required parameter; it serves the agents' retrieval path, not ingest. """ import asyncio import anyio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client def call_tool(server_ref: str, tool: str, credential: str | None) -> str: command = os.environ.get(server_ref) if not command: raise IngestError( f"MCP server_ref {server_ref!r} is not set in the environment " "(the manifest names a reference; the command resolves at run time — spec §4)", code="mcp_server_ref_unset", ) env = dict(os.environ) if credential is not None: env["MCP_CREDENTIAL"] = credential # Populated with OUR OWN cancel scope once `run()` enters it — read back after # `asyncio.run` returns/raises, to tell "our deadline fired" apart from any other # `TimeoutError` (builtin `TimeoutError` is also `socket.timeout`/`asyncio.TimeoutError` # since 3.10/3.11, so the exception TYPE alone does not prove ownership). deadline: dict[str, anyio.CancelScope] = {} async def run() -> str: # The deadline is an anyio cancel scope, not `asyncio.wait_for`, and it wraps BOTH # nested task groups (`stdio_client`, `ClientSession`) from the INSIDE. `wait_for` # cancels from outside a structure anyio itself owns, and the two cancellation # mechanisms do not compose: measured (2026-08-03), that mismatch surfaced as an # `anyio.BrokenResourceError` wrapped in a `BaseExceptionGroup` — never a `TimeoutError` # — because a background reader task lost its write end mid-teardown. anyio's own scope # is what the task groups already coordinate cancellation through, so nesting inside it # tears down cleanly and raises a plain `TimeoutError` at the `with` statement. with anyio.fail_after(timeout_seconds) as scope: deadline["scope"] = scope params = StdioServerParameters(command=command, args=list(args), env=env) async with stdio_client(params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() result = await session.call_tool(tool, {}) if result.isError: raise IngestError( f"MCP tool {tool!r} on {server_ref!r} returned an error result", code="mcp_tool_error", ) return _body_from_content(result.content, tool=tool, server_ref=server_ref) try: return asyncio.run(run()) except TimeoutError as exc: # Mirrors `_unwrap_ingest_error`'s ownership rule: own it, wrap it; otherwise, # untouched. Only OUR scope hitting ITS OWN deadline earns `mcp_timeout` — a # `TimeoutError` from elsewhere (there is no live source today: MCP's own internal # read-timeout converts to `McpError` before reaching us, and a server-side # `TimeoutError` becomes an ordinary `isError` result, both measured — but the type # alone does not guarantee it) is re-raised exactly as `_unwrap_ingest_error` would. scope = deadline.get("scope") if scope is not None and scope.cancelled_caught: raise IngestError( f"MCP tool {tool!r} on {server_ref!r} did not respond within " f"{timeout_seconds}s", code="mcp_timeout", ) from exc raise except BaseException as exc: # noqa: BLE001 — re-raised unchanged unless we own it owned = _unwrap_ingest_error(exc) if owned is None: raise raise owned from exc return call_tool def _unwrap_ingest_error(exc: BaseException) -> IngestError | None: """Recover an :class:`IngestError` that anyio re-packaged into a task-group exception group. ``stdio_client`` and ``ClientSession`` are each an anyio task group, and anyio wraps ANYTHING leaving one in a ``BaseExceptionGroup`` — nested once per group. So every error this transport raises from inside the session (``mcp_tool_error``, ``mcp_non_text_content``) reached callers as an exception group, never as the ``IngestError`` the whole Door A path catches and switches on by ``code``. Found only by running against a real server; the canned-tool tests never enter a task group. Duck-typed on ``.exceptions`` rather than ``except*`` / ``ExceptionGroup``: both are 3.11+, and this project supports ``>=3.10``, where anyio raises the ``exceptiongroup`` backport instead. Returns ``None`` for anything we do not own, so the caller re-raises it untouched — this narrows an exception group, it is never a blanket re-raise. """ if isinstance(exc, IngestError): return exc for sub in getattr(exc, "exceptions", ()): owned = _unwrap_ingest_error(sub) if owned is not None: return owned return None def _body_from_content(content: list[Any], *, tool: str, server_ref: str) -> str: """Concatenate an MCP tool result's TEXT blocks into the verbatim body (§5). Non-text content (images, embedded resources) is an ERROR rather than a silent drop: the body is rendered verbatim into the concept file, and quietly discarding part of an extraction would produce a bundle that looks complete and is not. """ parts: list[str] = [] for block in content: text = getattr(block, "text", None) if text is None: raise IngestError( f"MCP tool {tool!r} on {server_ref!r} returned non-text content " f"({type(block).__name__}); ingest renders bodies verbatim and never drops content", code="mcp_non_text_content", ) parts.append(text) return "".join(parts)