"""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 import uuid 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( # Unique per message, not per conversation. Reusing one id # across the turns of a conversation is a protocol error, and # the failure it invites — a peer treating the second call as # a replay of the first — would look like a stale answer # rather than a bug. message_id=f"{context_id}-{peer.name}-{uuid.uuid4().hex[:8]}", 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)"