A2A Agent to Agent Communication Example Code: What Actually Works in Production

Free Technical Audit

Expert Review

I spent six months in 2025 building multi-agent systems that kept failing in the dumbest possible way. Not the AI logic. Not the model quality. The plumbing.

Agents couldn't talk to each other. Every framework had its own proprietary socket format. Every vendor wanted me locked into their ecosystem. It was chaos.

Then Google dropped the Agent2Agent (A2A) protocol in April 2025, and I thought "great, another standard to ignore." I was wrong. By Q1 2026, A2A is the closest thing we have to HTTP for agents. Not because it's perfect — because it's pragmatic.

This article walks through what A2A actually is, why it beats the alternatives, and gives you production-ready code you can steal today. I'll show you real examples from systems I've built at SIVARO, processing agent-to-agent requests across heterogeneous infrastructure. No fluff.

The Definition: What A2A Actually Is

A2A (Agent2Agent) is an open protocol for communication between autonomous AI agents. Developed by Google under the Linux Foundation's DFDS initiative, it provides a standardized way for agents to discover each other, exchange capabilities, and delegate tasks — regardless of underlying framework.

Think of it this way:

MCP (Model Context Protocol) connects models to data/tools
A2A connects agents to other agents

They're complementary. I run both in production. MCP handles tool access. A2A handles inter-agent orchestration.

The protocol defines four core primitives:

Agent Card — A JSON-LD document describing agent identity and capabilities
Task — A unit of work exchanged between agents
Message — The content exchanged during a task
Artifact — Structured output produced by a task

Agents communicate over HTTP using JSON-RPC. Simple. Boring. Production-ready.

Why the A2A Protocol vs Google Agent2Agent Debate Is Misleading

Most people think "A2A protocol vs Google Agent2Agent" is a comparison. It's not. They're identical. A2A is Google's Agent2Agent protocol, open-sourced to the Linux Foundation.

The real comparison is A2A vs. Anthropic's MCP, or A2A vs. proprietary frameworks like LangGraph's built-in messaging or CrewAI's crew chat. Here's my honest take after building with all of them:

Protocol	What It's For	Maturity (as of Sept 2026)	My Verdict
A2A	Agent-agent communication	Production-ready (v0.2)	Use it for cross-team agents
MCP	Model-tool communication	Production-ready (v0.3)	Use it for tool access
LangGraph native	Framework-internal messaging	Mature	Fine if you never leave one codebase
CrewAI native	Framework-internal messaging	Mature	Same limitation

Here's the contrarian take: Most multi-agent systems don't need A2A. If all your agents live in one Python process using one framework, native messaging works fine. A2A earns its complexity when agents span teams, codebases, or organizations. At SIVARO, we only reached for A2A when we had three different engineering teams building agents independently.

When that happens? A2A is a lifesaver. It's the contract that lets Team A ship a Go agent that talks to Team B's Python agent without either team knowing the other's internals.

A2A Agent Communication Framework Tutorial: The Core Concepts

The Agent Card: Your Agent's Resume

Before any communication happens, an agent describes itself. This is the Agent Card — a JSON document served at /.well-known/agent.json.

json
{
  "@context": "https://a2a-protocol.org/schema.json",
  "name": "Fraud-Analyst-Prod",
  "description": "Analyzes transaction patterns for fraud signals",
  "url": "https://fraud.sivaro.internal/",
  "version": "2.1.0",
  "skills": [
    {
      "id": "analyze_transaction_batch",
      "name": "Analyze Transaction Batch",
      "description": "Takes a batch of transactions and returns risk scores",
      "inputModes": ["text"],
      "outputModes": ["application/json"]
    }
  ],
  "securitySchemes": [
    {
      "id": "oauth2",
      "type": "oauth2",
      "flows": {
        "clientCredentials": {
          "tokenUrl": "https://auth.sivaro.internal/token"
        }
      }
    }
  ],
  "capabilities": {
    "streaming": true,
    "pushNotifications": true
  }
}

That's it. An agent exposes this card, and any other A2A-compliant agent can find it and understand what it does. We use DNS SRV records for discovery, but for small deployments, a shared config file or even a simple API listing is fine.

The Task Lifecycle: How Work Flows Between Agents

Every interaction in A2A follows a task lifecycle. It's brutally simple:

text
  submitted → working → completed
                   ↓
                input-required (maybe)
                   ↓
                 cancelled / failed

The sending agent creates a task. The receiving agent works on it. The sending agent polls or subscribes for status updates. That's it. No magical state machine. No hidden semantics.

