3.6. A2A

What is A2A, and why is it not just MCP again?

Agent2Agent (A2A) is an open protocol for one agent to talk to another agent across a network boundary. It was created at Google, donated to the Agentic AI Foundation, and — like MCP — belongs to no single vendor.

The obvious question, having just built MCP tools in 3.3. MCP, is why a second protocol exists. The answer is that they connect different things, and the distinction is not academic:

Put plainly: MCP is how an agent uses something; A2A is how an agent delegates to someone. A tool does what it is told. A peer agent is told what you want and figures out how — it may take thirty seconds, ask a clarifying question, or come back partly done. A request/response function call cannot express that, which is why A2A is task-shaped rather than call-shaped.

They are complementary, not competing. This course's agent is an A2A server to its clients and an MCP client to its tools, simultaneously.

flowchart LR
    Client[A2A client<br/>another agent or app] -->|A2A: task| Agent[AgentOps Agent]
    Agent -->|MCP: tool call| Tools[Read tools]
    Agent -->|in-process| Writes[Guarded writes]

How does an A2A interaction work?

Three ideas carry the protocol:

1. The agent card. A JSON document at a well-known URL (/.well-known/agent-card.json) that advertises identity, skills, capabilities, and auth requirements. It is how a client discovers what an agent can do before trusting it with anything — the agent equivalent of an OpenAPI document.
2. The task. The unit of work, with a lifecycle: submitted → working → (input-required) → completed or failed. Tasks have IDs, so a long job can be polled, resumed, or streamed rather than held open on one blocking call.
3. Messages and artifacts. Conversation turns, and the outputs a task produces.

Transport is deliberately boring. JSON-RPC over HTTP is an ordinary POST whose JSON body names the method to run. Server-Sent Events (SSE) keeps one HTTP response open while the server appends lines to it. Ordinary infrastructure can therefore govern an A2A endpoint.

That is what lets Chapter 5 put agentgateway in front of it, and it is why the input-required state matters here: an approval pause (4.5. Guardrails) is a first-class protocol state, not a hack.

Run the task-lifecycle and session-mapping regressions before opening a listener:

cd agents/python
uv run pytest tests/test_server.py -q

How do you inspect the A2A contract?

Stop reading here and fetch the card. Every field the sections below explain is one you will already have printed.

cd agents/python
mise run a2a

In another terminal:

curl -fsS http://localhost:8080/.well-known/agent-card.json | jq '{name,url,skills}'

Expected name: AgentOps Agent; the card should list incident triage and guarded remediation.

What does the agent card declare?

The card you just fetched is built by one constructor in server.py.

# simplified
agent_card = AgentCard(
    name="AgentOps Agent",
    description="Runbook-grounded incident triage and guarded remediation for the AgentOps Open Course.",
    # a2a-sdk 1.x advertises endpoints as a list of transport interfaces, not a single url.
    supported_interfaces=[
        AgentInterface(
            url=f"{settings.a2a_protocol}://{settings.a2a_host}:{settings.a2a_port}/",
            protocol_binding="JSONRPC",
        )
    ],
    version=version("agentops-agent"),
    capabilities=AgentCapabilities(streaming=True),
    default_input_modes=["text/plain"],
    default_output_modes=["text/plain"],
    skills=[
        AgentSkill(
            id="incident-triage",
            name="Incident triage",
            description="Prioritize incidents using service state, logs, and deterministic severity rules.",
            tags=["incident", "triage", "operations"],
            examples=["Triage the open incidents."],
        ),
    ],
)

The source constructs both AgentSkill records inline; the excerpt shows the first. The served agent-card.json still exposes a top-level url alongside supportedInterfaces for client compatibility, which is why the jq filter above found one.

The interface URL is built from three separate settings in config.py that also matter operationally:

# simplified
a2a_bind_host: str = Field(default="127.0.0.1", min_length=1)
a2a_host: str = Field(default="localhost", min_length=1)
a2a_port: int = Field(default=8080, ge=1, le=65535)
a2a_protocol: str = Field(default="http", pattern=r"^https?$")

a2a_host (what a client dials) and a2a_bind_host (what the process listens on) are deliberately distinct.

• main() binds Uvicorn to a2a_bind_host, which defaults to loopback 127.0.0.1.
• Only the container image opts into AGENT_A2A_BIND_HOST=0.0.0.0 to accept traffic from other pods, and tests/test_server.py asserts the Dockerfile sets exactly that.
• The card, however, is always built from a2a_host/a2a_protocol/a2a_port, so it advertises a client-reachable service address.

