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

feat(mcp): concrete MCP servers become tools the agents can call during a run

Browse source 
Krav 3, and the operator chose the run path explicitly: the external service must
be reachable WHILE the run works, not only when documents are ingested. Until now
the run path had one in-process tool against a local folder — and on the bundle
path the agents had no tools at all.

MAF already ships the client (MCPStdioTool / MCPStreamableHTTPTool, verified in
the pinned 1.9.0 with allowed_tools and request_timeout), so `mcp_tools.py` owns
only what MAF cannot decide for us: which servers a run may contact, which of
their tools it may call, how long it waits, and where the credential comes from.
This is a DIFFERENT seam from ingest_mcp.py on purpose — that one pulls source
documents before a run and speaks to null-argument tools. Same protocol, different
job.

Every refusal is a live hazard, not tidiness. An empty allowlist would let the far
end decide what the agents may call, so naming the tools is mandatory. A
non-positive timeout is an unbounded wait against a third party. An unknown field
is refused rather than ignored, which is also what keeps a literal secret from
being parked in the config — there is no field for one, only the NAME of an env
var. A named-but-unset credential refuses instead of calling anonymously, because
an anonymous call can succeed with the wrong scope.

Egress is declared, always. Every server and permitted tool is named in the run
announcement before the first call — including when no --mandate is given, which
was a real hole: the announcement only printed with a commission, so configuring
servers without one would have contacted third parties with nothing printed at
all. --live-dry-run still opens nothing, because the tools are entered after the
dry-run cut: the promise to stop before the first call now covers egress too.

Threaded through BOTH modes. A flag accepted in one mode and silently dropped in
the other is the defect class this CLI refuses by name.

Load-bearing MEASURED against the whole 744-test suite, four mutations all red:
build the tools but never hand them to the agents (2) · never enter the
AsyncExitStack, so they are constructed and useless (1) · never declare the egress
(2) · drop the allowlist on the built client (1).

Two live docs claimed MCP was unwired in the run path; both corrected rather than
left to rot.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ULCqjLF61rehj5cZmdUoR3
This commit is contained in:
Kjell Tore Guttormsen 2026-08-05 16:53:07 +02:00
commit 9668e17f2f
8 changed files with 706 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

177
src/portfolio_optimiser/mcp_tools.py Normal file
View file

