Build the Catalogue agent: A2A in, MCP out, inference through Kong

First of the six agents, and the one that gives the A2A facade for the
others to interoperate with. A turn exchanges the user's token once per
audience, builds one MCP toolset per server with those tokens, runs the
model through the Kong AI gateway, and drops the toolsets.

Per-turn MCP clients are the point, not an implementation detail: a token
that outlives its turn can be replayed after the session ended. Several
narrow tokens rather than one broad one, so a prompt injection that talks
the agent into the wrong domain is refused at Kong rather than succeeding
quietly.

a2a-sdk rather than FastA2A, revising the May decision. That decision
rested on Agent.to_a2a() being removed in PydanticAI v2 — which has now
happened, so the argument no longer separates them. With four other
runtimes due to interoperate, the protocol author's implementation is the
safer base. httpx2 rather than httpx, which PydanticAI deprecates for
OpenAI-compatible providers and drops in v3.

Four things about a2a-sdk 1.1 that no documentation gave and that were
established against the running library, all recorded where they bite:
a2a.types are protobuf, not pydantic; DefaultRequestHandler speaks 0.3 and
V2 is needed for 1.0; a client must send A2A-Version: 1.0 or the server
assumes 0.3 and answers VERSION_NOT_SUPPORTED on a perfectly valid
request; and an executor must enqueue a Task before any status update.

Verified: the app builds with no deprecation warnings, serves its Agent
Card, and fails a turn cleanly with "no bearer token" rather than
answering from the model's own knowledge — which would hide a broken
identity chain behind a plausible reply. Seven tests cover the exchange
shape (scope not audience, per-audience caching, refusal) and the card.

The Keycloak client does not exist yet, so no end-to-end run has happened.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

.gitignore

@@ -0,0 +1,5 @@
+.venv/
+__pycache__/
+*.pyc
+.pytest_cache/
+.env

README.md

