# Agent-to-Agent (A2A) Communication Pattern

## Overview

The **Agent-to-Agent Communication Pattern** (A2A) enables multiple autonomous agents to exchange messages, coordinate activities, and collaborate on complex tasks through structured communication protocols. Unlike monolithic multi-agent systems where agents are tightly coupled, A2A focuses on the communication infrastructure that allows agents to discover each other, negotiate responsibilities, share information, and coordinate their actions dynamically.

Think of it as building a **communication network** for AI agents—similar to how microservices communicate via APIs, or how humans collaborate through email, chat, and meetings. Each agent maintains its own autonomy while participating in a larger collaborative ecosystem.

## Why Use This Pattern?

Traditional approaches to multi-agent systems often face challenges:

- **Tight coupling**: Agents are hardcoded to work together in specific ways
- **Limited flexibility**: Adding or removing agents requires system redesign
- **Poor scalability**: Communication patterns don't scale beyond a few agents
- **No discovery**: Agents can't dynamically find and collaborate with others
- **Opaque coordination**: Difficult to understand how agents are working together

Agent-to-Agent Communication solves these problems by:

- **Loose coupling**: Agents communicate through standard message protocols
- **Dynamic discovery**: Agents can find and connect with others at runtime
- **Scalable architecture**: Communication patterns work with 2 or 200 agents
- **Explicit protocols**: Clear message formats and interaction patterns
- **Observable communication**: Every message can be logged, traced, and analyzed
- **Fault tolerance**: System continues if individual agents fail

### Real-World Analogy

Imagine a software development company:
- **Without A2A**: The manager directly controls every developer, tester, and designer in a rigid hierarchy
- **With A2A**: Team members communicate via Slack, email, and meetings. They can form working groups, ask questions across teams, negotiate deadlines, and self-organize—while management monitors progress through communication channels

## How It Works

A2A communication involves four key elements:

### 1. Message Protocol

A standard format for agent messages:

```python
class Message:
    sender: str          # Who sent this message
    recipient: str       # Who should receive it (or "broadcast")
    message_type: str    # request, response, notification, query
    content: dict        # The actual message payload
    timestamp: datetime  # When it was sent
    conversation_id: str # Link related messages together
```

### 2. Agent Registry

A directory of available agents and their capabilities:

```python
class AgentRegistry:
    def register(agent_id, capabilities, metadata)
    def discover(required_capability) -> list[Agent]
    def get_agent(agent_id) -> Agent
```

### 3. Communication Channels

Mechanisms for message delivery:
- **Direct messaging**: Point-to-point communication
- **Broadcast**: One-to-many announcements
- **Pub-sub**: Topic-based message distribution
- **Request-reply**: Synchronous question-answer
- **Message queue**: Asynchronous task distribution

### 4. Coordination Patterns

Higher-level interaction protocols:
- **Negotiation**: Agents bid on tasks or resources
- **Consensus**: Agents agree on decisions
- **Workflow**: Structured task handoffs
- **Collaboration**: Joint problem-solving

## Architecture Flow

```
┌─────────────────────────────────────────────────────────────┐
│                    Agent Registry                            │
│  - Agent directory and capabilities                         │
│  - Discovery and lookup services                            │
└─────────────────────────────────────────────────────────────┘
                           │
                           ↓
┌─────────────────────────────────────────────────────────────┐
│                   Message Bus / Broker                       │
│  - Routes messages between agents                           │
│  - Manages pub-sub topics                                   │
│  - Queues and delivery guarantees                           │
└─────────────────────────────────────────────────────────────┘
        │               │               │               │
        ↓               ↓               ↓               ↓
    ┌───────┐      ┌───────┐      ┌───────┐      ┌───────┐
    │Agent A│◄────►│Agent B│◄────►│Agent C│◄────►│Agent D│
    └───────┘      └───────┘      └───────┘      └───────┘
        │               │               │               │
        ↓               ↓               ↓               ↓
    [Tools &        [Tools &        [Tools &        [Tools &
     Actions]        Actions]        Actions]        Actions]
```

