A2A Protocol Reference
The Agent-to-Agent (A2A) Protocol is an open standard for secure, interoperable AI agent communication.
Overview
A2A is supported by Google, Linux Foundation, and 100+ technology partners. It enables AI agents to:
- Discover each other's capabilities
- Communicate securely
- Collaborate on complex, multi-step tasks
Core Characteristics
| Characteristic | Description |
|---|---|
| Transport | JSON-RPC 2.0 over HTTP(S) with TLS encryption |
| Design | Stateful, multi-turn dialogue supporting long-running tasks |
| Authentication | OAuth 2.0, API keys, mutual TLS |
| Discovery | Agent Cards mechanism for capability advertisement |
| Async Support | Push notifications and disconnected operation support |
Protocol Flow
┌─────────────────┐
│ Client Agent │
│ (Supervisor) │
└────────┬────────┘
│
│ 1. POST /a2a HTTP/1.1
│ Content-Type: application/json
│
│ {
│ "jsonrpc": "2.0",
│ "id": "req-001",
│ "method": "execute_task",
│ "params": {
│ "task": "provision_cluster",
│ "config": {...}
│ }
│ }
│
▼
┌──────────────────────┐
│ Server Agent │
│ (Specialized Agent) │
└────────┬─────────────┘
│
│ 2. Response
│ {
│ "jsonrpc": "2.0",
│ "id": "req-001",
│ "result": {
│ "status": "success",
│ "cluster_id": "eks-prod-01"
│ }
│ }
│
▼
┌─────────────────────┐
│ Client Agent │
│ (Receives Result) │
└─────────────────────┘
Five Pillars of A2A
1. Unified Transport
All communication uses JSON-RPC 2.0 over HTTPS:
{
"jsonrpc": "2.0",
"id": "unique-request-id",
"method": "method_name",
"params": {
"key": "value"
}
}
Response:
{
"jsonrpc": "2.0",
"id": "unique-request-id",
"result": {...}
}
Error:
{
"jsonrpc": "2.0",
"id": "unique-request-id",
"error": {
"code": -32600,
"message": "Invalid Request"
}
}
2. Agent Discovery
Agents publish Agent Cards describing their capabilities:
{
"name": "Cloud Orchestration Agent",
"version": "1.0.0",
"description": "Provisions and manages cloud infrastructure",
"capabilities": [
{
"name": "provision_vpc",
"description": "Create Virtual Private Cloud",
"inputs": {
"provider": "aws|azure|gcp",
"cidr_block": "string",
"region": "string"
},
"outputs": {
"vpc_id": "string",
"status": "success|failure"
}
},
{
"name": "provision_cluster",
"description": "Create Kubernetes cluster",
"inputs": {
"provider": "aws|azure|gcp",
"node_count": "integer",
"instance_type": "string"
},
"outputs": {
"cluster_arn": "string",
"endpoint": "string"
}
}
],
"authentication": ["oauth2", "api_key"],
"endpoint": "https://cloud-agent.talkops.local/a2a"
}
3. Task Management
A2A supports multi-turn workflows with state tracking:
Session Start
│
▼
┌─────────────────┐
│ Task: Deploy │
│ Status: pending │
└────────┬────────┘
│
▼
┌─────────────────┐
│ Subtask: Build │──→ complete
└────────┬────────┘
│
▼
┌─────────────────┐
│ Subtask: Test │──→ complete
└────────┬────────┘
│
▼
┌─────────────────┐
│ Subtask: Deploy │──→ awaiting_approval
└────────┬────────┘
│
Human Approval
│
▼
┌─────────────────┐
│ Task: Deploy │
│ Status: success │
└─────────────────┘
4. Multi-modal Data
A2A supports various data types:
| Type | Example |
|---|---|
| Text | Natural language messages |
| Structured Data | JSON objects, arrays |
| Files | Base64-encoded or URLs |
| References | URIs to external resources |
5. Enterprise Security
| Control | Implementation |
|---|---|
| Authentication | OAuth 2.0, API keys, mTLS |
| Authorization | RBAC, scope-based permissions |
| Encryption | TLS 1.3 in transit |
| Audit | Request/response logging |
| Rate Limiting | Per-agent quotas |
Standard Methods
| Method | Description |
|---|---|
discover | Get agent capabilities (Agent Card) |
execute_task | Execute a capability |
get_status | Check task status |
cancel_task | Cancel running task |
list_tasks | List agent's active tasks |
Error Codes
| Code | Name | Description |
|---|---|---|
| -32700 | Parse error | Invalid JSON |
| -32600 | Invalid Request | Missing required fields |
| -32601 | Method not found | Unknown method |
| -32602 | Invalid params | Invalid method parameters |
| -32603 | Internal error | Agent internal error |
| -32000 | Authentication error | Auth failed |
| -32001 | Authorization error | Insufficient permissions |
| -32002 | Rate limited | Too many requests |
Implementation Example
Python A2A Client
import httpx
import json
class A2AClient:
def __init__(self, endpoint: str, api_key: str):
self.endpoint = endpoint
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
async def discover(self) -> dict:
"""Get agent capabilities"""
return await self._request("discover", {})
async def execute_task(self, task: str, params: dict) -> dict:
"""Execute a task on the agent"""
return await self._request("execute_task", {
"task": task,
"params": params
})
async def _request(self, method: str, params: dict) -> dict:
payload = {
"jsonrpc": "2.0",
"id": str(uuid.uuid4()),
"method": method,
"params": params
}
async with httpx.AsyncClient() as client:
response = await client.post(
self.endpoint,
headers=self.headers,
json=payload
)
return response.json()
Python A2A Server
from fastapi import FastAPI, Request
from pydantic import BaseModel
app = FastAPI()
class A2ARequest(BaseModel):
jsonrpc: str
id: str
method: str
params: dict
@app.post("/a2a")
async def handle_a2a(request: A2ARequest):
if request.method == "discover":
return {
"jsonrpc": "2.0",
"id": request.id,
"result": AGENT_CARD
}
elif request.method == "execute_task":
result = await execute_task(request.params)
return {
"jsonrpc": "2.0",
"id": request.id,
"result": result
}
else:
return {
"jsonrpc": "2.0",
"id": request.id,
"error": {"code": -32601, "message": "Method not found"}
}
Best Practices
- ✅ Always use HTTPS with TLS 1.3
- ✅ Implement exponential backoff for retries
- ✅ Use Agent Cards for dynamic discovery
- ✅ Log all requests/responses for audit
- ✅ Set appropriate timeouts (30-120s)
- ✅ Support both sync and async operations
- ✅ Validate all inputs against schemas
- ✅ Rotate credentials regularly