a2a-protocol-python

name	A2A Protocol (Python)
description	This skill should be used when the user asks to "create an A2A server", "build an A2A client", "implement agent-to-agent protocol", "use the a2a-sdk", "expose an agent over A2A", "connect to an A2A agent", "write an AgentExecutor", "publish an AgentCard", or mentions a2a-protocol.org or the a2aproject Python SDK. Provides Python patterns and ready-to-copy templates for both A2A server and client implementations using `a2a-sdk`.
version	0.1.0

A2A Protocol — Python Server & Client

This skill provides procedural knowledge for building agent-to-agent (A2A) systems in Python using the official a2a-sdk from a2aproject/a2a-python. It covers the two halves of any A2A integration: servers (an agent that exposes itself over the wire) and clients (callers that discover, message, and stream from remote agents).

When to Use This Skill

Trigger whenever the user wants to:

• Wrap an existing Python agent (LangGraph / ADK / CrewAI / plain function) so other agents can call it.
• Build a client that talks to an A2A endpoint (fetch its card, send messages, consume streamed events).
• Connect multiple agents into an orchestrated workflow over A2A.
• Debug AgentCard, AgentExecutor, EventQueue, Task, Message, or transport (JSON-RPC / gRPC / REST) issues.

Mental Model

A2A communication has three core artifacts:

Artifact	Role	Lives in
AgentCard	Public manifest — name, version, transports, skills, capabilities. Discovered via the well-known path /.well-known/agent-card.json.	server publishes, client resolves
AgentExecutor	Server-side adapter that turns incoming RequestContext + user Message into events written to an EventQueue (status updates, artifacts, completion).	server
Client (ClientFactory.create(card))	Sends Message requests, iterates async events back. Streaming vs non-streaming controlled by ClientConfig.	client

Servers run as ASGI apps (Starlette/FastAPI) on top of a DefaultRequestHandler wired to the executor and a TaskStore. Clients use httpx.AsyncClient + A2ACardResolver to load the card, then ClientFactory to create a transport-specific client.

Setup

Always use Python 3.10+. Install the SDK with the extras the project actually needs:

# Core (JSON-RPC over HTTP via Starlette)
pip install "a2a-sdk[http-server]"

# Add gRPC transport
pip install "a2a-sdk[grpc]"

# Everything (HTTP, gRPC, telemetry, etc.)
pip install "a2a-sdk[all]"

Use load_dotenv() at the top of any module reading .env (per project convention). For MCP-backed agents inside an A2A server, follow the project's standing rule: import from mcp.server.fastmcp import FastMCP and serve over streamable HTTP.

Core Workflows

Workflow A — Build an A2A Server

To expose any Python agent over A2A, follow these steps in order:

1. Define the business agent. Plain async class with one or more methods that take a string (or structured input) and return text or structured output. Keep agent logic separate from the executor — the executor is just an adapter.
2. Implement AgentExecutor. Subclass a2a.server.agent_execution.AgentExecutor, implement async def execute(context, event_queue) and async def cancel(...). Inside execute:
   • Pull or create a task: task = context.current_task or new_task(context.message); enqueue it.
   • Optionally enqueue intermediate TaskStatusUpdateEvent(final=False, status=TaskStatus(state=TaskState.working, ...)) for streaming progress.
   • Run the agent, then enqueue a TaskArtifactUpdateEvent carrying the result artifact.
   • Finish with TaskStatusUpdateEvent(final=True, status=TaskStatus(state=TaskState.completed)).
3. Describe the agent with AgentSkill + AgentCard. Each AgentSkill has id, name, description, tags, examples. The AgentCard carries a single url string (the externally reachable endpoint), default_input_modes / default_output_modes, capabilities=AgentCapabilities(streaming=True, push_notifications=False), and — if you expose a separate authenticated card — supports_authenticated_extended_card=True.
4. Wire request handler + ASGI app. Build DefaultRequestHandler(agent_executor=..., task_store=InMemoryTaskStore()), then A2AStarletteApplication(agent_card=card, http_handler=request_handler). Run with uvicorn.run(server.build(), host=..., port=...).

A complete minimal server lives at examples/helloworld_server.py. A streaming + LLM-style executor lives at examples/streaming_agent_executor.py. Adapt by changing the agent class and the skill/card metadata; the executor + server scaffolding stays the same.

Workflow B — Build an A2A Client

To call a remote A2A agent, follow these steps in order:

1. Resolve the card. Open an httpx.AsyncClient, instantiate A2ACardResolver(httpx_client=..., base_url=...), then card = await resolver.get_agent_card(). This hits /.well-known/agent-card.json by default.
2. Construct a client. client = ClientFactory(config=ClientConfig(streaming=False)).create(card). Pass streaming=True to consume incremental events as they're produced.
3. Send a message. Build a Message(role=Role.user, parts=[Part(root=TextPart(text=...))], message_id=uuid4().hex) and pass it directly to client.send_message(message) — the SDK wraps the JSON-RPC envelope internally. Iterate the async generator: async for chunk in client.send_message(message): task = chunk[0] if isinstance(chunk, tuple) else chunk.
4. Close. Close the underlying httpx.AsyncClient (typically via async with httpx.AsyncClient() — the Client object borrows it and does not need a separate close).

A complete minimal client lives at examples/helloworld_client.py. To fetch the optional authenticated extended card, check card.supports_authenticated_extended_card and call await client.get_extended_agent_card() (no arguments).