## When to Use This Pattern

### ✅ Ideal Use Cases

- **Complex problem-solving**: Tasks requiring multiple specialized skills (research, analysis, writing, review)
- **Distributed systems**: Agents running on different machines or services
- **Dynamic workflows**: Task sequences not known in advance
- **Collaborative tasks**: Multiple agents working toward shared goals
- **Skill-based delegation**: Routing work based on agent capabilities
- **Scalable agent ecosystems**: Systems with many agents joining/leaving
- **Observable multi-agent systems**: Need to track and debug agent interactions
- **Fault-tolerant systems**: Continue operating when agents fail

### ❌ When NOT to Use

- **Simple sequential tasks**: Single agent or basic prompt chain suffices
- **Predetermined workflows**: Use Multi-Agent Collaboration with fixed roles
- **Minimal interaction**: Agents work independently without communication
- **Real-time requirements**: Message passing adds latency overhead
- **Simple tool orchestration**: Regular Tool Use pattern is sufficient
- **Two-agent systems**: Direct communication is simpler than infrastructure

## Rule of Thumb

**Use A2A Communication when:**
1. You have **3+ agents** that need to coordinate dynamically
2. Agent roles and interactions are **not predetermined**
3. You need **observable communication flows** for debugging
4. Agents need to **discover and select** collaborators at runtime
5. System must **scale and adapt** as agents are added/removed

**Don't use A2A Communication when:**
1. You have a **fixed workflow** with known steps (use Multi-Agent Collaboration)
2. Agents **never interact** (use parallel execution)
3. **Latency is critical** (direct calls are faster)
4. You only have **1-2 agents** (overhead not justified)

## Core Components

### 1. Message Types

**Request**: Agent asks another for information or action
```python
{
    "type": "request",
    "action": "research_topic",
    "params": {"topic": "quantum computing"},
    "reply_to": "researcher_001"
}
```

**Response**: Reply to a request
```python
{
    "type": "response",
    "request_id": "req_123",
    "status": "success",
    "data": {...}
}
```

**Notification**: Broadcast information without expecting reply
```python
{
    "type": "notification",
    "event": "task_completed",
    "details": {...}
}
```

**Query**: Ask about agent capabilities or status
```python
{
    "type": "query",
    "query": "who_can",
    "capability": "code_review"
}
```

### 2. Communication Patterns

#### Direct Messaging (Point-to-Point)
```
Agent A ──[message]──> Agent B
Agent A <──[reply]──── Agent B
```

**Use when**: Specific agent-to-agent interaction needed

#### Broadcast (One-to-Many)
```
Agent A ──[announcement]──┬──> Agent B
                          ├──> Agent C
                          └──> Agent D
```

**Use when**: Information relevant to multiple agents

#### Pub-Sub (Topic-Based)
```
Publisher ──[topic:research]──> Broker ──> Subscriber 1
                                       └──> Subscriber 2
```

**Use when**: Agents subscribe to specific types of information

#### Request-Reply (Synchronous)
```
Requester ──[query]──> Service
Requester <──[result]─┘
```

**Use when**: Need immediate response from specific agent

#### Negotiation (Multi-Round)
```
Agent A ──[proposal]──> Agent B
Agent A <──[counter]─── Agent B
Agent A ──[accept]───> Agent B
```

**Use when**: Agents need to agree on terms, pricing, or allocation

### 3. Agent Discovery

Agents can find collaborators dynamically:

```python
# Agent needs a code reviewer
available_reviewers = registry.discover(capability="code_review")

# Select based on criteria (load, expertise, etc.)
best_reviewer = select_best(available_reviewers, criteria="python_expert")

# Send request
send_message(best_reviewer, request="review_code", code=...)
```

