a2a-development

name	a2a-development
description	A2A (Agent-to-Agent) protocol v1.0 development guide for Python. Use when building A2A agents, creating multi-agent systems, implementing A2A servers or clients, working with Agent Cards, handling Task lifecycle, implementing streaming or push notifications, integrating frameworks (LangGraph, Google ADK, BeeAI, AG2, CrewAI, Semantic Kernel, Microsoft Agent Framework) with A2A, configuring OAuth2/PKCE/mTLS security, implementing multi-tenancy, using extensions, or migrating from v0.3 to v1.0. Triggers: 'a2a', 'agent-to-agent', 'agent card', 'TaskUpdater', 'AgentExecutor', 'A2AStarletteApplication', 'A2AClient', 'ClientFactory', 'a2a-sdk', 'HandoffTool', 'RemoteA2aAgent', 'langgraph-a2a-server', 'to_a2a'. Do NOT use for MCP (Model Context Protocol) tool-calling patterns or non-A2A agent communication.
compatibility	Requires Python 3.10+ and uv or pip. Network access needed for A2A inter-agent communication.
metadata	{"author":"MaYiding","version":"2.0"}

A2A (Agent-to-Agent) Protocol Development

Build production-grade multi-agent systems with the Python SDK (pip install "a2a-sdk[http-server]").

• Protocol: v1.0.0 | SDK: a2a-sdk v1.0.0
• Normative source: specification/a2a.proto + specification/a2a.json

---

1. Gotchas — Fatal Errors to Avoid

These are the most common mistakes. Violating any causes immediate runtime failures.

# --- Part construction (v1.0 SDK) ---
Part(text="Hello")                     # CORRECT: v1.0 unified Part
Part(root=TextPart(text="Hello"))      # WRONG: v0.3 legacy syntax, removed in v1.0

# --- Role and TaskState enums (SCREAMING_SNAKE_CASE) ---
Role.ROLE_USER                         # CORRECT
Role.user                              # WRONG: v0.3 legacy

TaskState.TASK_STATE_COMPLETED         # CORRECT
TaskState.completed                    # WRONG: v0.3 legacy

# --- AgentCard construction ---
AgentCard(
    supported_interfaces=[             # CORRECT: v1.0 multi-interface
        AgentInterface(url="http://localhost:9999",
                       protocol_binding="JSONRPC",
                       protocol_version="1.0"),
    ], ...
)
AgentCard(url="http://localhost:9999", ...)  # WRONG: v0.3 single-url style

# --- Response parsing ---
response.root.result                   # CORRECT: RootModel requires .root
response.result                        # WRONG: AttributeError

# --- JSON-RPC envelope ---
SendMessageRequest(
    id=str(uuid4()),                   # REQUIRED: JSON-RPC request ID
    params=SendMessageParams(...)      # params, not direct message
)

# --- Params class name ---
SendMessageParams(...)                 # CORRECT: v1.0
MessageSendParams(...)                 # WRONG: v0.3 legacy

# --- httpx lifecycle ---
async with httpx.AsyncClient(timeout=30) as httpx_client:  # MUST manage lifecycle
    resolver = A2ACardResolver(httpx_client=httpx_client, base_url=url)

# --- New Client API (recommended) ---
client = await ClientFactory.connect("http://localhost:9999")  # Auto card+transport
async with client:
    async for response in client.send_message(message):  # Unified streaming/non-streaming
        ...

---

2. Agent Types

Type	Response	Scenario	Client Handling
Message-only	Always Message	Simple Q&A, FAQ bot	Read message.parts directly
Task-generating	Always Task	Long-running, cancelable	Check task.status.state
Hybrid	Message then Task	Complex negotiation	Check type, branch

---

3. Minimum Viable Scaffolding

Server (3 components)

import uvicorn
from a2a.server.agent_execution import AgentExecutor
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.server.context import RequestContext
from a2a.server.events import EventQueue
from a2a.types import AgentCard, AgentSkill, AgentCapabilities, AgentInterface
from a2a.utils import new_agent_text_message

