diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..2f6fbff --- /dev/null +++ b/.dockerignore @@ -0,0 +1,8 @@ +.venv/ +.git/ +.pytest_cache/ +__pycache__/ +*.pyc +tests/ +k8s/ +dev/ diff --git a/CLAUDE.md b/CLAUDE.md index 5fdc90c..af241cf 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -33,6 +33,16 @@ rebuilding it loses in-memory checkpoints (langchain-mcp-adapters #172, #194, #235). This agent has no MCP servers, so it is not affected — but do not copy an MCP pattern from here into a specialist. +**"No scopes at all" was too absolute.** The agent has to reason, so it holds +`llm-proxy-audience` as well as one audience per specialist. Inference is not a +domain. What the rule means, and what the realm enforces, is that no *domain* +audience is reachable from here. + +`Answer.needs_the_user` is the other thing not to lose: a specialist that parks +a decision must have its message relayed unchanged. Paraphrasing a proposed +write hides what is about to happen, and approving it here would move the +decision to the component with the weakest standing to make it. + ## Before writing code Read `classic-models-agent-platform/CLAUDE.md` first. It holds the identity @@ -41,10 +51,13 @@ blockers. This file only covers what is specific to this agent. ## Non-negotiables -- **Inference goes through the Kong AI gateway**, `https://llm-dev.sttlab.local/llm/v1/chat/completions`. - Calling a provider directly bypasses the rate limits and prompt guards. -- **One MCP client per conversation turn**, built with freshly exchanged tokens. - Do not hold a long-lived MCP connection with a fixed token. +- **Inference goes through the estate's LLM endpoint**, `https://llm-dev.sttlab.local/llm/v1/chat/completions`. + Calling a model provider directly bypasses the rate limits and prompt guards. +- **One client per conversation turn**, built with freshly exchanged tokens. + Here that is the A2A client and the graph itself: both carry the user's token. +- **Never name the middleware.** The agent talks to URLs that honour a contract. + Write "the MCP server", "the LLM endpoint", "the gateway" — never a product + name, in code, comments, commits or docs. - **A2A calls return answers, not data.** Expose narrow decision-shaped skills, never a generic `query`. - Deliverables in English. diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..46315c5 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,42 @@ +# Two stages so the runtime image carries no build toolchain and no uv. +FROM python:3.12-slim AS build + +COPY --from=ghcr.io/astral-sh/uv:0.9.7 /uv /bin/uv + +ENV UV_COMPILE_BYTECODE=1 \ + UV_LINK_MODE=copy \ + UV_PYTHON_DOWNLOADS=never + +WORKDIR /app + +COPY pyproject.toml README.md ./ +COPY src ./src + +# One step, no lockfile: the version floors in pyproject.toml are what pins this, +# and they are deliberately recent (August 2026) to keep the vulnerability +# surface small. Add a uv.lock when the build needs to be byte-reproducible. +RUN --mount=type=cache,target=/root/.cache/uv \ + uv venv /app/.venv && uv pip install --python /app/.venv/bin/python . + +FROM python:3.12-slim AS runtime + +# Non-root, no shell login, no home to write to. The agent holds a Keycloak +# client secret in its environment; a container that cannot be written to is one +# less way to read it back out. +RUN useradd --system --uid 10001 --no-create-home --shell /usr/sbin/nologin agent + +WORKDIR /app +COPY --from=build /app/.venv /app/.venv +ENV PATH="/app/.venv/bin:$PATH" \ + PYTHONUNBUFFERED=1 \ + PYTHONDONTWRITEBYTECODE=1 + +USER 10001 +EXPOSE 8080 + +# The Agent Card is the natural health endpoint: it is served by the same app, +# needs no auth, and a 200 means the A2A routes are actually mounted. +HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \ + CMD ["python", "-c", "import urllib.request,sys; sys.exit(0 if urllib.request.urlopen('http://127.0.0.1:8080/.well-known/agent-card.json', timeout=4).status==200 else 1)"] + +ENTRYPOINT ["python", "-m", "assistant_agent"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..9ed6b22 --- /dev/null +++ b/README.md @@ -0,0 +1,108 @@ +# Assistant — the orchestrating agent + +The single entry point for users. Understands a request, consults the +specialists that can settle it, and composes one answer. + +**Framework:** LangGraph · **Mode:** conversational · **Client id:** +`classic-models-agent-assistant` + +## It holds nothing, and that is enforced by the realm + +This is the component that sees raw user text, so it is the prompt-injection +target. It has no MCP server, no domain scope, and no data of its own. That is +not a convention in the prompt — it is what the realm grants: + +``` +OBTAINED classic-models-agent-catalogue-audience aud=[classic-models-agent-catalogue] +refused mcp-catalog-audience → Invalid scopes +refused mcp-sales-audience → Invalid scopes +refused catalog-api-audience → Invalid scopes +``` + +An injection that succeeds against this agent finds no path to a domain, +because there is none to find. Its two optional scopes are one specialist and +the inference endpoint — inference is not a domain, and the agent has to reason. + +Adding a specialist is an entry in `specialists.py` plus an audience scope +granted in the realm. Add the entry without the grant and every call to it fails +at the token exchange with `Invalid scopes`, loudly, before any data moves. + +## The chain + +Four hops, `sub` unchanged throughout — which is what makes the end user's own +roles apply at the far end: + +``` +1. user sub=pm-user azp=librechat aud=[…-agent-assistant] +2. assistant→cat sub=pm-user azp=classic-models-agent-assistant aud=[…-agent-catalogue] +3. cat→mcp sub=pm-user azp=classic-models-agent-catalogue aud=[mcp-catalog, token-exchange-broker] +4. MCP call → 200 +``` + +`azp` names the last agent in the chain, not the first. Keycloak accepts +`actor_token` and emits no `act` claim, so an orchestrator's identity does not +survive a second exchange — the delegation is visible in the traces, not in the +token. + +## What it does with an answer + +A specialist returns an answer, never data. The orchestrator composes; it does +not aggregate records it was not given. + +Two behaviours worth naming, both verified end to end: + +- **Out of remit → it says so.** Asked for a company-wide revenue figure, it + reports that no specialist can settle it rather than answering from what the + model knows. An invented answer is indistinguishable from one the estate gave. +- **A held write is relayed verbatim.** Catalogue parks a proposed write in + `input-required`; this agent passes that message through unchanged and stops. + The user's `approve` comes back through the same conversation and reaches the + specialist, which owns the decision. The orchestrator never approves on + anyone's behalf. + +## The graph + +`plan → delegate → plan → …` until the model stops calling tools. One tool per +specialist, described by what it *decides*. Delegations in the same step run +concurrently — that is the point of fanning out at all — and are capped by +`ASSISTANT_MAX_DELEGATIONS` (8), counted in calls rather than model turns, +because one specialist consulted twice for the same question is the loop worth +catching. + +The graph is rebuilt per turn: its tools carry the user's token, and a graph +held across turns would be a token held across turns. + +## Running it + +``` +uv venv && uv pip install -e '.[dev]' +ASSISTANT_CLIENT_SECRET=… ASSISTANT_CA_BUNDLE=… \ + ASSISTANT_URL_CATALOGUE=http://localhost:8099/ \ + python -m assistant_agent +``` + +`ASSISTANT_URL_` overrides a specialist's address, for running outside the +cluster against a port-forward. In the cluster nothing sets it. + +macOS resolves every `.local` name through mDNS before falling back to +`/etc/hosts`, at a flat ~5s per host. `ASSISTANT_HTTP_TIMEOUT` defaults to 60s +for that reason; in the cluster the names resolve instantly. + +## Deployment + +``` +gitea.sttlab.eu/stt/classic-models-agent-assistant:v0.1.0 +``` + +`k8s/` is what runs, `dev/` is the Flux plumbing. Two manual applies close the +loop once — the `agents` namespace and the `sttlab-ca` ConfigMap already exist, +created by the catalogue agent's repo, and are deliberately not declared here so +two Kustomizations do not fight over pruning them. + +``` +kubectl apply -f dev/gitrepository.yaml +kubectl apply -f dev/bootstrap-kustomization.yaml +``` + +`replicas: 1` is a constraint: conversation history lives in the process, and a +specialist's approval round trip passes back through here. diff --git a/dev/bootstrap-kustomization.yaml b/dev/bootstrap-kustomization.yaml new file mode 100644 index 0000000..4f25d35 --- /dev/null +++ b/dev/bootstrap-kustomization.yaml @@ -0,0 +1,33 @@ +# Watches ./dev, so the GitRepository and the Kustomization below it reconcile +# from git like any other manifest. It lives in ./dev itself, so it also manages +# itself — the arrangement `flux bootstrap` sets up, which flux-operator does +# not create, and the same one the platform repos use. +# +# It sits in the agents namespace, not flux-system: a sourceRef resolves in the +# Kustomization's own namespace, and this is the agent team's to own. +# +# Two manual applies, once, to close the loop — nothing can watch ./dev until +# the source that serves it exists. The `agents` namespace already exists; it is +# created by the catalogue agent's repo. +# +# kubectl apply -f dev/gitrepository.yaml +# kubectl apply -f dev/bootstrap-kustomization.yaml +# +# After that every file in this repo, this one included, reconciles from git. +# +# Sharp edge: with prune enabled, deleting flux-kustomization.yaml from git +# deletes the Kustomization it defines, and that Kustomization's own prune then +# removes the whole deployment. Correct GitOps behaviour, blast radius two +# levels deep. +apiVersion: kustomize.toolkit.fluxcd.io/v1 +kind: Kustomization +metadata: + name: assistant-agent-bootstrap + namespace: agents +spec: + interval: 10m + sourceRef: + kind: GitRepository + name: assistant-agent + path: ./dev + prune: true diff --git a/dev/flux-kustomization.yaml b/dev/flux-kustomization.yaml new file mode 100644 index 0000000..c25eb3e --- /dev/null +++ b/dev/flux-kustomization.yaml @@ -0,0 +1,14 @@ +apiVersion: kustomize.toolkit.fluxcd.io/v1 +kind: Kustomization +metadata: + name: assistant-agent + namespace: agents +spec: + interval: 10m + sourceRef: + kind: GitRepository + name: assistant-agent + path: ./k8s + prune: true + wait: true + timeout: 3m diff --git a/dev/gitrepository.yaml b/dev/gitrepository.yaml new file mode 100644 index 0000000..8393a25 --- /dev/null +++ b/dev/gitrepository.yaml @@ -0,0 +1,18 @@ +# HTTPS with no credentials, unlike the platform repos which use SSH and a +# deploy key from Vault. Deliberate: this repo is already anonymously readable over +# HTTPS on gitea, so a deploy key would guard nothing while adding a Vault entry +# and a gitea key to create by hand before anything deploys. TLS still gives +# integrity and authenticity of the source. +# +# If the gitea repos are ever made private, this becomes an SSH sourceRef with +# an ExternalSecret holding the deploy key — the pattern the platform repos use. +apiVersion: source.toolkit.fluxcd.io/v1 +kind: GitRepository +metadata: + name: assistant-agent + namespace: agents +spec: + interval: 5m + url: https://gitea.sttlab.eu/stt/classic-models-agent-assistant.git + ref: + branch: main diff --git a/k8s/configmap.yaml b/k8s/configmap.yaml new file mode 100644 index 0000000..a77fb41 --- /dev/null +++ b/k8s/configmap.yaml @@ -0,0 +1,27 @@ +# Everything here is an endpoint or a public identifier. The one secret the +# agent holds arrives through assistant-agent-secrets; nothing in this file is +# sensitive, which is the point of keeping the split visible. +apiVersion: v1 +kind: ConfigMap +metadata: + name: assistant-agent-config + namespace: agents +data: + ASSISTANT_KEYCLOAK_REALM_URL: https://keycloak.sttlab.local/realms/classic-models + ASSISTANT_CLIENT_ID: classic-models-agent-assistant + + ASSISTANT_LLM_BASE_URL: https://llm-dev.sttlab.local/llm/v1 + ASSISTANT_LLM_MODEL: google/gemini-3.7-flash + + # Trust the lab CA rather than disabling verification. Mounted from the + # sttlab-ca ConfigMap. + ASSISTANT_CA_BUNDLE: /etc/sttlab-ca/ca.crt + + # What callers dial, so it is the Service DNS name and not the bind address. + # Advertised in the Agent Card; get it wrong and A2A callers are pointed at + # nothing. + ASSISTANT_PUBLIC_URL: http://assistant-agent.agents.svc.cluster.local:8080/ + + # Specialists are reached at their own Service names, compiled into + # specialists.py. ASSISTANT_URL_ overrides one, and exists only for + # running the orchestrator outside the cluster against a port-forward. diff --git a/k8s/deployment.yaml b/k8s/deployment.yaml new file mode 100644 index 0000000..7cc47b2 --- /dev/null +++ b/k8s/deployment.yaml @@ -0,0 +1,74 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: assistant-agent + namespace: agents +spec: + # One replica, and that is a constraint rather than a default. Conversation + # history lives in the process, keyed by A2A contextId — and a specialist's + # approval round trip passes back through here, so a second replica would + # route the user's "approve" to a pod holding none of the exchange. Scaling + # out needs sticky routing on contextId or a shared store first. + replicas: 1 + selector: + matchLabels: + app: assistant-agent + template: + metadata: + labels: + app: assistant-agent + spec: + securityContext: + runAsNonRoot: true + runAsUser: 10001 + seccompProfile: + type: RuntimeDefault + containers: + - name: agent + # gitea's own registry: the image sits beside the source it was built + # from and inherits the repo's public visibility, so the pull needs no + # credentials at all. + image: gitea.sttlab.eu/stt/classic-models-agent-assistant:v0.1.0 + ports: + - name: http + containerPort: 8080 + envFrom: + - configMapRef: + name: assistant-agent-config + - secretRef: + name: assistant-agent-secrets + volumeMounts: + - name: sttlab-ca + mountPath: /etc/sttlab-ca + readOnly: true + # The Agent Card needs no auth and is served by the same app, so a 200 + # means the A2A routes are genuinely mounted — not merely that the + # process is alive. + readinessProbe: + httpGet: + path: /.well-known/agent-card.json + port: http + initialDelaySeconds: 5 + periodSeconds: 10 + livenessProbe: + httpGet: + path: /.well-known/agent-card.json + port: http + initialDelaySeconds: 15 + periodSeconds: 20 + resources: + requests: + cpu: 50m + memory: 192Mi + limits: + cpu: "1" + memory: 512Mi + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: ["ALL"] + volumes: + - name: sttlab-ca + configMap: + name: sttlab-ca diff --git a/k8s/externalsecret.yaml b/k8s/externalsecret.yaml new file mode 100644 index 0000000..87207f1 --- /dev/null +++ b/k8s/externalsecret.yaml @@ -0,0 +1,24 @@ +# The agent's Keycloak client secret, and nothing else. It comes from the same +# Vault entry the realm import reads, so there is exactly one copy of this +# secret in the estate and rotating it in Vault rotates it everywhere. +# +# vault-keycloak rather than a store of the agents' own: this value is a +# Keycloak client secret, owned by the realm, and inventing a second AppRole to +# read the same path would add a credential without adding a boundary. +apiVersion: external-secrets.io/v1 +kind: ExternalSecret +metadata: + name: assistant-agent-secrets + namespace: agents +spec: + refreshInterval: 1h + secretStoreRef: + kind: ClusterSecretStore + name: vault-keycloak + target: + name: assistant-agent-secrets + data: + - secretKey: ASSISTANT_CLIENT_SECRET + remoteRef: + key: keycloak/classic-models-realm + property: KC_SECRET_CLASSIC_MODELS_AGENT_ASSISTANT diff --git a/k8s/kustomization.yaml b/k8s/kustomization.yaml new file mode 100644 index 0000000..df34af9 --- /dev/null +++ b/k8s/kustomization.yaml @@ -0,0 +1,10 @@ +# The `agents` namespace and the sttlab-ca ConfigMap are created by the +# catalogue agent's repo, which got there first. Declaring them here too +# would give two Kustomizations the same objects to prune. +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization +resources: + - externalsecret.yaml + - configmap.yaml + - deployment.yaml + - service.yaml diff --git a/k8s/service.yaml b/k8s/service.yaml new file mode 100644 index 0000000..8086d76 --- /dev/null +++ b/k8s/service.yaml @@ -0,0 +1,12 @@ +apiVersion: v1 +kind: Service +metadata: + name: assistant-agent + namespace: agents +spec: + selector: + app: assistant-agent + ports: + - name: http + port: 8080 + targetPort: http diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..f8b7a4c --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,36 @@ +[project] +name = "assistant-agent" +version = "0.1.0" +description = "Assistant — the orchestrating agent for the classic-models estate" +readme = "README.md" +requires-python = ">=3.12" +dependencies = [ + # The A2A facade, and the client for calling specialists. The official SDK + # rather than a framework's own: 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", + # The graph. The orchestrator is where routing is worth making explicit — + # which specialist is consulted when, and where the answers are reconciled. + "langgraph>=1.2.11,<2", + # The model client. The inference endpoint speaks the OpenAI protocol. + "langchain-openai>=1.6.0,<2", + "langchain-core>=1.6.1,<2", + "httpx>=0.28.1,<1", + "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/assistant_agent"] + +[tool.pytest.ini_options] +pythonpath = ["src"] +asyncio_mode = "auto" diff --git a/src/assistant_agent/__init__.py b/src/assistant_agent/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/assistant_agent/__main__.py b/src/assistant_agent/__main__.py new file mode 100644 index 0000000..cbc625b --- /dev/null +++ b/src/assistant_agent/__main__.py @@ -0,0 +1,20 @@ +"""Entry point: `python -m assistant_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/assistant_agent/__pycache__/__init__.cpython-312.pyc b/src/assistant_agent/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..9e445cf Binary files /dev/null and b/src/assistant_agent/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/assistant_agent/__pycache__/__main__.cpython-312.pyc b/src/assistant_agent/__pycache__/__main__.cpython-312.pyc new file mode 100644 index 0000000..23b5b86 Binary files /dev/null and b/src/assistant_agent/__pycache__/__main__.cpython-312.pyc differ diff --git a/src/assistant_agent/__pycache__/a2a_server.cpython-312.pyc b/src/assistant_agent/__pycache__/a2a_server.cpython-312.pyc new file mode 100644 index 0000000..24af0e5 Binary files /dev/null and b/src/assistant_agent/__pycache__/a2a_server.cpython-312.pyc differ diff --git a/src/assistant_agent/__pycache__/config.cpython-312.pyc b/src/assistant_agent/__pycache__/config.cpython-312.pyc new file mode 100644 index 0000000..f483d9a Binary files /dev/null and b/src/assistant_agent/__pycache__/config.cpython-312.pyc differ diff --git a/src/assistant_agent/__pycache__/delegation.cpython-312.pyc b/src/assistant_agent/__pycache__/delegation.cpython-312.pyc new file mode 100644 index 0000000..44e80db Binary files /dev/null and b/src/assistant_agent/__pycache__/delegation.cpython-312.pyc differ diff --git a/src/assistant_agent/__pycache__/graph.cpython-312.pyc b/src/assistant_agent/__pycache__/graph.cpython-312.pyc new file mode 100644 index 0000000..43d0d83 Binary files /dev/null and b/src/assistant_agent/__pycache__/graph.cpython-312.pyc differ diff --git a/src/assistant_agent/__pycache__/identity.cpython-312.pyc b/src/assistant_agent/__pycache__/identity.cpython-312.pyc new file mode 100644 index 0000000..692e52d Binary files /dev/null and b/src/assistant_agent/__pycache__/identity.cpython-312.pyc differ diff --git a/src/assistant_agent/__pycache__/orchestrator.cpython-312.pyc b/src/assistant_agent/__pycache__/orchestrator.cpython-312.pyc new file mode 100644 index 0000000..37db43f Binary files /dev/null and b/src/assistant_agent/__pycache__/orchestrator.cpython-312.pyc differ diff --git a/src/assistant_agent/__pycache__/specialists.cpython-312.pyc b/src/assistant_agent/__pycache__/specialists.cpython-312.pyc new file mode 100644 index 0000000..208c012 Binary files /dev/null and b/src/assistant_agent/__pycache__/specialists.cpython-312.pyc differ diff --git a/src/assistant_agent/a2a_server.py b/src/assistant_agent/a2a_server.py new file mode 100644 index 0000000..10dc156 --- /dev/null +++ b/src/assistant_agent/a2a_server.py @@ -0,0 +1,184 @@ +"""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 httpx +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 .config import Settings +from .orchestrator import Assistant + +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: + """One skill, and it is deliberately about routing rather than data. + + This agent holds nothing. What it offers is that a caller can ask one + question and have the right specialists consulted — which is a capability, + not a data surface. Naming it `answer_about_the_business` rather than + `query` keeps the distinction visible. + """ + return AgentCard( + name="assistant", + description=( + "The entry point for the Classic Models estate. Understands a request, " + "consults the specialists that can settle it, and composes one answer. " + "Holds no data and no domain scope of its own." + ), + 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="answer_about_the_business", + name="Answer a question about the business", + description=( + "Route a question to the specialists that can settle it and compose " + "their answers. Every figure comes from a specialist, within what the " + "asking user is allowed to see." + ), + tags=["orchestration", "routing"], + examples=[ + "How many products are in the Classic Cars line, and which is dearest?", + "Can we ship order 10248?", + ], + ) + ], + ) + + +class AssistantExecutor(AgentExecutor): + def __init__(self, agent: Assistant) -> 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 holds no data at all, so an answer produced without a + # specialist is necessarily invented — and would be indistinguishable + # from one that came from the estate. + await updater.failed( + message=updater.new_agent_message( + parts=[Part(text="No bearer token on the request; cannot consult any specialist.")] + ) + ) + return + + try: + answer = await self._agent.run( + context.get_user_input(), subject_token, context.context_id + ) + 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 + + # Always `completed`, never `input-required`. A specialist holding a + # write parks its own task and its message is relayed here verbatim — + # the user answers that specialist through this agent's next turn, and + # the orchestrator never owns the decision. + 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: + http_client = httpx.AsyncClient(verify=settings.verify, timeout=settings.http_timeout) + card = build_agent_card(settings) + handler = DefaultRequestHandlerV2( + agent_executor=AssistantExecutor(Assistant(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/assistant_agent/config.py b/src/assistant_agent/config.py new file mode 100644 index 0000000..a44b91f --- /dev/null +++ b/src/assistant_agent/config.py @@ -0,0 +1,57 @@ +"""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 one secret — this agent's Keycloak client secret — has +no default and fails 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="ASSISTANT_", env_file=".env") + + # --- identity ----------------------------------------------------------- + keycloak_realm_url: str = "https://keycloak.sttlab.local/realms/classic-models" + client_id: str = "classic-models-agent-assistant" + client_secret: str = Field(..., description="Keycloak client secret; no default on purpose") + + # --- inference ---------------------------------------------------------- + # This agent has no data of its own. The only thing it may reach besides + # other agents is the inference endpoint, and the realm enforces exactly + # that: `classic-models-agent-assistant` is granted two optional scopes, one + # per specialist plus this one. Every domain audience it could ask for comes + # back `Invalid scopes`. + llm_base_url: str = "https://llm-dev.sttlab.local/llm/v1" + llm_model: str = "google/gemini-3.7-flash" + llm_scope: str = "llm-proxy-audience" + + # --- transport ---------------------------------------------------------- + # The lab uses a private CA. Point this at the bundle rather than disabling + # verification. See the note on `.local` names in the README: macOS resolves + # them through mDNS and takes ~5s per host, which is why the timeout is not + # the usual 5 or 10. + ca_bundle: str | None = None + http_timeout: float = 60.0 + + # A turn that fans out and loops has to stop somewhere. Counted in + # delegations, not in model calls — one specialist consulted twice for the + # same question is the loop worth catching. + max_delegations: int = 8 + + host: str = "0.0.0.0" + port: int = 8080 + # Advertised in the Agent Card, so it must be the URL other callers 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/assistant_agent/delegation.py b/src/assistant_agent/delegation.py new file mode 100644 index 0000000..ef491fc --- /dev/null +++ b/src/assistant_agent/delegation.py @@ -0,0 +1,99 @@ +"""Calling a specialist over A2A, with a token minted for that specialist alone. + +One token per callee, exchanged per turn. Several narrow tokens rather than one +broad one: a token minted for Catalogue is refused by every other agent, so a +prompt injection that talks this orchestrator into consulting the wrong +specialist fails at the realm rather than succeeding quietly. + +The client is built per call for the same reason the Catalogue agent builds its +MCP client per turn — a client is where the header lives, and a long-lived +client is a long-lived token. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass + +import httpx +from a2a import types as a2a_types +from a2a.client import ClientConfig, ClientFactory + +from .config import Settings +from .identity import TokenExchanger +from .specialists import Specialist + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class Answer: + """What a specialist said, and whether it finished saying it.""" + + specialist: str + text: str + state: str + + @property + def needs_the_user(self) -> bool: + """The specialist is holding something for a decision. + + Catalogue parks a proposed write in `input-required` rather than making + it. The orchestrator must relay that verbatim and stop — it has no + standing to approve a write on the user's behalf, and paraphrasing the + proposal would hide what is actually about to happen. + """ + return self.state == "TASK_STATE_INPUT_REQUIRED" + + +def _text_of(message: a2a_types.Message) -> str: + return "".join(part.text for part in message.parts if part.text) + + +class Delegator: + """Sends one question to one specialist and waits for its answer.""" + + def __init__(self, settings: Settings, exchanger: TokenExchanger) -> None: + self._settings = settings + self._exchanger = exchanger + + async def ask( + self, specialist: Specialist, question: str, subject_token: str, context_id: str + ) -> Answer: + token = await self._exchanger.exchange(subject_token, specialist.audience_scope) + + async with httpx.AsyncClient( + headers={"Authorization": f"Bearer {token}"}, + verify=self._settings.verify, + timeout=self._settings.http_timeout, + ) as http: + factory = ClientFactory(ClientConfig(httpx_client=http, streaming=False)) + client = await factory.create_from_url(specialist.url) + try: + request = a2a_types.SendMessageRequest( + message=a2a_types.Message( + message_id=f"{context_id}-{specialist.name}", + # Carried across so the specialist keeps one conversation + # per user conversation rather than one per question — + # which is what lets its own approval flow resume. + context_id=context_id, + role=a2a_types.Role.ROLE_USER, + parts=[a2a_types.Part(text=question)], + ) + ) + text, state = "", "TASK_STATE_UNSPECIFIED" + async for event in client.send_message(request): + if event.HasField("task"): + state = a2a_types.TaskState.Name(event.task.status.state) + if event.task.status.HasField("message"): + text = _text_of(event.task.status.message) + elif event.HasField("message"): + text = _text_of(event.message) + state = "TASK_STATE_COMPLETED" + finally: + await client.close() + + if not text: + text = f"({specialist.name} returned no answer; state {state})" + logger.info("delegated to %s -> %s", specialist.name, state) + return Answer(specialist=specialist.name, text=text, state=state) diff --git a/src/assistant_agent/graph.py b/src/assistant_agent/graph.py new file mode 100644 index 0000000..33e852b --- /dev/null +++ b/src/assistant_agent/graph.py @@ -0,0 +1,184 @@ +"""The orchestration graph: decide who to consult, consult them, compose. + +A graph rather than a loop because the routing is the interesting part here. +Which specialist is asked, whether two are asked at once, and where their +answers are reconciled are the decisions this agent exists to make — and a graph +states them where a chain of calls would bury them. + +The graph is built per turn. Its tools carry the user's token, and a graph held +across turns would be a token held across turns. +""" + +from __future__ import annotations + +import asyncio +import logging +from typing import Annotated, Any, TypedDict + +from langchain_core.messages import AIMessage, AnyMessage, SystemMessage, ToolMessage +from langchain_openai import ChatOpenAI +from langgraph.graph import END, START, StateGraph +from langgraph.graph.message import add_messages + +from .config import Settings +from .delegation import Answer, Delegator +from .specialists import SPECIALISTS, by_name, describe + +logger = logging.getLogger(__name__) + +SYSTEM_PROMPT = """\ +You are the Assistant for the Classic Models estate. You are the single entry +point: you understand what the user wants, consult the specialists that can +settle it, and compose one answer. + +You hold no data yourself. You cannot read the catalog, sales, customers, +payments or workforce — the specialists can, each within what the user is +allowed to see. When something is outside every specialist's remit, say so +rather than answering from memory: an answer you invent is indistinguishable +from one they gave you, and that is the failure mode this estate is built to +avoid. + +How to work: +- Consult a specialist when the answer depends on data. Do not guess a figure, + a name or a price you were not given, even if it seems obvious. +- Ask several specialists in the same step when the question has independent + parts. Do not chain calls that could have been made together. +- Ask a complete question in plain language. A specialist answers questions; it + does not run queries for you. +- Report figures exactly as a specialist returned them, and say which one said + what when it matters. +- If a specialist asks for a decision — a write it is holding, a missing detail + — relay what it said, unchanged, and stop. You do not approve on the user's + behalf. +- If a specialist refuses, report the refusal. It is the user's own permissions + being applied, not an obstacle to work around. +""" + + +class State(TypedDict): + """What flows through the graph. + + `delegations` counts calls rather than model turns: one specialist consulted + twice for the same question is the loop worth catching, and it is invisible + in a message count. + """ + + messages: Annotated[list[AnyMessage], add_messages] + subject_token: str + context_id: str + delegations: int + + +def _tool_schemas() -> list[dict[str, Any]]: + """One tool per specialist, described by what it decides.""" + return [ + { + "type": "function", + "function": { + "name": f"ask_{s.name}", + "description": describe(s), + "parameters": { + "type": "object", + "properties": { + "question": { + "type": "string", + "description": ( + "A complete question in plain language, carrying every detail " + "the specialist needs. It cannot see the conversation." + ), + } + }, + "required": ["question"], + }, + }, + } + for s in SPECIALISTS + ] + + +def build_graph(settings: Settings, model: ChatOpenAI, delegator: Delegator) -> Any: + """Compile the graph for one turn.""" + bound = model.bind_tools(_tool_schemas()) + + async def plan(state: State) -> dict[str, Any]: + """Decide who to consult, or answer.""" + reply = await bound.ainvoke([SystemMessage(SYSTEM_PROMPT), *state["messages"]]) + return {"messages": [reply]} + + async def delegate(state: State) -> dict[str, Any]: + """Consult every specialist the last step named, concurrently. + + Concurrency is the point of fanning out at all: "can we ship order + 10248?" asks two specialists about independent things, and waiting for + one before starting the other buys nothing. + """ + last = state["messages"][-1] + assert isinstance(last, AIMessage) + calls = last.tool_calls + + budget = settings.max_delegations - state["delegations"] + allowed, refused = calls[:budget], calls[budget:] + + async def run(call: dict[str, Any]) -> ToolMessage: + specialist = by_name(call["name"].removeprefix("ask_")) + if specialist is None: + # The model named something that is not a specialist. Answer the + # tool call rather than raising: the run continues and the model + # is told, which is more useful than a stack trace. + return ToolMessage( + content=f"There is no specialist called {call['name']!r}.", + tool_call_id=call["id"], + ) + answer = await delegator.ask( + specialist, + call["args"].get("question", ""), + state["subject_token"], + state["context_id"], + ) + return ToolMessage( + content=_render(answer), tool_call_id=call["id"], name=call["name"] + ) + + results = list(await asyncio.gather(*(run(c) for c in allowed))) + results.extend( + ToolMessage( + content=( + "Not consulted: this turn has reached its limit of " + f"{settings.max_delegations} delegations. Answer with what you have, " + "and say what you could not check." + ), + tool_call_id=call["id"], + name=call["name"], + ) + for call in refused + ) + return {"messages": results, "delegations": state["delegations"] + len(allowed)} + + def next_step(state: State) -> str: + last = state["messages"][-1] + if isinstance(last, AIMessage) and last.tool_calls: + return "delegate" + return END + + graph = StateGraph(State) + graph.add_node("plan", plan) + graph.add_node("delegate", delegate) + graph.add_edge(START, "plan") + graph.add_conditional_edges("plan", next_step, {"delegate": "delegate", END: END}) + graph.add_edge("delegate", "plan") + return graph.compile() + + +def _render(answer: Answer) -> str: + """What the model sees of a specialist's reply. + + A specialist waiting on a decision is labelled as such, because the model's + instruction is to relay it unchanged — and an unlabelled reply reads like a + finished answer. + """ + if answer.needs_the_user: + return ( + f"{answer.specialist} is waiting for the user to decide. Relay this " + f"verbatim and stop:\n\n{answer.text}" + ) + return answer.text diff --git a/src/assistant_agent/identity.py b/src/assistant_agent/identity.py new file mode 100644 index 0000000..7893a2c --- /dev/null +++ b/src/assistant_agent/identity.py @@ -0,0 +1,111 @@ +"""Token exchange: the end user's token in, one narrowly-scoped token per +specialist out. + +Copied from the Catalogue agent rather than shared. `classic-models-agent-platform` +holds documentation and no code today, and two copies of forty lines is cheaper +than a package that exists before anyone knows what belongs in it. Move it there +when a third agent needs it unchanged. + +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 base64 +import json +import time +from dataclasses import dataclass + +import httpx + +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: httpx.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 + + +def caller_of(access_token: str) -> str: + """The username inside a token, for the prompt and nothing else. + + Read without verifying the signature on purpose: this is a label shown to + the model, never a decision. Every decision that depends on this identity is + taken by Keycloak or by the gateway, both of which do verify. + """ + try: + payload = access_token.split(".")[1] + payload += "=" * (-len(payload) % 4) + claims = json.loads(base64.urlsafe_b64decode(payload)) + except Exception: # noqa: BLE001 — a malformed token is the gateway's problem + return "an unidentified caller" + return claims.get("preferred_username") or claims.get("sub") or "an unidentified caller" diff --git a/src/assistant_agent/orchestrator.py b/src/assistant_agent/orchestrator.py new file mode 100644 index 0000000..0ddee4f --- /dev/null +++ b/src/assistant_agent/orchestrator.py @@ -0,0 +1,84 @@ +"""One turn of the Assistant: exchange, build, run, remember. + +Everything token-bearing is built per turn and dropped after it. What survives is +the conversation, keyed by A2A `contextId` — kept in memory and bounded, which is +why the deployment runs a single replica. +""" + +from __future__ import annotations + +from collections import OrderedDict + +import httpx +from langchain_core.messages import AnyMessage, HumanMessage +from langchain_openai import ChatOpenAI + +from .config import Settings +from .delegation import Delegator +from .graph import build_graph +from .identity import TokenExchanger + +MAX_CONVERSATIONS = 200 + + +class Assistant: + """Holds what is stable across turns; rebuilds what must not be.""" + + def __init__(self, settings: Settings, http_client: httpx.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._delegator = Delegator(settings, self._exchanger) + self._history: OrderedDict[str, list[AnyMessage]] = OrderedDict() + + async def run(self, prompt: str, subject_token: str, conversation_id: str) -> str: + llm_token = await self._exchanger.exchange(subject_token, self._settings.llm_scope) + model = ChatOpenAI( + model=self._settings.llm_model, + base_url=self._settings.llm_base_url, + # An exchanged token, not a static key. The endpoint validates it as + # a bearer, so inference is attributable to the same (agent, end + # user) pair as every delegation. + api_key=llm_token, + timeout=self._settings.http_timeout, + # No retries here: the estate's endpoint answers 429 with a + # Retry-After the client library does not honour, and a blind retry + # would burn the budget faster. Handled at the transport in the + # specialists; left visible here rather than hidden behind a default. + max_retries=0, + ) + graph = build_graph(self._settings, model, self._delegator) + + history = self._history.get(conversation_id, []) + state = await graph.ainvoke( + { + "messages": [*history, HumanMessage(prompt)], + "subject_token": subject_token, + "context_id": conversation_id, + "delegations": 0, + } + ) + self._remember(conversation_id, state["messages"]) + + final = state["messages"][-1] + return _text_of(final) + + def _remember(self, conversation_id: str, messages: list[AnyMessage]) -> None: + self._history[conversation_id] = messages + self._history.move_to_end(conversation_id) + while len(self._history) > MAX_CONVERSATIONS: + self._history.popitem(last=False) + + +def _text_of(message: AnyMessage) -> str: + content = message.content + if isinstance(content, str): + return content + # Some providers return content as a list of blocks; keep the text ones. + return "".join( + block.get("text", "") for block in content if isinstance(block, dict) + ) or str(content) diff --git a/src/assistant_agent/specialists.py b/src/assistant_agent/specialists.py new file mode 100644 index 0000000..b92cac5 --- /dev/null +++ b/src/assistant_agent/specialists.py @@ -0,0 +1,82 @@ +"""The specialists this agent may consult, and how to reach one. + +A specialist is described here by what it *decides*, not by what data it holds. +That is the difference the estate is built on: an A2A call returns an answer, so +the orchestrator learns "38 products, the dearest is S10_1949" and never gains +read access to the catalog domain. + +Adding a specialist is three things: an entry below, an audience scope granted to +`classic-models-agent-assistant` in the realm, and nothing else. If a specialist +appears here without the realm grant, every call to it fails at the token +exchange with `Invalid scopes` — loudly, and before any data moves. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass, replace + + +@dataclass(frozen=True) +class Specialist: + """One agent this orchestrator can delegate to.""" + + name: str + url: str + audience_scope: str + serves: str + """What it is for, in the words the model routing on it will read.""" + + decides: tuple[str, ...] + """The questions it can settle. Deliberately narrow. + + These become the tool description. A specialist advertised as "ask it + anything about products" turns the orchestrator into a data proxy by the + back door — the very thing the topology exists to prevent. + """ + + +CATALOGUE = Specialist( + name="catalogue", + url="http://catalogue-agent.agents.svc.cluster.local:8080/", + audience_scope="classic-models-agent-catalogue-audience", + serves="product management: products, product lines, pricing, discontinuation", + decides=( + "how many products a product line holds, and which are the dearest or cheapest", + "a product's details: name, vendor, scale, stock, buy price, MSRP", + "whether a product sells well enough to justify a pricing or discontinuation decision", + "creating a product, which it holds for the user's approval before writing", + ), +) + + +#: Everything the design foresees, minus what is not built. Ventes, Support, +#: Approvisionnement and Recouvrement belong here as they arrive; the graph +#: needs no change for them, only this list and a realm grant. +SPECIALISTS: tuple[Specialist, ...] = (CATALOGUE,) + + +def by_name(name: str) -> Specialist | None: + """The specialist called `name`, with its URL overridable per environment. + + `ASSISTANT_URL_` exists for running the orchestrator outside the + cluster against a port-forwarded specialist. In the cluster nothing sets it + and the Service DNS name below is used as written. + """ + found = next((s for s in SPECIALISTS if s.name == name), None) + if found is None: + return None + override = os.environ.get(f"ASSISTANT_URL_{found.name.upper()}") + return replace(found, url=override) if override else found + + +def describe(specialist: Specialist) -> str: + """The tool description the routing model reads.""" + lines = [f"Consult the {specialist.name} agent. It serves {specialist.serves}."] + lines.append("It can settle:") + lines.extend(f"- {d}" for d in specialist.decides) + lines.append( + "Ask it a complete question in plain language, as you would ask a colleague. " + "It answers; it does not hand back raw records." + ) + return "\n".join(lines) diff --git a/tests/__pycache__/test_identity.cpython-312-pytest-9.1.1.pyc b/tests/__pycache__/test_identity.cpython-312-pytest-9.1.1.pyc new file mode 100644 index 0000000..60fc6da Binary files /dev/null and b/tests/__pycache__/test_identity.cpython-312-pytest-9.1.1.pyc differ diff --git a/tests/__pycache__/test_routing.cpython-312-pytest-9.1.1.pyc b/tests/__pycache__/test_routing.cpython-312-pytest-9.1.1.pyc new file mode 100644 index 0000000..6bf604d Binary files /dev/null and b/tests/__pycache__/test_routing.cpython-312-pytest-9.1.1.pyc differ diff --git a/tests/test_identity.py b/tests/test_identity.py new file mode 100644 index 0000000..50141cf --- /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 assistant_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") diff --git a/tests/test_routing.py b/tests/test_routing.py new file mode 100644 index 0000000..da33712 --- /dev/null +++ b/tests/test_routing.py @@ -0,0 +1,56 @@ +"""What the orchestrator must do without a model in the loop. + +The interesting properties here are structural: which specialists exist, what +they advertise, and what happens to a delegation that is refused or that comes +back waiting on a decision. None of that needs inference to test, and testing it +without inference is what makes it a check rather than an observation. +""" + +from __future__ import annotations + +import pytest + +from assistant_agent.delegation import Answer +from assistant_agent.specialists import SPECIALISTS, by_name, describe + + +class TestRegistry: + def test_every_specialist_is_reachable_by_name(self) -> None: + for specialist in SPECIALISTS: + assert by_name(specialist.name) is specialist or by_name(specialist.name) == specialist + + def test_an_unknown_name_yields_none_rather_than_raising(self) -> None: + # The model can emit a tool call for something that no longer exists. + # That must reach it as an answer, not as a stack trace. + assert by_name("ventes") is None + + def test_a_specialist_url_is_overridable_for_local_runs(self, monkeypatch) -> None: + monkeypatch.setenv("ASSISTANT_URL_CATALOGUE", "http://localhost:8099/") + assert by_name("catalogue").url == "http://localhost:8099/" + + def test_every_specialist_names_a_distinct_audience_scope(self) -> None: + # One token per callee is the whole point: a token minted for one + # specialist must be refused by every other. + scopes = [s.audience_scope for s in SPECIALISTS] + assert len(scopes) == len(set(scopes)) + + @pytest.mark.parametrize("specialist", SPECIALISTS, ids=lambda s: s.name) + def test_a_specialist_advertises_decisions_not_a_query(self, specialist) -> None: + # An agent advertised as "ask it anything" has become a data proxy and + # defeats the per-domain scoping the estate is built on. + text = describe(specialist).casefold() + assert "it answers; it does not hand back raw records" in text + assert specialist.decides, "a specialist with no stated decisions is a query endpoint" + + +class TestAnswer: + def test_input_required_is_recognised_as_needing_the_user(self) -> None: + answer = Answer("catalogue", "…", "TASK_STATE_INPUT_REQUIRED") + assert answer.needs_the_user is True + + @pytest.mark.parametrize( + "state", + ["TASK_STATE_COMPLETED", "TASK_STATE_FAILED", "TASK_STATE_CANCELLED"], + ) + def test_every_other_state_is_the_orchestrator_s_to_compose(self, state: str) -> None: + assert Answer("catalogue", "…", state).needs_the_user is False