Agent-to-Agent Communication: Protocols, Patterns, and Performance

Agents rarely work in isolation. They need to share information, delegate work, and coordinate. Here’s how they do it.

The Communication Spectrum

Pattern Latency Throughput Coupling Best For
Direct Call Low Medium High Tight coordination
Message Queue Medium High Low Decoupled workflows
Shared Blackboard Variable Medium Medium Coordination via state
Event Stream Low-Medium High Low Real-time updates
Request/Response RPC Low Medium Medium Synchronous delegation

1. Direct Agent-to-Agent Call

# Agent A calls Agent B directly
result = await agent_b.run({
    "task": "analyze_data",
    "data": my_data,
    "context": {"priority": "high"}
})

Pros

  • Lowest latency
  • Simple to implement
  • Strong typing possible

Cons

  • Tight coupling (A must know B’s interface)
  • Failure propagation (B down → A blocked)
  • Difficult to scale B independently

When to Use

  • Agents in the same trust domain
  • Low-latency requirements (<100ms)
  • When you control both agents

2. Message Queue Pattern

# Agent A publishes to queue
await message_queue.publish("task.analyze_data", {
    "data": my_data,
    "callback_queue": "results.agent_a"
})

# Agent B consumes from queue
@message_queue.subscribe("task.analyze_data")
async def handle_analyze(data):
    result = await analyze(data["data"])
    await message_queue.publish(data["callback_queue"], result)

Pros

  • Decoupled (A and B never touch)
  • Built-in buffering and retry
  • Horizontal scaling of consumers
  • Dead letter queues for poison messages

Cons

  • Higher latency (queue hop)
  • Need to manage queue infrastructure
  • At-least-once delivery requires idempotency

When to Use

  • Workloads with variable demand
  • When you want independent scaling
  • Fault-tolerant pipelines

3. Shared Blackboard

# Shared state (could be Redis, database, or even file)
class Blackboard:
    async def post(self, key: str, value: Any):
        await self.redis.set(key, json.dumps(value))
    
    async def get(self, key: str) -> Any:
        data = await self.redis.get(key)
        return json.loads(data) if data else None

# Agent A posts result
blackboard.post("analysis:result123", my_analysis)

# Agent B polls for result
while True:
    result = await blackboard.get("analysis:result123")
    if result:
        break
    await asyncio.sleep(0.1)

Pros

  • Simple mental model
  • Works across trust domains
  • Natural for coordination via state

Cons

  • Polling waste (unless using pub/sub)
  • Potential for race conditions
  • Can become bottleneck

When to Use

  • Coordination via shared state
  • When agents need to agree on a value
  • Simple workflows with few participants

4. Event Stream (Pub/Sub)

# Agent A publishes event
await event_stream.publish("topic.analysis_complete", {
    "topic_id": "abc123",
    "result": my_analysis,
    "timestamp": time.time()
})

# Agent B subscribes
@event_stream.subscribe("topic.analysis_complete")
async def handle_analysis(event):
    await process_result(event["result"])

Pros

  • Real-time propagation
  • Many-to-many communication
  • Built-in fan-out
  • Persistent streams for replay

Cons

  • More complex infrastructure
  • Need to manage consumer groups
  • Potential for thundering herd

When to Use

  • Broadcasting state changes
  • Audit trails and replay
  • Multiple interested parties

5. Request/Response RPC (with Middleware)

# Using gRPC, HTTP+JSON, or MCP
class AgentClient:
    async def call(self, method: str, params: dict) -> Any:
        response = await self.http.post(
            f"{self.agent_url}/mcp",
            json={
                "jsonrpc": "2.0",
                "method": method,
                "params": params,
                "id": self.next_id()
            }
        )
        return response.json()["result"]

# Agent A calls Agent B via RPC
client = AgentClient("https://agent-b.example.com")
result = await client.call("analyze_topic", {
    "topic_id": "abc123",
    "depth": "deep"
})

Pros

  • Familiar request/response semantics
  • Strong contract (IDL/schema)
  • Works across networks and languages
  • Middleware for auth, logging, retries

Cons

  • Requires network roundtrip
  • More complex than direct call
  • Need to handle timeouts and retries

When to Use

  • Cross-service agent communication
  • When you need strong typing/contracts
  • Heterogeneous agent ecosystems

Performance Considerations

1. Latency Budget

Total Latency = Network + Serialization + Queue Wait + Processing
Pattern Typical Latency (ms)
Direct Call 1-10
RPC/HTTP 10-100
Message Queue 50-500
Event Stream 10-200
Blackboard (polling) 100-1000+

2. Throughput Limits

  • Direct Call: Limited by connection pool and agent concurrency
  • Message Queue: Limited by queue throughput and consumer count
  • Event Stream: Limited by broker throughput and partition count
  • RPC: Limited by server concurrency and connection pool

3. Optimization Techniques

  • Connection pooling for RPC/direct calls
  • Batch processing for queues and streams
  • Compression for large payloads
  • Caching for frequently requested data
  • Async everywhere — never block threads

Reliability Patterns

1. Retry with Exponential Backoff

async def resilient_call(func, *args, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await func(*args)
        except TemporaryError:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt + random.uniform(0, 1))

2. Circuit Breaker

class CircuitBreaker:
    def __init__(self, failure_threshold=5, timeout=60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failure_count = 0
        self.last_failure = None
        self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN
    
    async def call(self, func, *args):
        if self.state == "OPEN":
            if time.time() - self.last_failure > self.timeout:
                self.state = "HALF_OPEN"
            else:
                raise CircuitBreakerOpen()
        
        try:
            result = await func(*args)
            if self.state == "HALF_OPEN":
                self.state = "CLOSED"
                self.failure_count = 0
            return result
        except Exception:
            self.failure_count += 1
            self.last_failure = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = "OPEN"
            raise

3. Bulkhead Pattern

# Limit concurrent calls to prevent resource exhaustion
class Bulkhead:
    def __init__(self, max_concurrent=10):
        self.semaphore = asyncio.Semaphore(max_concurrent)
    
    async def call(self, func, *args):
        async with self.semaphore:
            return await func(*args)

The Takeaway

Choose your communication pattern based on your constraints, not trends.

  1. Direct calls for low-latency, tightly-coupled agents
  2. Message queues for decoupled, scalable workloads
  3. Event streams for real-time fan-out and replay
  4. Shared blackboard for simple coordination via state
  5. RPC/MCP for cross-service contracts and strong typing

The pattern affects everything: latency, throughput, reliability, observability, and team autonomy. Choose wisely.

“The best agent-to-agent communication is the one you don’t need — because the agent can do the work alone.” But when you do need to communicate, make it intentional.