class MyExecutor(AgentExecutor):
    async def execute(self, context: RequestContext, event_queue: EventQueue):
        await event_queue.enqueue_event(new_agent_text_message("Hello"))
    async def cancel(self, context, event_queue):
        raise Exception("Not supported")

agent_card = AgentCard(
    name="My Agent", description="...", version="1.0.0",
    supported_interfaces=[
        AgentInterface(
            url="http://localhost:9999",
            protocol_binding="JSONRPC",
            protocol_version="1.0",
        ),
    ],
    capabilities=AgentCapabilities(streaming=True),
    default_input_modes=["text/plain"], default_output_modes=["text/plain"],
    skills=[AgentSkill(id="s1", name="Skill", description="...", tags=["t"])],
)
handler = DefaultRequestHandler(agent_executor=MyExecutor(), task_store=InMemoryTaskStore())
app = A2AStarletteApplication(agent_card=agent_card, http_handler=handler)
uvicorn.run(app.build(), host="0.0.0.0", port=9999)

Client — New API (recommended)

from a2a.client import ClientFactory, Client
from a2a.client.helpers import create_text_message_object
from a2a.utils import get_message_text
from a2a.types import Message, Task

async with await ClientFactory.connect("http://localhost:9999") as client:
    message = create_text_message_object(text="Hello")
    async for response in client.send_message(message):
        if isinstance(response, Message):
            print(get_message_text(response))
        elif isinstance(response, tuple):
            task: Task = response[0]
            if task.artifacts:
                print(get_message_text(task.artifacts[0]))

Client — Legacy API

import httpx
from uuid import uuid4
from a2a.client import A2ACardResolver, A2AClient
from a2a.types import (
    SendMessageRequest, SendMessageParams, Message, Part, Role,
    SendMessageSuccessResponse, JSONRPCErrorResponse, Task,
)

async with httpx.AsyncClient(timeout=30) as httpx_client:
    resolver = A2ACardResolver(httpx_client=httpx_client, base_url="http://localhost:9999")
    card = await resolver.get_agent_card()
    client = A2AClient(httpx_client=httpx_client, agent_card=card)

    response = await client.send_message(SendMessageRequest(
        id=str(uuid4()),
        params=SendMessageParams(
            message=Message(role=Role.ROLE_USER, parts=[Part(text="Hello")], message_id=uuid4().hex),
        ),
    ))
    if isinstance(response.root, SendMessageSuccessResponse):
        result = response.root.result  # Task | Message

---

4. Task State Machine

SUBMITTED -> WORKING -> COMPLETED | FAILED | CANCELED | REJECTED  (terminal)
                     -> INPUT_REQUIRED -> WORKING  (loop)
                     -> AUTH_REQUIRED  -> WORKING  (loop)

• Terminal states are irreversible — create new Task under same context_id to continue
• task_id is server-generated only — never create on client side
• context_id = framework session — maps to thread_id (LangGraph), session_id (ADK)
• Task refinement: use reference_task_ids to reference completed Tasks for modification
• Parallel Tasks: multiple independent Tasks within same context_id

---

5. Hard Rules

Rule	Detail
Part construction	Part(text="..."), Part(data={...}), Part(raw="b64", media_type="...") — unified Part, no more TextPart/FilePart/DataPart wrappers
Enum format	SCREAMING_SNAKE: TaskState.TASK_STATE_COMPLETED, Role.ROLE_USER
Artifact for outputs	TaskStatus.message = progress; Artifact = final result
DefaultRequestHandler params	push_config_store / push_sender (NOT push_notification_*)
BasePushNotificationSender	Requires httpx_client + config_store (share store with handler)
PushNotificationAuthenticationInfo	.schemes is list[str] (plural), not .scheme: str
Terminal Task	Cannot receive messages; use reference_task_ids + new Task
CancelTask	Idempotent; safe to retry; no-op on terminal Tasks
Version header	A2A-Version: 1.0 (Major.Minor, no patch)
Webhook security	3 methods: Token, JWT+JWKS, HMAC-SHA256
IANA registered	application/a2a+json media type, A2A-Version/A2A-Extensions headers
Stream event type detection	JSON member existence (if "task" in event), NOT kind discriminator