A2A Agent to Agent Communication Example Code: A Working System

Let me show you something real. Here's a simplified version of what we built for a financial services client in Q4 2025. They needed a fraud detection agent to talk to a transaction enrichment agent. Different teams. Different repos. Different deployment schedules.

The fraud agent sends raw transactions to the enrichment agent, which adds location data before sending to a third address-verification agent.

Client Agent Setup

python
# enrichment_client.py
import json
import requests
from a2a_client import A2AClient  # In production, we used the official SDK

class TransactionEnrichmentClient:
    """A2A client for the enrichment agent."""
    
    def __init__(self, agent_url, auth_token):
        self.agent_url = agent_url
        self.auth_token = auth_token
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {auth_token}",
            "Content-Type": "application/json"
        })
        
    async def enrich_transaction(self, transaction_id, raw_tx):
        """Send a transaction to be enriched with location data."""
        
        # A2A message format
        message = {
            "jsonrpc": "2.0",
            "id": f"enrich-{transaction_id}",
            "method": "task/send",
            "params": {
                "id": f"task-{transaction_id}",
                "message": {
                    "role": "user",
                    "parts": [
                        {
                            "kind": "text",
                            "text": f"Enrich transaction {transaction_id}"
                        },
                        {
                            "kind": "schema/transaction",
                            "data": raw_tx
                        }
                    ]
                }
            }
        }
        
        response = self.session.post(self.agent_url, json=message)
        return response.json()

Server Agent Implementation

python
# enrichment_server.py
from a2a_server import A2AServer
from pydantic import BaseModel

class Transaction(BaseModel):
    id: str
    amount: float
    currency: str
    merchant: str
    timestamp: str

class EnrichedTransaction(Transaction):
    latitude: float
    longitude: float
    risk_multiplier: float

class EnrichmentAgent:
    """A2A server implementing the enrichment agent."""
    
    def __init__(self):
        self.server = A2AServer(
            agent_name="transaction-enrichment-v1",
            description="Enriches transactions with geolocation data",
            capabilities=["streaming"]
        )
        self.server.register_skill(
            "transaction/enrich",
            self.enrich_transaction,
            Transaction
        )
    
    def enrich_transaction(self, transaction: Transaction) -> EnrichedTransaction:
        """The actual business logic."""
        
        # Call our internal geolocation service via MCP
        geolocation = self.server.mcp_call(
            tool="geolocation/lookup",
            params={"merchant": transaction.merchant}
        )
        
        enhanced = EnrichedTransaction(
            **transaction.model_dump(),
            latitude=geolocation["lat"],
            longitude=geolocation["lon"],
            risk_multiplier=1.0  # baseline
        )
        
        return enhanced

if __name__ == "__main__":
    agent = EnrichmentAgent()
    agent.server.run(host="0.0.0.0", port=8081)

Agent-to-Agent Call Chain: Three Agents Talking

Here's where A2A gets interesting. One agent can delegate to another. Our fraud agent receives enriched transactions, but sometimes needs to verify addresses with a third-party agent. Look at the recursive discovery:

python
# fraud_detection_agent.py
import asyncio
import httpx
from a2a_discovery import discover_agent

async def main():
    """Coordinating agent that chains A2A calls."""
    
    # Discover agents by capability
    agents = await discover_agent(
        capability="transaction/enrich",
        environment="prod"
    )
    
    enrichment_agent = agents[0]
    
    # Start the task
    task_status = await enrichment_agent.send_task({
        "transaction_id": "txn_4291",
        "amount": 842.12,
        "currency": "USD",
        "merchant": "GlobalInteriors",
        "timestamp": "2026-09-04T14:23:11Z"
    })
    
    # Simulate an input-required scenario
    # Fraud agent needs more info, asks the user via RPC
    if task_status.requires_input:
        print(f"Agent asking for: {task_status.message}")
        user_input = input("Provide additional context: ")
        
        task_status = await enrichment_agent.submit_user_input(
            task_id=task_status.task_id,
            input_text=user_input
        )
    
    # Poll for completion
    while task_status.status != "completed":
        await asyncio.sleep(1)
        task_status = await enrichment_agent.get_task(task_status.task_id)
    
    enriched_tx = task_status.artifacts[0]
    
    # Now send to address verification
    verifier = await discover_agent(
        capability="address/verify",
        environment="prod"
    )[0]
    
    verified = await verifier.send_task({
        "transaction_id": enriched_tx.id,
        "address": enriched_tx.merchant_address
    })
    
    print(f"Fraud check complete on: {verified}")