It never advertises the listener address 0.0.0.0, which means "every interface," not somewhere a client can call back.

How does the server preserve tasks?

Tasks and sessions are written to a SQLite file, not held in process memory, so a restart does not lose them. create_app() builds an explicit runner with a DatabaseSessionService and an A2A DatabaseTaskStore, both backed by .state/runtime.db.

The durability argument is three values in server.py:

session_service = _A2ASessionService(
    db_url=database_url,
    pool_size=1,
    max_overflow=0,
    connect_args={"timeout": 30},
)
# Build the App here rather than importing the module-level one, so an injected agent
# (tests, embedding) is governed by the same policy plugin as the shipped composition.
# ``app=`` is ADK's recommended Runner form; the deprecated ``agent=``/``plugins=`` pair
# would attach the policy to only this Runner instead of to the application.
runner = _SessionSerializingRunner(
    app=build_app(selected_agent),
    session_service=session_service,
)
task_engine = session_service.db_engine
runtime = Runtime(
    runner=runner,
    session_service=session_service,
    task_engine=task_engine,
    task_store=DatabaseTaskStore(engine=task_engine),
)

The task store does not open its own database — task_engine = session_service.db_engine reuses the session service's engine, so both stores literally share one connection. _A2ASessionService recovers the executor's concurrent first-use get-or-create race without discarding requested state. _SessionSerializingRunner then queues overlapping turns for one user and session before ADK reads its detached state snapshot; other sessions remain concurrent. The lifespan closes the runner and the session service that owns that engine. This is a single-process course durability model, not a horizontally scalable database design.

What bounds one A2A task?

A networked task can run away in ways an in-process call cannot. The caller is another agent that may itself be looping, and the work is open-ended by design. Two bounds cap a single A2A task in this repository, both configurable and both with safe defaults.

1. A per-task model-call ceiling. _bounded_request overrides whatever max_llm_calls ADK's converter produced with a fixed budget:

# simplified
a2a_max_llm_calls: int = Field(default=12, ge=1, le=100)

Every A2A task therefore gets at most 12 model round-trips (clamped to 1–100) before ADK aborts the run, so a delegated goal cannot silently loop forever on your token budget. Change it with AGENT_A2A_MAX_LLM_CALLS; tests/test_server.py asserts the override survives even the streaming path. 1. A wall-clock drain bound. SIGTERM is the polite stop signal a runtime sends before it kills a process. drain_timeout_s (default 10s, AGENT_DRAIN_TIMEOUT_S) is how long an in-flight task may keep running after that signal before Uvicorn forces shutdown. A turn that overruns it is cut, which is why every guarded write is transactional and never relies on drain time for correctness; a container's terminationGracePeriodSeconds must exceed this value.

The lifecycle those bounds act on is small and testable:

stateDiagram-v2
    [*] --> submitted
    submitted --> working
    state "input-required" as input_required
    working --> input_required: guarded action needs approval
    input_required --> working: confirmed with rationale
    working --> completed: final true
    working --> failed: final true, adk_error_code
    completed --> [*]
    failed --> [*]
    note right of working
        AGENT_A2A_MAX_LLM_CALLS caps model calls (12)
        AGENT_DRAIN_TIMEOUT_S cuts a run on SIGTERM (10s)
    end note

test_a2a_confirmation_response_resumes_the_guarded_action_with_audit_identity drives the left path end to end:

1. The task pauses at input-required, carrying an adk_request_confirmation call.
2. The client resumes the same taskId/contextId with a confirmed DataPart — one structured JSON chunk of an A2A message — and a rationale.
3. The task reaches completed with an attributable audit row.

That pause is final: false in a2a-sdk 1.x. The response ends there; the task does not, because the same task resumes. The failed edge is exercised by the stream-interruption test discussed below.

{
  "kind": "data",
  "data": {
    "id": "<the paused call id>",
    "name": "adk_request_confirmation",
    "response": { "confirmed": true, "payload": { "rationale": "runbook-backed; INC-002 verified" } }
  }
}

The state machine above is what the server tracks; the two-request handshake below is what a client actually implements. Approval is a second message/send that reuses the same identifiers:

sequenceDiagram
    participant C as A2A client
    participant S as A2A server
    C->>S: message/send (new task)
    S-->>C: working, then input-required (final:false)<br/>+ adk_request_confirmation
    Note over C: user approves with a rationale
    C->>S: message/send — FunctionResponse DataPart<br/>same taskId + contextId
    S-->>C: working → completed + attributable audit row

