Most of the MCP articles in this repo connect an agent downward to tools: an agent is a client, and a server exposes functions it can call. That is the vertical axis. The Agent2Agent (A2A) protocol covers the horizontal axis, where one agent talks to another agent as a peer. Instead of “call this function,” the message is “here is a task, you handle it and report back.” The two protocols are complementary: an agent can be an MCP client (reaching down to tools) and an A2A server (answering peers sideways) at the same time.
A2A is an open protocol, originally from Google and now hosted by the Linux
Foundation. It defines how an agent advertises itself with an Agent Card
(served at /.well-known/agent-card.json), and how peers send it work over
JSON-RPC. A calling agent discovers a remote agent, reads its card to learn what
it can do, and sends a message. The remote agent runs and streams back results.
Crucially, the peer is a black box: you talk to its published skills, not its
internals, so the two agents can be different frameworks, languages, or vendors.
In this tutorial you will build both halves on macOS with uv:
- A Currency Agent served over A2A. It advertises a
convert_currencyskill in its Agent Card and answers conversion requests. The logic is deterministic (a small rate table), so the whole system runs with no API key and no LLM. - A client that plays the role of a coordinator agent: it discovers the Currency Agent from its card and delegates questions to it over A2A.
- A
pytestsuite that drives the real A2A app in-process (full protocol, no open port) plus the pure conversion logic. - A help-first
Makefile.
Version note. This article pins the
a2a-sdk0.3.x line, which ships the ergonomicA2AStarletteApplication/ pydanticAgentCardAPI used by A2A’s official Python quickstart. The 1.x line is a newer, protobuf-based rewrite with a different surface; pinning>=0.3,<0.4keeps this tutorial reproducible. The A2A protocol concepts (Agent Card,message/send) are the same across both.
Prerequisites
- macOS 13+ with Homebrew (brew.sh).
- uv 0.5+ —
brew install uv; verify withuv --version. - Python 3.11+ (managed by
uvvia.python-version). - Xcode Command Line Tools (
xcode-select --install) formake. - Basic familiarity with
async/await. No API key is needed anywhere.
Everything runs through uv and make. Nothing reaches the public internet:
the client talks to a server on 127.0.0.1, and the tests use an in-process
transport.
Step 1: Scaffold and lock down hygiene
Create the workspace and its .gitignore first, before any other file. This
project needs no secrets, but keeping the habit means a stray .env can never
be committed later.
Create the files
mkdir -p a2a-agent-to-agent-protocol-macos/tests
cd a2a-agent-to-agent-protocol-macos
echo '3.11' > .python-version
touch tests/__init__.py
touch .gitignore
Add the code: .gitignore
# Python
__pycache__/
*.py[cod]
.venv/
# uv
.uv/
# Secrets (none required here, but keep the habit)
.env
# Tooling caches
.pytest_cache/
.ruff_cache/
.mypy_cache/
# OS / editor noise
.DS_Store
*.log
Detailed breakdown
.python-versionpins the interpreteruvprovisions for this project (3.11), so contributors resolve the same environment..venv/and the tool caches (.pytest_cache/, etc.) are regenerated byuvandpytest, so they are ignored rather than committed.tests/__init__.pymarks the test directory as a package. Creating it now means Step 1 leaves a clean, importable layout before any code exists.
Step 2: Initialize the project and add dependencies
Turn the folder into a managed uv project, then add the A2A SDK and the two
supporting libraries.
Initialize and install
uv init --name a2a-currency --python 3.11
rm -f main.py
uv add "a2a-sdk[http-server]>=0.3,<0.4" uvicorn httpx
uv add --dev pytest pytest-asyncio
Add the code: pyproject.toml (generated, shown for reference)
[project]
name = "a2a-currency"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
"a2a-sdk[http-server]>=0.3,<0.4",
"httpx>=0.28.1",
"uvicorn>=0.51.0",
]
[dependency-groups]
dev = [
"pytest>=9.1.1",
"pytest-asyncio>=1.4.0",
]
Detailed breakdown
a2a-sdk[http-server]is the key dependency. The barea2a-sdkpackage ships the types and client, but the[http-server]extra pulls instarletteandsse-starlette, whichA2AStarletteApplicationneeds to serve the agent. Omitting the extra raises anImportErrorthe moment you build the app, so install it up front.uvicornis the ASGI server that actually runs the app on a port.httpxis the async HTTP client the A2A client uses to fetch the Agent Card and send messages. The SDK depends on it internally too.pytest-asynciolets the test suiteawaita real A2A round-trip. It is a dev dependency, kept out of the runtime set.rm -f main.pydeletes the sample fileuv initgenerates; this project uses named modules instead.
Step 3: Write the agent’s logic and its Agent Card
This module is the heart of the server. It holds three things: a pure convert
function (easy to test), an AgentExecutor that adapts that function to A2A, and
a factory for the Agent Card that advertises the agent to peers.
Create the file
touch currency_agent.py
Add the code: currency_agent.py
"""A deterministic A2A specialist: a currency-conversion agent.
The conversion logic is a pure function so it is trivial to test, and the
`AgentExecutor` is the thin A2A adapter around it. No LLM or API key is
involved, which keeps the whole tutorial reproducible and free to run.
"""
import re
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.types import AgentCapabilities, AgentCard, AgentSkill
from a2a.utils import new_agent_text_message
# Value of 1 unit of each currency in USD terms is derived from this table,
# which lists how many units of the currency one USD buys.
RATES = {"USD": 1.0, "EUR": 0.92, "JPY": 156.0, "GBP": 0.79}
# Matches "100 USD to EUR", "convert 50 usd in jpy", etc.
_PATTERN = re.compile(
r"(\d+(?:\.\d+)?)\s*([A-Za-z]{3})\s*(?:to|in|into)\s*([A-Za-z]{3})"
)
def convert(text: str) -> str:
"""Parse a free-text conversion request and return a one-line answer."""
match = _PATTERN.search(text)
if not match:
return "Ask me like: '100 USD to EUR'."
amount, src, dst = float(match.group(1)), match.group(2).upper(), match.group(3).upper()
if src not in RATES or dst not in RATES:
return f"I don't have a rate for {src} or {dst}."
usd = amount / RATES[src]
converted = round(usd * RATES[dst], 2)
return f"{amount:g} {src} = {converted:g} {dst}"
class CurrencyAgentExecutor(AgentExecutor):
"""Adapts the pure `convert` function to the A2A executor interface."""
async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
user_text = context.get_user_input()
answer = convert(user_text)
await event_queue.enqueue_event(new_agent_text_message(answer))
async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
raise Exception("Currency conversion is instantaneous; nothing to cancel.")
def build_agent_card(url: str = "http://localhost:9999/") -> AgentCard:
"""Build the public Agent Card that advertises this agent to peers."""
skill = AgentSkill(
id="convert_currency",
name="Convert currency",
description="Convert an amount from one currency to another.",
tags=["finance", "currency"],
examples=["100 USD to EUR", "convert 50 USD in JPY"],
)
return AgentCard(
name="Currency Agent",
description="Converts amounts between USD, EUR, JPY, and GBP.",
url=url,
version="1.0.0",
default_input_modes=["text"],
default_output_modes=["text"],
capabilities=AgentCapabilities(streaming=False),
skills=[skill],
)
Detailed breakdown
convertis a pure function: text in, one line of text out. It normalizes currency codes, rejects unparseable input, and computes the result by pivoting through USD (amount / RATES[src]gives USD, then* RATES[dst]). Keeping the arithmetic here, free of any A2A types, is what makes the logic unit-testable in isolation.CurrencyAgentExecutoris the A2A contract. Every A2A agent implementsAgentExecutorwith two coroutines.executereads the caller’s text withcontext.get_user_input(), computes the answer, and pushes it onto theevent_queuewithnew_agent_text_message(...). The queue is how the SDK streams events back to the caller; this agent enqueues exactly one message and finishes.cancelis required by the interface; because a conversion is instantaneous, there is nothing to interrupt, so it raises.build_agent_cardproduces the agent’s public description. TheAgentSkillis the advertised capability: anid, human-readablename/description,tags, andexamplesthat tell a calling agent how to phrase a request. TheAgentCardwraps the skills with the agent’sname,url(where peers reach it),version, supported input/output modes, andcapabilities. Settingstreaming=Falsetells callers this agent returns a single message rather than a stream of incremental updates.- The
urlis a parameter so the same card can be built for a real port (the server) or a dummy host (the tests), which you will use in Step 6.
Step 4: Serve the agent over A2A
Wire the executor and card into an A2A application and run it with uvicorn.
Factoring the wiring into build_app() lets the test suite reuse the exact same
app without opening a port.
Create the file
touch server.py
Add the code: server.py
"""Serve the Currency Agent over the A2A protocol.
`build_app()` wires the executor and Agent Card into an A2A Starlette app so it
can be reused by the test suite with an in-process transport (no open port).
"""
import uvicorn
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from currency_agent import CurrencyAgentExecutor, build_agent_card
HOST = "127.0.0.1"
PORT = 9999
def build_app():
"""Return the ASGI app that speaks A2A for the Currency Agent."""
handler = DefaultRequestHandler(
agent_executor=CurrencyAgentExecutor(),
task_store=InMemoryTaskStore(),
)
card = build_agent_card(url=f"http://{HOST}:{PORT}/")
return A2AStarletteApplication(agent_card=card, http_handler=handler).build()
if __name__ == "__main__":
uvicorn.run(build_app(), host=HOST, port=PORT, log_level="warning")
Detailed breakdown
DefaultRequestHandleris the SDK’s standard request handler. It takes theagent_executor(your logic) and atask_store. A2A models longer jobs as tasks with state;InMemoryTaskStorekeeps that state in memory, which is right for a stateless demo agent. A production agent would swap in a persistent store.A2AStarletteApplication(...).build()assembles the ASGI app: it mounts the JSON-RPC endpoint at/and publishes the Agent Card at/.well-known/agent-card.json. The returned object is a standard Starlette app, so anything that runs ASGI can serve it.build_app()is separate from theuvicorn.runcall on purpose. The__main__block runs the app on a real port for interactive use; the tests importbuild_appand mount the same app on an in-process transport. Neither path duplicates the wiring.log_level="warning"quiets uvicorn’s per-request logs so the client’s output stays readable in the terminal.
Step 5: Run the server and inspect the Agent Card
Start the server so you can see what it advertises before writing a client.
Run the server
uv run python server.py
Leave it running. In a second terminal, fetch the Agent Card the way any peer would discover this agent:
curl -s http://127.0.0.1:9999/.well-known/agent-card.json | uv run python -m json.tool
You should see the card, including the advertised skill:
{
"capabilities": {
"streaming": false
},
"defaultInputModes": [
"text"
],
"defaultOutputModes": [
"text"
],
"description": "Converts amounts between USD, EUR, JPY, and GBP.",
"name": "Currency Agent",
"preferredTransport": "JSONRPC",
"protocolVersion": "0.3.0",
"skills": [
{
"description": "Convert an amount from one currency to another.",
"examples": [
"100 USD to EUR",
"convert 50 USD in JPY"
],
"id": "convert_currency",
"name": "Convert currency",
"tags": [
"finance",
"currency"
]
}
],
"url": "http://127.0.0.1:9999/",
"version": "1.0.0"
}
You can also send a raw A2A message/send call to confirm the JSON-RPC endpoint
works end to end:
curl -s http://127.0.0.1:9999/ -H 'content-type: application/json' \
-d '{"jsonrpc":"2.0","id":"1","method":"message/send","params":{"message":{"role":"user","messageId":"m1","parts":[{"kind":"text","text":"10 USD to GBP"}]}}}' \
| uv run python -m json.tool
The result carries the agent’s reply as a text part:
{
"id": "1",
"jsonrpc": "2.0",
"result": {
"kind": "message",
"messageId": "<random-uuid>",
"parts": [
{
"kind": "text",
"text": "10 USD = 7.9 GBP"
}
],
"role": "agent"
}
}
Detailed breakdown
preferredTransport: "JSONRPC"andprotocolVersionare filled in by the SDK, not by your card factory. They tell callers how to talk to the agent and which A2A revision it speaks, so different SDK versions can interoperate.- The
messageIdis a fresh UUID on every call, so the exact value in your output will differ from the placeholder above. Theroleflips to"agent"on the reply. Everything else (10 USD = 7.9 GBP, since10 * 0.79 = 7.9) is deterministic. - Hand-driving the endpoint with
curlshows there is nothing magic here: A2A is JSON-RPC over HTTP with a discovery document. The client in the next step just automates the discover-then-send dance.
Step 6: Write the client (a coordinator agent)
The client is the other agent in the conversation. It discovers the Currency Agent from its card, builds a transport-appropriate client, and delegates questions over A2A.
Create the file
touch client.py
Add the code: client.py
"""A minimal A2A client (a stand-in coordinator agent).
It discovers the Currency Agent by fetching its Agent Card, then delegates a
few questions over A2A and prints the replies. This is the "agent talking to
another agent" half of the system.
"""
import asyncio
import httpx
from a2a.client import (
A2ACardResolver,
ClientConfig,
ClientFactory,
create_text_message_object,
)
from a2a.types import Message
BASE_URL = "http://127.0.0.1:9999"
async def ask(text: str) -> str:
"""Send one question to the remote agent and collect its text reply."""
async with httpx.AsyncClient(timeout=30) as httpx_client:
# 1. Discover the peer from its published Agent Card.
card = await A2ACardResolver(httpx_client, base_url=BASE_URL).get_agent_card()
# 2. Build a transport-appropriate client from the card.
client = ClientFactory(
ClientConfig(httpx_client=httpx_client, streaming=False)
).create(card)
# 3. Send a user message and gather the agent's reply parts.
message = create_text_message_object(content=text)
replies: list[str] = []
async for event in client.send_message(message):
if isinstance(event, Message):
replies.append(
"".join(p.root.text for p in event.parts if p.root.kind == "text")
)
else: # (Task, update) tuple — unused by this non-streaming agent
task, _ = event
replies.append(f"[task {task.status.state}]")
return "\n".join(replies)
async def main() -> None:
for question in ["100 USD to EUR", "convert 50 USD in JPY", "hello there"]:
answer = await ask(question)
print(f"> {question}\n {answer}")
if __name__ == "__main__":
asyncio.run(main())
Detailed breakdown
A2ACardResolver(...).get_agent_card()performs discovery: it fetches/.well-known/agent-card.jsonfromBASE_URLand returns a typedAgentCard. A real coordinator would inspect the card’s skills here to decide whether this agent can handle the task.ClientFactory(ClientConfig(...)).create(card)builds a client configured for the transport the card advertises. Passing the sharedhttpx_clientreuses one connection pool;streaming=Falsematches the agent’s non-streaming capability. The factory pattern is what lets the same client code talk to a gRPC-based or REST-based agent by reading the card instead of hardcoding a transport.send_messagereturns an async iterator, because A2A supports streaming agents that emit many events. Each yielded item is either aMessage(a direct reply, this agent’s case) or a(Task, update)tuple for long-running work. The loop pulls the text out of each message part (p.root.textwherep.root.kind == "text") and ignores the task branch, which this non-streaming agent never takes.create_text_message_object(content=...)builds therole="user"message the SDK sends. This is the client-side mirror ofnew_agent_text_messageon the server.
Step 7: Run the two agents together
With the server still running from Step 5, run the client in the second terminal:
uv run python client.py
Expected output:
> 100 USD to EUR
100 USD = 92 EUR
> convert 50 USD in JPY
50 USD = 7800 JPY
> hello there
Ask me like: '100 USD to EUR'.
The first two questions round-trip through the A2A protocol and come back
converted; the third exercises the agent’s fallback when it cannot parse a
request. Stop the server with Ctrl+C when you are done.
Detailed breakdown
100 USD = 92 EURbecause100 * 0.92 = 92;50 USD = 7800 JPYbecause50 * 156.0 = 7800. The%gformatting drops trailing zeros, so you see92, not92.00.- The
hello thereline never matches the regex, so the agent returns its usage hint. That the server decides this (not the client) is the point of A2A: the caller delegates the whole task and trusts the peer’s judgment. - Each
askcall opens its ownhttpx.AsyncClient, rediscovers the card, and sends one message. In a real coordinator you would discover once and reuse the client; the per-call structure here keeps the example self-contained.
Step 8: Test the agent without a network
You can validate the pure logic and a full A2A round-trip without opening a port,
using httpx’s in-process ASGI transport. That drives the real app object
directly, so the test covers Agent Card discovery and message/send for real.
Create the files
touch pytest.ini
touch tests/test_currency_agent.py
Add the code: pytest.ini
[pytest]
testpaths = tests
asyncio_mode = auto
Add the code: tests/test_currency_agent.py
"""Tests for the currency agent: the pure logic and a real A2A round-trip.
The round-trip test drives the actual A2A app over an in-process ASGI
transport, so it exercises the full protocol (Agent Card discovery + a
`message/send` call) without opening a port or hitting the network.
"""
import sys
from pathlib import Path
import httpx
import pytest
from a2a.client import (
A2ACardResolver,
ClientConfig,
ClientFactory,
create_text_message_object,
)
from a2a.types import Message
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from currency_agent import build_agent_card, convert # noqa: E402
from server import build_app # noqa: E402
def test_convert_known_pairs():
assert convert("100 USD to EUR") == "100 USD = 92 EUR"
assert convert("convert 50 USD in JPY") == "50 USD = 7800 JPY"
assert convert("10 USD into GBP") == "10 USD = 7.9 GBP"
def test_convert_unparseable_input():
assert convert("hello there").startswith("Ask me")
def test_convert_unknown_currency():
assert "don't have a rate" in convert("100 USD to XYZ")
def test_agent_card_advertises_the_skill():
card = build_agent_card()
assert card.name == "Currency Agent"
assert [s.id for s in card.skills] == ["convert_currency"]
@pytest.mark.asyncio
async def test_end_to_end_over_a2a():
app = build_app()
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
card = await A2ACardResolver(hc, base_url="http://test").get_agent_card()
assert card.name == "Currency Agent"
client = ClientFactory(
ClientConfig(httpx_client=hc, streaming=False)
).create(card)
message = create_text_message_object(content="10 USD to GBP")
replies: list[str] = []
async for event in client.send_message(message):
if isinstance(event, Message):
replies.append(
"".join(p.root.text for p in event.parts if p.root.kind == "text")
)
assert replies == ["10 USD = 7.9 GBP"]
Detailed breakdown
asyncio_mode = autoinpytest.iniletspytest-asynciorunasync deftests without decorating each one;testpaths = testsscopes discovery to thetests/directory.- The
sys.path.insertline adds the project root to the import path so the test file can importcurrency_agentandserverwhen run fromtests/. The# noqa: E402comments acknowledge that those imports intentionally follow the path setup. - The three
converttests cover the deterministic branches: known pairs, unparseable input, and an unknown currency. Becauseconvertis pure, they run instantly with no A2A machinery. test_agent_card_advertises_the_skillchecks the card contract, that the agent still advertises exactly theconvert_currencyskill, catching an accidental rename before a peer ever fails to discover it.test_end_to_end_over_a2ais the important one.httpx.ASGITransport(app=app)routes requests straight into the app object in memory, soA2ACardResolverandsend_messageexercise the genuine protocol path (discovery, JSON-RPC, executor, event queue) without a socket. It asserts the reply is exactly10 USD = 7.9 GBP.
Run the tests
uv run pytest -v
Expected output:
tests/test_currency_agent.py::test_convert_known_pairs PASSED [ 20%]
tests/test_currency_agent.py::test_convert_unparseable_input PASSED [ 40%]
tests/test_currency_agent.py::test_convert_unknown_currency PASSED [ 60%]
tests/test_currency_agent.py::test_agent_card_advertises_the_skill PASSED [ 80%]
tests/test_currency_agent.py::test_end_to_end_over_a2a PASSED [100%]
Five passing tests, no network, no API key.
Step 9: Add a Makefile with a help-first default
A Makefile makes the workflow discoverable. Running plain make must print a
help screen listing every target.
Create the file
touch Makefile
Add the code: Makefile
.DEFAULT_GOAL := help
.PHONY: help install serve ask test
help: ## Show this help screen
@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) \
| awk 'BEGIN {FS = ":.*?## "}; {printf " \033[36m%-10s\033[0m %s\n", $$1, $$2}'
install: ## Sync runtime and dev dependencies
uv sync
serve: ## Run the Currency Agent A2A server on 127.0.0.1:9999
uv run python server.py
ask: ## Run the client against a running server (start `make serve` first)
uv run python client.py
test: ## Run the test suite (pure logic + in-process A2A round-trip)
uv run pytest -v
Detailed breakdown
.DEFAULT_GOAL := helpmakes baremakerun thehelptarget, satisfying the requirement thatmakewith no arguments prints a help screen.- The
helptarget scans theMakefilefor targets annotated with a## descriptioncomment and prints them in aligned, colored columns. Any new annotated target shows up automatically. serveandaskare two halves of one demo: startmake servein one terminal, thenmake askin another. The help text says so, because the client fails fast if nothing is listening on the port..PHONYdeclares these are commands, not files, somakealways runs them.
Validate the default target
make
Confirm the output lists help, install, serve, ask, and test, each with
its description.
A2A and MCP, side by side
Now that both protocols are in play, the division of labor is concrete:
- MCP is agent-to-tool (vertical). An agent is the client; a server exposes typed tools, resources, and prompts that the agent invokes. The agent drives; the tool is passive. Use it to give one agent capabilities.
- A2A is agent-to-agent (horizontal). Both sides are agents. A caller delegates a task, and the remote agent decides how to fulfill it, possibly running its own model, its own tools, or its own sub-agents. The peer is autonomous; you talk to its advertised skills, not its internals.
They compose. The Currency Agent here could itself be an MCP client, calling a
live exchange-rate MCP server to replace the static RATES table, while still
presenting an A2A face to the coordinator. From the caller’s side nothing
changes: it discovers the Agent Card and sends a message. That layering, A2A
across the top for delegation, MCP underneath for tools, is the common shape for
multi-agent systems.
Troubleshooting
ImportError: Packages 'starlette' and 'sse-starlette' are required— you installeda2a-sdkwithout the server extra. Runuv add "a2a-sdk[http-server]>=0.3,<0.4".httpx.ConnectErrorfrom the client — the server is not running. Startuv run python server.py(ormake serve) in another terminal first, and confirm it is on127.0.0.1:9999.ModuleNotFoundError: No module named 'currency_agent'— run commands from the project root, and keepserver.py,client.py, andcurrency_agent.pyin the same directory. The test suite adds the root tosys.pathitself.pytestcollects zero tests, or async tests are skipped — confirmpytest.inihasasyncio_mode = autoand thatpytest-asynciois installed (uv add --dev pytest-asyncio).- A different
a2a-sdkAPI than shown — you resolved a 1.x release. This article targets the 0.3.x line; pin"a2a-sdk[http-server]>=0.3,<0.4"and runuv sync.
Recap
You built both sides of an agent-to-agent conversation:
- A Currency Agent served over the A2A protocol, advertising a
convert_currencyskill in an Agent Card at/.well-known/agent-card.json, with anAgentExecutorwrapping deterministic logic. - A client that discovers the agent from its card and delegates questions
over A2A, handling the
send_messageevent stream. - A
pytestsuite that drives the real A2A app in-process (full protocol, no port) plus the pure conversion logic. - A help-first
Makefile, and a clear picture of how A2A (agent-to-agent) and MCP (agent-to-tool) compose in one system.
Next improvements
- Make the agent stream by setting
capabilities=AgentCapabilities(streaming=True)and enqueuingTaskStatusUpdateEvents, so the caller sees incremental progress on longer jobs. - Back the rates with a live MCP server so the A2A agent is itself an MCP client, demonstrating both axes in one process.
- Give the coordinator real routing by inspecting each discovered card’s
skills and picking an agent per request, instead of hardcoding one
BASE_URL. - Add authentication with A2A security schemes so only authorized peers can call the agent.
Sources
- A2A protocol site and specification
- a2aproject/a2a-python on GitHub
- A2A Python quickstart / helloworld example
- Model Context Protocol — the complementary agent-to-tool protocol