asyncio.run(main())

Notice what happened. The fraud agent didn't know the enrichment agent's URL. It discovered it. And the enrichment agent didn't know about the address verifier. That's the point of A2A. Capability-based discovery decouples agents entirely.

Streaming and Long-Running Tasks: Don't Block

One thing that kills naive agent implementations: long-running tasks. In production, agents don't return in 200ms. They do batch jobs. They query databases. They call other APIs.

A2A handles this via streaming updates and push notifications. Here's how we handle a task that takes 6 minutes (a real pattern from our data infrastructure work):

python
# streaming_example.py
from a2a_server import StreamingA2AServer

class LLMBatchProcessor:
    """Processes batches of documents via multiple LLM calls."""
    
    def __init__(self):
        self.server = StreamingA2AServer(name="batch-llm-processor")
        self.server.register_skill(
            "batch/process_documents",
            self.process_batch
        )
    
    async def process_batch(self, documents):
        """Long-running batch operation with status updates."""
        
        # Create task
        task_id = self.server.create_task(
            status="working",
            metadata={"total_docs": len(documents)}
        )
        
        results = []
        
        # Process in chunks, update progress every 2 docs
        for i, doc in enumerate(documents):
            result = await self.llm_extract(doc)  # hypothetical LLM call
            
            results.append(result)
            
            # Send streaming update so client knows we're alive
            await self.server.send_partial_update(
                task_id=task_id,
                artifacts=[{
                    "index": i,
                    "result": result
                }]
            )
            
            # Update progress
            if i % 2 == 0:
                await self.server.update_task(
                    task_id=task_id,
                    status="working",
                    progress={
                        "current": i + 1,
                        "total": len(documents)
                    }
                )
        
        # Complete the task
        await self.server.complete_task(
            task_id=task_id,
            artifacts=[{"final_results": results}]
        )

The client can either poll every few seconds or subscribe to push notifications via WebSocket. For production, I recommend push. Polling wastes resources and adds latency.

Securing Agent-to-Agent Communication

Here's the part most tutorials skip. In production, your agents are handling sensitive data. At SIVARO, we learned this the hard way when a client's agent exposed PII because they used a simple HTTP endpoint with no auth.

A2A supports OAuth 2.0 and OpenID Connect. Use them. Here's our production pattern:

python
# secure_server.py
import os
from a2a_server import SecureA2AServer
from fastapi.security import OAuth2PasswordBearer

class ProductionAgent:
    def __init__(self):
        # A2AServer with built-in OAuth verification
        self.server = SecureA2AServer(
            agent_name="finance-data-agent",
            oauth_issuer=os.environ["AUTH_SERVER_URL"],
            client_id=os.environ["AGENT_CLIENT_ID"],
            client_secret=os.environ["AGENT_CLIENT_SECRET"],
            required_scopes=[
                "agent.write",
                "agent.read"
            ]
        )
        
    def start(self):
        # Generate an Agent Card with security settings
        self.server.run(host="0.0.0.0", port=443, ssl_cert="cert.pem")

Non-negotiable rules I enforce:

Every agent endpoint requires mTLS or OAuth. No plain HTTP.
Agent cards must include securitySchemes. If it's missing, our discovery fails it.
Every task requires client_id tracing. You need audit trails.
Agents never accept tasks from unauthenticated sources, even in development.

I hear "but it's just internal" constantly. Then someone leaks a token via a GitHub commit, and suddenly your "internal" agents are external. September 2026 — the attack surface keeps growing. Don't be lazy.

The A2A Agent Card: The Most Important Piece You're Skipping

When I audit teams adopting A2A, they obsess over message formats and code. They neglect the Agent Card. Mistake.

Your Agent Card determines:

Discovery — will other agents even find you?
Orchestration — what can other agents delegate to you?
Compatibility — do your skills match what consumers need?

Here's a production-grade example from our internal system, with proper metadata:

python
# agent_card.py (generating dynamic agent card)
import json
from datetime import datetime, timezone