A missing rationale, or a resume on a fresh taskId, does not continue the paused action — the guarded write only fires when the confirmation is tied back to the exact task that requested it.

How do I stream a response through the gateway?

Call message/stream instead of message/send. The response is an SSE channel: every data: line is one JSON-RPC response carrying a whole task event (submitted, working, completed, failed).

With the A2A server (mise run a2a), the gateway (5.3. A2A Gateway), and a model backend running:

curl -N -fsS http://127.0.0.1:3001/ \
  -H 'Content-Type: application/json' \
  -d '{
    "jsonrpc": "2.0", "id": "1", "method": "message/stream",
    "params": {"message": {"kind": "message", "messageId": "demo-1",
      "role": "user", "parts": [{"kind": "text", "text": "Triage the open incidents."}]}}
  }'

You always get this event stream — it reports task progress, not model tokens. Whether the model also streams is a separate, opt-in decision made server-side in server.py:

# simplified
updates: dict[str, object] = {"max_llm_calls": settings.a2a_max_llm_calls}
if settings.a2a_streaming:
    updates["streaming_mode"] = StreamingMode.SSE
converted.run_config = run_config.model_copy(update=updates)

By default the model runs non-streaming and the stream carries whole events. AGENT_A2A_STREAMING=true adds partial per-token events for lower perceived latency, at the redaction cost described below — which is why it defaults to false.

How does a caller cancel an active task?

A client cancels an active task by sending its taskId through the A2A tasks/cancel method.

{
  "jsonrpc": "2.0",
  "id": "cancel-1",
  "method": "tasks/cancel",
  "params": { "id": "<task-id>" }
}

The pinned ADK executor does not yet implement its cancellation hook. The course supplies only the missing terminal event:

class _CancelableA2AExecutor(A2aAgentExecutor):
    """Add the terminal cancellation event omitted by ADK's executor."""

    @override
    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        if not context.task_id or not context.context_id:
            raise ValueError("A2A cancellation requires task and context ids.")
        # The A2A handler cancels the producer before calling this method. ADK
        # 2.4 leaves cancel() unimplemented, so publish the terminal protocol
        # event here instead of turning a user cancellation into a failed task.
        await TaskUpdater(event_queue, context.task_id, context.context_id).cancel()

The A2A server cancels the running producer before calling this hook. _CancelableA2AExecutor then emits canceled, so the task store and every connected client observe one terminal state instead of a failed request.

The browser client enables Cancel only while it knows a non-terminal task id. Its stream, approval, reconnect, cancellation, error, and session-continuity paths all use the same task/context state machine.

What happens when a stream fails mid-answer?

The stream terminates explicitly, not silently. When the run raises mid-stream, the executor emits a terminal failed status-update event (final: true) on the same channel, so a client sees the task fail rather than a hung connection.

# simplified
def _agent_executor(runner: Runner) -> A2aAgentExecutor:
    """Create the maintained A2A executor with the bounded request policy."""
    return A2aAgentExecutor(
        runner=runner,
        config=A2aAgentExecutorConfig(
            request_converter=_bounded_request,
            execute_interceptors=[_error_code_interceptor()],
        ),
        # ADK's legacy result aggregator mutates terminal failures back to
        # ``working`` before enqueueing them. The maintained executor preserves
        # the A2A terminal state and emits partial model output as artifacts.
        force_new_version=True,
    )

Redaction has a precise contract under streaming. redact_response_pii is an after_model_callback, and ADK runs it on every partial chunk and on the final aggregated response, so enabling streaming does not bypass the PII guardrail (4.5. Guardrails).

Per-chunk redaction is still best-effort:

• An entity split across two chunk boundaries may not match either fragment.
• Fragments already sent cannot be retracted.

The final aggregate is redacted as a whole, so the durable answer is clean; only the transient partial view carries that residual risk. This asymmetry is the reason AGENT_A2A_STREAMING defaults to false and is an explicit trade-off, not a free feature.

What is in-process delegation?

The cheap version of delegation never touches the network. An ADK coordinator can transfer control to a named sub-agent in the same process and session.

The course's diagnosis specialist owns only read and runbook tools, the remediation specialist owns only the guarded actions, and the coordinator owns triage tools and decides when to delegate:

