diff --git a/README.md b/README.md index 306c020..422d085 100644 --- a/README.md +++ b/README.md @@ -59,7 +59,7 @@ succeeding quietly. |---|---| | Reads | catalog (products, product lines), sales (read-only, to judge whether a product sells) | | Writes | catalog: `create-product`, and only after the user confirms the exact record | -| Calls | nothing yet. `Approvisionnement.check_availability` comes with discontinuation decisions | +| Calls | `Approvisionnement.check_availability`, before recommending a discontinuation | | Exposes | `describe_catalog` | Who may write is decided by the MCP server, not by this agent. It authorises diff --git a/k8s/deployment.yaml b/k8s/deployment.yaml index 7d0b7f6..51bd036 100644 --- a/k8s/deployment.yaml +++ b/k8s/deployment.yaml @@ -30,7 +30,7 @@ spec: # repo is public, so the pull needs no credentials — which removes an # imagePullSecret that would otherwise have to be created by hand in # this namespace and kept in step with a GitHub token. - image: gitea.sttlab.eu/stt/classic-models-agent-catalogue:v0.2.0 + image: gitea.sttlab.eu/stt/classic-models-agent-catalogue:v0.3.0 ports: - name: http containerPort: 8080 diff --git a/pyproject.toml b/pyproject.toml index 1b82ab3..360dfec 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "classic-models-agent-catalogue" -version = "0.2.0" +version = "0.3.0" description = "Product management agent for the Classic Models catalog domain" requires-python = ">=3.12" diff --git a/src/catalogue_agent/agent.py b/src/catalogue_agent/agent.py index 9a7e322..9fa9bd2 100644 --- a/src/catalogue_agent/agent.py +++ b/src/catalogue_agent/agent.py @@ -27,6 +27,7 @@ from .approval import validate_draft from .config import Settings from .identity import TokenExchanger, caller_of from .mcp_clients import build_toolsets +from .peers import ask_peer, peers @dataclass class CallerDeps: @@ -52,8 +53,11 @@ How to work: 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. +- When a decision depends on availability, ask Approvisionnement with + `check_availability`. It answers; it does not hand back the stock table, and + you must not infer availability from the catalog's own quantity field. +- Never recommend discontinuing a reference without asking. A reference about to + be reordered is not a reference to drop. Writing to the catalog: - You may create products. Every field you write comes from the user or from a @@ -267,6 +271,7 @@ class CatalogueAgent: instructions=[SYSTEM_PROMPT, caller_instructions], deps_type=CallerDeps, toolsets=toolsets, + tools=[self._availability_tool(subject_token)], # A turn may end with an answer or with a decision to make. Without # DeferredToolRequests in the union, a held tool call is an error # rather than a state. @@ -274,6 +279,35 @@ class CatalogueAgent: ) return agent, CallerDeps(user=caller_of(subject_token)) + def _availability_tool(self, subject_token: str) -> Any: + """One narrow tool, not a generic `call_agent`. + + A generic one would let the model reach any peer with any prompt, which + is how an agent becomes a proxy onto its neighbours. This one asks a + single question of a single peer. + """ + peer = peers(self._settings)["approvisionnement"] + settings, exchanger = self._settings, self._exchanger + + async def check_availability(productcode: str) -> str: + """Ask the supply agent whether a product can be promised. + + Args: + productcode: the reference, as it appears in the catalog, e.g. S10_1949. + """ + return await ask_peer( + settings, + exchanger, + peer, + f"Peut-on promettre le produit {productcode} ?", + subject_token, + # One conversation per peer per turn. The peer keeps its own + # thread, which is what lets it resume anything it parks. + f"cat-{abs(hash(subject_token)) % 10**8}", + ) + + return check_availability + def _finish(self, conversation_id: str, result: Any) -> TurnResult: self._remember(conversation_id, result.all_messages()) output = result.output diff --git a/src/catalogue_agent/config.py b/src/catalogue_agent/config.py index 787ad42..9b6ee28 100644 --- a/src/catalogue_agent/config.py +++ b/src/catalogue_agent/config.py @@ -34,6 +34,13 @@ class Settings(BaseSettings): mcp_catalog_url: str = "https://catalog-dev.sttlab.local/catalog-mcp" mcp_catalog_scope: str = "mcp-catalog-audience token-exchange-broker-audience" mcp_sales_url: str = "https://sales-dev.sttlab.local/sales-mcp" + + # --- peers --------------------------------------------------------------- + # An A2A call returns an answer, not data. Consulting this one is how the + # agent learns whether a reference is about to be reordered before it + # recommends dropping it — without gaining any sight of the supply domain. + approvisionnement_url: str = "http://approvisionnement-agent.agents.svc.cluster.local:8080/" + approvisionnement_scope: str = "classic-models-agent-approvisionnement-audience" mcp_sales_scope: str = "mcp-sales-audience token-exchange-broker-audience" # --- inference ---------------------------------------------------------- diff --git a/src/catalogue_agent/peers.py b/src/catalogue_agent/peers.py new file mode 100644 index 0000000..bcbc72a --- /dev/null +++ b/src/catalogue_agent/peers.py @@ -0,0 +1,114 @@ +"""Calling another agent, as a tool the model can reach for. + +The rule this agent lives by: **an A2A call returns an answer, not data.** It +asks Approvisionnement "can S10_1949 be promised?" and receives a decision. It +never gains read access to whatever that agent can see, and it must not — the +whole per-domain scoping rests on a specialist answering rather than handing +over its records. + +So this is deliberately one narrow tool, not a generic `call_agent(name, text)`. +A generic one would let the model reach any peer with any prompt, which is how +an agent becomes a proxy onto its neighbours. + +The peer's token is minted per call, for that peer alone, from the end user's +own token. A token for Approvisionnement is refused by every other agent, so a +prompt injection that talks this agent into consulting the wrong peer fails at +the realm. +""" + +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 + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class Peer: + """An agent this one may consult.""" + + name: str + url: str + audience_scope: str + + +def peers(settings: Settings) -> dict[str, Peer]: + """Approvisionnement, and nothing else. + + The design has this agent consult it before a discontinuation decision: + a reference that is about to be reordered is not a reference to drop. + """ + return { + "approvisionnement": Peer( + "approvisionnement", + settings.approvisionnement_url, + settings.approvisionnement_scope, + ) + } + + +def _text_of(message: a2a_types.Message) -> str: + return "".join(part.text for part in message.parts if part.text) + + +async def ask_peer( + settings: Settings, + exchanger: TokenExchanger, + peer: Peer, + question: str, + subject_token: str, + context_id: str, +) -> str: + """Put one question to one peer and return what it said. + + A peer that ends its task in any state but completed is reported as such + rather than smoothed over. `input-required` and `failed` both mean the + caller should stop and say so — a specialist that escalated has not answered, + and presenting its escalation as an answer is how a refusal turns into a + fact. + """ + token = await exchanger.exchange(subject_token, peer.audience_scope) + + async with httpx.AsyncClient( + headers={"Authorization": f"Bearer {token}"}, + verify=settings.verify, + timeout=settings.http_timeout, + ) as http: + factory = ClientFactory(ClientConfig(httpx_client=http, streaming=False)) + client = await factory.create_from_url(peer.url) + try: + request = a2a_types.SendMessageRequest( + message=a2a_types.Message( + message_id=f"{context_id}-{peer.name}", + 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() + + logger.info("asked %s -> %s", peer.name, state) + if state != "TASK_STATE_COMPLETED": + return ( + f"{peer.name} did not answer — it ended in {state} and said: " + f"{text or '(nothing)'}. Report this and do not decide without it." + ) + return text or f"({peer.name} returned no answer)"