@@ -0,0 +1,113 @@
+# Catalogue
+
+Product management agent for the Classic Models catalog domain. Serves a product
+manager: what exists, how it is priced, whether it sells, what to discontinue.
+
+Reached over **A2A**; reads data over **MCP** through the Kong gateways; thinks
+through the **Kong AI gateway**. It is one MCP client among others, exactly as
+LibreChat is — nothing under it changes to accommodate it.
+
+## Running it
+
+```bash
+uv venv && uv pip install -e ".[dev]"
+export CATALOGUE_CLIENT_SECRET=...        # Keycloak client secret, no default
+export CATALOGUE_CA_BUNDLE=/path/to/sttlab-local-ca.crt
+python -m catalogue_agent
+```
+
+Everything else has a working default in `config.py` and is overridable with
+`CATALOGUE_`-prefixed variables.
+
+```bash
+curl localhost:8080/.well-known/agent-card.json
+
+curl -X POST localhost:8080/ \
+  -H 'Content-Type: application/json' \
+  -H 'A2A-Version: 1.0' \
+  -H "Authorization: Bearer $USER_TOKEN" \
+  -d '{"jsonrpc":"2.0","id":1,"method":"SendMessage",
+       "params":{"message":{"messageId":"m1","role":"ROLE_USER",
+                            "parts":[{"text":"How many product lines are there?"}]}}}'
+```
+
+**`A2A-Version: 1.0` is required.** Without it the SDK assumes 0.3 and answers
+`VERSION_NOT_SUPPORTED`, which reads as a protocol error even though the request
+is fine.
+
+## How a turn works
+
+```
+A2A SendMessage  ──►  exchange the user's token, once per audience
+                 ──►  build one MCP toolset per server, with that turn's tokens
+                 ──►  PydanticAI runs the model through the Kong AI gateway
+                 ──►  answer, toolsets dropped
+```
+
+The MCP client is built **per turn**. A token that outlives the turn it was
+minted for is a token that can be replayed after the user's session ended, and
+holding one long-lived connection would mean one long-lived token.
+
+Several narrow tokens rather than one broad one: 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.
+
+## What it can and cannot do
+
+| | |
+|---|---|
+| Reads | catalog (products, product lines), sales (read-only, to judge whether a product sells) |
+| Writes | **none yet** — the agent says so when asked |
+| Calls | nothing yet. `Approvisionnement.check_availability` comes with discontinuation decisions |
+| Exposes | `describe_catalog` |
+
+Skills are decision-shaped on purpose. An agent exposing a generic `query` has
+become a data proxy and defeats the per-domain scoping the estate is built on,
+so skills get added when another agent has a specific question — not in advance.
+
+## Two Keycloak facts this depends on
+
+Both measured against Keycloak 26.6, both costly to rediscover:
+
+**`actor_token` is silently ignored.** Standard Token Exchange accepts it,
+returns a valid token and emits no `act` claim — no error. Delegation cannot be
+built on RFC 8693 `act`; the actor is identified by `azp`, which carries this
+agent's client id while `sub` stays the human.
+
+**The audience is requested as a scope.** `audience=mcp-catalog` returns
+"Requested audience not available". `scope=openid mcp-catalog-audience` works.
+
+## Library choices
+
+Pinned to what was current on 2026-08-28, ranges below the next major so patch
+fixes arrive but an upgrade stays deliberate.
+
+**`a2a-sdk` rather than FastA2A.** The earlier design said FastA2A, on the
+grounds that `Agent.to_a2a()` was going away in PydanticAI v2 — which has since
+happened, so that argument no longer distinguishes them. This estate exists to
+prove A2A interoperability across five runtimes, and the reference
+implementation is the safer thing for four other stacks to interoperate with.
+
+**`httpx2` rather than `httpx`.** PydanticAI deprecates httpx clients for
+OpenAI-compatible providers and removes them in v3.
+
+Notes on `a2a-sdk` 1.1 that are not guessable, all established against the
+running library:
+
+- `a2a.types` are protobuf messages, not pydantic models.
+- Use `DefaultRequestHandlerV2`; `DefaultRequestHandler` speaks 0.3.
+- JSON-RPC methods are `SendMessage`, `GetTask`… The 0.3 spelling
+  (`message/send`) still works — `enable_v0_3_compat` is on.
+- An executor must enqueue a `Task` before any status update when
+  `context.current_task` is None.
+
+## Not done yet
+
+- **The Keycloak client does not exist.** `classic-models-agent-catalogue` has
+  to be declared in `infra-keycloak/k8s/keycloak/configmap-realm.yaml`, with
+  `mcp-catalog-audience` and `mcp-sales-audience` as optional scopes. Until then
+  the exchange fails and the agent says so.
+- Writes, and the A2A call to Approvisionnement.
+- Deployment manifests — they belong in `classic-models-agent-platform` once a
+  second agent shows what is actually shared.

pyproject.toml

@@ -0,0 +1,36 @@
+[project]
+name = "classic-models-agent-catalogue"
+version = "0.1.0"
+description = "Product management agent for the Classic Models catalog domain"
+requires-python = ">=3.12"
+
+# Pinned to what was current on 2026-08-28 and resolved together in one
+# environment. Ranges rather than exact pins so patch-level security fixes are
+# picked up, capped below the next major so an upgrade is a deliberate act.
+dependencies = [
+    # The A2A facade. The official SDK rather than FastA2A: this estate exists
+    # to prove A2A interoperability across five runtimes, so the reference
+    # implementation is the safer base for the others to interoperate with.
+    "a2a-sdk[http-server]>=1.1.2,<2",
+    # Reasoning and tool calling. -slim keeps the dependency surface to the
+    # providers actually used; [openai] because the Kong AI gateway speaks the
+    # OpenAI protocol, [mcp] for the MCP client.
+    "pydantic-ai-slim[openai,mcp]>=2.35.3,<3",
+    "httpx2>=2.12.0,<3",
+    "pydantic-settings>=2.7",
+    "uvicorn>=0.52.4",
+]
+
+[project.optional-dependencies]
+dev = ["pytest>=8", "pytest-asyncio>=0.24"]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/catalogue_agent"]
+
+[tool.pytest.ini_options]
+asyncio_mode = "auto"
+testpaths = ["tests"]

src/catalogue_agent/__init__.py

@@ -0,0 +1,4 @@
+"""Catalogue — product management agent for the Classic Models catalog domain."""
+
+__all__ = ["__version__"]
+__version__ = "0.1.0"

src/catalogue_agent/__main__.py

