"""Reference solution — A2A delegation, and a protocol-agnostic core. MCP answers "what tools do I have". A2A answers "who else can do this, and how do I hand it to them". The second question brings problems the first does not: long-running work, a task lifecycle that outlives a request, artifacts, cancellation, push notifications, and — the one that actually matters in a bank — an identity chain across organizational boundaries. The design rule this file exists to demonstrate: the KERNEL speaks an internal task model; A2A and ACP are EDGE ADAPTERS. Otherwise every protocol revision is a rewrite. Deterministic: no clock, no randomness, no network. ``python solution.py`` runs a full delegation, a cancellation, a rejected cycle and an ACP round-trip. """ from __future__ import annotations from dataclasses import dataclass, field, replace from enum import Enum from typing import Callable, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple A2A_VERSION = "0.3" # ====================================================================================== # 1. The parts model — what agents actually exchange # ====================================================================================== class PartKind(str, Enum): TEXT = "text" FILE = "file" DATA = "data" @dataclass(frozen=True) class Part: """One piece of a message or artifact. Three kinds, because agents exchange more than strings: prose, files (by reference or bytes), and STRUCTURED data. The data kind is the one that makes agent-to-agent different from chat — a delegated task returns a typed result, not a paragraph. """ kind: PartKind text: Optional[str] = None uri: Optional[str] = None mime_type: Optional[str] = None data: Optional[Mapping[str, object]] = None def __post_init__(self) -> None: if self.kind is PartKind.TEXT and self.text is None: raise ValueError("a text part needs text") if self.kind is PartKind.FILE and not self.uri: raise ValueError("a file part needs a uri") if self.kind is PartKind.DATA and self.data is None: raise ValueError("a data part needs data") @classmethod def text_part(cls, text: str) -> "Part": return cls(PartKind.TEXT, text=text) @classmethod def file_part(cls, uri: str, mime_type: str) -> "Part": return cls(PartKind.FILE, uri=uri, mime_type=mime_type) @classmethod def data_part(cls, data: Mapping[str, object]) -> "Part": return cls(PartKind.DATA, data=dict(data)) class Role(str, Enum): USER = "user" # the CALLER, which for a delegated task is another agent AGENT = "agent" @dataclass(frozen=True) class Message: message_id: str role: Role parts: Tuple[Part, ...] task_id: Optional[str] = None context_id: Optional[str] = None def text(self) -> str: return " ".join(p.text for p in self.parts if p.kind is PartKind.TEXT and p.text) @dataclass(frozen=True) class Artifact: """A durable OUTPUT of a task, distinct from the conversation about it. The distinction matters for evidence: the audit record wants the artifact ("the sanctions screening report"), not the chat that produced it. """ artifact_id: str name: str parts: Tuple[Part, ...] description: str = "" # ====================================================================================== # 2. The task lifecycle # ====================================================================================== class TaskState(str, Enum): SUBMITTED = "submitted" WORKING = "working" INPUT_REQUIRED = "input-required" AUTH_REQUIRED = "auth-required" COMPLETED = "completed" CANCELED = "canceled" FAILED = "failed" REJECTED = "rejected" TERMINAL_STATES: frozenset = frozenset({ TaskState.COMPLETED, TaskState.CANCELED, TaskState.FAILED, TaskState.REJECTED }) #: Legal transitions. A delegated task is long-running by design, so unlike a tool call #: it has real intermediate states — and unlike a tool call, the CALLER may need to act #: (input-required, auth-required) while the callee waits. TASK_TRANSITIONS: Mapping[TaskState, frozenset] = { TaskState.SUBMITTED: frozenset({TaskState.WORKING, TaskState.REJECTED, TaskState.CANCELED, TaskState.FAILED, TaskState.AUTH_REQUIRED}), TaskState.WORKING: frozenset({TaskState.INPUT_REQUIRED, TaskState.AUTH_REQUIRED, TaskState.COMPLETED, TaskState.CANCELED, TaskState.FAILED}), TaskState.INPUT_REQUIRED: frozenset({TaskState.WORKING, TaskState.CANCELED, TaskState.FAILED}), TaskState.AUTH_REQUIRED: frozenset({TaskState.WORKING, TaskState.CANCELED, TaskState.FAILED, TaskState.REJECTED}), } class IllegalTaskTransition(RuntimeError): pass def advance(state: TaskState, target: TaskState) -> TaskState: if state in TERMINAL_STATES: raise IllegalTaskTransition(f"{state.value} is terminal; cannot move to {target.value}") if target not in TASK_TRANSITIONS.get(state, frozenset()): raise IllegalTaskTransition(f"{state.value} -> {target.value} is not a legal transition") return target @dataclass(frozen=True) class TaskStatus: state: TaskState message: Optional[Message] = None timestamp: int = 0 @dataclass(frozen=True) class Task: """The unit of delegated work. ``context_id`` groups related tasks — a single investigation may delegate three tasks to three agents, and the context is what ties them into one story in the audit record. """ task_id: str context_id: str status: TaskStatus history: Tuple[Message, ...] = () artifacts: Tuple[Artifact, ...] = () delegation_chain: Tuple[str, ...] = () @property def is_terminal(self) -> bool: return self.status.state in TERMINAL_STATES # ====================================================================================== # 3. Agent Cards and discovery # ====================================================================================== @dataclass(frozen=True) class Skill: skill_id: str name: str description: str tags: Tuple[str, ...] = () examples: Tuple[str, ...] = () @dataclass(frozen=True) class AgentCard: """A2A's discovery document: OpenAPI plus a capability manifest, for agents. The bank-relevant fields are the ones people skip: ``security_schemes`` (how do I authenticate to you), ``tenants`` (may I even see you), and ``data_classification`` (what may I send you). """ name: str description: str url: str version: str protocol_version: str = A2A_VERSION skills: Tuple[Skill, ...] = () streaming: bool = False push_notifications: bool = False default_input_modes: Tuple[str, ...] = ("text/plain",) default_output_modes: Tuple[str, ...] = ("text/plain",) security_schemes: Tuple[str, ...] = ("oauth2",) owner: str = "unknown" tenants: Tuple[str, ...] = () max_data_classification: str = "internal" def to_json(self) -> Dict[str, object]: return { "protocolVersion": self.protocol_version, "name": self.name, "description": self.description, "url": self.url, "version": self.version, "capabilities": { "streaming": self.streaming, "pushNotifications": self.push_notifications, }, "defaultInputModes": list(self.default_input_modes), "defaultOutputModes": list(self.default_output_modes), "securitySchemes": list(self.security_schemes), "skills": [ {"id": s.skill_id, "name": s.name, "description": s.description, "tags": list(s.tags), "examples": list(s.examples)} for s in self.skills ], } _CLASSIFICATION_ORDER = ("public", "internal", "confidential", "restricted") def classification_rank(name: str) -> int: try: return _CLASSIFICATION_ORDER.index(name) except ValueError: raise ValueError(f"unknown data classification: {name!r}") from None @dataclass(frozen=True) class CallerContext: """Who is delegating. Everything about discovery and admission is relative to this.""" agent_id: str tenant: str user_id: str scopes: Tuple[str, ...] = () data_classification: str = "internal" delegation_chain: Tuple[str, ...] = () class AgentDirectory: """Discovery over Agent Cards — the A2A analogue of the tool registry. Same rule as Phase 02: filter BEFORE you list. An agent you may not delegate to must not appear, because its card's description enters your model's context and its name is something an injected instruction can then name. """ def __init__(self) -> None: self._cards: Dict[str, AgentCard] = {} def register(self, card: AgentCard) -> None: if card.name in self._cards: raise ValueError(f"agent already registered: {card.name}") self._cards[card.name] = card def get(self, name: str) -> Optional[AgentCard]: return self._cards.get(name) def discover(self, caller: CallerContext, *, tags: Sequence[str] = ()) -> List[AgentCard]: """Cards this caller may see, ranked by skill-tag overlap then name.""" wanted = set(tags) allowed_rank = classification_rank(caller.data_classification) visible: List[Tuple[int, AgentCard]] = [] for name in sorted(self._cards): card = self._cards[name] if card.name == caller.agent_id: continue # never discover yourself if card.tenants and caller.tenant not in card.tenants: continue if classification_rank(card.max_data_classification) < allowed_rank: continue # cannot handle our data overlap = 0 for skill in card.skills: overlap = max(overlap, len(wanted & set(skill.tags))) if wanted and overlap == 0: continue visible.append((overlap, card)) visible.sort(key=lambda pair: (-pair[0], pair[1].name)) return [card for _, card in visible] # ====================================================================================== # 4. Delegation admission — the part A2A does not specify # ====================================================================================== MAX_DELEGATION_DEPTH = 4 @dataclass(frozen=True) class DelegationDenial: code: str reason: str def check_delegation( caller: CallerContext, card: AgentCard, *, max_depth: int = MAX_DELEGATION_DEPTH, ) -> List[DelegationDenial]: """Everything the protocol leaves to you. A2A defines how to send a task. It does not define whether you may — and in a multi-agent flow that is where the entire risk lives: unbounded delegation depth, cycles, cross-tenant hops, and data flowing to an agent not cleared for it. """ denials: List[DelegationDenial] = [] chain = caller.delegation_chain if len(chain) >= max_depth: denials.append(DelegationDenial( "DEPTH_EXCEEDED", f"delegation chain is {len(chain)} deep; max is {max_depth}")) if card.name in chain: denials.append(DelegationDenial( "CYCLE_DETECTED", f"{card.name} already appears in the chain {list(chain)}")) if card.tenants and caller.tenant not in card.tenants: denials.append(DelegationDenial( "TENANT_NOT_PERMITTED", f"{card.name} does not serve tenant {caller.tenant!r}")) if classification_rank(card.max_data_classification) < classification_rank(caller.data_classification): denials.append(DelegationDenial( "CLASSIFICATION_EXCEEDED", f"{card.name} handles up to {card.max_data_classification!r}; " f"the task carries {caller.data_classification!r}")) if "oauth2" not in card.security_schemes and "mtls" not in card.security_schemes: denials.append(DelegationDenial( "NO_ACCEPTABLE_AUTH", f"{card.name} offers only {list(card.security_schemes)}")) return denials # ====================================================================================== # 5. Streaming events # ====================================================================================== @dataclass(frozen=True) class TaskStatusUpdate: task_id: str context_id: str status: TaskStatus final: bool = False @dataclass(frozen=True) class TaskArtifactUpdate: task_id: str context_id: str artifact: Artifact append: bool = False last_chunk: bool = False StreamEvent = object # TaskStatusUpdate | TaskArtifactUpdate | Task @dataclass(frozen=True) class PushNotificationConfig: """Where to call back when a long-running task changes state. The security note that belongs in every design review: this is a URL supplied by a CALLER, which your server will then request. That is an SSRF by construction unless the URL is validated against an allow-list and the callback carries a token the receiver can verify. """ url: str token: str = "" authentication: str = "bearer" # ====================================================================================== # 6. The A2A server # ====================================================================================== AgentHandler = Callable[["A2AServer", Task, Message], Iterable[StreamEvent]] class A2AError(Exception): def __init__(self, code: str, message: str) -> None: super().__init__(message) self.code = code self.message = message class A2AServer: """An agent exposed over A2A. The handler is a generator of stream events, which is the only shape that supports both ``message/send`` (drain it, return the final task) and ``message/stream`` (yield as they come) without writing the agent twice. """ def __init__( self, *, card: AgentCard, handler: AgentHandler, allowed_callback_hosts: Sequence[str] = (), ) -> None: self.card = card self.handler = handler self.allowed_callback_hosts = tuple(allowed_callback_hosts) self.tasks: Dict[str, Task] = {} self.push_configs: Dict[str, PushNotificationConfig] = {} self.pushed: List[Tuple[str, TaskStatus]] = [] self._tick = 0 self._counter = 0 # -- helpers ------------------------------------------------------------------ def _next_tick(self) -> int: self._tick += 1 return self._tick def _next_id(self, prefix: str) -> str: self._counter += 1 return f"{prefix}-{self._counter}" def _store(self, task: Task) -> Task: self.tasks[task.task_id] = task return task def set_status(self, task: Task, state: TaskState, message: Optional[Message] = None) -> Task: """Advance a task, enforcing the lifecycle and pushing a notification if configured.""" new_state = advance(task.status.state, state) status = TaskStatus(state=new_state, message=message, timestamp=self._next_tick()) updated = replace(task, status=status, history=task.history + ((message,) if message else ())) self._store(updated) config = self.push_configs.get(task.task_id) if config is not None: self.pushed.append((config.url, status)) return updated def add_artifact(self, task: Task, artifact: Artifact) -> Task: updated = replace(task, artifacts=task.artifacts + (artifact,)) return self._store(updated) # -- methods ------------------------------------------------------------------ def message_send(self, message: Message, *, caller: CallerContext, context_id: Optional[str] = None) -> Task: """Blocking form: drain the stream, return the final task.""" final: Optional[Task] = None for event in self.message_stream(message, caller=caller, context_id=context_id): if isinstance(event, Task): final = event if final is None: # pragma: no cover - handler contract raise A2AError("INTERNAL", "handler produced no final task") return final def message_stream(self, message: Message, *, caller: CallerContext, context_id: Optional[str] = None) -> Iterable[StreamEvent]: """Streaming form. Yields updates, then the final Task as the last event.""" denials = check_delegation(caller, self.card) if denials: raise A2AError(denials[0].code, denials[0].reason) if message.task_id is not None: task = self.tasks.get(message.task_id) if task is None: raise A2AError("TASK_NOT_FOUND", f"no such task: {message.task_id}") if task.is_terminal: raise A2AError("TASK_TERMINAL", f"task {task.task_id} is {task.status.state.value}") task = replace(task, history=task.history + (message,)) self._store(task) else: task_id = self._next_id("task") task = Task( task_id=task_id, context_id=context_id or self._next_id("ctx"), status=TaskStatus(TaskState.SUBMITTED, timestamp=self._next_tick()), history=(message,), delegation_chain=caller.delegation_chain + (caller.agent_id,), ) self._store(task) yield TaskStatusUpdate(task.task_id, task.context_id, task.status) for event in self.handler(self, self.tasks[task.task_id], message): yield event yield self.tasks[task.task_id] def tasks_get(self, task_id: str) -> Task: task = self.tasks.get(task_id) if task is None: raise A2AError("TASK_NOT_FOUND", f"no such task: {task_id}") return task def tasks_cancel(self, task_id: str) -> Task: task = self.tasks_get(task_id) if task.is_terminal: raise A2AError("TASK_TERMINAL", f"task {task_id} is already {task.status.state.value}") return self.set_status(task, TaskState.CANCELED) def set_push_config(self, task_id: str, config: PushNotificationConfig) -> PushNotificationConfig: """Register a callback. Validates the host against the allow-list, because a caller-supplied URL that your server will fetch is an SSRF by construction.""" if not self.card.push_notifications: raise A2AError("UNSUPPORTED", "this agent does not support push notifications") self.tasks_get(task_id) host = _host_of(config.url) if self.allowed_callback_hosts and host not in self.allowed_callback_hosts: raise A2AError("CALLBACK_NOT_ALLOWED", f"callback host {host!r} is not in the allow-list") if not config.token: raise A2AError("CALLBACK_UNAUTHENTICATED", "a push notification config must carry a token the receiver can verify") self.push_configs[task_id] = config return config def _host_of(url: str) -> str: without_scheme = url.split("://", 1)[-1] return without_scheme.split("/", 1)[0].split("@")[-1] # ====================================================================================== # 7. The A2A client # ====================================================================================== class A2AClient: """The delegating side. Owns the identity chain — the callee cannot be trusted to.""" def __init__(self, server: A2AServer) -> None: self.server = server self._counter = 0 def _next_message_id(self) -> str: self._counter += 1 return f"msg-{self._counter}" def card(self) -> AgentCard: return self.server.card def delegate(self, caller: CallerContext, text: str, *, data: Optional[Mapping[str, object]] = None, context_id: Optional[str] = None) -> Task: parts = [Part.text_part(text)] if data is not None: parts.append(Part.data_part(data)) message = Message(self._next_message_id(), Role.USER, tuple(parts), context_id=context_id) return self.server.message_send(message, caller=caller, context_id=context_id) def delegate_streaming(self, caller: CallerContext, text: str, context_id: Optional[str] = None) -> List[StreamEvent]: message = Message(self._next_message_id(), Role.USER, (Part.text_part(text),), context_id=context_id) return list(self.server.message_stream(message, caller=caller, context_id=context_id)) def reply(self, caller: CallerContext, task: Task, text: str) -> Task: message = Message(self._next_message_id(), Role.USER, (Part.text_part(text),), task_id=task.task_id, context_id=task.context_id) return self.server.message_send(message, caller=caller) # ====================================================================================== # 8. The protocol-agnostic core, and an ACP adapter # ====================================================================================== @dataclass(frozen=True) class InternalTask: """What the kernel actually stores. Deliberately not A2A's shape and not ACP's. If the kernel stored A2A objects, an A2A revision would be a data migration. This type is the seam that makes protocol churn an adapter change. """ task_id: str context_id: str state: str # our vocabulary, mapped at the edge input_text: str output_text: str = "" structured: Mapping[str, object] = field(default_factory=dict) delegation_chain: Tuple[str, ...] = () artifacts: Tuple[str, ...] = () #: Our states are a superset chosen for OUR needs; each protocol maps onto them. A2A_STATE_TO_INTERNAL: Mapping[TaskState, str] = { TaskState.SUBMITTED: "queued", TaskState.WORKING: "running", TaskState.INPUT_REQUIRED: "awaiting_input", TaskState.AUTH_REQUIRED: "awaiting_auth", TaskState.COMPLETED: "succeeded", TaskState.CANCELED: "cancelled", TaskState.FAILED: "failed", TaskState.REJECTED: "rejected", } INTERNAL_TO_A2A_STATE: Mapping[str, TaskState] = { v: k for k, v in A2A_STATE_TO_INTERNAL.items() } #: ACP is REST-shaped with its own vocabulary. Same internal states, different words. ACP_STATUS_TO_INTERNAL: Mapping[str, str] = { "created": "queued", "in-progress": "running", "awaiting": "awaiting_input", "completed": "succeeded", "cancelled": "cancelled", "failed": "failed", } INTERNAL_TO_ACP_STATUS: Mapping[str, str] = { "queued": "created", "running": "in-progress", "awaiting_input": "awaiting", "awaiting_auth": "awaiting", "succeeded": "completed", "cancelled": "cancelled", "failed": "failed", "rejected": "failed", } def a2a_to_internal(task: Task) -> InternalTask: structured: Dict[str, object] = {} output_bits: List[str] = [] for artifact in task.artifacts: for part in artifact.parts: if part.kind is PartKind.TEXT and part.text: output_bits.append(part.text) elif part.kind is PartKind.DATA and part.data: structured.update(part.data) inputs = [m for m in task.history if m.role is Role.USER] return InternalTask( task_id=task.task_id, context_id=task.context_id, state=A2A_STATE_TO_INTERNAL[task.status.state], input_text=inputs[0].text() if inputs else "", output_text=" ".join(output_bits), structured=structured, delegation_chain=task.delegation_chain, artifacts=tuple(a.name for a in task.artifacts), ) def internal_to_acp(task: InternalTask) -> Dict[str, object]: """An ACP-shaped run envelope: REST resource, multipart message, explicit status.""" parts: List[Dict[str, object]] = [] if task.output_text: parts.append({"content_type": "text/plain", "content": task.output_text}) if task.structured: parts.append({"content_type": "application/json", "content": dict(task.structured)}) return { "run_id": task.task_id, "session_id": task.context_id, "status": INTERNAL_TO_ACP_STATUS[task.state], "input": [{"content_type": "text/plain", "content": task.input_text}], "output": [{"role": "agent", "parts": parts}], "metadata": {"delegation_chain": list(task.delegation_chain), "artifacts": list(task.artifacts)}, } def acp_to_internal(envelope: Mapping[str, object]) -> InternalTask: status = str(envelope.get("status", "")) if status not in ACP_STATUS_TO_INTERNAL: raise ValueError(f"unknown ACP status: {status!r}") inputs = envelope.get("input") or [] input_text = " ".join( str(p.get("content", "")) for p in inputs if isinstance(p, dict) and p.get("content_type") == "text/plain" ) output_text_bits: List[str] = [] structured: Dict[str, object] = {} for message in envelope.get("output") or []: for part in message.get("parts", []) if isinstance(message, dict) else []: if part.get("content_type") == "text/plain": output_text_bits.append(str(part.get("content", ""))) elif part.get("content_type") == "application/json": content = part.get("content") if isinstance(content, dict): structured.update(content) metadata = envelope.get("metadata") or {} return InternalTask( task_id=str(envelope.get("run_id", "")), context_id=str(envelope.get("session_id", "")), state=ACP_STATUS_TO_INTERNAL[status], input_text=input_text, output_text=" ".join(output_text_bits), structured=structured, delegation_chain=tuple(metadata.get("delegation_chain", []) if isinstance(metadata, dict) else []), artifacts=tuple(metadata.get("artifacts", []) if isinstance(metadata, dict) else []), ) # ====================================================================================== # Worked example # ====================================================================================== def _sanctions_agent(server: A2AServer, task: Task, message: Message): """A callee that works, produces an artifact, and completes.""" task = server.set_status(task, TaskState.WORKING) yield TaskStatusUpdate(task.task_id, task.context_id, task.status) name = "Acme Trading FZE" for part in message.parts: if part.kind is PartKind.DATA and part.data and "beneficiary" in part.data: name = str(part.data["beneficiary"]) artifact = Artifact( artifact_id="art-1", name="sanctions-screening", description="Screening result with match detail", parts=(Part.text_part(f"{name}: 1 possible match (score 0.83) against SDN list"), Part.data_part({"matches": 1, "top_score": 0.83, "list": "SDN", "recommendation": "manual_review"}))) task = server.add_artifact(task, artifact) yield TaskArtifactUpdate(task.task_id, task.context_id, artifact, last_chunk=True) done = server.set_status( task, TaskState.COMPLETED, Message("m-out", Role.AGENT, (Part.text_part("Screening complete."),), task_id=task.task_id, context_id=task.context_id)) yield TaskStatusUpdate(done.task_id, done.context_id, done.status, final=True) def _clarifying_agent(server: A2AServer, task: Task, message: Message): """A callee that needs more information first — the input-required path.""" if task.status.state is TaskState.SUBMITTED: task = server.set_status(task, TaskState.WORKING) asked = server.set_status( task, TaskState.INPUT_REQUIRED, Message("m-ask", Role.AGENT, (Part.text_part("Which jurisdiction should I screen against?"),), task_id=task.task_id, context_id=task.context_id)) yield TaskStatusUpdate(asked.task_id, asked.context_id, asked.status) return task = server.set_status(task, TaskState.WORKING) done = server.set_status( task, TaskState.COMPLETED, Message("m-done", Role.AGENT, (Part.text_part("Screened. No matches."),), task_id=task.task_id, context_id=task.context_id)) yield TaskStatusUpdate(done.task_id, done.context_id, done.status, final=True) def main() -> None: # pragma: no cover - narrative output directory = AgentDirectory() sanctions_card = AgentCard( name="sanctions-screening-agent", description="Screens counterparties against sanctions and watch lists.", url="https://agents.bank.ae/sanctions", version="3.1.0", skills=(Skill("screen", "Screen a counterparty", "Screen a legal entity or individual against SDN, UN and local lists.", tags=("sanctions", "compliance", "screening"), examples=("Screen Acme Trading FZE",)),), streaming=True, push_notifications=True, owner="group-compliance", tenants=("wholesale", "retail"), max_data_classification="restricted") kyc_card = AgentCard( name="kyc-refresh-agent", description="Runs periodic KYC refresh workflows.", url="https://agents.bank.ae/kyc", version="1.4.0", skills=(Skill("refresh", "Refresh KYC", "Collect and verify refreshed KYC data.", tags=("kyc", "compliance")),), owner="retail-onboarding", tenants=("retail",), max_data_classification="confidential") marketing_card = AgentCard( name="marketing-copy-agent", description="Writes marketing copy.", url="https://agents.bank.ae/marketing", version="0.9.0", skills=(Skill("copy", "Write copy", "Draft campaign copy.", tags=("marketing",)),), owner="brand", max_data_classification="internal", security_schemes=("apikey",)) for card in (sanctions_card, kyc_card, marketing_card): directory.register(card) caller = CallerContext(agent_id="payments-investigator", tenant="wholesale", user_id="u-42", scopes=("delegate",), data_classification="restricted", delegation_chain=("orchestrator",)) print("=" * 78) print("1. DISCOVERY — WHO CAN DO THIS?") print("=" * 78) for card in directory.discover(caller, tags=["sanctions"]): print(f" {card.name:<28} v{card.version:<8} skills=" f"{[s.skill_id for s in card.skills]} streaming={card.streaming}") print(" not visible to this caller:") print(" kyc-refresh-agent — serves tenant 'retail' only") print(" marketing-copy-agent — handles up to 'internal'; our task is 'restricted'") print(f" full visible set (no tag filter): " f"{[c.name for c in directory.discover(caller)]}") print() print("=" * 78) print("2. DELEGATION ADMISSION — WHAT A2A DOES NOT SPECIFY") print("=" * 78) for card in (sanctions_card, kyc_card, marketing_card): denials = check_delegation(caller, card) verdict = "ALLOW" if not denials else ", ".join(d.code for d in denials) print(f" {card.name:<28} -> {verdict}") deep = replace(caller, delegation_chain=("a", "b", "c", "d")) print(f" chain depth 4 -> " f"{[d.code for d in check_delegation(deep, sanctions_card)]}") cyclic = replace(caller, delegation_chain=("sanctions-screening-agent",)) print(f" cycle -> " f"{[d.code for d in check_delegation(cyclic, sanctions_card)]}") print() print("=" * 78) print("3. A STREAMED DELEGATION") print("=" * 78) server = A2AServer(card=sanctions_card, handler=_sanctions_agent, allowed_callback_hosts=("agents.bank.ae",)) client = A2AClient(server) events = client.delegate_streaming(caller, "Screen the beneficiary of PMT-771") for event in events: if isinstance(event, TaskStatusUpdate): print(f" status {event.status.state.value:<16} final={event.final}") elif isinstance(event, TaskArtifactUpdate): print(f" artifact {event.artifact.name:<16} last_chunk={event.last_chunk}") else: print(f" task {event.task_id} -> {event.status.state.value}, " f"{len(event.artifacts)} artifact(s), chain={list(event.delegation_chain)}") final = events[-1] for part in final.artifacts[0].parts: print(f" {part.kind.value}: {part.text or part.data}") print() print("=" * 78) print("4. INPUT-REQUIRED: A TASK THAT OUTLIVES A REQUEST") print("=" * 78) clar = A2AServer(card=replace(sanctions_card, name="clarifier"), handler=_clarifying_agent) clar_client = A2AClient(clar) task = clar_client.delegate(caller, "Screen this entity") print(f" after first send : {task.status.state.value} — " f"{task.status.message.text()!r}") task = clar_client.reply(caller, task, "UAE and UN lists") print(f" after reply : {task.status.state.value} — " f"{task.status.message.text()!r}") print(f" history : {[m.role.value for m in task.history]}") print() print("=" * 78) print("5. CANCELLATION AND TERMINAL STATES") print("=" * 78) cancel_server = A2AServer(card=sanctions_card, handler=_clarifying_agent) cancel_client = A2AClient(cancel_server) pending = cancel_client.delegate(caller, "Screen something slow") cancelled = cancel_server.tasks_cancel(pending.task_id) print(f" cancelled : {cancelled.status.state.value}") try: cancel_server.tasks_cancel(pending.task_id) except A2AError as exc: print(f" cancel again : [{exc.code}] {exc.message}") try: advance(TaskState.COMPLETED, TaskState.WORKING) except IllegalTaskTransition as exc: print(f" illegal move : {exc}") print() print("=" * 78) print("6. PUSH NOTIFICATIONS — AND THE SSRF THEY INVITE") print("=" * 78) live = A2AServer(card=sanctions_card, handler=_clarifying_agent, allowed_callback_hosts=("agents.bank.ae",)) live_client = A2AClient(live) t = live_client.delegate(caller, "Screen slowly") try: live.set_push_config(t.task_id, PushNotificationConfig("https://evil.example/hook", token="x")) except A2AError as exc: print(f" external host : [{exc.code}] {exc.message}") try: live.set_push_config(t.task_id, PushNotificationConfig("https://agents.bank.ae/cb")) except A2AError as exc: print(f" no token : [{exc.code}] {exc.message}") live.set_push_config(t.task_id, PushNotificationConfig("https://agents.bank.ae/cb", token="s3cret")) live_client.reply(caller, t, "UAE lists") print(f" pushes delivered : " f"{[(url, status.state.value) for url, status in live.pushed]}") print() print("=" * 78) print("7. PROTOCOL-AGNOSTIC CORE: A2A -> INTERNAL -> ACP -> INTERNAL") print("=" * 78) internal = a2a_to_internal(final) print(f" internal state : {internal.state}") print(f" structured : {dict(internal.structured)}") envelope = internal_to_acp(internal) print(f" ACP envelope : run_id={envelope['run_id']} " f"status={envelope['status']} parts={len(envelope['output'][0]['parts'])}") round_tripped = acp_to_internal(envelope) print(f" round-trips : {round_tripped == internal}") print(f" A2A state map : {A2A_STATE_TO_INTERNAL[TaskState.INPUT_REQUIRED]!r} " f"<- input-required ; ACP 'awaiting' -> " f"{ACP_STATUS_TO_INTERNAL['awaiting']!r}") print(" -> the kernel never learns a protocol's vocabulary; adapters translate.") print() print("=" * 78) print("8. THE AGENT CARD ON THE WIRE") print("=" * 78) card_json = sanctions_card.to_json() for key in ("protocolVersion", "name", "url", "version", "capabilities", "securitySchemes"): print(f" {key:<18} {card_json[key]}") print(f" {'skills':<18} {[s['id'] for s in card_json['skills']]}") print(" note: tenants, owner and max_data_classification are PLATFORM metadata") print(" and are deliberately absent from the published card.") if __name__ == "__main__": # pragma: no cover main()