### 4. Conversation Tracking

Group related messages into conversations:

```python
conversation_id = "conv_789"

# Initial request
send(recipient="agent_b", conversation_id=conversation_id,
     content={"action": "start_research"})

# Follow-up messages
send(recipient="agent_b", conversation_id=conversation_id,
     content={"action": "provide_more_context", ...})

# Get conversation history
history = get_conversation(conversation_id)
```

## Implementation Approaches

### Approach 1: Simple Message Passing

Direct agent-to-agent communication with a central coordinator:

```python
class MessageBus:
    def __init__(self):
        self.agents = {}

    def register(self, agent_id, agent):
        self.agents[agent_id] = agent

    def send(self, sender, recipient, message):
        if recipient in self.agents:
            self.agents[recipient].receive(sender, message)

# Usage
bus = MessageBus()
bus.register("researcher", ResearchAgent())
bus.register("writer", WriterAgent())

bus.send("researcher", "writer",
         {"type": "data", "findings": [...]})
```

### Approach 2: LangGraph with Custom State

Use LangGraph's state management for message passing:

```python
class AgentCommunicationState(TypedDict):
    messages: Annotated[list, add_messages]
    message_bus: dict  # sender -> recipient -> messages
    agent_statuses: dict
    current_conversation: str

def agent_node(agent_name: str):
    def node(state: AgentCommunicationState):
        # Check for messages to this agent
        inbox = state["message_bus"].get(agent_name, [])

        # Process messages and generate response
        response = agent.process(inbox)

        # Send messages to other agents
        outbox = response.get("outgoing_messages", [])

        return {
            "message_bus": update_message_bus(outbox),
            "messages": [response["output"]]
        }
    return node
```

### Approach 3: Pub-Sub Architecture

Topic-based message distribution:

```python
class PubSubBroker:
    def __init__(self):
        self.subscribers = {}  # topic -> [agents]

    def subscribe(self, agent_id, topic):
        if topic not in self.subscribers:
            self.subscribers[topic] = []
        self.subscribers[topic].append(agent_id)

    def publish(self, topic, message):
        for agent_id in self.subscribers.get(topic, []):
            self.deliver(agent_id, message)

# Usage
broker = PubSubBroker()
broker.subscribe("writer", "research_completed")
broker.subscribe("reviewer", "research_completed")

broker.publish("research_completed",
               {"findings": [...], "confidence": 0.95})
```

### Approach 4: Negotiation Protocol

Agents negotiate task assignment:

```python
def negotiation_protocol(task, available_agents):
    # Step 1: Broadcast task
    responses = broadcast_cfp(task)  # Call for Proposals

    # Step 2: Collect bids
    bids = []
    for agent, response in responses:
        if response["interested"]:
            bids.append({
                "agent": agent,
                "cost": response["cost"],
                "time": response["estimated_time"],
                "confidence": response["confidence"]
            })

    # Step 3: Select winner
    winner = select_best_bid(bids, criteria="balanced")

    # Step 4: Award contract
    send_message(winner["agent"],
                 {"type": "task_awarded", "task": task})

    return winner["agent"]
```

## Key Benefits

### 🔗 Loose Coupling
- **Independent development**: Agents can be built and updated separately
- **Mix and match**: Combine agents from different sources
- **Easy testing**: Test agents in isolation with mock messages
- **Version compatibility**: Agents with different implementations can coexist

### 📡 Dynamic Discovery
- **Runtime flexibility**: Agents find collaborators when needed
- **Capability-based routing**: Select agents by skills, not hardcoded IDs
- **Load balancing**: Distribute work to available agents
- **Fault tolerance**: Automatically find replacements if agents fail

### 📊 Observability
- **Message tracing**: Track every communication between agents
- **Conversation logs**: See full context of interactions
- **Performance monitoring**: Measure message latency and throughput
- **Debugging support**: Replay conversations to diagnose issues