@@ -0,0 +1,20 @@
+"""Entry point: `python -m catalogue_agent`."""
+
+from __future__ import annotations
+
+import logging
+
+import uvicorn
+
+from .a2a_server import build_app
+from .config import Settings
+
+
+def main() -> None:
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s")
+    settings = Settings()  # type: ignore[call-arg]  # client_secret comes from the environment
+    uvicorn.run(build_app(settings), host=settings.host, port=settings.port)
+
+
+if __name__ == "__main__":
+    main()

src/catalogue_agent/a2a_server.py

@@ -0,0 +1,178 @@
+"""The A2A facade: how other agents reach this one.
+
+Notes on a2a-sdk 1.1, none of which are guessable and all of which were
+established against the running library:
+
+- `a2a.types` are **protobuf** messages, not pydantic models.
+- Use `DefaultRequestHandlerV2`. `DefaultRequestHandler` speaks A2A 0.3 and
+  rejects 1.0 traffic.
+- The JSON-RPC methods are `SendMessage`, `GetTask`, … The 0.3 spelling
+  (`message/send`) is available through `enable_v0_3_compat=True`, kept on here
+  so a client written against the older spelling still works.
+- **A client must send `A2A-Version: 1.0`.** With the header absent the server
+  assumes 0.3 and answers `VERSION_NOT_SUPPORTED` — a confusing failure, since
+  nothing in the request is actually wrong.
+- An executor must enqueue a `Task` before any status update when
+  `context.current_task` is None, or the SDK answers
+  "Agent should enqueue Task before TaskStatusUpdateEvent".
+"""
+
+from __future__ import annotations
+
+import contextlib
+import logging
+from collections.abc import AsyncIterator
+
+import httpx2
+from a2a.server.agent_execution import AgentExecutor, RequestContext
+from a2a.server.events import EventQueue
+from a2a.server.request_handlers import DefaultRequestHandlerV2
+from a2a.server.routes.agent_card_routes import create_agent_card_routes
+from a2a.server.routes.jsonrpc_routes import create_jsonrpc_routes
+from a2a.server.tasks import InMemoryTaskStore, TaskUpdater
+from a2a.types import (
+    AgentCapabilities,
+    AgentCard,
+    AgentInterface,
+    AgentSkill,
+    Part,
+    Task,
+    TaskState,
+    TaskStatus,
+)
+from starlette.applications import Starlette
+
+from .agent import CatalogueAgent
+from .config import Settings
+
+logger = logging.getLogger(__name__)
+
+# The bearer the caller presents, which this agent exchanges for MCP tokens.
+# It is the END USER's token, not the calling agent's: the whole scoping model
+# rests on `sub` staying the human throughout the chain while `azp` names the
+# agent doing the asking.
+SUBJECT_TOKEN_HEADER = "authorization"
+
+
+def build_agent_card(settings: Settings) -> AgentCard:
+    """Skills are decision-shaped, never a generic `query`.
+
+    Only one skill for now, and it is read-only. An agent that exposes
+    "ask me anything about my domain" has become a data proxy and defeats the
+    scoping the estate is built on — so skills are added when another agent has
+    a specific question, not in advance.
+    """
+    return AgentCard(
+        name="catalogue",
+        description=(
+            "Product management for the Classic Models catalog: products, product "
+            "lines, pricing and discontinuation decisions."
+        ),
+        version="0.1.0",
+        default_input_modes=["text/plain"],
+        default_output_modes=["text/plain"],
+        capabilities=AgentCapabilities(streaming=False),
+        supported_interfaces=[
+            AgentInterface(
+                url=settings.public_url,
+                protocol_binding="JSONRPC",
+                protocol_version="1.0",
+            )
+        ],
+        skills=[
+            AgentSkill(
+                id="describe_catalog",
+                name="Describe the catalog",
+                description=(
+                    "Answer questions about products and product lines: what exists, "
+                    "how it is priced, how it is positioned, and whether it sells."
+                ),
+                tags=["catalog", "product", "pricing"],
+                examples=[
+                    "Which product lines have fewer than five references?",
+                    "What is the margin on the 1969 Harley Davidson Ultimate Chopper?",
+                ],
+            )
+        ],
+    )
+
+
+class CatalogueExecutor(AgentExecutor):
+    def __init__(self, agent: CatalogueAgent) -> None:
+        self._agent = agent
+
+    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
+        updater = TaskUpdater(event_queue, context.task_id, context.context_id)
+
+        # The SDK requires the Task to exist before any status update.
+        if context.current_task is None:
+            await event_queue.enqueue_event(
+                Task(
+                    id=context.task_id,
+                    context_id=context.context_id,
+                    status=TaskStatus(state=TaskState.TASK_STATE_SUBMITTED),
+                )
+            )
+        await updater.start_work()
+
+        subject_token = _subject_token(context)
+        if subject_token is None:
+            # Refused rather than answered from the model's own knowledge: this
+            # agent has nothing useful to say without reaching the catalog, and
+            # answering anyway would hide a broken identity chain behind a
+            # plausible reply.
+            await updater.failed(
+                message=updater.new_agent_message(
+                    parts=[Part(text="No bearer token on the request; cannot reach the catalog.")]
+                )
+            )
+            return
+
+        try:
+            answer = await self._agent.run(context.get_user_input(), subject_token)
+        except Exception as exc:  # noqa: BLE001 — surfaced to the caller, logged here
+            logger.exception("turn failed")
+            await updater.failed(
+                message=updater.new_agent_message(parts=[Part(text=f"Turn failed: {exc}")])
+            )
+            return
+
+        await updater.complete(message=updater.new_agent_message(parts=[Part(text=answer)]))
+
+    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
+        await TaskUpdater(event_queue, context.task_id, context.context_id).cancel()
+
+
+def _subject_token(context: RequestContext) -> str | None:
+    headers = {}
+    call = context.call_context
+    if call is not None:
+        headers = {k.lower(): v for k, v in (call.state.get("headers") or {}).items()}
+    raw = headers.get(SUBJECT_TOKEN_HEADER)
+    if not raw:
+        return None
+    return raw[7:] if raw.lower().startswith("bearer ") else raw
+
+
+def build_app(settings: Settings) -> Starlette:
+    # httpx2, not httpx: PydanticAI deprecates httpx clients for
+    # OpenAI-compatible providers and drops them in v3.
+    http_client = httpx2.AsyncClient(verify=settings.verify, timeout=60.0)
+    card = build_agent_card(settings)
+    handler = DefaultRequestHandlerV2(
+        agent_executor=CatalogueExecutor(CatalogueAgent(settings, http_client)),
+        task_store=InMemoryTaskStore(),
+        agent_card=card,
+    )
+
+    # lifespan rather than on_shutdown: Starlette 1.x removed the latter.
+    @contextlib.asynccontextmanager
+    async def lifespan(_app: Starlette) -> AsyncIterator[None]:
+        yield
+        await http_client.aclose()
+
+    return Starlette(
+        routes=create_agent_card_routes(card)
+        + create_jsonrpc_routes(handler, "/", enable_v0_3_compat=True),
+        lifespan=lifespan,
+    )