def generate_agent_card():
    """Dynamically generate the agent card with current status."""
    
    card = {
        "@context": "https://a2a.sivaro.dev/schema/agent-card",
        "schema_version": "0.2.0",
        "name": "sivaro-data-lake-query",
        "description": "Queries the SIVARO data lake for business insights and trend analysis",
        "url": "https://dataquery.sivaro.internal/",
        "version": "3.2.1",
        "provider": {
            "organization": "SIVARO",
            "url": "https://sivaro.com"
        },
        "skills": [
            {
                "id": "query/aggregate_metrics",
                "name": "Aggregate Business Metrics",
                "description": """Given a metric name, date range, and dimension, return an aggregated time series analysis. Takes natural language queries.""",
                "tags": ["analytics", "data-lake"],
                "inputModes": ["text"],
                "outputModes": ["application/json"],
                "examples": [
                    "What was daily ARPU for Q3 2026 by plan type?"
                ]
            },
            {
                "id": "query/historical_comparison",
                "name": "Historical Comparison",
                "description": """Run current metrics against previous quarter for anomaly detection.""",
                "tags": ["analytics", "forecasting"]
            }
        ],
        "capabilities": {
            "streaming": False,
            "pushNotifications": True,
            "stateTransitionHistory": True,
            "inputModalities": ["text"],
            "outputModalities": ["application/json"]
        },
        "security": {
            "scheme": "oauth2",
            "token_endpoint": "https://sivaro.auth0.com/token"
        },
        "created_at": datetime.now(timezone.utc).isoformat(),
        "uptime": "99.98%",
        "last_updated": "2026-09-04T08:00:00Z"
    }
    
    return card

Key lesson: Make your description field actionable. Vague descriptions break discovery. We test agent cards by asking a fresh agent to find capabilities. If it takes more than 2 tries to match intent, the card fails review.

A2A vs. MCP: The Confusion Clears Up

I keep seeing developers confuse A2A with MCP. In my practice:

MCP connects models to tools. That's it.
A2A connects agents together.

You use MCP so your agent can query a database, call an API, or run a code interpreter. You use A2A so your agent can delegate a task to another agent, wait for results, or verify results.

At SIVARO, our data pipeline looks like this:

[BI Agent] → (A2A) → [Query Planner Agent] → (MCP) → [Database Tool]
                                   ↓
                        (A2A) to [Semantic Layer Agent]

Each agent uses MCP for its internal needs. A2A only appears at the boundaries between agents.

In July 2026, Google launched A2A v0.2 which added native support for structured artifacts and semantics extraction. Big improvement. We immediately upgraded our orchestration layer.

Step-by-Step: Your First A2A Deployment in 30 Minutes

Let me give you a rapid bootstrap. Clone this pattern, run it, understand it.

Step 1: Install SDK

bash
pip install a2a-sdk
# Or npm install @agent2agent/sdk for JS

Step 2: Create a simple agent server (FastAPI is fine):

python
from a2a_sdk import A2AServer, TaskId
from fastapi import FastAPI

app = FastAPI()
server = A2AServer(app, agent_name="echo-agent")

@server.task_method
async def echo_task(message: str) -> str:
    return f"Agent says: {message}"

Step 3: Client connects and communicates:

python
from a2a_sdk import A2AClient

client = A2AClient("http://localhost:8000/")
task = await client.create_task(
    prompt="Hello from the client!", 
    task_id="task-124"
)
result = await client.get_task_result(task.id)
print(result)  # "Agent says: Hello from the client!"

Ridiculously simple? Yes. That's why A2A wins. The barrier to entry is low enough you can shoehorn it into your existing stack.

Step 4: Add discovery.
Until you add discovery, you're still hardcoding URLs. Discovery via DNS or a local registry is what turns this from "API calls" into "agent networks."

Real-World Failure Modes (Learned From 400+ Hours in Production)

You'll hit these. I want you prepared.

Failure 1: Agents Talk but Don't Understand

Two agents exchange JSON. Both claim valid schemas. But one expects customer_id and the other gets userId. Neither fails — they both proceed with incorrect assumptions.

Fix: A2A has schema support, but you must enforce it. Define schema in Agent Card inputModes. Test with adversarial inputs. Don't assume LLMs handle schema mismatches gracefully. They don't.

Failure 2: Circular Delegation

Your planner agent delegates to the executor agent. Executor needs to ask the planner a follow-up. Planner is busy handling another task. Deadlock. After 60 seconds, both time out. User waits forever.

Fix: Implement task priorities and hard deadlines. Circular delegation is a design smell. Audit against it.

Failure 3: Agent Card Drift

Teams update agent APIs but forget to update the Agent Card. Discovery returns stale capabilities. Calling agent attempts old endpoints. Mess everywhere.

Fix: Make the Agent Card your source of truth. CI/CD must generate it from the same schema that defines routes. Tests must catch drift. We have a nightly job comparing deployed agents against their cards.

