Multi-Agent Communication Protocols

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

Why Standardized Communication?

Message Format

Envelope

{
  "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)

Content Types

Transport Protocols

1. HTTP/REST (Request-Response)

# 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)

# 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)

# 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)

// 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

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

Conversation Patterns

1. Request-Response

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

2. Streaming/Progress

# 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

# 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

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

Security

Authentication

# 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

# 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

# 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

# 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

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

Related Notes

• Coordination Patterns — Coordination using communication
• Negotiation Consensus — Negotiation protocols
• Delegation Handoff — Handoff patterns
• Safety Coordination — Safe communication
• Control Flow — Control flow with messages
• LangGraph — LangGraph send()
• AutoGen — AutoGen GroupChat
• OpenAI Agents SDK — Handoffs
• MCP Protocol — MCP as transport
• Function Calling — Tools as messages
• Observability — Communication observability
• Security — Communication security
• Agent Societies — 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.