open/portfolio-optimiser
1
0
Fork 0
Code Issues Releases 1 Activity

feat(ingest): add the MCP connector as a transport inside the http family (S2.2)

Browse source 
Commons settled on 2026-08-01 that MCP is an extension of the `http` source
family, not a fourth family (`shared/ingest-spec.md` §4). This implements it
with ZERO schema change and zero spec amendment.

The transport discriminator lives in `base_url`, not in new manifest fields:
the shared library rejects unknown manifest keys fail-fast, and we consume it
pull-only at a pinned v0.3.1, so `server_ref`/`tool` as fields would have meant
a spec amendment plus a library release. It buys nothing — the library already
joins `base_url` + `/` + `query`, so `mcp+stdio://<server_ref>` + `<tool>`
reproduces exactly the two-part structure the (now stale) reference plan wanted.

Staying inside the family INHERITS what a fourth family would have had to write
and could have forgotten: the §8 network grant (measured to fire before any tool
call), the `max_rows` cap, §5 verbatim fenced rendering, and the §7 provenance
stamp. The discriminator gates rather than labels — `mcp_get` refuses a URL it
does not own, so an MCP transport can never quietly serve an `https://` manifest
and leave the bundle's provenance claiming a transport that was never used.

Parsing is string-based, not `urlsplit`-based: `urlsplit().hostname` lowercases
the host, which would silently break the case-sensitive env lookup `server_ref`
depends on.

`ingest.py` is untouched — it is AST-guarded mcp-free, so the transport lives in
its own module and is opt-in at the call site. `ingest_mcp.py` imports the open
`mcp` protocol client but never `agent_framework`, keeping the seam D7-portable.

Load-bearing, six mutations all measured RED: detach the scheme guard · make the
refusal unconditional · swap parsing to `urlsplit().hostname` · skip non-text
content instead of raising · force `allow_network=True` · smuggle in a MAF
import. Both source files restored byte-identical (`shasum -c`) after each.

Honesty: `stdio_call_tool` (the real stdio path) is written but never executed
end to end — every test injects a canned tool call, so the suite spawns no
subprocess and opens no socket. No golden fixture, and MCP stays unwired in the
optimiser run path. Stated in docs/extending.md rather than implied away.

555 -> 578 tests; ruff + mypy green.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0112FPR5TX6pDLiNicBPzE8i
This commit is contained in:
Kjell Tore Guttormsen 2026-08-02 21:14:57 +02:00
commit ddd6338f02
4 changed files with 683 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

186
src/portfolio_optimiser/ingest_mcp.py Normal file
View file

@ -0,0 +1,186 @@
"""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.1``, 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://<server_ref>" query: "<tool>"
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}<server_ref>/<tool>, 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 — every contract test injects a canned
``McpCallTool`` instead, so the suite spawns no subprocess (cost discipline: no heavy runs).
"""
import asyncio
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
async def run() -> str:
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)
return asyncio.run(asyncio.wait_for(run(), timeout=timeout_seconds))
return call_tool
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)