Stéphane Tailland
ba8892b0ad
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>
81 lines
2.9 KiB
Python
81 lines
2.9 KiB
Python
"""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")
|