@ -0,0 +1,177 @@
"""Concrete MCP servers as tools on the RUN path (Trekk B1, krav 3).
MAF already ships the MCP client (``MCPStdioTool`` / ``MCPStreamableHTTPTool``, verified present in
the pinned 1.9.0 alongside ``allowed_tools`` / ``request_timeout``), so this module owns only what
MAF cannot decide for us: **which** servers a run may contact, **which** of their tools it may call,
how long it waits, and where the credential comes from.
MAF-SPECIFIC by design, and therefore NOT in ``_MAF_FREE_MODULES``: it builds framework client
objects. That is the deliberate split from ``mandate.py``, which stays framework-neutral and
D7-portable. The D7 sibling will need its own transport module against the same config shape.
**This is a different seam from ``ingest_mcp.py``, on purpose.** That one pulls SOURCE DOCUMENTS
into a bundle before a run and speaks to null-argument tools; this one hands live tools to the
agents DURING the debate. They share a protocol, not a job.
The transport set is CLOSED (``stdio`` | ``http``): a config file can never name arbitrary
machinery to load — the same rule ``--embedder-config`` follows, for the same reason.
Every refusal below exists because its absence is a live hazard:
* an EMPTY ``allowed_tools`` hands the agents whatever the far end chooses to serve — authority
granted by the other party. You must name what you allow;
* a non-positive timeout is an unbounded wait against a third party, which is the fail-fast
invariant this repo applies to every loop and cap;
* an UNKNOWN field is refused rather than ignored, which is also what keeps a literal secret from
being parked in the file and carried along unnoticed — there is no field for one;
* a named credential env var that is NOT set refuses the run rather than calling the service
anonymously: an unauthenticated call can still succeed, with the wrong scope, which is worse
than not calling at all.
"""
from __future__ import annotations
import os
from pathlib import Path
from typing import Any, Literal
from agent_framework import MCPStdioTool, MCPStreamableHTTPTool
from pydantic import BaseModel, ConfigDict, Field, model_validator
class McpServerConfig(BaseModel):
"""One external MCP server a run is permitted to contact.
``extra="forbid"`` is load-bearing, not tidiness: it is what turns a stray ``"credential":
"sk-..."`` into a refusal instead of an ignored key sitting in a config file.
"""
model_config = ConfigDict(extra="forbid", frozen=True)
name: str = Field(min_length=1)
transport: Literal["stdio", "http"]
allowed_tools: tuple[str, ...] = Field(min_length=1)
timeout_seconds: int = Field(gt=0)
command: str | None = None
args: tuple[str, ...] = ()
url: str | None = None
#: The NAME of an environment variable holding the credential — never the credential itself.
credential_env: str | None = None
@model_validator(mode="after")
def _coordinates_match_the_transport(self) -> McpServerConfig:
if self.transport == "stdio":
if not self.command:
raise ValueError(f"{self.name}: stdio transport requires 'command'")
if self.url:
raise ValueError(
f"{self.name}: stdio transport must not carry 'url' — it is ambiguous which "
"endpoint would actually be contacted"
)
else:
if not self.url:
raise ValueError(f"{self.name}: http transport requires 'url'")
if self.command:
raise ValueError(
f"{self.name}: http transport must not carry 'command' — it is ambiguous "
"which endpoint would actually be contacted"
)
return self
class McpConfig(BaseModel):
"""The full set of servers a run may contact."""
model_config = ConfigDict(extra="forbid")
servers: tuple[McpServerConfig, ...] = ()
@model_validator(mode="after")
def _names_are_unique(self) -> McpConfig:
seen: set[str] = set()
for server in self.servers:
if server.name in seen:
raise ValueError(f"duplicate MCP server name: {server.name!r}")
seen.add(server.name)
return self
def load_mcp_config(path: str | Path) -> tuple[McpServerConfig, ...]:
"""Fail-fast loader (mirrors ``load_dimension`` / ``load_mandate``).
Authoritative startup input: missing -> ``FileNotFoundError``, malformed -> ``ValidationError``.
A tolerant read here would be the worst of the three loaders in this repo — degrading a broken
egress config to "no external services" would leave the announcement telling the truth about a
run that was never configured, while a degraded one that *partially* parsed could contact a
subset nobody chose.
"""
p = Path(path)
if not p.is_file():
raise FileNotFoundError(f"MCP config not found: {str(path)!r}")
return McpConfig.model_validate_json(p.read_text(encoding="utf-8")).servers
def _credential(config: McpServerConfig) -> str | None:
"""The credential value, read from the environment at BUILD time. A named-but-unset variable
refuses: calling an external service anonymously can succeed with the wrong scope."""
if config.credential_env is None:
return None
value = os.environ.get(config.credential_env)
if not value:
raise ValueError(
f"{config.name}: credential env var {config.credential_env!r} is not set — refusing "
"to contact the service unauthenticated"
)
return value
def build_mcp_tools(configs: tuple[McpServerConfig, ...]) -> list[Any]:
"""Build one MAF MCP client per configured server, with the allowlist and timeout applied.
NOTE for the caller: an ``MCPTool`` is an async context manager. These objects are constructed
here but NOT connected — the run path must enter them around the debate and exit afterwards.
"""
tools: list[Any] = []
for config in configs:
credential = _credential(config)
if config.transport == "stdio":
# ``credential`` is non-None only when ``credential_env`` named a variable, so the key
# is a str here — narrowed explicitly rather than asserted, so mypy sees it too.
env = (
{config.credential_env: credential}
if credential and config.credential_env
else None
)
tools.append(
MCPStdioTool(
name=config.name,
command=config.command or "",
args=list(config.args),
env=env,
allowed_tools=config.allowed_tools,
request_timeout=config.timeout_seconds,
)
)
else:
header_provider = (
(lambda _ctx, token=credential: {"Authorization": f"Bearer {token}"})
if credential
else None
)
tools.append(
MCPStreamableHTTPTool(
name=config.name,
url=config.url or "",
allowed_tools=config.allowed_tools,
request_timeout=config.timeout_seconds,
header_provider=header_provider,
)
)
return tools
def service_labels(configs: tuple[McpServerConfig, ...]) -> tuple[str, ...]:
"""The egress declaration, one label per server: what will be contacted and which tools are
permitted. Feeds ``mandate.announce(external_services=...)``, so an operator sees the full
reach of a run BEFORE the first call — a run never contacts a service it did not announce."""
return tuple(f"{c.name} ({', '.join(c.allowed_tools)})" for c in configs)

