Files
classic-models-agent-catalogue/src/catalogue_agent/mcp_clients.py
T
Stéphane Tailland 6ffab873e2 feat: enable catalog writes, with the conversation history they require
The gateway already authorised writes for a product_manager — ai-mcp-proxy
filters tool discovery by consumer group, so create-product is simply absent
from a read-only user's turn. Only the system prompt was holding the agent
back, and it now describes how to write instead: restate the record, ask for
confirmation, never invent a missing field, report the identifier, and stop on
a refusal rather than working around it.

That flow cannot terminate without history. Each turn built a fresh Agent with
no message_history, so the user's "yes" arrived at an agent that had no idea
what it was agreeing to. History is now kept per A2A contextId in a bounded
in-memory map — enough for confirm-then-act, deliberately not durable, and
documented as needing sticky routing beyond one replica.

Verified end to end: pm-user creates S99_TEST1 across two turns and the row
appears in MySQL; ro-user asking for the same thing is told no write tool
exists, because for that token none does. Both test rows removed afterwards.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-28 17:39:51 +02:00

86 lines
3.2 KiB
Python

"""MCP toolsets, rebuilt for every conversation turn.
The rule this enforces is the one thing the identity design depends on: **an MCP
client is built per turn, with tokens exchanged for that turn**, never held open
with a fixed token.
That is not a limitation of any framework — it is what keeps the blast radius of
a leaked token to one turn, and what makes `azp`/`sub` in the audit trail mean
something. PydanticAI takes `headers` at toolset construction, which is exactly
the shape this needs; the same constraint on other runtimes (langchain-mcp-adapters
issues #172/#194/#235, CrewAI's MCPServerAdapter) has the same answer.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
import httpx
from pydantic_ai.mcp import MCPToolset, StreamableHttpTransport
from .config import Settings
from .identity import TokenExchanger
@dataclass(frozen=True)
class McpTarget:
"""One MCP server this agent is allowed to reach."""
name: str
url: str
audience_scope: str
def targets(settings: Settings) -> list[McpTarget]:
"""The catalog domain read/write, plus sales read-only.
sales is here to answer "does this product actually sell?" before a pricing
or discontinuation decision. Read only — this agent never touches orders,
and the scope it requests reflects that.
"""
return [
McpTarget("catalog", settings.mcp_catalog_url, settings.mcp_catalog_scope),
McpTarget("sales", settings.mcp_sales_url, settings.mcp_sales_scope),
]
async def build_toolsets(
settings: Settings,
exchanger: TokenExchanger,
subject_token: str,
) -> list[MCPToolset]:
"""One toolset per MCP server, each carrying its own audience-scoped token.
Several narrow tokens rather than one broad one, deliberately: a token
minted for `mcp-catalog` is refused by the sales gateway, so a prompt
injection that talks the agent into reaching the wrong domain fails at
Kong rather than succeeding quietly.
"""
def client_factory(**kwargs: Any) -> httpx.AsyncClient:
# fastmcp builds this client itself and would otherwise take httpx's
# 5s default connect timeout. See Settings.http_timeout for why that is
# not survivable against `.local` names, and what the failure looks like
# when it bites: a bare "Failed to initialize server session".
kwargs["verify"] = settings.verify
kwargs["timeout"] = httpx.Timeout(settings.http_timeout, connect=settings.http_timeout)
return httpx.AsyncClient(**kwargs)
toolsets: list[MCPToolset] = []
for target in targets(settings):
token = await exchanger.exchange(subject_token, target.audience_scope)
toolsets.append(
MCPToolset(
StreamableHttpTransport(
url=target.url,
headers={"Authorization": f"Bearer {token}"},
# No `verify=` here: the factory below sets it, and passing
# both makes fastmcp ignore this one with a warning.
httpx_client_factory=client_factory,
),
id=target.name,
init_timeout=settings.http_timeout,
)
)
return toolsets