Google's A2A Protocol — How AI Agents Communicate in Production Multi-Agent Systems

Introduction

Why This Matters

Multi-agent AI systems fail when agents are tightly coupled to their orchestrators. Hard-coded API calls, custom message formats, and one-off integrations make it impossible to swap agents, add new capabilities, or debug failures at scale.

Google's A2A (Agent-to-Agent) protocol solves this by giving agents a standard way to declare their capabilities, accept tasks, and report results. Teams using A2A can add new specialist agents to a fleet without modifying the orchestrator.

What Is the A2A Protocol?

A2A defines four core concepts:

Agent Cards: JSON manifests declaring what an agent does, its inputs, outputs, and endpoint
Task Lifecycle: Submitted → Working → Completed, with webhook notifications at each transition
Discovery Service: A registry where orchestrators find agents by capability
Streaming: SSE endpoints for real-time task progress

This decouples orchestrators from agents. An orchestrator doesn't care which team built an agent — only that it conforms to the schema declared in its agent card.

Agent Cards: Declaring Capabilities

An agent card is a JSON document hosted at GET /agent-card on every A2A-compatible agent:

{
  "name": "email-summarizer",
  "version": "1.2.0",
  "description": "Summarizes email threads into structured decision logs",
  "capabilities": [
    {
      "id": "summarize_email_thread",
      "name": "Summarize Email Thread",
      "description": "Analyzes an email conversation and produces a structured summary"
    },
    {
      "id": "extract_action_items",
      "name": "Extract Action Items",
      "description": "Identifies and returns all action items from a thread"
    }
  ],
  "input_schema": {
    "type": "object",
    "properties": {
      "thread_id": { "type": "string", "description": "Email thread ID" },
      "max_words": { "type": "integer", "description": "Maximum summary length in words" }
    },
    "required": ["thread_id"]
  },
  "output_schema": {
    "type": "object",
    "properties": {
      "summary": { "type": "string" },
      "decisions": { "type": "array", "items": { "type": "string" } },
      "action_items": { "type": "array", "items": { "type": "string" } }
    }
  },
  "endpoint": "https://agents.internal.company.com/email-summarizer",
  "timeout_seconds": 120,
  "retry_policy": {
    "max_retries": 3,
    "backoff_multiplier": 2,
    "initial_delay_ms": 1000
  }
}

Task Lifecycle and Type Definitions

Tasks are the core unit of A2A communication. An orchestrator submits a task and receives a webhook when it completes:

// a2a-types.ts
export type TaskStatus = 'submitted' | 'working' | 'completed' | 'failed'
 
export interface A2ATask {
  task_id: string
  agent_id: string
  capability_id: string
  status: TaskStatus
  input: Record<string, unknown>
  output?: Record<string, unknown>
  error?: string
  created_at: string
  started_at?: string
  completed_at?: string
  webhook_url?: string
}
 
export interface TaskSubmission {
  task_id: string
  capability_id: string
  input: Record<string, unknown>
  webhook_url?: string
  priority?: 'low' | 'normal' | 'high'
}
 
export interface TaskWebhookPayload {
  task_id: string
  agent_id: string
  status: 'success' | 'failure'
  output?: Record<string, unknown>
  error?: string
  completed_at: string
  duration_ms: number
}

Building an A2A-Compatible Agent in Express

// agent-server.ts
import express, { Request, Response } from 'express'
import { A2ATask, TaskSubmission } from './a2a-types'
 
const app = express()
app.use(express.json())
 
// In-memory task store — use Redis or PostgreSQL in production
const tasks = new Map<string, A2ATask>()
 
// Agent card
app.get('/agent-card', (_req: Request, res: Response) => {
  res.json({
    name: 'email-summarizer',
    version: '1.2.0',
    capabilities: [{ id: 'summarize_email_thread' }],
    input_schema: {
      type: 'object',
      properties: {
        thread_id: { type: 'string' },
      },
      required: ['thread_id'],
    },
    endpoint: process.env.AGENT_ENDPOINT_URL,
    timeout_seconds: 120,
  })
})
 
// Health check
app.get('/health', (_req: Request, res: Response) => {
  res.json({ status: 'healthy', agent: 'email-summarizer' })
})
 