src/catalogue_agent/agent.py

@@ -0,0 +1,76 @@
+"""The agent itself: reasoning, tools, and the system prompt that bounds it."""
+
+from __future__ import annotations
+
+import httpx2
+from pydantic_ai import Agent
+from pydantic_ai.models.openai import OpenAIChatModel
+from pydantic_ai.providers.openai import OpenAIProvider
+
+from .config import Settings
+from .identity import TokenExchanger
+from .mcp_clients import build_toolsets
+
+SYSTEM_PROMPT = """\
+You are Catalogue, the product management agent for the Classic Models catalog.
+
+You serve a product manager: you answer questions about products and product
+lines, and you help decide on pricing, positioning and discontinuation.
+
+How to work:
+- Use the catalog tools to read product and product line data.
+- Use the sales tools only to judge whether a product actually sells, when that
+  bears on a pricing or discontinuation decision. You never act on orders.
+- State figures as the tools returned them. If a tool returns nothing, say so
+  rather than estimating.
+- When a decision depends on stock or open orders, say that it does and what is
+  missing. Do not infer availability from what you can see.
+
+You currently have read access only. If asked to change anything, say plainly
+that writes are not enabled yet.
+"""
+
+
+def build_model(settings: Settings, http_client: httpx2.AsyncClient) -> OpenAIChatModel:
+    """Inference through the Kong AI gateway, never a provider directly.
+
+    The gateway speaks the OpenAI protocol, hence this provider. Going straight
+    to OpenRouter would bypass the token rate limits, the prompt guards and the
+    audit trail that records what was sent to the model — all of which are the
+    point of routing inference through Kong.
+    """
+    return OpenAIChatModel(
+        settings.llm_model,
+        provider=OpenAIProvider(
+            base_url=settings.llm_base_url,
+            api_key=settings.llm_api_key,
+            http_client=http_client,
+        ),
+    )
+
+
+class CatalogueAgent:
+    """Holds what is stable across turns; rebuilds what must not be."""
+
+    def __init__(self, settings: Settings, http_client: httpx2.AsyncClient) -> None:
+        self._settings = settings
+        self._exchanger = TokenExchanger(
+            http_client,
+            token_endpoint=settings.token_endpoint,
+            client_id=settings.client_id,
+            client_secret=settings.client_secret,
+        )
+        self._model = build_model(settings, http_client)
+
+    async def run(self, prompt: str, subject_token: str) -> str:
+        """One turn: exchange tokens, build the toolsets, answer, drop them.
+
+        The Agent is constructed here rather than once in __init__ because its
+        toolsets carry this turn's tokens. Construction is cheap; a long-lived
+        agent holding a long-lived token is not.
+        """
+        toolsets = await build_toolsets(self._settings, self._exchanger, subject_token)
+        agent = Agent(self._model, instructions=SYSTEM_PROMPT, toolsets=toolsets)
+        async with agent:
+            result = await agent.run(prompt)
+        return result.output