### ⚡ Scalability
- **Add agents dynamically**: No system redesign needed
- **Horizontal scaling**: Run more agent instances for load
- **Distributed deployment**: Agents can run on different machines
- **Efficient resource use**: Agents activate only when needed

### 🛡️ Fault Tolerance
- **Graceful degradation**: System continues if agents fail
- **Message queuing**: Don't lose messages if recipient is busy
- **Retry logic**: Automatically resend failed messages
- **Circuit breakers**: Detect and isolate failing agents

## Trade-offs

### ⚠️ Increased Complexity

**Issue**: More moving parts to manage and coordinate

**Impact**: Harder to understand system behavior, more debugging surface

**Mitigation**:
- Start simple with direct messaging before adding pub-sub
- Use visualization tools to map agent interactions
- Implement comprehensive logging and tracing
- Create clear documentation of message protocols
- Use typed message schemas for validation

### 💬 Message Overhead

**Issue**: Message serialization, routing, and delivery add latency

**Impact**: Slower than direct function calls, more network traffic

**Mitigation**:
- Use efficient message formats (Protocol Buffers, MessagePack)
- Implement message batching for bulk operations
- Cache frequently accessed agent information
- Use async messaging to avoid blocking
- Optimize hot paths with direct connections

### 🔄 Coordination Challenges

**Issue**: Deadlocks, race conditions, inconsistent state

**Impact**: Agents may wait forever or produce incorrect results

**Mitigation**:
- Implement timeouts on all requests
- Use conversation IDs to track related messages
- Add explicit coordination protocols (locks, semaphores)
- Design idempotent message handlers
- Use saga patterns for distributed transactions

### 🐛 Debugging Difficulty

**Issue**: Errors spread across multiple agents with async communication

**Impact**: Hard to trace root cause of failures

**Mitigation**:
- Implement distributed tracing (correlation IDs)
- Use centralized logging with structured data
- Create message replay capabilities
- Build visualization tools for agent communication
- Add health checks and status endpoints

## Best Practices

### 1. Design Clear Message Protocols

```python
from typing import Literal
from pydantic import BaseModel

class AgentMessage(BaseModel):
    """Strongly typed message format"""
    sender: str
    recipient: str
    message_type: Literal["request", "response", "notification"]
    content: dict
    conversation_id: str
    timestamp: datetime

    def validate_content(self):
        """Validate message-type-specific content"""
        if self.message_type == "request":
            assert "action" in self.content
        elif self.message_type == "response":
            assert "request_id" in self.content
```

### 2. Implement Timeouts and Retries

```python
async def send_with_timeout(recipient, message, timeout=30):
    """Send message with timeout"""
    try:
        response = await asyncio.wait_for(
            send_message(recipient, message),
            timeout=timeout
        )
        return response
    except asyncio.TimeoutError:
        logger.error(f"Timeout sending to {recipient}")
        # Retry with exponential backoff
        return await retry_with_backoff(recipient, message)
```

### 3. Use Conversation IDs

```python
class ConversationManager:
    def __init__(self):
        self.conversations = {}

    def start_conversation(self, participants):
        """Start new conversation thread"""
        conv_id = generate_uuid()
        self.conversations[conv_id] = {
            "participants": participants,
            "messages": [],
            "started": datetime.now()
        }
        return conv_id

    def add_message(self, conv_id, message):
        """Add message to conversation history"""
        self.conversations[conv_id]["messages"].append(message)

    def get_history(self, conv_id):
        """Retrieve full conversation"""
        return self.conversations[conv_id]["messages"]
```

### 4. Implement Agent Discovery