Workflow C — Multi-Agent Orchestration

When one A2A agent must call another (a common pattern in samples like the healthcare concierge), the orchestrator is both a server (to its caller) and a client (to its downstream agents). Inside its AgentExecutor.execute, instantiate downstream clients (cache them across requests if the executor is long-lived), forward the user's message, and stream the downstream events back through event_queue. See references/advanced-patterns.md for the orchestration recipe and a downstream-call pattern.

Key Decisions

When building a server, consult this table before writing code:

Question	Default	When to change
Streaming or one-shot?	Streaming on (AgentCapabilities(streaming=True))	Turn off only for hard one-shot tools to simplify the executor.
Task store	InMemoryTaskStore()	Use DatabaseTaskStore (PostgreSQL/MySQL/SQLite) for persistence across restarts.
Transport	JSON-RPC over HTTP via A2AStarletteApplication	Use gRPC for low-latency internal traffic (pip install "a2a-sdk[grpc]").
Public vs extended card	One public card	Set AgentCard(supports_authenticated_extended_card=True) and serve a richer card from the authenticated endpoint for skills only authenticated callers can see.
Push notifications	Off	Enable for long-running tasks where polling/streaming aren't viable.

Common Pitfalls

• Mixing type imports. With a2a-sdk[http-server] (the default), all types are pydantic models in the single module a2a.types — AgentCard, AgentSkill, AgentCapabilities, Message, Part, Role, Task, TaskState, TaskStatus, TaskStatusUpdateEvent, TaskArtifactUpdateEvent, etc. The a2a.types.a2a_pb2 protobuf module only exists when the [grpc] extra is installed. Don't invent imports — they all come from a2a.types.
• Wrong enum casing. Values are lowercase strings in this SDK: Role.user (not ROLE_USER), TaskState.working, TaskState.completed, TaskState.failed, TaskState.canceled (one L), TaskState.input_required.
• Forgetting to enqueue the task first. The very first event for a turn must be the task itself (new_task(context.message)); status/artifact events reference its task.id and task.context_id.
• Returning from execute without a terminal status. Always emit TaskStatusUpdateEvent(status=TaskStatus(state=TaskState.completed), final=True) (or failed) so clients know the turn ended.
• Wrong client send_message shape. Client.send_message(message: Message) takes a Message directly — not a SendMessageRequest envelope. The JSON-RPC framing happens internally. Returns an async iterator of (Task, event_or_None) tuples (sometimes a bare Message).
• Hard-coding localhost. AgentCard.url (a single string in this SDK, not a list of AgentInterface) is what clients dial. Use the externally reachable URL (or set it from an env var) when deploying.
• pi / LLM call timeouts. The SDK's underlying httpx defaults to a 5s read timeout. For agents that wrap slow LLMs or CLIs, pass httpx.AsyncClient(timeout=httpx.Timeout(180.0)) on both sides (caller and intermediate orchestrator).

Bundled Resources

Reference Files (load when relevant)

• references/server-architecture.md — Full breakdown of AgentExecutor, RequestContext, EventQueue, task store options, push-notification config, ASGI app variants (Starlette vs FastAPI vs gRPC).
• references/client-usage.md — A2ACardResolver, ClientFactory/ClientConfig, streaming vs non-streaming, extended cards, auth interceptors, error handling.
• references/types-reference.md — Frequently used types: AgentCard, AgentSkill, AgentCapabilities, Message, Part, TextPart, Role, Task, TaskState, TaskStatus, TaskStatusUpdateEvent, TaskArtifactUpdateEvent, Artifact, plus helper utilities (new_task, new_agent_text_message, new_text_artifact).
• references/advanced-patterns.md — Streaming progress updates, multi-turn conversations (continuing a task by context_id), multi-agent orchestration, push notifications, authentication, persistent task stores.

Example Files (copy and adapt)

• examples/helloworld_server.py — Minimal end-to-end server with one skill.
• examples/helloworld_client.py — Minimal client: resolve card → non-streaming call → streaming call.
• examples/streaming_agent_executor.py — Executor that emits multiple working-state updates before the final artifact (template for LLM-backed agents).
• examples/llm_agent_executor.py — Pattern for wrapping a real LLM call (placeholder for any provider) inside an executor with streamed token-style updates.
• examples/orchestrator_executor.py — Executor that calls a downstream A2A agent, streaming its events back to the original caller.
• examples/pyproject.toml — Reference dependencies / extras.

Scripts (run directly)

• scripts/scaffold_server.py — Generate a new A2A server project (__main__.py + agent_executor.py + pyproject.toml) from a name/description/skill prompt.
• scripts/scaffold_client.py — Generate a new A2A client script targeting a given base URL.
• scripts/inspect_card.py — Fetch and pretty-print a remote agent's card to verify discovery and transports before writing client code.

Suggested Implementation Order

For a from-scratch new A2A integration, work in this order:

1. Read references/types-reference.md to get the vocabulary right.
2. Copy examples/helloworld_server.py + examples/helloworld_client.py, run them locally to confirm the SDK works.
3. Replace the HelloWorldAgent with the real business agent; update AgentSkill + AgentCard metadata.
4. If the agent is LLM-backed or slow, switch to the examples/streaming_agent_executor.py pattern.
5. If the agent must call other agents, layer in examples/orchestrator_executor.py.
6. Before deployment, swap InMemoryTaskStore for a persistent store (see references/advanced-patterns.md) and set AgentCard.url to the externally reachable endpoint.