59
src/portfolio_optimiser/run.py
View file

@ -28,6 +28,7 @@ from __future__ import annotations
import asyncio
import json
from collections.abc import Awaitable, Callable, Iterable, Sequence
from contextlib import AsyncExitStack
from dataclasses import dataclass, replace
from pathlib import Path
from typing import Any, Literal, cast
@ -64,6 +65,12 @@ from portfolio_optimiser.mandate import (
load_mandate,
settle,
)
from portfolio_optimiser.mcp_tools import (
McpServerConfig,
build_mcp_tools,
load_mcp_config,
service_labels,
)
from portfolio_optimiser.provenance import ProvenanceStamp
from portfolio_optimiser.reference_domain import Project, load_reference_projects
from portfolio_optimiser.validator import Rejection, ValidatedProposal, baseline_from_project
@ -412,6 +419,7 @@ async def run_project(
semantic_retrieval: bool = False,
embedder: Embedder | None = None,
mandate: Mandate | None = None,
mcp_servers: tuple[McpServerConfig, ...] = (),
) -> RunResult | DryRunReport:
"""Run the vertical slice for ONE project. ``client_factory`` is the test-injection seam
(defaults to the real backend). ``verdict_input`` carries the expert decision/rationale
@ -491,6 +499,14 @@ async def run_project(
citations = [chunk_dict_to_citation(c) for c in chunks]
context = "\n".join(c["snippet"] for c in chunks)
debate_tools = [make_retrieval_tool(docs_dir, top_k=top_k)]
# Trekk B2 (krav 3): configured MCP servers become tools the AGENTS can call during the debate.
# Appended to BOTH paths — on the bundle path they are the first tools that path has ever had.
# Constructed here but NOT connected: an ``MCPTool`` is an async context manager, so the run
# enters them around ``debate.run`` below and exits afterwards. Empty tuple -> the tool list is
# byte-identical to the pre-Trekk-B one, and no network call is possible.
live_mcp_tools = build_mcp_tools(mcp_servers) if mcp_servers else []
debate_tools = debate_tools + live_mcp_tools
if not citations:
raise ValueError(f"no citable content in docs_dir: {docs_dir!r}")
@ -539,7 +555,16 @@ async def run_project(
max_tokens=max_tokens,
top_k=top_k,
)
result = await debate.run(f"Find a cost-saving measure for {project.id}.\nContext:\n{context}")
# The MCP lifecycle (Trekk B2): entered HERE, after the dry-run cut above, so a dry run never
# opens a connection — its promise to stop before the first call covers egress too. Constructed
# tools that are never entered expose nothing, and ones never exited leave the process hanging,
# so the stack owns both halves.
async with AsyncExitStack() as mcp_stack:
for live_tool in live_mcp_tools:
await mcp_stack.enter_async_context(live_tool)
result = await debate.run(
f"Find a cost-saving measure for {project.id}.\nContext:\n{context}"
)
# F1: the candidate must derive from the DEBATE. Feed the proposer's converged output into
# generation (retrieval context is the last-resort fallback only). The checker's verdict
# (Step 3/4) is parsed from the SAME debate result and gates the outcome below.
@ -853,6 +878,7 @@ async def run_portfolio(
semantic_retrieval: bool = False,
embedder: Embedder | None = None,
mandate: Mandate | None = None,
mcp_servers: tuple[McpServerConfig, ...] = (),
) -> PortfolioResult:
"""Fan out over a portfolio of independent projects SEQUENTIALLY, composing ``run_project``
as-is (every project's execution state — meter, debate, retrieval context — is built fresh
@ -1050,6 +1076,7 @@ async def run_portfolio(
semantic_retrieval=semantic_retrieval,
embedder=embedder,
mandate=mandate,
mcp_servers=mcp_servers,
meter=_run_meter(meter_factory, portfolio_meter, max_rounds),
)
for pid, snapshot in snapshots
@ -1182,6 +1209,16 @@ def main(argv: list[str] | None = None) -> int:
"afterwards, one row per approach. Valid in both modes; in portfolio mode it applies to "
"every project in the pass",
)
parser.add_argument(
"--mcp-config",
default=None,
metavar="FILE",
help="external MCP servers this run may contact (JSON, fail-fast): name, transport "
"(stdio|http), coordinates, the ALLOWED tool names, a timeout, and optionally the NAME of "
"an env var holding the credential (never the credential itself). Every server and tool is "
"named in the run announcement BEFORE the first call; without this flag no network call is "
"made at all",
)
parser.add_argument(
"--embedder-config",
default=None,
@ -1432,6 +1469,17 @@ def main(argv: list[str] | None = None) -> int:
print(f"run refused: {exc}", file=sys.stderr)
return 1
# The egress config, loaded fail-fast alongside the commission. Degrading a broken one to "no
# external services" would make the announcement describe a run nobody configured, and a
# partially-parsed one could contact a subset nobody chose.
mcp_servers: tuple[McpServerConfig, ...] = ()
if args.mcp_config is not None:
try:
mcp_servers = load_mcp_config(args.mcp_config)
except (FileNotFoundError, ValidationError, ValueError) as exc:
print(f"run refused: {exc}", file=sys.stderr)
return 1
# The scripted door (offline WHOLE-loop run over the caller's own data). Resolved BEFORE the
# dry-run branch so the two offline modes cannot both be honoured — and BEFORE the portfolio
# dispatch, because the door serves BOTH modes. It originally sat below that dispatch, which
@ -1487,8 +1535,14 @@ def main(argv: list[str] | None = None) -> int:
max_rounds=_DEFAULT_MAX_ROUNDS,
max_tokens=_DEFAULT_MAX_TOKENS,
dimension_label=dimension_label,
external_services=service_labels(mcp_servers),
)
)
elif mcp_servers:
# The egress declaration must NOT depend on a commission being present. Without this
# branch, configuring servers and omitting --mandate would contact third parties with
# nothing printed at all — silent egress, which this repo forbids outright.
print("Contacts: " + ", ".join(service_labels(mcp_servers)))
if args.portfolio:
# Portfolio mode (Step 3): dispatch to the EXISTING run_portfolio via the fail-fast loaders
@ -1515,6 +1569,7 @@ def main(argv: list[str] | None = None) -> int:
semantic_retrieval=args.semantic_retrieval,
client_factory=scripted_client_factory,
mandate=mandate,
mcp_servers=mcp_servers,
)
)
except (ValueError, FileNotFoundError, ValidationError) as exc:
@ -1590,6 +1645,7 @@ def main(argv: list[str] | None = None) -> int:
outbox_dir=args.outbox_dir,
run_id=args.run_id,
verdict_input={"decision": args.decision, "rationale": args.rationale},
mcp_servers=mcp_servers,
live_dry_run=True,
)
)
@ -1639,6 +1695,7 @@ def main(argv: list[str] | None = None) -> int:
semantic_retrieval=args.semantic_retrieval,
client_factory=scripted_client_factory,
mandate=mandate,
mcp_servers=mcp_servers,
)
),
)