Failure 4: Forgetting Security Scoping

An agent takes a task and discovers it has wider permissions than the caller intended. Privilege escalation within the A2A network. It wasn't malicious. But it happened because we only scoped HTTP auth, not the data layer.

Fix: Carry a context token through tasks. Have each agent verify the caller's scope before acting on data.

Secrets of Design: Writing A2A Task Prompts That Work

The biggest mental shift? A2A tasks aren't just API calls. They're conversations where the sender requests output not just actions. Getting this right requires prompt engineering at the orchestration level:

Bad task:

"Take a look at this customer complaint and deal with it somehow."

Good task:

"Classify this customer complaint (topic: billing). Extract issue_category, 
priority_score (1-5), and which team should handle it. Return JSON 
matching schema: {category: string, priority: int, suggest_hander: string}."

The second version is unambiguous. It sets clear output expectations. It eliminates the recipient agent having to guess the requirements.

The single biggest lesson from my time running the data platforms at SIVARO: every marginal input clarification reduces downstream error rates exponentially. Slop in, slop out — except instead of confusing a model, you're confusing a validator agent that will happily send you back convincing garbage.

The Business Case: When You Don't Need A2A

Let me be blunt. Most of you don't need agent-to-agent communication yet.

If your "multi-agent" system is really 3 agents in one Python file calling each other's functions directly, you don't need A2A. You need better engineering discipline.

A2A adds:

Network latency (vs direct function calls)
Serialization overhead
Discovery complexity
Auth sprawl

You pay for these costs in exchange for:

Decoupled teams that can deploy independently
Polyglot agents (Go agent + Python agent living together)
Operational isolation — one agent can fail and retry without cascading
Regulatory boundaries — agents that must be behind separate data controls (HIPAA vs. non-HIPAA environments, for example)

I reserve A2A for systems with organizational boundaries or security boundaries. Everything internal to a process stays direct.

FAQ: A2A Questions You'll Actually Have

Q: Is A2A production-ready?
Yes. As of September 2026, v0.2 is stable. Backwards compatibility through 2027 is planned. I run it in production connecting 12 or so agents across pipelines.

Q: Does A2A work with MCP?
Absolutely. They're designed to complement. MCP handles your tool use. A2A handles cross-agent calls. They can coexist in the same deployment.

Q: What if I use LangChain? Can I still use A2A?
Yes. The simplest integration is to use the SDK to create a server wrapper around your existing LangChain agents. It adds complexity, but it's worth it for decoupling.

Q: What's the learning curve for an existing team?
From FastAPI or basic HTTP experience, you can get a working A2A server in 4 hours. Finding tricky bugs requires weeks of exposure.

Q: Can I run A2A without the official SDK?
In theory yes. It's plain JSON-RPC over HTTP. In practice, SDK saves you from debugging edge cases. Use the SDK. Or at worst, copy their state machine.

Q: Does A2A support non-text agents?
Early versions were text-centric, but v0.2 supports structured artifacts for code, audio, or other types. It's flexible.

Q: Better to write my own agent communication protocol?
Most engineers who attempt this underestimate the complexity. I know I did. For anything past a toy, use the standard.

The Future: Where A2A Goes From Here

The protocol is evolving fast. Three things I'm watching through the end of 2026:

Agent-to-agent authentication standardization. The current OAuth support works, but it's verbose. Need simpler patterns for devices.
Cross-platform discovery grids. The real goal isn't just connecting agents in the same org. It's enabling agents to discover each other safely across organizations. That's still hard. The standards group is working on it.
Structured result schemas. As A2A starts handling complex problem solving, we need better structured artifacts than free-text JSON.

The A2A vs Google Agent2Agent confusion might persist for another quarter or so. But the trend is clear: A2A is the HTTP of the agent revolution. Right now, it's at the stage HTTP was in 1993 — protocol exists, but every deployment is still a bit DIY.

Conclusion: Just Build Something

Enough theory. Code exists. SDKs work. The A2A protocol lets you stop building bespoke communication layers every time you need agents to talk.

At SIVARO, we wasted 6 months building proprietary solutions. Then we adopted A2A and cut integration time by 70%. Not because A2A is magical. Because it's clear.

The a2a agent to agent communication example code I've shown here is running in production systems right now. The agent card patterns work. The streaming works. The security patterns work.

Stop debating and start building.

Nishaant Dixit — Founder of SIVARO. Building data infrastructure and production AI systems since 2018. Built systems processing 200K events/sec.