// Accept task
app.post('/tasks', async (req: Request, res: Response) => {
  const submission = req.body as TaskSubmission
 
  if (!submission.input?.thread_id) {
    return res.status(400).json({ error: 'thread_id is required in input' })
  }
 
  const task: A2ATask = {
    task_id: submission.task_id,
    agent_id: 'email-summarizer',
    capability_id: submission.capability_id,
    status: 'submitted',
    input: submission.input,
    webhook_url: submission.webhook_url,
    created_at: new Date().toISOString(),
  }
 
  tasks.set(task.task_id, task)
 
  // Acknowledge immediately, process asynchronously
  res.status(202).json({ task_id: task.task_id, status: 'submitted' })
 
  setImmediate(() => processTask(task))
})
 
// Get task status
app.get('/tasks/:taskId', (req: Request, res: Response) => {
  const task = tasks.get(req.params.taskId)
  if (!task) return res.status(404).json({ error: 'Task not found' })
  res.json(task)
})
 
async function processTask(task: A2ATask): Promise<void> {
  const startedAt = new Date().toISOString()
  task.status = 'working'
  task.started_at = startedAt
  tasks.set(task.task_id, task)
 
  const startMs = Date.now()
 
  try {
    const threadId = task.input.thread_id as string
    const maxWords = task.input.max_words as number | undefined
 
    const output = await summarizeEmailThread(threadId, maxWords)
 
    task.status = 'completed'
    task.output = output
    task.completed_at = new Date().toISOString()
    tasks.set(task.task_id, task)
 
    if (task.webhook_url) {
      await sendWebhook(task.webhook_url, {
        task_id: task.task_id,
        agent_id: 'email-summarizer',
        status: 'success',
        output,
        completed_at: task.completed_at,
        duration_ms: Date.now() - startMs,
      })
    }
  } catch (err) {
    task.status = 'failed'
    task.error = (err as Error).message
    task.completed_at = new Date().toISOString()
    tasks.set(task.task_id, task)
 
    if (task.webhook_url) {
      await sendWebhook(task.webhook_url, {
        task_id: task.task_id,
        agent_id: 'email-summarizer',
        status: 'failure',
        error: task.error,
        completed_at: task.completed_at,
        duration_ms: Date.now() - startMs,
      })
    }
  }
}
 
async function sendWebhook(url: string, payload: object): Promise<void> {
  try {
    await fetch(url, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify(payload),
    })
  } catch (err) {
    console.error('Webhook delivery failed:', err)
    // Implement retry with exponential backoff in production
  }
}
 
async function summarizeEmailThread(
  threadId: string,
  maxWords?: number
): Promise<Record<string, unknown>> {
  // Call your LLM or internal service here
  return {
    summary: `Summary of thread ${threadId}`,
    decisions: ['Decision 1', 'Decision 2'],
    action_items: ['Action 1'],
  }
}
 
app.listen(3000, () => console.log('A2A agent running on port 3000'))

Agent Discovery Service

A discovery service lets orchestrators find agents by capability without hard-coded URLs:

// discovery-service.ts
import express from 'express'
 
interface AgentRegistration {
  agent_id: string
  name: string
  endpoint: string
  card_url: string
  capabilities: string[]
  registered_at: string
  last_health_check?: string
  status: 'healthy' | 'unhealthy' | 'unknown'
}
 
const app = express()
app.use(express.json())
 
const registry = new Map<string, AgentRegistration>()
 
// Register a new agent
app.post('/agents/register', (req, res) => {
  const { agent_id, name, endpoint, card_url, capabilities } = req.body
 
  const registration: AgentRegistration = {
    agent_id,
    name,
    endpoint,
    card_url,
    capabilities,
    registered_at: new Date().toISOString(),
    status: 'unknown',
  }
 
  registry.set(agent_id, registration)
  console.log(`Agent registered: ${name} (${agent_id})`)
  res.status(201).json({ agent_id, status: 'registered' })
})
 
// Find agents by capability
app.get('/agents', (req, res) => {
  const { capability } = req.query as { capability?: string }
 
  const results = [...registry.values()].filter((agent) => {
    if (!capability) return true
    return agent.capabilities.includes(capability)
  })
 
  res.json({
    agents: results.filter((a) => a.status !== 'unhealthy'),
    total: results.length,
  })
})
 
