# Multi-Agent Communication Protocols

Standardized communication between agents: message formats, protocols, and patterns.

---

## Why Standardized Communication?

| Without Standard | With Standard |
|---|---|
| Custom protocols per framework | Interoperable agents |
| Tight coupling | Loose coupling |
| Hard to compose | Composable systems |
| Vendor lock-in | Portable agents |
| Debugging nightmare | Observable, traceable |

---

## Message Format

### Envelope
```json
{
  "protocol": "a2a/1.0",
  "message_id": "uuid-v4",
  "timestamp": "2024-01-15T10:30:00Z",
  "from": "agent://researcher-001",
  "to": "agent://coder-002",
  "reply_to": "agent://orchestrator-001",
  "performative": "request",
  "content_type": "application/json",
  "content": {
    "task": "Implement user authentication",
    "context": {...},
    "constraints": [...]
  },
  "metadata": {
    "priority": "normal",
    "timeout_seconds": 300,
    "requires_ack": true,
    "correlation_id": "workflow-123"
  }
}
```

### Performatives (Speech Acts)
| Performative | Meaning | Response Expected |
|---|---|---|
| **request** | Ask agent to perform action | Yes (ack + result) |
| **inform** | Share information/facts | No (optional ack) |
| **query** | Ask question | Yes (answer) |
| **propose** | Offer a plan/deal | Yes (accept/reject/counter) |
| **accept** | Accept proposal | No |
| **reject** | Reject proposal | No |
| **counter** | Counter-proposal | Yes (accept/reject) |
| **subscribe** | Request notifications | Yes (confirmation) |
| **notify** | Push notification | No |
| **cancel** | Cancel previous request | Yes (confirmation) |

### Content Types
| Type | Schema | Use Case |
|---|---|---|
| `task` | TaskSchema | Delegate work |
| `query` | QuerySchema | Ask question |
| `result` | ResultSchema | Return result |
| `error` | ErrorSchema | Report failure |
| `plan` | PlanSchema | Share plan |
| `context` | ContextSchema | Share context |
| `approval` | ApprovalSchema | Request approval |

---

## Transport Protocols

### 1. HTTP/REST (Request-Response)
```python
# Agent as HTTP server
@app.post("/a2a/message")
async def receive_message(message: A2AMessage):
    # Process message
    result = await agent.handle(message)
    return {"status": "accepted", "message_id": message.message_id}

# Client
async def send_message(to_url: str, message: A2AMessage):
    async with httpx.AsyncClient() as client:
        response = await client.post(f"{to_url}/a2a/message", json=message.model_dump())
        return response.json()
```

### 2. WebSocket (Persistent, Bidirectional)
```python
# Server
@app.websocket("/a2a/ws/{agent_id}")
async def websocket_endpoint(websocket: WebSocket, agent_id: str):
    await websocket.accept()
    connection_manager.connect(agent_id, websocket)
    try:
        while True:
            data = await websocket.receive_json()
            message = A2AMessage(**data)
            await handle_message(message)
    except WebSocketDisconnect:
        connection_manager.disconnect(agent_id)

# Client
async def connect_agent(agent_url: str):
    async with websockets.connect(f"{agent_url}/a2a/ws/my-agent") as ws:
        await ws.send_json(message.model_dump())
        response = await ws.recv_json()
```

### 3. Message Queue (Async, Durable)
```python
# Using Redis Streams / Kafka / RabbitMQ
class MessageQueueTransport:
    def __init__(self, redis: redis.Redis):
        self.redis = redis
        self.consumer_group = "agents"
    
    async def send(self, to_agent: str, message: A2AMessage):
        stream = f"a2a:{to_agent}"
        await self.redis.xadd(stream, message.model_dump())
    
    async def receive(self, agent_id: str, consumer: str) -> AsyncIterator[A2AMessage]:
        stream = f"a2a:{agent_id}"
        while True:
            messages = await self.redis.xreadgroup(
                self.consumer_group, consumer, {stream: ">"}, count=10, block=5000
            )
            for stream_name, messages in messages:
                for msg_id, data in messages:
                    yield A2AMessage(**data)
                    await self.redis.xack(stream, self.consumer_group, msg_id)
```

### 4. gRPC (High Performance, Typed)
```protobuf
// a2a.proto
syntax = "proto3";

package a2a;

service AgentCommunication {
  rpc SendMessage(Message) returns (Ack);
  rpc StreamMessages(stream Message) returns (stream Message);
}

message Message {
  string protocol = 1;
  string message_id = 2;
  int64 timestamp = 3;
  string from_agent = 4;
  string to_agent = 5;
  string reply_to = 6;
  string performative = 7;
  string content_type = 8;
  bytes content = 9;
  map<string, string> metadata = 10;
}

message Ack {
  string message_id = 1;
  string status = 2;
}
```

---

## Agent Discovery