src/catalogue_agent/config.py

@@ -0,0 +1,55 @@
+"""Runtime configuration, entirely from the environment.
+
+Nothing here has a value that would be wrong to commit, so the defaults are the
+lab's real endpoints. The two secrets — the agent's Keycloak client secret and
+nothing else — have no default and fail loudly if unset.
+"""
+
+from __future__ import annotations
+
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_prefix="CATALOGUE_", env_file=".env")
+
+    # --- identity -----------------------------------------------------------
+    keycloak_realm_url: str = "https://keycloak.sttlab.local/realms/classic-models"
+    client_id: str = "classic-models-agent-catalogue"
+    client_secret: str = Field(..., description="Keycloak client secret; no default on purpose")
+
+    # --- data ---------------------------------------------------------------
+    # One entry per MCP server this agent may reach. The audience is requested
+    # as a SCOPE, not through the `audience` parameter — Keycloak answers
+    # "Requested audience not available" to the latter. See identity.py.
+    mcp_catalog_url: str = "https://catalog-dev.sttlab.local/catalog-mcp"
+    mcp_catalog_scope: str = "mcp-catalog-audience"
+    mcp_sales_url: str = "https://sales-dev.sttlab.local/sales-mcp"
+    mcp_sales_scope: str = "mcp-sales-audience"
+
+    # --- inference ----------------------------------------------------------
+    # Always the Kong AI gateway, never a provider directly: it carries the rate
+    # limits, the prompt guards and the audit trail. It speaks the OpenAI
+    # protocol, which is why the OpenAI provider is used against it.
+    llm_base_url: str = "https://llm-dev.sttlab.local/llm/v1"
+    llm_model: str = "google/gemini-3.7-flash"
+    llm_api_key: str = Field("unused", description="Kong authenticates the caller, not this key")
+
+    # --- transport ----------------------------------------------------------
+    # The lab uses a private CA (sttlab-local-ca). Point this at the CA bundle
+    # rather than disabling verification.
+    ca_bundle: str | None = None
+    host: str = "0.0.0.0"
+    port: int = 8080
+    # Advertised in the Agent Card, so it must be the URL other agents reach,
+    # not the bind address.
+    public_url: str = "http://localhost:8080/"
+
+    @property
+    def token_endpoint(self) -> str:
+        return f"{self.keycloak_realm_url}/protocol/openid-connect/token"
+
+    @property
+    def verify(self) -> str | bool:
+        return self.ca_bundle if self.ca_bundle else True

src/catalogue_agent/identity.py