```python
class AgentRegistry:
    def __init__(self):
        self.agents = {}  # agent_id -> metadata

    def register(self, agent_id, capabilities, metadata):
        """Register agent with capabilities"""
        self.agents[agent_id] = {
            "capabilities": capabilities,
            "metadata": metadata,
            "status": "available",
            "registered_at": datetime.now()
        }

    def discover(self, required_capability):
        """Find agents with specific capability"""
        return [
            agent_id for agent_id, info in self.agents.items()
            if required_capability in info["capabilities"]
            and info["status"] == "available"
        ]

    def get_agent_info(self, agent_id):
        """Get agent metadata"""
        return self.agents.get(agent_id)
```

### 5. Log All Communication

```python
class MessageLogger:
    def __init__(self):
        self.logs = []

    def log_message(self, message):
        """Log message with metadata"""
        log_entry = {
            "timestamp": datetime.now(),
            "sender": message["sender"],
            "recipient": message["recipient"],
            "type": message["message_type"],
            "conversation_id": message["conversation_id"],
            "content_summary": summarize(message["content"])
        }
        self.logs.append(log_entry)
        logger.info(f"Message: {message['sender']} -> {message['recipient']}")

    def get_conversation_log(self, conv_id):
        """Get all messages in conversation"""
        return [
            log for log in self.logs
            if log["conversation_id"] == conv_id
        ]
```

### 6. Handle Failures Gracefully

```python
def send_message_safely(recipient, message):
    """Send with error handling"""
    try:
        response = send_message(recipient, message)
        return {"status": "success", "response": response}

    except AgentNotFoundError:
        logger.error(f"Agent {recipient} not found")
        # Try to find alternative agent
        alternatives = registry.discover(required_capability)
        if alternatives:
            return send_message_safely(alternatives[0], message)
        return {"status": "error", "reason": "no_agent_available"}

    except TimeoutError:
        logger.error(f"Timeout sending to {recipient}")
        return {"status": "error", "reason": "timeout"}

    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        return {"status": "error", "reason": str(e)}
```

## Performance Metrics

Track these metrics for A2A communication systems:

- **Message latency**: Time from send to receive (p50, p95, p99)
- **Message throughput**: Messages per second across system
- **Agent response time**: Time for agent to process and reply
- **Conversation duration**: Time to complete multi-turn interactions
- **Discovery latency**: Time to find and connect to agents
- **Success rate**: % of messages delivered successfully
- **Error rate**: Failed messages, timeouts, retries
- **Queue depth**: Pending messages waiting for delivery
- **Agent utilization**: % of time agents are actively processing
- **Coordination efficiency**: Time in coordination vs. actual work

## Example Scenarios

### Scenario 1: Research Collaboration

**Task**: Write a research report on AI safety

**Agents**:
- Research Agent: Gather information from sources
- Analysis Agent: Analyze and synthesize findings
- Writer Agent: Draft the report
- Reviewer Agent: Review for quality and accuracy

**Communication Flow**:
```
1. Coordinator broadcasts: "New task: AI safety research"
2. Research Agent subscribes to topic "AI_safety"
3. Research Agent: Gathers information, publishes "research_completed"
4. Analysis Agent: Receives research, sends request to Research Agent: "clarify_point"
5. Research Agent: Responds with clarification
6. Analysis Agent: Publishes "analysis_completed"
7. Writer Agent: Receives analysis, drafts report
8. Writer Agent: Sends "review_request" to Reviewer
9. Reviewer: Sends "revision_needed" with feedback
10. Writer Agent: Revises and sends "final_draft"
11. Reviewer: Sends "approved"
```

### Scenario 2: Software Development Team

**Task**: Implement a new feature

**Agents**:
- Product Manager: Defines requirements
- Developer: Writes code
- Tester: Tests functionality
- Reviewer: Reviews code quality

**Negotiation Protocol**:
```
1. PM: Broadcasts "feature_request" to topic "development"
2. Multiple Developers: Send bids with time estimates
3. PM: Selects developer based on expertise and availability
4. PM: Sends "task_assigned" to selected Developer
5. Developer: Periodically publishes "progress_update"
6. Developer: Sends "ready_for_review" to Reviewer
7. Reviewer: Sends "changes_requested" with feedback
8. Developer: Makes changes, sends "review_again"
9. Reviewer: Sends "approved"
10. Developer: Sends "ready_for_testing" to Tester
11. Tester: Sends "test_results" (pass/fail)
12. If pass, Tester sends "deployment_approved"
```