### Registry
```python
class AgentRegistry:
    def __init__(self):
        self.agents: dict[str, AgentInfo] = {}
    
    def register(self, agent: AgentInfo):
        self.agents[agent.agent_id] = agent
    
    def discover(self, capability: str = None, tags: list[str] = None) -> list[AgentInfo]:
        candidates = self.agents.values()
        if capability:
            candidates = [a for a in candidates if capability in a.capabilities]
        if tags:
            candidates = [a for a in candidates if set(tags).issubset(a.tags)]
        return list(candidates)
    
    def resolve(self, agent_id: str) -> AgentInfo | None:
        return self.agents.get(agent_id)

@dataclass
class AgentInfo:
    agent_id: str
    name: str
    description: str
    endpoint: str  # HTTP, WebSocket, gRPC
    capabilities: list[str]
    tags: list[str]
    version: str
    metadata: dict
```

### Discovery Protocols
| Protocol | Mechanism |
|---|---|
| **mDNS/Bonjour** | Local network discovery |
| **Consul/etcd** | Service registry |
| **Kubernetes** | Service discovery via DNS |
| **Custom HTTP** | `/a2a/agents` endpoint |
| **DNS-SD** | DNS-based |

---

## Conversation Patterns

### 1. Request-Response
```python
# Simple delegation
orchestrator -> researcher: request(task="Find papers on transformers")
researcher -> orchestrator: result(papers=[...])
```

### 2. Streaming/Progress
```python
# Long-running task with updates
orchestrator -> coder: request(task="Build API", stream=true)
coder -> orchestrator: progress(step="writing tests", percent=25)
coder -> orchestrator: progress(step="implementing", percent=60)
coder -> orchestrator: result(code="...", tests_pass=true)
```

### 3. Publish-Subscribe
```python
# Event broadcasting
data_agent -> topic "data.updated": notify(data_id="123", changes={...})
analytics_agent <- topic "data.updated": notify (subscribed)
dashboard_agent <- topic "data.updated": notify (subscribed)
```

### 4. Negotiation
```python
# Contract negotiation
buyer -> seller: propose(price=100, quantity=10)
seller -> buyer: counter(price=120, quantity=10)
buyer -> seller: accept
```

---

## Security

### Authentication
```python
# Mutual TLS
ssl_context = ssl.create_default_context()
ssl_context.load_cert_chain("agent.pem", "agent.key")
ssl_context.load_verify_locations("ca.pem")

# Bearer tokens
headers = {"Authorization": "Bearer <agent-token>"}

# Message signing
signed_message = sign(message, private_key)
```

### Authorization
```python
# Capability-based
message.metadata["capabilities"] = ["read:data", "write:reports"]

# Role-based
message.metadata["roles"] = ["analyst", "reporter"]

# Policy engine
if not policy_engine.check(sender, message.performative, message.content):
    raise PermissionError("Not authorized")
```

### Encryption
```python
# End-to-end encryption
encrypted_content = encrypt(message.content, recipient_public_key)
message.content = encrypted_content
message.content_type = "application/encrypted"

# Decrypt on receive
decrypted = decrypt(message.content, private_key)
```

---

## Observability

### Tracing
```python
# Distributed tracing headers
message.metadata["trace_id"] = "abc123"
message.metadata["span_id"] = "def456"
message.metadata["parent_span_id"] = "parent789"

# OpenTelemetry integration
with tracer.start_as_current_span("a2a.send") as span:
    span.set_attribute("a2a.to", to_agent)
    span.set_attribute("a2a.performative", performative)
    await transport.send(message)
```

### Metrics
```python
MESSAGES_SENT = Counter("a2a_messages_sent_total", "Messages sent", ["to", "performative", "status"])
MESSAGE_LATENCY = Histogram("a2a_message_latency_seconds", "Message latency", ["to", "performative"])
MESSAGE_SIZE = Histogram("a2a_message_size_bytes", "Message size", ["performative"])
```

---

## Frameworks Implementing A2A

| Framework | Protocol | Features |
|---|---|---|
| **LangGraph** | Custom | Subgraphs, `send()`, interrupts |
| **AutoGen** | Custom | GroupChat, nested chats |
| **OpenAI Agents SDK** | Custom | Handoffs, function tools |
| **Custom** | A2A/JSON-RPC | Full control |

---

## Related Notes

- [[MultiAgent/Coordination-Patterns.md]] — Coordination using communication
- [[MultiAgent/Negotiation-Consensus.md]] — Negotiation protocols
- [[MultiAgent/Delegation-Handoff.md]] — Handoff patterns
- [[MultiAgent/Safety-Coordination.md]] — Safe communication
- [[Architectures/Control-Flow.md]] — Control flow with messages
- [[Frameworks/LangGraph.md]] — LangGraph `send()`
- [[Frameworks/AutoGen.md]] — AutoGen GroupChat
- [[Frameworks/OpenAI-Agents-SDK.md]] — Handoffs
- [[ToolUse/MCP-Protocol.md]] — MCP as transport
- [[ToolUse/Function-Calling.md]] — Tools as messages
- [[Production/Observability.md]] — Communication observability
- [[Production/Security.md]] — Communication security
- [[Frontiers/Agent-Societies.md]] — Large-scale agent communication

---

*Sources: FIPA ACL, A2A protocol proposals, LangGraph `send()`, AutoGen GroupChat, OpenAI Agents SDK handoffs, gRPC, JSON-RPC, message queue patterns, distributed systems communication patterns.*