---

6. Getting Started Workflow

• Install SDK: pip install "a2a-sdk[http-server]"
• Implement AgentExecutor.execute() with your agent logic
• Define AgentCard with supported_interfaces, skills, capabilities
• Wire up DefaultRequestHandler + A2AStarletteApplication
• Test with the Test Client template (## Template 4)
• Add streaming support if needed (read references/streaming-and-push.md)
• Add security (OAuth2/Bearer) for production (read references/production-guide.md)
• Run deployment checklist (read references/production-guide.md ## Deployment Checklist)

---

7. Decision Router — Which Reference to Load

Use this table to decide which reference file to read for the current task. Each file is self-contained. Load only what is needed.

Starting a New Project

Task	Reference File	Grep Pattern
Understand data types (Task, Message, Part, Artifact)	references/protocol-fundamentals.md	"## Core Data Types"
Agent types (Message-only / Task-generating / Hybrid)	references/protocol-fundamentals.md	"## Agent Types"
Define AgentCard, skills, JWS signing + verification	references/protocol-fundamentals.md	"## Agent Card"
AgentCard complete field reference	references/protocol-fundamentals.md	"### AgentCard Field Reference"
Task state transitions, context_id rules	references/protocol-fundamentals.md	"## Task Lifecycle"
context_id / task_id validation matrix	references/protocol-fundamentals.md	"### Context and Task ID Validation"
Task refinement and parallel execution	references/protocol-fundamentals.md	"## Task Refinement"

Building Server

Task	Reference File	Grep Pattern
Implement AgentExecutor with TaskUpdater	references/server-development.md	"## TaskUpdater"
Custom TaskStore (Redis / PostgreSQL / DatabaseTaskStore)	references/server-development.md	"## Custom Components"
Custom ID generator	references/server-development.md	"### Custom ID Generator"
Throw errors (ServerError)	references/server-development.md	"## ServerError"
Use FastAPI instead of Starlette	references/server-development.md	"### FastAPI Alternative"

Building Client

Task	Reference File	Grep Pattern
New Client API (ClientFactory + Client)	references/client-development.md	"## New Client API"
Client middleware / interceptors	references/client-development.md	"## Client Middleware"
Send messages (non-streaming)	references/client-development.md	"## Sending Messages"
Stream messages (SSE)	references/client-development.md	"## Streaming Messages"
Multi-turn dialog (INPUT_REQUIRED loop)	references/client-development.md	"## Multi-Turn Dialog"
Get / list / cancel Task (cursor pagination)	references/client-development.md	"## Get / List / Cancel"
SubscribeToTask (reconnection)	references/client-development.md	"## SubscribeToTask"
returnImmediately / blocking semantics	references/client-development.md	"## Blocking Semantics"
Fetch extended Agent Card	references/client-development.md	"## Extended Agent Card"

Advanced Features

Task	Reference File	Grep Pattern
Server-side streaming implementation	references/streaming-and-push.md	"## Streaming Implementation"
Artifact chunked transfer	references/streaming-and-push.md	"### Artifact Chunked Transfer"
Push notifications (server + webhook)	references/streaming-and-push.md	"## Push Notifications"
Webhook security (Token / JWT / HMAC)	references/streaming-and-push.md	"### Webhook Security"
SSRF protection for webhook URLs	references/streaming-and-push.md	"### SSRF Protection"
SSE format + StreamResponse types	references/streaming-and-push.md	"### SSE Format"
Multi-agent orchestration (5 patterns)	references/multi-agent-and-frameworks.md	"## Orchestration Patterns"
Pattern selection guide (when to use each)	references/multi-agent-and-frameworks.md	"### Pattern Selection"
Framework integration (12+ frameworks)	references/multi-agent-and-frameworks.md	"## Framework Integrations"
Agent Gateway Protocol (AGP) extension	references/multi-agent-and-frameworks.md	"### Agent Gateway Protocol"
Extension mechanism + governance	references/multi-agent-and-frameworks.md	"## Extensions"
Official extensions (Timestamp/Traceability/Passport)	references/multi-agent-and-frameworks.md	"### Official Extension"

Production & Operations

Task	Reference File	Grep Pattern
Security schemes (OAuth2, PKCE, DeviceCode, mTLS)	references/production-guide.md	"## Security and Authentication"
Distributed tracing (OpenTelemetry, W3C Trace Context)	references/production-guide.md	"### Distributed Tracing"
Data privacy (GDPR, CCPA, HIPAA)	references/production-guide.md	"### Data Privacy"
API management and governance	references/production-guide.md	"### API Management"
Rate limiting	references/production-guide.md	"### Rate Limiting"
Multi-tenancy	references/production-guide.md	"## Multi-Tenancy"
Error codes and handling	references/production-guide.md	"## Error Handling"
Deployment checklist (IANA, W3C Trace Context)	references/production-guide.md	"## Deployment Checklist"
Unit / integration / multi-agent tests	references/production-guide.md	"## Testing"
Debugging and logging	references/production-guide.md	"## Debugging"

Templates & Reference

Task	Reference File	Grep Pattern
Complete runnable code templates (4 types)	references/code-templates.md	"## Template"
SDK vs Proto difference table	references/api-reference.md	"### SDK vs Proto Differences"
SDK utility functions + framework converters	references/api-reference.md	"## SDK Utility Functions"
All gotchas and anti-patterns	references/api-reference.md	"## Gotchas"
v0.3 to v1.0 migration (complete guide)	references/api-reference.md	"## v0.3 to v1.0 Migration"
Spec edge behaviors (undefined/ambiguous)	references/api-reference.md	"## Edge Behaviors"
RPC method mapping (JSON-RPC / gRPC / HTTP)	references/api-reference.md	"### RPC Method Mapping"
gRPC binding details (Proto3 service, client/server)	references/api-reference.md	"### gRPC Binding"
HTTP+JSON/REST binding paths	references/api-reference.md	"### HTTP+JSON Binding"
JSON request/response examples	references/api-reference.md	"### JSON Request/Response Examples"
Quick reference card	references/api-reference.md	"## Quick Reference Card"

---

8. Import Quick Reference

# Server
from a2a.server.agent_execution import AgentExecutor
from a2a.server.apps import A2AStarletteApplication  # or A2AFastAPIApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore, DatabaseTaskStore
from a2a.server.context import RequestContext
from a2a.server.events import EventQueue

# Client — New API (recommended)
from a2a.client import ClientFactory, ClientConfig, Client
from a2a.client.helpers import create_text_message_object
from a2a.client.middleware import ClientCallInterceptor
from a2a.client.auth import AuthInterceptor, InMemoryContextCredentialStore

# Client — Legacy API
from a2a.client import A2ACardResolver, A2AClient

# Types (import what you need)
from a2a.types import (
    AgentCard, AgentSkill, AgentCapabilities, AgentInterface,
    Task, TaskState, TaskStatus, Message, Part, Role,
    Artifact,
    SendMessageRequest, SendStreamingMessageRequest,
    SendMessageParams, SendMessageConfiguration,
    SendMessageSuccessResponse, JSONRPCErrorResponse,
    TaskStatusUpdateEvent, TaskArtifactUpdateEvent,
    PushNotificationConfig, PushNotificationAuthenticationInfo,
)

# Utilities
from a2a.utils import new_agent_text_message, get_message_text
from a2a.utils.task_builder import TaskUpdater, new_task

# Signing
from a2a.utils.signing import create_agent_card_signer, create_signature_verifier

---

9. SDK Install Options

a2a-sdk                   Core types only
a2a-sdk[http-server]      Starlette/FastAPI server
a2a-sdk[grpc]             gRPC transport
a2a-sdk[signing]          Agent Card JWS signing
a2a-sdk[telemetry]        OpenTelemetry integration
a2a-sdk[sql]              All SQL drivers
a2a-sdk[postgresql]       PostgreSQL
a2a-sdk[mysql]            MySQL
a2a-sdk[sqlite]           SQLite
a2a-sdk[all]              All optional dependencies