### Scenario 3: Customer Support Escalation

**Task**: Handle complex customer issue

**Agents**:
- Frontline Agent: Initial customer interaction
- Specialist Agents: Domain experts (billing, technical, account)
- Escalation Agent: Handles difficult cases

**Dynamic Routing**:
```
1. Customer: Sends query to Frontline Agent
2. Frontline: Classifies issue type (technical problem)
3. Frontline: Discovers agents with capability "technical_support"
4. Frontline: Sends "handoff" to Technical Agent
5. Technical: Requests information from Account Agent: "get_customer_history"
6. Account: Responds with customer data
7. Technical: Attempts solution, sends "resolution_proposed"
8. Customer: "Not satisfied"
9. Technical: Sends "escalate" to Escalation Agent
10. Escalation: Broadcasts "urgent_help" to all specialists
11. Multiple agents collaborate on solution
12. Escalation: Sends final resolution to customer
```

## Advanced Patterns

### 1. Consensus Protocol

Agents vote on decisions:

```python
def consensus_protocol(proposal, agents, threshold=0.66):
    """Get consensus from multiple agents"""
    # Phase 1: Broadcast proposal
    votes = []
    for agent in agents:
        response = send_message(agent, {
            "type": "vote_request",
            "proposal": proposal
        })
        votes.append(response["vote"])  # "approve" or "reject"

    # Phase 2: Count votes
    approvals = sum(1 for v in votes if v == "approve")
    approval_rate = approvals / len(votes)

    # Phase 3: Decision
    if approval_rate >= threshold:
        # Notify all agents of decision
        for agent in agents:
            send_message(agent, {
                "type": "notification",
                "decision": "approved",
                "approval_rate": approval_rate
            })
        return True
    else:
        return False
```

### 2. Workflow Orchestration

Define multi-stage workflows:

```python
class WorkflowOrchestrator:
    def __init__(self, workflow_definition):
        self.workflow = workflow_definition
        self.state = {}

    def execute(self, initial_input):
        """Execute workflow stages"""
        current_data = initial_input

        for stage in self.workflow["stages"]:
            # Find capable agent
            agent = self.find_agent(stage["required_capability"])

            # Send work to agent
            response = send_message(agent, {
                "type": "request",
                "action": stage["action"],
                "data": current_data
            })

            # Update data for next stage
            current_data = response["output"]
            self.state[stage["name"]] = response

        return current_data

# Usage
workflow = {
    "stages": [
        {"name": "research", "required_capability": "research", "action": "gather_info"},
        {"name": "analyze", "required_capability": "analysis", "action": "synthesize"},
        {"name": "write", "required_capability": "writing", "action": "draft_report"},
        {"name": "review", "required_capability": "review", "action": "quality_check"}
    ]
}

orchestrator = WorkflowOrchestrator(workflow)
result = orchestrator.execute({"topic": "AI safety"})
```

### 3. Blackboard System

Shared knowledge space for collaboration:

```python
class Blackboard:
    """Shared knowledge space for agents"""
    def __init__(self):
        self.knowledge = {}
        self.subscribers = {}

    def write(self, key, value, author):
        """Write to blackboard"""
        self.knowledge[key] = {
            "value": value,
            "author": author,
            "timestamp": datetime.now()
        }
        # Notify subscribers
        self.notify_subscribers(key)

    def read(self, key):
        """Read from blackboard"""
        return self.knowledge.get(key)

    def subscribe(self, agent_id, key_pattern):
        """Subscribe to updates"""
        if key_pattern not in self.subscribers:
            self.subscribers[key_pattern] = []
        self.subscribers[key_pattern].append(agent_id)

    def notify_subscribers(self, key):
        """Notify agents of updates"""
        for pattern, agents in self.subscribers.items():
            if matches_pattern(key, pattern):
                for agent in agents:
                    send_notification(agent, {
                        "event": "blackboard_updated",
                        "key": key
                    })
```