# simplified
coordinator_agent = Agent(
    model=build_model(),
    name="coordinator_agent",
    description="On-call coordinator that triages incidents and delegates diagnosis and remediation.",
    tools=ALL_TOOLS,
    sub_agents=[diagnosis_agent, remediation_agent],
)

In-process delegation is simple and low latency, but both agents share deployment, failure, trust, and scaling boundaries. The next section prices that shared boundary.

Owned by 3.7. Multi-Agent, which covers the delegation pattern and its least-privilege boundaries in depth: each specialist holds only the tools its own job needs. The definitions in delegation.py attach each specialist's instructions and tools. The enclosing App supplies redaction and stable error handling through AgentOpsPolicyPlugin.

When is A2A worth its cost?

A2A buys a real boundary: independent deployment, ownership, language, scaling, and failure domain. Two teams can ship agents on separate schedules and only agree on a card.

That boundary is not free. You inherit:

• identity and authorization between agents;
• network policy;
• versioning and compatibility;
• retries and idempotency;
• task persistence across restarts;
• distributed tracing to explain what happened.

The in-process alternative in the section above has none of those problems, because it has no network.

So: use in-process delegation until an organizational or operational boundary forces your hand. A different team, a different release cadence, a different scaling profile, a different trust level, or a different language are good reasons. "It feels more like microservices" is not — you would be buying a distributed system to solve a function-call problem.

In this repository, the in-process coordinator demonstrates delegation semantics and agent.server exposes the root AgentOps Agent over A2A. It does not pretend that the in-process specialist is already a separately deployed remote agent.

When should an agent become a separate service?

Split it when ownership, data classification, scaling, release cadence, or blast radius differs enough to justify the network boundary. Blast radius is how much else breaks when this one part fails. Do not split merely to draw a multi-agent diagram: a function or in-process agent is cheaper when lifecycle and trust are identical.

How would you write a minimal A2A client?

Exercise: go past curl and script the full task-vs-session handshake yourself, so the difference between a contextId and a taskId becomes concrete rather than described.

• Mode: temporary experiment.
• Goal: from a short script, discover the agent, submit a guarded request, handle the approval pause, and resume it on the same task — the exact flow the reference clients/web browser client implements, reduced to its essentials.
• Files to touch: .agents/tmp/a2a-handshake.py only; the live request writes disposable runtime state under agents/python/.state.
• Preflight: require test ! -e .agents/tmp/a2a-handshake.py, mise run doctor:gateway passing, and the documented A2A plus gateway processes already running.
• Steps to implement: (1) GET /.well-known/agent-card.json through the :3001 gateway listener and read whether the card advertises streaming; (2) send restart_service(...) for a down service with message/send, keeping the returned contextId; (3) observe the task settle in input-required rather than completing; (4) resume by sending the approval on the same taskId, and confirm the audit row names the approver (3.1. Tools).
• Gate that proves completion: the second (resume) request must reuse the taskId from step 2 — a new task id means you started a fresh conversation instead of answering the pending approval, which is the one mistake this exercise exists to catch. Read clients/web/index.html (it comments exactly when taskId is set and cleared) to check your handling against the reference.
• Final state: stop only the processes you started, run rm -- .agents/tmp/a2a-handshake.py, then cd agents/python && mise run data:reset; the script is absent and the disposable state matches the seed again.

What proves this page worked?

cd agents/python
uv run pytest tests/test_server.py tests/test_delegation.py

Verify card fields, advertised address, persistent service/task construction, cancellation, and shutdown cleanup. The same run covers the coordinator's sub-agent wiring and the real input-required confirmation response resuming the same task with attributable audit evidence. Deterministic fake models drive cancellation and approval, so no live model call is needed.

The focused server test should finish without a network service left behind. Apply the complete mise run test coverage gate at chapter exit.

After inspecting the card, return to the terminal running mise run a2a and press Ctrl-C. The later gateway lab starts it again with the configuration that lab owns.

You are done when:

• mise run a2a served a card whose name is AgentOps Agent, then stopped cleanly with Ctrl-C.
• uv run pytest tests/test_server.py tests/test_delegation.py passes.
• You can say which identifier carries the session (contextId) and which one resumes a paused task (taskId).
• You can cancel a working task and observe a terminal canceled state.
• You can name one boundary — team, release cadence, scaling, trust, or language — that would justify moving an in-process sub-agent behind A2A.

Continue to 3.7. Multi-Agent when in-process delegation is your default and you can say what would force you onto the network.