@@ -0,0 +1,88 @@
+"""Token exchange: the end user's token in, one narrowly-scoped token per MCP
+server out.
+
+Two measured facts about Keycloak 26.6 shape this, both of which cost time to
+discover and neither of which is obvious from the RFC:
+
+**`actor_token` is silently ignored.** Standard Token Exchange accepts the
+parameter, returns a valid token, and emits no `act` claim — no error, no
+warning. Delegation therefore cannot be built on RFC 8693 `act`. The actor is
+identified by `azp`, which carries this agent's client id.
+
+**The audience is requested as a scope.** Passing `audience=mcp-catalog` returns
+"Requested audience not available". Passing `scope=openid mcp-catalog-audience`
+works, and the resulting token carries `aud: [mcp-catalog]`.
+"""
+
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass
+
+import httpx2
+
+GRANT_TYPE = "urn:ietf:params:oauth:grant-type:token-exchange"
+SUBJECT_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:access_token"
+
+
+class TokenExchangeError(RuntimeError):
+    """Keycloak refused the exchange. Carries its response, which names the cause."""
+
+
+@dataclass(frozen=True)
+class ExchangedToken:
+    access_token: str
+    expires_at: float
+
+    @property
+    def expired(self) -> bool:
+        # A conversation turn lasts seconds and a token five minutes, so this
+        # margin only matters for a turn that stalls on a slow model.
+        return time.monotonic() >= self.expires_at - 30
+
+
+class TokenExchanger:
+    """Exchanges a user token for audience-scoped tokens.
+
+    Caches per (subject token, scope) for the lifetime of one turn. Deliberately
+    not a long-lived cache keyed by user: a token that outlives the turn it was
+    minted for is a token that can be replayed after the user's session ended.
+    """
+
+    def __init__(self, client: httpx2.AsyncClient, *, token_endpoint: str, client_id: str, client_secret: str) -> None:
+        self._client = client
+        self._endpoint = token_endpoint
+        self._client_id = client_id
+        self._client_secret = client_secret
+        self._cache: dict[tuple[str, str], ExchangedToken] = {}
+
+    async def exchange(self, subject_token: str, audience_scope: str) -> str:
+        key = (subject_token, audience_scope)
+        cached = self._cache.get(key)
+        if cached and not cached.expired:
+            return cached.access_token
+
+        response = await self._client.post(
+            self._endpoint,
+            data={
+                "grant_type": GRANT_TYPE,
+                "client_id": self._client_id,
+                "client_secret": self._client_secret,
+                "subject_token": subject_token,
+                "subject_token_type": SUBJECT_TOKEN_TYPE,
+                # Not `audience=...`. See the module docstring.
+                "scope": f"openid {audience_scope}",
+            },
+        )
+        if response.status_code != 200:
+            raise TokenExchangeError(
+                f"exchange for {audience_scope!r} failed: HTTP {response.status_code} {response.text[:200]}"
+            )
+
+        payload = response.json()
+        token = ExchangedToken(
+            access_token=payload["access_token"],
+            expires_at=time.monotonic() + float(payload.get("expires_in", 60)),
+        )
+        self._cache[key] = token
+        return token.access_token

src/catalogue_agent/mcp_clients.py

@@ -0,0 +1,71 @@
+"""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 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.
+    """
+    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}"},
+                    verify=settings.verify,
+                ),
+                id=target.name,
+            )
+        )
+    return toolsets

tests/test_a2a_server.py

