Your agents can't talk to each other: build two interoperating agents with the A2A protocol — hands-on
MCP gave every agent the same way to use tools. A2A — the Agent2Agent protocol, now a Linux Foundation standard — gives agents a way to use each other. This tutorial builds two real, interoperating A2A agents with the official Python SDK: a FactFinder research agent, and a BriefWriter that discovers it, delegates to it, and assembles the result. No API key, no GPU, every command executed.
Every serious agent stack in 2026 has the same hole. Your research agent can't ask your data agent for help. Your company's support agent can't delegate to the vendor's troubleshooting agent. Each one is brilliant in isolation and mute in company — so teams glue them together with bespoke REST endpoints, shared databases, or one mega-agent that does everything badly.
A2A, the Agent2Agent protocol, is the industry's answer. Announced by Google in April 2025, donated to the Linux Foundation that June, and now at a stable v1.0 with 150+ organizations behind it, A2A is an open standard for one agent to discover and delegate work to another — across frameworks, vendors, and company boundaries. The one-line mental model the ecosystem settled on: MCP connects agents to tools; A2A connects agents to agents. An agent's internal tools, memory, and prompts stay private; what crosses the wire is a task with a lifecycle.
This tutorial builds the real thing, not a diagram. You'll create FactFinder, an A2A research agent, then BriefWriter, a second agent that discovers FactFinder over the protocol, delegates research to it, and returns a finished brief as an artifact. Every command below was executed on October 1, 2026 against a2a-sdk 1.2.1 (the official Python SDK), and every output shown is real. No API key, no GPU, no cloud account — the agents are deterministic so the whole loop is free and perfectly reproducible. When you're done, you'll know exactly where to plug in a real LLM.
What you'll need #
- Python 3.10 or newer — check with
python3 --version. The SDK requires 3.10+. - pip and a terminal. A virtual environment is strongly recommended.
- No API key, no account, no GPU. Both agents in this tutorial are deterministic — the protocol is the point, not the model. A note at the end shows where a real LLM plugs in.
- About 20 minutes, most of it reading. The running takes seconds.
Step 1 — Install the SDK #
The official Python SDK lives on PyPI as a2a-sdk (source at a2aproject/a2a-python). The base package covers the client and protocol types; the fastapi extra adds the Starlette-based server pieces you'll need. Pin the version — the SDK moved fast from 0.x to 1.x and old tutorials no longer match the API:
python3 -m venv a2a-tutorial && source a2a-tutorial/bin/activate
pip install "a2a-sdk[fastapi]==1.2.1" httpx uvicornVerify the install and take a first look at the API surface — this is the map for everything that follows:
python -c "
from a2a.server.agent_execution import AgentExecutor
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.server.routes import create_jsonrpc_routes, create_agent_card_routes
from a2a.client.client_factory import ClientFactory
from a2a.client.card_resolver import A2ACardResolver
print('SDK surface OK')
"
# SDK surface OKSix imports, two halves: a2a.server.* builds agents, a2a.client.* talks to them. An agent that delegates — like the BriefWriter you'll build in Step 5 — uses both halves at once.
Step 2 — Five concepts that run the whole protocol #
A2A has a small vocabulary. Learn these five nouns and the rest of the tutorial is just spelling:
- AgentCard — a JSON document every A2A server publishes at
/.well-known/agent-card.json. It names the agent, lists its skills (id, name, description, tags, example prompts), declares its capabilities (streaming, push notifications), its input/output modes, and its supported interfaces — the URLs and protocol bindings (JSON-RPC, gRPC) it speaks. This is the discovery primitive: a client never needs out-of-band knowledge of an agent beyond its base URL. - Message — the unit of communication. A message has a role (
ROLE_USERorROLE_AGENT) and a list of parts: text, files, or structured data. Multimodality is built in, not bolted on. - Task — the unit of work. Sending a message creates a task with an id; the task moves through a lifecycle:
submitted→working→completed(orfailed,canceled,input-required). This is the core difference from a tool call: the remote side is running its own agent loop, which may take seconds or hours, stream partial results, and ask for clarification. - Artifact — the task's deliverable. A completed task can carry artifacts (documents, files, data) alongside its final message — the thing the agent produced, distinct from what it said.
- JSON-RPC transport — the default binding. The client POSTs JSON-RPC 2.0 requests (
message/send,tasks/get,tasks/cancel) to the URL from the agent card. Streaming uses server-sent events; everything else is plain request/response.
Notice what the protocol doesn't standardize: the agent's internals. Framework, model, tools, prompts, memory — all private. Two agents built by different teams on different stacks interoperate because the contract is only about tasks, messages, and cards.
Step 3 — Build FactFinder, your first A2A agent #
An A2A server has three layers: the agent card (who am I), the executor (what I do — your code), and the routes (the HTTP plumbing, provided by the SDK). Save this as factfinder_agent.py:
"""FactFinder: a minimal A2A agent that returns curated facts about a topic."""
import uuid
import uvicorn
from starlette.applications import Starlette
from a2a.server.agent_execution import AgentExecutor
from a2a.server.agent_execution.context import RequestContext
from a2a.server.events.event_queue import EventQueue
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
from a2a.server.tasks import InMemoryTaskStore
from a2a.types.a2a_pb2 import (
AgentCapabilities, AgentCard, AgentInterface, AgentSkill,
Message, Part, Role, Task, TaskState, TaskStatus,
)
# Deterministic knowledge base: no LLM, no API key, fully reproducible.
KNOWLEDGE = {
"photosynthesis": [
"Photosynthesis converts light energy into chemical energy stored as glucose.",
"It occurs in chloroplasts, driven by chlorophyll absorbing red and blue light.",
"Net equation: 6CO2 + 6H2O + light -> C6H12O6 + 6O2.",
],
"rust": [
"Rust is a systems language guaranteeing memory safety without a garbage collector.",
"Ownership, borrowing, and lifetimes are enforced at compile time.",
"Cargo is its build tool and package manager.",
],
}
class FactFinderExecutor(AgentExecutor):
"""Your agent logic lives here. The SDK handles everything else."""
async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
topic = context.get_user_input().strip().lower()
facts = KNOWLEDGE.get(
topic,
[f"No curated facts for '{topic}'.",
"FactFinder only knows about: " + ", ".join(sorted(KNOWLEDGE))],
)
body = "FACTS about '%s':\n" % topic + "\n".join(f"- {f}" for f in facts)
reply = Message(
message_id=uuid.uuid4().hex,
role=Role.ROLE_AGENT,
parts=[Part(text=body)],
)
# The task is the protocol's unit of work: hand the handler a
# completed Task and it takes care of the lifecycle bookkeeping.
task = Task(
id=context.task_id,
context_id=context.context_id,
status=TaskStatus(state=TaskState.TASK_STATE_COMPLETED, message=reply),
history=[context.message],
)
await event_queue.enqueue_event(task)
async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
task = Task(
id=context.task_id,
context_id=context.context_id,
status=TaskStatus(state=TaskState.TASK_STATE_CANCELED),
)
await event_queue.enqueue_event(task)
agent_card = AgentCard(
name="FactFinder",
description="Returns a short list of curated facts about a requested topic.",
version="1.0.0",
supported_interfaces=[
AgentInterface(url="http://localhost:10001/", protocol_binding="JSONRPC")
],
capabilities=AgentCapabilities(streaming=False, push_notifications=False),
default_input_modes=["text"],
default_output_modes=["text"],
skills=[
AgentSkill(
id="lookup-facts",
name="Look up facts",
description="Return curated facts about one topic.",
tags=["research", "facts"],
examples=["facts about photosynthesis", "facts about rust"],
)
],
)
request_handler = DefaultRequestHandler(
agent_executor=FactFinderExecutor(),
task_store=InMemoryTaskStore(),
agent_card=agent_card,
)
app = Starlette(routes=[
*create_jsonrpc_routes(request_handler, rpc_url="/"),
*create_agent_card_routes(agent_card),
])
if __name__ == "__main__":
uvicorn.run(app, host="127.0.0.1", port=10001, log_level="warning")Three things to notice. First, your code is only the AgentExecutor — about 30 lines. The request handler manages the task lifecycle, the task store remembers tasks, and the route factories expose the JSON-RPC endpoint and the agent card. Second, context.get_user_input() extracts the text from the incoming message, and context.task_id / context.context_id are generated by the handler when the client doesn't supply them. Third, the executor signals completion by enqueueing a Task whose status is TASK_STATE_COMPLETED — the handler streams or returns it to the client from there.
Run it:
python factfinder_agent.pyIn another terminal, fetch the agent card — the discovery primitive from Step 2, served live:
curl -s http://localhost:10001/.well-known/agent-card.json | python3 -m json.tool | head -30
# {
# "name": "FactFinder",
# "description": "Returns a short list of curated facts about a requested topic.",
# "supportedInterfaces": [
# {
# "url": "http://localhost:10001/",
# "protocolBinding": "JSONRPC"
# }
# ],
# "version": "1.0.0",
# "capabilities": {"streaming": false, "pushNotifications": false},
# "defaultInputModes": ["text"],
# "defaultOutputModes": ["text"],
# "skills": [
# {
# "id": "lookup-facts",
# ...
# }
# ],
# ...
# }That JSON is the whole pitch of A2A: any client, in any language, can learn everything it needs to work with your agent from this one document.
Step 4 — Talk to it: the client #
A client does three things: resolve the card, build requests from its own types, and read the response stream. Save this as client_demo.py and run it while FactFinder is up:
"""Minimal A2A client: resolve the card, send a message, print the result."""
import asyncio
import uuid
import httpx
from a2a.client.card_resolver import A2ACardResolver
from a2a.client.client import ClientConfig
from a2a.client.client_factory import ClientFactory
from a2a.types.a2a_pb2 import Message, Part, Role, SendMessageRequest
async def main() -> None:
http = httpx.AsyncClient()
# 1. Discovery: fetch and parse the agent card. The factory then picks
# a compatible transport from the card's supported interfaces.
card = await A2ACardResolver(http, "http://localhost:10001").get_agent_card()
print("resolved card:", card.name, "| skills:", [s.id for s in card.skills])
client = ClientFactory(ClientConfig(httpx_client=http)).create(card)
# 2. Send a message. The SDK POSTs a JSON-RPC message/send request.
request = SendMessageRequest(
message=Message(
message_id=uuid.uuid4().hex,
role=Role.ROLE_USER,
parts=[Part(text="photosynthesis")],
)
)
async for response in client.send_message(request):
# 3. Read the stream. A StreamResponse carries one of: task,
# message, status_update, artifact_update.
if response.HasField("task"):
task = response.task
state = task.status.state # TASK_STATE_COMPLETED == 3
print("task id:", task.id, "| state:", state)
msg = task.status.message
if msg and msg.parts:
print("agent reply:\n" + msg.parts[0].text)
await client.close()
asyncio.run(main())python client_demo.py
# resolved card: FactFinder | skills: ['lookup-facts']
# task id: 04ddfc90-4c4b-499d-a44e-d9f974c2342f | state: 3
# agent reply:
# FACTS about 'photosynthesis':
# - Photosynthesis converts light energy into chemical energy stored as glucose.
# - It occurs in chloroplasts, driven by chlorophyll absorbing red and blue light.
# - Net equation: 6CO2 + 6H2O + light -> C6H12O6 + 6O2.That output is real, captured from the run. A few details worth pocketing: send_message returns an async iterator even when the server doesn't stream — the protocol is stream-shaped by default, so a non-streaming server just yields one task event. State 3 is TASK_STATE_COMPLETED (the enum also defines SUBMITTED, WORKING, FAILED, CANCELED, INPUT_REQUIRED, REJECTED, AUTH_REQUIRED). And note the client never imported anything from the server file — the card was the entire contract.
One environment note: if you run this behind a corporate proxy and httpx throws URL-parsing errors on startup, construct the client with httpx.AsyncClient(trust_env=False) so it ignores the proxy variables for local addresses.
Step 5 — BriefWriter: an agent that delegates to another agent #
Now the payoff. BriefWriter is an A2A server to its own clients and an A2A client of FactFinder — the same SDK, both halves, in one process. When asked for a brief, it delegates the research step over the protocol, then assembles the result into an artifact:
"""BriefWriter: delegates research to FactFinder, then writes a brief.
Requires FactFinder on http://localhost:10001/ — run that first.
This file runs its own server on http://localhost:10002/.
"""
import uuid
import httpx
import uvicorn
from starlette.applications import Starlette
from a2a.client.card_resolver import A2ACardResolver
from a2a.client.client import ClientConfig
from a2a.client.client_factory import ClientFactory
from a2a.server.agent_execution import AgentExecutor
from a2a.server.agent_execution.context import RequestContext
from a2a.server.events.event_queue import EventQueue
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
from a2a.server.tasks import InMemoryTaskStore
from a2a.types.a2a_pb2 import (
AgentCapabilities, AgentCard, AgentInterface, AgentSkill, Artifact,
Message, Part, Role, SendMessageRequest,
Task, TaskState, TaskStatus,
)
FACTFINDER_URL = "http://localhost:10001"
async def fetch_facts(topic: str) -> str:
"""Act as an A2A client: ask FactFinder for facts about the topic."""
http = httpx.AsyncClient()
try:
card = await A2ACardResolver(http, FACTFINDER_URL).get_agent_card()
client = ClientFactory(ClientConfig(httpx_client=http)).create(card)
request = SendMessageRequest(
message=Message(
message_id=uuid.uuid4().hex,
role=Role.ROLE_USER,
parts=[Part(text=topic)],
)
)
async for response in client.send_message(request):
if response.HasField("task"):
task = response.task
if task.status.state == TaskState.TASK_STATE_COMPLETED:
return task.status.message.parts[0].text
raise RuntimeError("FactFinder returned no completed task")
finally:
await http.aclose()
class BriefWriterExecutor(AgentExecutor):
async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
topic = context.get_user_input().strip()
facts_text = await fetch_facts(topic) # <-- agent-to-agent delegation
brief = (
f"# Brief: {topic}\n\n"
f"{facts_text}\n\n"
"---\n"
"Researched by FactFinder via A2A. Written by BriefWriter."
)
artifact = Artifact(
artifact_id=uuid.uuid4().hex,
name=f"brief-{topic}.md",
description=f"A short brief about {topic}",
parts=[Part(text=brief)],
)
reply = Message(
message_id=uuid.uuid4().hex,
role=Role.ROLE_AGENT,
parts=[Part(text=f"Brief on '{topic}' is ready (see artifact).")],
)
task = Task(
id=context.task_id,
context_id=context.context_id,
status=TaskStatus(state=TaskState.TASK_STATE_COMPLETED, message=reply),
artifacts=[artifact],
history=[context.message],
)
await event_queue.enqueue_event(task)
async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
task = Task(
id=context.task_id,
context_id=context.context_id,
status=TaskStatus(state=TaskState.TASK_STATE_CANCELED),
)
await event_queue.enqueue_event(task)
agent_card = AgentCard(
name="BriefWriter",
description="Writes a short brief on a topic, delegating research to FactFinder over A2A.",
version="1.0.0",
supported_interfaces=[
AgentInterface(url="http://localhost:10002/", protocol_binding="JSONRPC")
],
capabilities=AgentCapabilities(streaming=False, push_notifications=False),
default_input_modes=["text"],
default_output_modes=["text"],
skills=[
AgentSkill(
id="write-brief",
name="Write brief",
description="Research a topic via FactFinder and return a written brief.",
tags=["writing", "research", "delegation"],
examples=["brief on photosynthesis", "brief on rust"],
)
],
)
request_handler = DefaultRequestHandler(
agent_executor=BriefWriterExecutor(),
task_store=InMemoryTaskStore(),
agent_card=agent_card,
)
app = Starlette(routes=[
*create_jsonrpc_routes(request_handler, rpc_url="/"),
*create_agent_card_routes(agent_card),
])
if __name__ == "__main__":
uvicorn.run(app, host="127.0.0.1", port=10002, log_level="warning")The delegation is exactly one function call — fetch_facts(topic) — but notice what it buys: BriefWriter knows nothing about FactFinder's framework, model, or knowledge base. It discovered the agent from its card at runtime and spoke only the protocol. Swap FactFinder tomorrow for a different team's agent, in another language, behind another company's firewall, and this code doesn't change.
Step 6 — Run the whole chain #
Start both servers (two terminals), then point the client at BriefWriter instead of FactFinder:
# terminal 1
python factfinder_agent.py # http://localhost:10001/
# terminal 2
python briefwriter_agent.py # http://localhost:10002/
# terminal 3 — same client_demo.py, but resolve BriefWriter:
# A2ACardResolver(http, "http://localhost:10002")# resolved card: BriefWriter | skills: ['write-brief']
# task id: a25794ee-0ae5-452e-b4dd-d1ff8ca12cfe | state: 3
# agent reply:
# Brief on 'rust' is ready (see artifact).The reply mentions an artifact — fetch it. The completed task carries a brief-rust.md artifact with the assembled brief, and you can re-fetch the task by id at any time with tasks/get:
# artifacts on the completed task:
# ARTIFACT: brief-photosynthesis.md
# # Brief: photosynthesis
#
# FACTS about 'photosynthesis':
# - Photosynthesis converts light energy into chemical energy stored as glucose.
# - It occurs in chloroplasts, driven by chlorophyll absorbing red and blue light.
# - Net equation: 6CO2 + 6H2O + light -> C6H12O6 + 6O2.
#
# ---
# Researched by FactFinder via A2A. Written by BriefWriter.
# re-fetching by id:
fetched = await client.get_task(GetTaskRequest(id=task_id))
# get_task state: 3 | artifacts: 1Trace what just happened across the wire: your client sent one message/send to BriefWriter. BriefWriter's executor opened its own client, resolved FactFinder's card, sent message/send to FactFinder, waited for the completed task, built an artifact, and completed its own task. Two independent agent processes, zero shared code, one protocol.
Which approach should you use? #
A2A is not the only way to connect agents. Pick by the shape of the relationship:
- Use A2A when the other side is an agent — something that runs its own loop, takes unpredictable time, and may stream, ask questions, or fail independently. Especially across teams, vendors, or trust boundaries: the agent card is a self-describing contract and neither side exposes internals.
- Use MCP when the other side is a tool or data source — deterministic, short-lived, shaped output. MCP is the vertical wire from an agent down to its tools; A2A is the horizontal wire between agents. They compose: one agent can use MCP tools internally while delegating to other agents over A2A.
- Use plain function calls when both sides live in the same process and latency matters. A2A's HTTP + task lifecycle is overhead you don't need for in-process subagents.
- Use raw REST when the remote side isn't agentic at all — a CRUD API doesn't need task states and artifacts.
The common mistake is wrapping an agent as an MCP tool to get "multi-agent" behavior. It works until the agent takes ten minutes, needs to stream progress, or must ask a clarifying question mid-run — exactly the lifecycle semantics A2A exists to provide.
Four production-hardening steps #
This tutorial optimizes for clarity. Before these agents touch real users, four changes matter:
- Turn on streaming. Set
streaming=TrueinAgentCapabilitiesand enqueueTaskStatusUpdateEvents as work progresses instead of one finalTask. Long-running agents that go silent look broken; streaming status updates are the difference. - Persist the task store.
InMemoryTaskStoreforgets everything on restart. The SDK's task-store interface is small — back it with Postgres or SQLite sotasks/getsurvives deploys and long-running tasks survive crashes. - Authenticate the card.
AgentCardcarriessecurity_schemesandsecurity_requirementsfields for exactly this: declare API keys, OAuth2, or mutual TLS on the card so clients know how to authenticate before the first call. An unsigned, unauthenticated card on the open internet is a prompt-injection drive-through. - Honor cancellation. The
cancel()method on your executor is wired totasks/cancel— implement it for real in long-running agents (check a flag in your loop, clean up, enqueueTASK_STATE_CANCELED). Clients will cancel; orphaned agent loops are how you burn money.
And the question everyone asks: where does the LLM go? Inside execute(). That method is your agent's entire world — call your model, run your tool loop, stream your thoughts as status updates, and enqueue the completed task when done. Everything around it (cards, transport, lifecycle, task storage) is already handled. The deterministic executors here exist so you can learn the protocol without a bill; the seam for intelligence is exactly one method wide.
The takeaway #
The interesting thing about this tutorial is how little of it is about AI. No prompts, no embeddings, no vector math — just two HTTP servers, a JSON discovery document, and a task lifecycle. That's the point the A2A designers got right: agent interoperability is an interface problem, not an intelligence problem, and interfaces are solved with boring, stable contracts.
Build one A2A agent this week — wrap something your team already runs, publish its card, and watch how quickly "can your agent just ask our agent?" stops being a three-sprint integration project.