// Periodic health checks
setInterval(async () => {
  for (const [agentId, agent] of registry.entries()) {
    try {
      const res = await fetch(`${agent.endpoint}/health`, { signal: AbortSignal.timeout(5000) })
      agent.status = res.ok ? 'healthy' : 'unhealthy'
      agent.last_health_check = new Date().toISOString()
      registry.set(agentId, agent)
    } catch {
      agent.status = 'unhealthy'
      registry.set(agentId, agent)
    }
  }
}, 30_000) // Every 30 seconds
 
app.listen(3001, () => console.log('Discovery service running on port 3001'))

Streaming Task Progress with Server-Sent Events

For long-running tasks, stream progress updates instead of polling:

// SSE endpoint on the agent
app.get('/tasks/:taskId/stream', (req, res) => {
  const taskId = req.params.taskId
 
  res.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    'Connection': 'keep-alive',
    'X-Accel-Buffering': 'no', // Disable nginx buffering
  })
 
  const sendUpdate = () => {
    const task = tasks.get(taskId)
    if (!task) {
      res.write(`event: error\ndata: ${JSON.stringify({ error: 'Task not found' })}\n\n`)
      res.end()
      return clearInterval(interval)
    }
 
    res.write(`event: task-update\ndata: ${JSON.stringify({
      task_id: task.task_id,
      status: task.status,
      output: task.output,
      error: task.error,
    })}\n\n`)
 
    if (task.status === 'completed' || task.status === 'failed') {
      clearInterval(interval)
      res.end()
    }
  }
 
  const interval = setInterval(sendUpdate, 1000)
  sendUpdate() // Send initial state immediately
 
  req.on('close', () => clearInterval(interval))
})
 
// Client usage
const eventSource = new EventSource('/tasks/task-abc123/stream')
eventSource.addEventListener('task-update', (e) => {
  const update = JSON.parse(e.data)
  console.log('Task status:', update.status, update.output)
  if (update.status === 'completed') eventSource.close()
})

A2A vs MCP: When to Use Each

Dimension	A2A	MCP
Communication style	Async, task-based	Sync, tool-call based
Client type	Multiple orchestrators	Single model/client
Discovery	Registry-based	Direct connection
Best for	Agent fleets	Single-model tool access
State management	Agent owns task state	Stateless tool calls

Use MCP when a single AI model needs to call tools (Claude calling search, code execution). Use A2A when independent agent teams need to delegate work to each other across organizational boundaries.

Common Mistakes

Using HTTP 200 for task submissions — always return 202 (Accepted) because processing is async
Storing task state in process memory — agent restarts lose all in-flight tasks
Skipping webhook retries — network failures are common, implement exponential backoff
Not versioning agent cards — breaking changes to input schemas will silently break orchestrators
Missing health check endpoints — discovery services need to monitor agent availability

Best Practices

Return 202 Accepted immediately on task submission; never block the HTTP response while processing
Store task state in Redis or PostgreSQL so tasks survive agent restarts
Implement webhook delivery with at-least-once semantics and idempotency keys on the receiver
Version agent cards with semantic versioning and maintain backward compatibility
Run health checks every 30 seconds and exclude unhealthy agents from discovery results
Add OpenTelemetry tracing with task_id as a span attribute for end-to-end visibility

Key Takeaways

A2A protocol decouples AI agents from their orchestrators using JSON agent cards that declare capabilities, inputs, and outputs
Tasks follow a strict lifecycle: submitted → working → completed, with webhook notifications at each state transition
A discovery service allows orchestrators to find agents by capability without hard-coded endpoint URLs
Always return HTTP 202 immediately on task submission — process tasks asynchronously
Server-Sent Events on /tasks/:id/stream allow real-time progress without polling
A2A is for agent-to-agent delegation across teams; MCP is for a single model calling tools directly
Store task state in Redis or PostgreSQL — in-memory state is lost on agent restarts
Health checks and automatic discovery exclusion prevent failed agents from receiving new tasks

Introduction#

Why This Matters#

What Is the A2A Protocol?#

Agent Cards: Declaring Capabilities#

Task Lifecycle and Type Definitions#

Building an A2A-Compatible Agent in Express#

Agent Discovery Service#

Streaming Task Progress with Server-Sent Events#

A2A vs MCP: When to Use Each#

Common Mistakes#

Best Practices#

Key Takeaways#