@@ -0,0 +1,56 @@
+"""The A2A surface: the Agent Card, and how the subject token is read."""
+
+from __future__ import annotations
+
+from catalogue_agent.a2a_server import _subject_token, build_agent_card
+from catalogue_agent.config import Settings
+
+
+class _Call:
+    def __init__(self, headers: dict[str, str]) -> None:
+        self.state = {"headers": headers}
+
+
+class _Context:
+    def __init__(self, headers: dict[str, str] | None) -> None:
+        self.call_context = _Call(headers) if headers is not None else None
+
+
+def _settings() -> Settings:
+    return Settings(client_secret="s3cret")  # type: ignore[call-arg]
+
+
+def test_agent_card_exposes_a_decision_shaped_skill() -> None:
+    """Never a generic `query`.
+
+    An agent that offers "ask me anything about my domain" has become a data
+    proxy, and the per-domain scoping the estate relies on stops meaning
+    anything.
+    """
+    card = build_agent_card(_settings())
+    ids = [s.id for s in card.skills]
+    assert ids == ["describe_catalog"]
+    assert not any(s.id in {"query", "search", "execute"} for s in card.skills)
+
+
+def test_agent_card_declares_protocol_1_0() -> None:
+    """Without this the SDK assumes 0.3 and refuses 1.0 traffic."""
+    iface = build_agent_card(_settings()).supported_interfaces[0]
+    assert iface.protocol_version == "1.0"
+    assert iface.protocol_binding == "JSONRPC"
+
+
+def test_bearer_prefix_is_stripped_case_insensitively() -> None:
+    assert _subject_token(_Context({"authorization": "Bearer abc"})) == "abc"
+    assert _subject_token(_Context({"Authorization": "bearer abc"})) == "abc"
+    assert _subject_token(_Context({"authorization": "abc"})) == "abc"
+
+
+def test_absent_token_yields_none_rather_than_an_empty_string() -> None:
+    """The executor branches on None to fail the task loudly.
+
+    An empty string would be sent to Keycloak and produce a confusing 400
+    instead of a clear "no token on the request".
+    """
+    assert _subject_token(_Context({})) is None
+    assert _subject_token(_Context(None)) is None

tests/test_identity.py

@@ -0,0 +1,80 @@
+"""The token exchange, which is where this agent's security actually lives."""
+
+from __future__ import annotations
+
+import pytest
+
+from catalogue_agent.identity import TokenExchanger, TokenExchangeError
+
+
+class _Response:
+    def __init__(self, status_code: int, payload: dict | None = None, text: str = "") -> None:
+        self.status_code = status_code
+        self._payload = payload or {}
+        self.text = text
+
+    def json(self) -> dict:
+        return self._payload
+
+
+class _Client:
+    """Records what was posted, so the assertions are about the request shape."""
+
+    def __init__(self, response: _Response) -> None:
+        self.response = response
+        self.calls: list[dict] = []
+
+    async def post(self, url: str, data: dict) -> _Response:
+        self.calls.append({"url": url, "data": data})
+        return self.response
+
+
+def _exchanger(client: _Client) -> TokenExchanger:
+    return TokenExchanger(
+        client,  # type: ignore[arg-type]
+        token_endpoint="https://kc/realms/r/protocol/openid-connect/token",
+        client_id="classic-models-agent-catalogue",
+        client_secret="s3cret",
+    )
+
+
+async def test_audience_is_requested_as_a_scope() -> None:
+    """Keycloak answers 'Requested audience not available' to `audience=...`.
+
+    The audience has to be asked for as a scope instead. This is the single
+    detail that decides whether the exchange works at all, so it is asserted
+    rather than left to a comment.
+    """
+    client = _Client(_Response(200, {"access_token": "t", "expires_in": 300}))
+    await _exchanger(client).exchange("user-token", "mcp-catalog-audience")
+
+    data = client.calls[0]["data"]
+    assert data["scope"] == "openid mcp-catalog-audience"
+    assert "audience" not in data
+    assert data["grant_type"] == "urn:ietf:params:oauth:grant-type:token-exchange"
+    assert data["subject_token"] == "user-token"
+
+
+async def test_tokens_are_cached_per_audience_not_globally() -> None:
+    """Two audiences must produce two exchanges.
+
+    Reusing one token across audiences would hand the agent a broader
+    credential than any single MCP server needs, which is the thing the
+    per-audience design exists to prevent.
+    """
+    client = _Client(_Response(200, {"access_token": "t", "expires_in": 300}))
+    exchanger = _exchanger(client)
+
+    await exchanger.exchange("user-token", "mcp-catalog-audience")
+    await exchanger.exchange("user-token", "mcp-catalog-audience")
+    await exchanger.exchange("user-token", "mcp-sales-audience")
+
+    scopes = [c["data"]["scope"] for c in client.calls]
+    assert scopes == ["openid mcp-catalog-audience", "openid mcp-sales-audience"]
+
+
+async def test_refusal_is_raised_with_keycloak_s_own_message() -> None:
+    """A failed exchange must not degrade into an unauthenticated call."""
+    client = _Client(_Response(400, text='{"error":"invalid_client"}'))
+    with pytest.raises(TokenExchangeError, match="invalid_client"):
+        await _exchanger(client).exchange("user-token", "mcp-catalog-audience")