roanbrasil Agent-to-Agent (A2A) Protocol
What A2A Is
Google's open protocol (released April 2025) for horizontal agent-to-agent communication.
MCP vs A2A: Complementary, Not Competing
==========================================
MCP (Model Context Protocol) A2A (Agent-to-Agent)
------------------------------- ---------------------------
Agent --> Tools / Resources Agent --> Agent
JSON-RPC 2.0 (stdio or SSE) HTTP/JSON REST
Tool invocation Task delegation
No standard discovery Agent Cards at well-known URL
Stateless server Task lifecycle (submitted->done)
Vertical: capability extension Horizontal: work distribution
Typical stack:
Orchestrator Agent
|--- MCP servers (tools: web search, code executor, DB)
|--- A2A --> Code Review Agent
|--- A2A --> Data Analysis Agent
|--- A2A --> Notification Agent
A2A enables agents to delegate work without exposing internal state, memory, tools, or system prompts to the caller. Each agent is a black box that accepts tasks and returns artifacts.
Core Concepts
Concept | Description
---------------|--------------------------------------------------------------
Agent Card | JSON manifest at /.well-known/agent.json
| Describes: name, URL, skills, auth, capabilities
Task | Unit of work; has ID, status, input message, output artifacts
Part | Content unit in a message: TextPart, FilePart, DataPart
Artifact | Output produced by a completed task (document, data, file)
Skill | Named capability declared in Agent Card with I/O schema
Push | Agent proactively sends updates (requires pushNotifications cap)
Streaming | Agent streams partial results via SSE during task execution
Task Status Lifecycle
+-----------> input-required <----------+
| | |
| | (multi-turn) |
v v |
submitted --> working ---------> completed |
| |
+-----------> failed |
| |
+-----------> canceled |
|
(agent needs more info from caller) ---+
Agent Card
The Agent Card is the discovery contract. Serve it at /.well-known/agent.json.
{
"name": "Code Review Agent",
"description": "Reviews pull requests for bugs, style violations, and security issues",
"url": "https://code-review.agents.example.com",
"version": "1.2.0",
"documentationUrl": "https://docs.example.com/agents/code-review",
"capabilities": {
"streaming": true,
"pushNotifications": false,
"stateTransitionHistory": true
},
"authentication": {
"schemes": ["Bearer"],
"credentials": null
},
"defaultInputModes": ["text"],
"defaultOutputModes": ["text", "data"],
"skills": [
{
"id": "review-pr",
"name": "Review Pull Request",
"description": "Analyzes a PR diff and returns structured findings with severity ratings",
"tags": ["code-review", "security", "quality"],
"examples": [
"Review this Python PR for security vulnerabilities",
"Check this diff for style issues per PEP 8"
],
"inputModes": ["text"],
"outputModes": ["text", "data"],
"inputSchema": {
"type": "object",
"properties": {
"diff": {"type": "string", "description": "The PR diff in unified diff format"},
"language": {"type": "string", "description": "Primary programming language"},
"focus": {
"type": "array",
"items": {"type": "string", "enum": ["security", "style", "bugs", "performance"]},
"description": "Areas to focus on; defaults to all"
}
},
"required": ["diff"]
}
},
{
"id": "explain-code",
"name": "Explain Code",
"description": "Generates a plain-English explanation of a code snippet",
"tags": ["documentation"],
"inputModes": ["text"],
"outputModes": ["text"]
}
]
}
A2A Endpoints
Endpoint | Method | Purpose
--------------------------------|--------|----------------------------------
/.well-known/agent.json | GET | Fetch Agent Card (discovery)
/tasks/send | POST | Send task; wait for completion
/tasks/sendSubscribe | POST | Send task; stream SSE updates
/tasks/{id} | GET | Get current task state
/tasks/{id}/cancel | POST | Cancel a running task
/tasks/{id}/send | POST | Continue multi-turn task
Request / Response Shapes
// POST /tasks/send
{
"id": "task-uuid-1234",
"message": {
"role": "user",
"parts": [
{
"type": "text",
"text": "Review this diff for security issues: ..."
},
{
"type": "data",
"data": {"language": "python", "focus": ["security"]}
}
]
},
"metadata": {
"caller": "orchestrator-agent-v2",
"priority": "normal"
}
}
// Response (completed task)
{
"id": "task-uuid-1234",
"status": {
"state": "completed",
"timestamp": "2025-06-17T12:34:56Z"
},
"artifacts": [
{
"name": "review-findings",
"description": "Security findings from code review",
"parts": [
{
"type": "data",
"data": {
"findings": [
{
"severity": "high",
"line": 42,
"issue": "SQL injection via string interpolation",
"recommendation": "Use parameterized queries"
}
],
"summary": "1 high severity finding"
}
}
]
}
]
}
Python Implementation: Full A2A Server
# a2a_server.py
import asyncio
import uuid
from datetime import datetime, UTC
from typing import AsyncIterator
from fastapi import FastAPI, HTTPException, Header
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import json
app = FastAPI(title="Code Review A2A Agent")
# --- Data models ---
class Part(BaseModel):
type: str # "text" | "data" | "file"
text: str | None = None
data: dict | None = None
class Message(BaseModel):
role: str
parts: list[Part]
class TaskRequest(BaseModel):
id: str
message: Message
metadata: dict | None = None
class TaskStatus(BaseModel):
state: str
timestamp: str
message: Message | None = None
class Artifact(BaseModel):
name: str
description: str | None = None
parts: list[Part]
class Task(BaseModel):
id: str
status: TaskStatus
artifacts: list[Artifact] = []
history: list[TaskStatus] = []
# In-memory store (use Redis/DB in production)
tasks: dict[str, Task] = {}
# --- Agent Card ---
AGENT_CARD = {
"name": "Code Review Agent",
"description": "Reviews code for bugs, security issues, and style violations",
"url": "https://code-review.agents.example.com",
"version": "1.0.0",
"capabilities": {"streaming": True, "pushNotifications": False},
"authentication": {"schemes": ["Bearer"]},
"skills": [
{
"id": "review-pr",
"name": "Review Pull Request",
"description": "Analyzes a diff and returns structured findings",
"inputModes": ["text"],
"outputModes": ["text", "data"],
}
],
}
@app.get("/.well-known/agent.json")
async def agent_card():
return AGENT_CARD
# --- Authentication ---
async def verify_bearer(authorization: str = Header(...)) -> str:
if not authorization.startswith("Bearer "):
raise HTTPException(401, "Missing Bearer token")
token = authorization.removeprefix("Bearer ")
# Validate token against your auth system (JWT, OAuth introspection, etc.)
if not await validate_token(token):
raise HTTPException(403, "Invalid or expired token")
return token
async def validate_token(token: str) -> bool:
# Replace with real JWT validation
return token == "demo-token"
# --- Task processing ---
async def process_review(task_request: TaskRequest) -> tuple[str, list[Artifact]]:
"""Core task logic. Returns (summary_text, artifacts)."""
# Extract diff from message parts
diff = ""
focus = ["security", "bugs", "style"]
for part in task_request.message.parts:
if part.type == "text":
diff = part.text or ""
elif part.type == "data" and part.data:
focus = part.data.get("focus", focus)
# Call your LLM / analysis logic here
# This is a stub:
findings = [
{
"severity": "high",
"line": 42,
"issue": "Potential SQL injection",
"recommendation": "Use parameterized queries",
}
]
summary = f"Found {len(findings)} issues in reviewed diff."
artifacts = [
Artifact(
name="review-findings",
description="Code review findings",
parts=[
Part(type="text", text=summary),
Part(type="data", data={"findings": findings, "summary": summary}),
],
)
]
return summary, artifacts
@app.post("/tasks/send")
async def send_task(task_request: TaskRequest, authorization: str = Header(...)):
await verify_bearer(authorization)
task = Task(
id=task_request.id,
status=TaskStatus(
state="working",
timestamp=datetime.now(UTC).isoformat(),
),
)
tasks[task.id] = task
try:
summary, artifacts = await process_review(task_request)
task.status = TaskStatus(
state="completed",
timestamp=datetime.now(UTC).isoformat(),
)
task.artifacts = artifacts
except Exception as e:
task.status = TaskStatus(
state="failed",
timestamp=datetime.now(UTC).isoformat(),
message=Message(role="agent", parts=[Part(type="text", text=str(e))]),
)
return task
@app.post("/tasks/sendSubscribe")
async def send_task_subscribe(
task_request: TaskRequest, authorization: str = Header(...)
):
await verify_bearer(authorization)
async def event_stream() -> AsyncIterator[str]:
task = Task(
id=task_request.id,
status=TaskStatus(state="submitted", timestamp=datetime.now(UTC).isoformat()),
)
tasks[task.id] = task
yield f"data: {task.model_dump_json()}\n\n"
# Update to "working"
task.status = TaskStatus(state="working", timestamp=datetime.now(UTC).isoformat())
yield f"data: {task.model_dump_json()}\n\n"
# Stream intermediate progress (optional)
for i in range(3):
await asyncio.sleep(1)
progress = Task(
id=task.id,
status=TaskStatus(
state="working",
timestamp=datetime.now(UTC).isoformat(),
message=Message(
role="agent",
parts=[Part(type="text", text=f"Analyzing section {i+1}/3...")],
),
),
)
yield f"data: {progress.model_dump_json()}\n\n"
summary, artifacts = await process_review(task_request)
task.status = TaskStatus(state="completed", timestamp=datetime.now(UTC).isoformat())
task.artifacts = artifacts
yield f"data: {task.model_dump_json()}\n\n"
return StreamingResponse(event_stream(), media_type="text/event-stream")
@app.get("/tasks/{task_id}")
async def get_task(task_id: str, authorization: str = Header(...)):
await verify_bearer(authorization)
if task_id not in tasks:
raise HTTPException(404, "Task not found")
return tasks[task_id]
@app.post("/tasks/{task_id}/cancel")
async def cancel_task(task_id: str, authorization: str = Header(...)):
await verify_bearer(authorization)
if task_id not in tasks:
raise HTTPException(404, "Task not found")
task = tasks[task_id]
if task.status.state not in ("submitted", "working", "input-required"):
raise HTTPException(400, f"Cannot cancel task in state: {task.status.state}")
task.status = TaskStatus(state="canceled", timestamp=datetime.now(UTC).isoformat())
return task
Python A2A Client
# a2a_client.py
import uuid
import httpx
import json
from typing import AsyncIterator
class A2AClient:
def __init__(self, agent_url: str, token: str):
self.base = agent_url.rstrip("/")
self.headers = {"Authorization": f"Bearer {token}"}
async def get_agent_card(self) -> dict:
async with httpx.AsyncClient() as client:
resp = await client.get(
f"{self.base}/.well-known/agent.json",
headers=self.headers,
)
resp.raise_for_status()
return resp.json()
def _supports_skill(self, card: dict, skill_id: str) -> bool:
return any(s["id"] == skill_id for s in card.get("skills", []))
async def send_task(self, text: str, data: dict | None = None) -> dict:
parts = [{"type": "text", "text": text}]
if data:
parts.append({"type": "data", "data": data})
payload = {
"id": str(uuid.uuid4()),
"message": {"role": "user", "parts": parts},
}
async with httpx.AsyncClient(timeout=120) as client:
resp = await client.post(
f"{self.base}/tasks/send",
json=payload,
headers=self.headers,
)
resp.raise_for_status()
return resp.json()
async def stream_task(
self, text: str, data: dict | None = None
) -> AsyncIterator[dict]:
parts = [{"type": "text", "text": text}]
if data:
parts.append({"type": "data", "data": data})
payload = {
"id": str(uuid.uuid4()),
"message": {"role": "user", "parts": parts},
}
async with httpx.AsyncClient(timeout=300) as client:
async with client.stream(
"POST",
f"{self.base}/tasks/sendSubscribe",
json=payload,
headers=self.headers,
) as resp:
resp.raise_for_status()
async for line in resp.aiter_lines():
if line.startswith("data: "):
yield json.loads(line[6:])
# Usage
async def main():
client = A2AClient("https://code-review.agents.example.com", "demo-token")
# Discover capabilities
card = await client.get_agent_card()
print(f"Agent: {card['name']} -- Skills: {[s['id'] for s in card['skills']]}")
# Stream a task
async for update in client.stream_task(
"Review this diff for security issues",
{"diff": "- raw_query = f'SELECT * FROM users WHERE id = {user_id}'", "focus": ["security"]},
):
print(f"[{update['status']['state']}]", end=" ")
if update["status"].get("message"):
print(update["status"]["message"]["parts"][0].get("text", ""))
if update["status"]["state"] == "completed":
for artifact in update.get("artifacts", []):
for part in artifact["parts"]:
if part["type"] == "data":
print("Findings:", part["data"])
Java / Spring Boot Implementation
Agent Card Endpoint
// AgentCardController.java
package com.example.a2a;
import org.springframework.web.bind.annotation.*;
import java.util.*;
@RestController
public class AgentCardController {
@GetMapping("/.well-known/agent.json")
public Map<String, Object> agentCard() {
var skill = Map.of(
"id", "review-pr",
"name", "Review Pull Request",
"description", "Analyzes a PR diff and returns structured findings",
"inputModes", List.of("text"),
"outputModes", List.of("text", "data")
);
return Map.of(
"name", "Code Review Agent",
"description", "Reviews code for bugs, security issues, and style violations",
"url", "https://code-review.agents.example.com",
"version", "1.0.0",
"capabilities", Map.of(
"streaming", true,
"pushNotifications", false
),
"authentication", Map.of("schemes", List.of("Bearer")),
"skills", List.of(skill)
);
}
}
Task Handler with Async Processing
// TaskController.java
package com.example.a2a;
import org.springframework.http.*;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.time.Instant;
import java.util.*;
import java.util.concurrent.*;
@RestController
@RequestMapping("/tasks")
public class TaskController {
private final ConcurrentHashMap<String, TaskEntity> taskStore = new ConcurrentHashMap<>();
private final ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
private final CodeReviewService reviewService;
public TaskController(CodeReviewService reviewService) {
this.reviewService = reviewService;
}
@PostMapping("/send")
public ResponseEntity<TaskEntity> sendTask(
@RequestBody TaskRequest request,
@RequestHeader("Authorization") String auth
) {
validateBearer(auth);
var task = new TaskEntity(request.id(), "working", Instant.now());
taskStore.put(task.id(), task);
try {
var result = reviewService.review(extractText(request));
task = task.withState("completed").withArtifacts(List.of(result));
taskStore.put(task.id(), task);
return ResponseEntity.ok(task);
} catch (Exception e) {
task = task.withState("failed").withErrorMessage(e.getMessage());
taskStore.put(task.id(), task);
return ResponseEntity.status(500).body(task);
}
}
@PostMapping("/sendSubscribe")
public SseEmitter sendTaskSubscribe(
@RequestBody TaskRequest request,
@RequestHeader("Authorization") String auth
) {
validateBearer(auth);
SseEmitter emitter = new SseEmitter(300_000L);
executor.submit(() -> {
try {
var task = new TaskEntity(request.id(), "submitted", Instant.now());
emitter.send(task);
task = task.withState("working");
emitter.send(task);
// Progress updates
for (int i = 1; i <= 3; i++) {
Thread.sleep(1000);
emitter.send(task.withProgress("Analyzing section " + i + "/3"));
}
var result = reviewService.review(extractText(request));
task = task.withState("completed").withArtifacts(List.of(result));
emitter.send(task);
emitter.complete();
} catch (Exception e) {
emitter.completeWithError(e);
}
});
return emitter;
}
@GetMapping("/{taskId}")
public ResponseEntity<TaskEntity> getTask(
@PathVariable String taskId,
@RequestHeader("Authorization") String auth
) {
validateBearer(auth);
var task = taskStore.get(taskId);
return task != null
? ResponseEntity.ok(task)
: ResponseEntity.notFound().build();
}
@PostMapping("/{taskId}/cancel")
public ResponseEntity<TaskEntity> cancelTask(
@PathVariable String taskId,
@RequestHeader("Authorization") String auth
) {
validateBearer(auth);
var task = taskStore.get(taskId);
if (task == null) return ResponseEntity.notFound().build();
var cancelable = Set.of("submitted", "working", "input-required");
if (!cancelable.contains(task.state())) {
return ResponseEntity.badRequest().build();
}
task = task.withState("canceled");
taskStore.put(taskId, task);
return ResponseEntity.ok(task);
}
private void validateBearer(String auth) {
if (auth == null || !auth.startsWith("Bearer ")) {
throw new ResponseStatusException(HttpStatus.UNAUTHORIZED);
}
// Validate token here
}
private String extractText(TaskRequest request) {
return request.message().parts().stream()
.filter(p -> "text".equals(p.type()))
.map(Part::text)
.findFirst()
.orElseThrow(() -> new IllegalArgumentException("No text part in message"));
}
}
Spring AI Integration
// A2AOrchestrator.java -- Orchestrator delegating to A2A agents
package com.example.a2a;
import org.springframework.ai.chat.client.ChatClient;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestClient;
import java.util.*;
@Service
public class A2AOrchestrator {
private final RestClient restClient;
private final AgentRegistry registry;
public A2AOrchestrator(AgentRegistry registry) {
this.registry = registry;
this.restClient = RestClient.create();
}
public Map<String, Object> delegateTask(String skillId, String taskText) {
// 1. Discover agent by skill
var agentUrl = registry.findBySkill(skillId)
.orElseThrow(() -> new NoSuchElementException("No agent for skill: " + skillId));
// 2. Verify agent card declares this skill
var card = restClient.get()
.uri(agentUrl + "/.well-known/agent.json")
.retrieve()
.body(Map.class);
boolean hasSkill = ((List<Map<String, ?>>) card.get("skills")).stream()
.anyMatch(s -> skillId.equals(s.get("id")));
if (!hasSkill) throw new IllegalStateException("Agent card does not declare skill: " + skillId);
// 3. Send task
var payload = Map.of(
"id", UUID.randomUUID().toString(),
"message", Map.of(
"role", "user",
"parts", List.of(Map.of("type", "text", "text", taskText))
)
);
return restClient.post()
.uri(agentUrl + "/tasks/send")
.header("Authorization", "Bearer " + getToken())
.body(payload)
.retrieve()
.body(Map.class);
}
private String getToken() {
// Load from secrets manager / OAuth client credentials flow
return System.getenv("A2A_BEARER_TOKEN");
}
}
Agent Registry Pattern
# registry.py
from dataclasses import dataclass, field
from datetime import datetime, UTC
import asyncio
import httpx
@dataclass
class RegisteredAgent:
url: str
card: dict
last_seen: datetime = field(default_factory=lambda: datetime.now(UTC))
healthy: bool = True
class AgentRegistry:
"""
In-memory registry. In production, back with Redis or a database.
Agents self-register or are registered by an admin.
"""
def __init__(self):
self._agents: dict[str, RegisteredAgent] = {} # url -> agent
async def register(self, agent_url: str) -> RegisteredAgent:
card = await self._fetch_card(agent_url)
entry = RegisteredAgent(url=agent_url, card=card)
self._agents[agent_url] = entry
return entry
def find_by_skill(self, skill_id: str) -> list[RegisteredAgent]:
return [
agent for agent in self._agents.values()
if agent.healthy
and any(s["id"] == skill_id for s in agent.card.get("skills", []))
]
async def health_check_all(self):
"""Run periodically (e.g. every 60s) to verify agent liveness."""
for url, agent in self._agents.items():
try:
card = await self._fetch_card(url)
agent.card = card
agent.last_seen = datetime.now(UTC)
agent.healthy = True
except Exception:
agent.healthy = False
async def _fetch_card(self, url: str) -> dict:
async with httpx.AsyncClient(timeout=10) as client:
resp = await client.get(f"{url.rstrip('/')}/.well-known/agent.json")
resp.raise_for_status()
return resp.json()
Multi-Turn Tasks (input-required)
# When your agent needs more information from the caller
async def handle_ambiguous_task(task_request: TaskRequest) -> Task:
"""
Return input-required status with a clarifying question.
Caller continues via POST /tasks/{id}/send.
"""
if needs_clarification(task_request):
return Task(
id=task_request.id,
status=TaskStatus(
state="input-required",
timestamp=datetime.now(UTC).isoformat(),
message=Message(
role="agent",
parts=[Part(
type="text",
text="Which programming language should I focus on? "
"Please reply with: python, java, go, or typescript",
)],
),
),
)
# ... process normally
Security Checklist for A2A Agents
Authentication and Authorization:
[ ] All endpoints require Bearer token (OAuth 2.1 recommended)
[ ] Token validated on every request (not cached without expiry)
[ ] Caller identity extracted from token and logged
[ ] Scope/permissions checked: not every caller gets every skill
Input validation:
[ ] Task input validated against skill's inputSchema before processing
[ ] Task ID is a valid UUID; reject requests with malformed IDs
[ ] Input size limits enforced (max message length, part count)
[ ] Parts with unexpected types rejected
Output safety:
[ ] Artifacts do not contain internal state, config, or secrets
[ ] Agent Card does not expose internal infrastructure details
[ ] Error messages do not leak stack traces to caller
Task lifecycle:
[ ] Task IDs are globally unique; collision check before accepting
[ ] Concurrent task limits per caller to prevent abuse
[ ] Long-running tasks have timeouts and automatic cancellation
[ ] All state transitions logged with timestamp and caller ID
Operational:
[ ] Rate limiting per caller token
[ ] Audit log: every task received, state change, artifact delivered
[ ] Health check: /.well-known/agent.json should return < 200ms
[ ] Graceful shutdown: drain in-flight tasks before stopping
A2A vs MCP Reference
| Dimension | MCP | A2A |
|---|---|---|
| Direction | Agent to Tools/Resources | Agent to Agent |
| Protocol | JSON-RPC 2.0 | HTTP/JSON REST |
| Transport | stdio or SSE | HTTP (SSE for streaming) |
| Discovery | No standard | Agent Cards at well-known URL |
| State management | Stateless server | Task lifecycle (submitted-done) |
| Content | Tool calls, resources | Tasks, artifacts, multi-turn |
| Use case | Capability extension | Work delegation |
| Spec owner | Anthropic | |
| Release | November 2024 | April 2025 |
Anti-Patterns
[ ] Exposing internal tool names or memory in task artifacts
--> Artifacts are external API; treat them like public endpoints
[ ] Hardcoding agent URLs instead of using registry + Agent Card discovery
--> URLs change; Agent Cards let you find agents by capability
[ ] Sending secrets or PII in task messages
--> Task messages are logged; use references (IDs) not raw data
[ ] Not validating skill capability before sending a task
--> Always fetch and check Agent Card; agent may not support the skill
[ ] Skipping authentication because "agents are internal"
--> A2A is designed for multi-tenant, cross-org agent networks; always auth
[ ] Using task ID generated by server instead of client
--> A2A spec: client generates task ID; enables idempotency
[ ] Polling /tasks/{id} in a tight loop instead of streaming
--> Use sendSubscribe for SSE; poll with backoff if SSE unavailable
[ ] Not handling input-required state in client
--> Multi-turn tasks are a first-class feature; implement the continue flow