diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..99b40f3 --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +.venv/ +__pycache__/ +*.pyc +.pytest_cache/ +.env diff --git a/README.md b/README.md new file mode 100644 index 0000000..9839b58 --- /dev/null +++ b/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. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..ef079b6 --- /dev/null +++ b/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"] diff --git a/src/catalogue_agent/__init__.py b/src/catalogue_agent/__init__.py new file mode 100644 index 0000000..fb66f38 --- /dev/null +++ b/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" diff --git a/src/catalogue_agent/__main__.py b/src/catalogue_agent/__main__.py new file mode 100644 index 0000000..7bad93c --- /dev/null +++ b/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() diff --git a/src/catalogue_agent/a2a_server.py b/src/catalogue_agent/a2a_server.py new file mode 100644 index 0000000..6b26608 --- /dev/null +++ b/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, + ) diff --git a/src/catalogue_agent/agent.py b/src/catalogue_agent/agent.py new file mode 100644 index 0000000..d11a952 --- /dev/null +++ b/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 diff --git a/src/catalogue_agent/config.py b/src/catalogue_agent/config.py new file mode 100644 index 0000000..099dc19 --- /dev/null +++ b/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 diff --git a/src/catalogue_agent/identity.py b/src/catalogue_agent/identity.py new file mode 100644 index 0000000..9a9017f --- /dev/null +++ b/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 diff --git a/src/catalogue_agent/mcp_clients.py b/src/catalogue_agent/mcp_clients.py new file mode 100644 index 0000000..d4c8ee3 --- /dev/null +++ b/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 diff --git a/tests/test_a2a_server.py b/tests/test_a2a_server.py new file mode 100644 index 0000000..8c8ec30 --- /dev/null +++ b/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 diff --git a/tests/test_identity.py b/tests/test_identity.py new file mode 100644 index 0000000..ce12208 --- /dev/null +++ b/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")