Stéphane Tailland
f79b58fc88
Two bugs from review, the second in both agents because identity.py is a copy. The exchanged-token cache grew for the life of the process. It is keyed on (subject token, scope) and the subject token is minted fresh every turn, so every turn added keys nothing would ever read again — and expiry was checked on read but never collected. A slow leak that only shows in a pod up for days, which is the worst way to find one. It now drops expired entries first and evicts oldest beyond a bound. Expired first on purpose: an expired entry is dead weight wherever it sits in the ordering, so evicting by age alone would keep it while discarding a live one. The docstring was wrong before the code was: it claimed the cache lasted one turn. It never did. Corrected rather than quietly made true. message_id reused the context id, which is the conversation — and a conversation makes several calls to the same peer, relaying an approval being exactly that. A peer treating the second call as a replay of the first would surface as a stale answer rather than an error. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
107 lines
4.3 KiB
Python
107 lines
4.3 KiB
Python
"""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
|
|
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
|
|
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(
|
|
# Unique per message. The context id is the
|
|
# conversation, and a conversation makes several calls to
|
|
# the same specialist — relaying an approval is exactly
|
|
# that. Reusing an id invites a peer to treat the second
|
|
# call as a replay of the first, which surfaces as a
|
|
# stale answer rather than as an error.
|
|
message_id=f"{context_id}-{specialist.name}-{uuid.uuid4().hex[:8]}",
|
|
# 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)
|