### 4. Agent Marketplace

Agents bid on tasks:

```python
class AgentMarketplace:
    def __init__(self):
        self.active_tasks = {}
        self.agent_profiles = {}

    def post_task(self, task):
        """Post task for bidding"""
        task_id = generate_uuid()
        self.active_tasks[task_id] = {
            "task": task,
            "bids": [],
            "status": "open"
        }

        # Broadcast to all agents
        broadcast_message({
            "type": "task_posted",
            "task_id": task_id,
            "task": task,
            "deadline": datetime.now() + timedelta(minutes=5)
        })

        return task_id

    def submit_bid(self, task_id, agent_id, bid):
        """Agent submits bid"""
        self.active_tasks[task_id]["bids"].append({
            "agent_id": agent_id,
            "cost": bid["cost"],
            "time": bid["estimated_time"],
            "quality": bid["quality_score"],
            "submitted_at": datetime.now()
        })

    def award_task(self, task_id, selection_criteria="balanced"):
        """Select winning bid"""
        bids = self.active_tasks[task_id]["bids"]

        # Score bids based on criteria
        winner = select_best_bid(bids, criteria=selection_criteria)

        # Award task
        send_message(winner["agent_id"], {
            "type": "task_awarded",
            "task_id": task_id,
            "task": self.active_tasks[task_id]["task"]
        })

        # Notify losers
        for bid in bids:
            if bid["agent_id"] != winner["agent_id"]:
                send_message(bid["agent_id"], {
                    "type": "bid_rejected",
                    "task_id": task_id
                })

        return winner["agent_id"]
```

## Comparison with Related Patterns

| Pattern | Focus | Communication | Agent Roles | When to Use |
|---------|-------|---------------|-------------|-------------|
| **A2A Communication** | Message infrastructure | Explicit protocol | Dynamic discovery | 3+ agents, dynamic coordination |
| **Multi-Agent Collaboration** | Team workflow | Implicit/predetermined | Fixed roles | Known workflow, stable team |
| **MCP (Model Context Protocol)** | Tool/resource access | Request-response | Client-server | External tool integration |
| **Goal Management** | Task planning | Hierarchical | Goal-oriented | Complex planning problems |
| **ReAct** | Reasoning + acting | Internal (single agent) | One agent | Single agent with tools |

### Key Differences

**A2A vs Multi-Agent Collaboration**:
- A2A: Focus on communication infrastructure, agents discover each other
- MAC: Focus on workflow, agents have predetermined roles

**A2A vs MCP**:
- A2A: Agent-to-agent communication, collaborative
- MCP: Client-tool communication, request-response

**A2A vs Goal Management**:
- A2A: How agents communicate
- Goal Management: What agents should achieve

## Common Pitfalls

### 1. Over-Engineering Communication

**Problem**: Building complex message systems for simple tasks

**Solution**: Start with direct messaging, add complexity only when needed

```python
# Bad: Over-engineered for simple task
broker = PubSubBroker()
registry = AgentRegistry()
queue = MessageQueue()
orchestrator = WorkflowEngine()

# Good: Simple direct communication
result = agent_b.process(agent_a.output())
```

### 2. Missing Error Handling

**Problem**: No handling for agent failures or timeouts

**Solution**: Implement comprehensive error handling

```python
# Bad: No error handling
response = send_message(agent_id, message)

# Good: Proper error handling
try:
    response = send_message_with_timeout(agent_id, message, timeout=30)
except AgentNotAvailableError:
    alternative = registry.find_alternative(required_capability)
    response = send_message_with_timeout(alternative, message, timeout=30)
except TimeoutError:
    logger.error(f"Timeout communicating with {agent_id}")
    response = None
```

