Google A2A Agent-to-Agent Protocol in Python 2026 — let agents from different stacks talk to each other over a shared protocol, not a one-off webhook.
In 2026, multi-agent systems rarely live in one framework. You might run a LangGraph supervisor, a CrewAI research crew, and a vendor specialist agent in another cloud. Agent2Agent (A2A) is the open protocol (with Google's a2a-sdk for Python) that standardizes how those agents discover each other, exchange tasks, and stream results.
Quick mental model: MCP is how an agent calls tools and data. A2A is how an agent talks to another agent. Pair both when you go to production — see our notes on deploying production agentic systems.
What you will build
- An A2A server that publishes an
AgentCardand answers JSON-RPC messages - A small client that resolves the card and sends a user message
- A clear map of A2A vs MCP vs LangGraph/CrewAI
Install (Python 3.10+)
pip install "a2a-sdk[http-server]"
# optional: pip install "a2a-sdk[fastapi]" "a2a-sdk[grpc]"
The official package is a2a-sdk (Protocol Spec 1.0). Docs live at a2a-protocol.org; source at a2aproject/a2a-python.
Core concepts in 60 seconds
- AgentCard — public identity: name, skills, capabilities, and
supported_interfaces(JSON-RPC, HTTP+JSON, or gRPC URLs). - AgentSkill — what the agent can do (tags, examples, I/O modes).
- DefaultRequestHandler — protocol glue: tasks, streaming, cards.
- AgentExecutor — your business logic: read the message, update task state, emit artifacts.
1) Minimal A2A server (Starlette + JSON-RPC)
Save as agent_executor.py:
from a2a.helpers import (
get_message_text,
new_task_from_user_message,
new_text_message,
new_text_part,
)
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.tasks import TaskUpdater
from a2a.types import TaskState
class EchoAgent:
async def invoke(self, user_request: str) -> str:
return f"A2A echo: I received ({user_request})"
class EchoAgentExecutor(AgentExecutor):
def __init__(self) -> None:
self.agent = EchoAgent()
async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
if context.current_task:
task = context.current_task
else:
task = new_task_from_user_message(context.message)
await event_queue.enqueue_event(task)
updater = TaskUpdater(
event_queue=event_queue, task_id=task.id, context_id=task.context_id
)
await updater.update_status(
state=TaskState.TASK_STATE_WORKING,
message=new_text_message("Working..."),
)
query = get_message_text(context.message) or ""
result = await self.agent.invoke(query) if query else "No text input."
await updater.add_artifact(
parts=[new_text_part(text=result, media_type="text/plain")]
)
await updater.update_status(
state=TaskState.TASK_STATE_COMPLETED,
message=new_text_message("Done"),
)
async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
raise NotImplementedError("Cancel is not supported in this demo.")
Then server.py:
import uvicorn
from starlette.applications import Starlette
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 import (
AgentCapabilities,
AgentCard,
AgentInterface,
AgentSkill,
)
from agent_executor import EchoAgentExecutor
skill = AgentSkill(
id="echo_bot",
name="Echo Bot",
description="Acknowledges a client request and returns an echo reply.",
input_modes=["text/plain"],
output_modes=["text/plain"],
tags=["a2a", "echo"],
examples=["hi", "ping"],
)
card = AgentCard(
name="PyInns Echo Agent",
description="Minimal A2A JSON-RPC agent for local demos",
version="0.1.0",
default_input_modes=["text/plain"],
default_output_modes=["text/plain"],
capabilities=AgentCapabilities(streaming=True, extended_agent_card=False),
supported_interfaces=[
AgentInterface(
protocol_binding="JSONRPC",
url="http://127.0.0.1:9999",
protocol_version="1.0",
)
],
skills=[skill],
)
handler = DefaultRequestHandler(
agent_executor=EchoAgentExecutor(),
task_store=InMemoryTaskStore(),
agent_card=card,
)
routes = []
routes.extend(create_agent_card_routes(card))
routes.extend(create_jsonrpc_routes(handler, "/"))
app = Starlette(routes=routes)
if __name__ == "__main__":
uvicorn.run(app, host="127.0.0.1", port=9999)
Run: python server.py. The agent card is discoverable from that base URL; JSON-RPC handles message send/stream.
2) Client: resolve the card and send a message
import asyncio
import httpx
from a2a.client import A2ACardResolver, ClientConfig, create_client
from a2a.helpers import new_text_message
from a2a.types import Role, SendMessageRequest
async def main() -> None:
async with httpx.AsyncClient() as httpx_client:
resolver = A2ACardResolver(
httpx_client=httpx_client,
base_url="http://127.0.0.1:9999",
)
card = await resolver.get_agent_card()
client = await create_client(
agent=card,
client_config=ClientConfig(streaming=False),
)
message = new_text_message("Hello from a peer agent", role=Role.ROLE_USER)
request = SendMessageRequest(message=message)
async for chunk in client.send_message(request):
print(chunk)
await client.close()
if __name__ == "__main__":
asyncio.run(main())
For streaming demos, set ClientConfig(streaming=True) and keep consuming the async iterator.
A2A vs MCP vs orchestration frameworks
| Layer | Job | Use when |
|---|---|---|
| MCP | Tools, resources, prompts | Agent needs DB, files, APIs |
| A2A | Agent to agent tasks | Cross-team / cross-vendor agents |
| LangGraph / CrewAI | Orchestration inside your app | Graphs, roles, checkpoints — see multi-agent collaboration patterns |
A practical 2026 stack: LangGraph (or CrewAI) owns the workflow, MCP exposes tools, A2A exposes your agent to other organizations' agents without sharing private graph state.
Production tips
- Publish a real HTTPS
AgentInterface.urland keep the card versioned. - Prefer durable task stores (SQL extras on
a2a-sdk) overInMemoryTaskStoreoutside demos. - Authenticate before serving an extended agent card with privileged skills.
- Observe task latency and failure rates the same way you would for any agent API.
Wrap-up
A2A gives Python teams a standard way to offer and call agent capabilities across process and vendor boundaries. Start with the echo server above, then swap EchoAgent.invoke for your real model and tools (and keep MCP for the tool plane). When you are ready to ship, harden deploy, auth, and monitoring using the production checklist linked above.