### 3. Unbounded Message Queues

**Problem**: Message queues grow without limit, causing memory issues

**Solution**: Implement queue limits and backpressure

```python
class BoundedMessageQueue:
    def __init__(self, max_size=1000):
        self.queue = deque(maxlen=max_size)
        self.dropped = 0

    def enqueue(self, message):
        if len(self.queue) >= self.queue.maxlen:
            self.dropped += 1
            logger.warning("Queue full, dropping message")
            return False
        self.queue.append(message)
        return True
```

### 4. No Message Ordering Guarantees

**Problem**: Messages arrive out of order, causing confusion

**Solution**: Use sequence numbers and conversation IDs

```python
class OrderedMessage:
    def __init__(self, content, conversation_id):
        self.content = content
        self.conversation_id = conversation_id
        self.sequence = next_sequence_number()

    def process_in_order(self, messages):
        """Process messages in sequence order"""
        sorted_messages = sorted(messages, key=lambda m: m.sequence)
        for message in sorted_messages:
            self.handle_message(message)
```

### 5. Circular Communication Dependencies

**Problem**: Agent A waits for B, B waits for C, C waits for A (deadlock)

**Solution**: Implement timeouts and detect cycles

```python
def detect_deadlock(waiting_graph):
    """Detect circular dependencies in agent communication"""
    def has_cycle(node, visited, rec_stack):
        visited.add(node)
        rec_stack.add(node)

        for neighbor in waiting_graph.get(node, []):
            if neighbor not in visited:
                if has_cycle(neighbor, visited, rec_stack):
                    return True
            elif neighbor in rec_stack:
                return True

        rec_stack.remove(node)
        return False

    visited = set()
    rec_stack = set()

    for node in waiting_graph:
        if node not in visited:
            if has_cycle(node, visited, rec_stack):
                logger.error(f"Deadlock detected involving {node}")
                return True
    return False
```

## Conclusion

The Agent-to-Agent Communication pattern provides the infrastructure for building scalable, flexible, and observable multi-agent systems. By focusing on explicit message protocols, dynamic discovery, and well-defined coordination patterns, it enables agents to collaborate effectively without tight coupling.

**Use A2A Communication when:**
- Building systems with multiple autonomous agents
- Need dynamic agent discovery and capability-based routing
- Require observable and traceable agent interactions
- Want loose coupling and independent agent development
- Need fault tolerance and graceful degradation
- Building distributed or scalable agent systems

**Implementation checklist:**
- ✅ Define clear message protocols and types
- ✅ Implement agent registry with discovery
- ✅ Choose appropriate communication patterns (direct, pub-sub, etc.)
- ✅ Add conversation tracking and history
- ✅ Implement timeouts and error handling
- ✅ Log all communication for debugging
- ✅ Monitor message latency and throughput
- ✅ Design for fault tolerance and recovery
- ✅ Start simple, add complexity as needed
- ✅ Test communication patterns in isolation

**Key Takeaways:**
- 📡 A2A enables dynamic, flexible agent collaboration
- 🔗 Loose coupling allows independent agent development
- 📊 Explicit messaging makes systems observable and debuggable
- ⚖️ Balance between flexibility and complexity
- 🛠️ LangGraph provides excellent state management for A2A
- 🎯 Start with simple patterns, evolve as needed
- 🔍 Always log and trace communications
- 🛡️ Design for failures and timeouts

---

*Agent-to-Agent Communication transforms isolated AI agents into collaborative ecosystems—enabling them to discover, coordinate, and work together dynamically while maintaining independence and observability.*

## Corporate SSL proxy note

If you're behind a corporate SSL-inspecting proxy, run examples with:

```bash
AGENTIC_DISABLE_SSL=1 bash run.sh
```

