Agent Communication Protocols

Why This Matters

Communication is the nervous system of multi-agent systems. Without well-designed protocols, agents become isolated islands—unable to coordinate, share state, or respond to events. Proper communication architecture determines whether your system scales gracefully or collapses under load.

Real-World Analogy: Think of a hospital. Doctors, nurses, pharmacists, and patients all need different communication channels. A doctor prescribes medication (message passing), the pharmacy broadcasts alerts about drug interactions (pub/sub), and patient records are accessed by multiple staff (shared memory). Each protocol serves a specific purpose—using the wrong one creates chaos.

Communication Architecture Overview

Message Queue Implementation

import asyncio
import uuid
import time
import logging
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Coroutine, Optional
from collections import defaultdict
import heapq

logger = logging.getLogger(__name__)


class MessagePriority(Enum):
    LOW = 0
    NORMAL = 1
    HIGH = 2
    CRITICAL = 3


@dataclass
class Message:
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    topic: str = ""
    payload: Any = None
    sender: str = ""
    priority: MessagePriority = MessagePriority.NORMAL
    timestamp: float = field(default_factory=time.time)
    ttl: float = 300.0
    reply_to: Optional[str] = None
    correlation_id: Optional[str] = None
    metadata: dict = field(default_factory=dict)

    @property
    def is_expired(self) -> bool:
        return time.time() - self.timestamp > self.ttl

    def __lt__(self, other):
        return self.priority.value > other.priority.value


@dataclass
class Subscription:
    subscriber_id: str
    topic: str
    callback: Callable[[Message], Coroutine[Any, Any, None]]
    filter_fn: Optional[Callable[[Message], bool]] = None
    max_retries: int = 3


class QueueFullError(Exception):
    pass


class MessageQueue:
    def __init__(self, max_size: int = 10000):
        self._queues: dict[str, list[Message]] = defaultdict(list)
        self._subscribers: dict[str, list[Subscription]] = defaultdict(list)
        self._max_size = max_size
        self._dead_letter_queue: list[Message] = []
        self._metrics = {
            "published": 0,
            "delivered": 0,
            "failed": 0,
            "expired": 0,
        }

    async def publish(self, message: Message) -> str:
        if len(self._queues[message.topic]) >= self._max_size:
            raise QueueFullError(f"Queue for topic '{message.topic}' is full")
        
        heapq.heappush(self._queues[message.topic], message)
        self._metrics["published"] += 1
        
        await self._deliver_messages(message.topic)
        return message.id

    async def _deliver_messages(self, topic: str):
        subscribers = self._subscribers.get(topic, [])
        if not subscribers:
            return

        queue = self._queues[topic]
        while queue:
            message = queue[0]
            if message.is_expired:
                heapq.heappop(queue)
                self._metrics["expired"] += 1
                continue
            break

        for subscription in subscribers:
            if subscription.filter_fn and not subscription.filter_fn(message):
                continue
            
            try:
                await subscription.callback(message)
                self._metrics["delivered"] += 1
            except Exception as e:
                self._metrics["failed"] += 1
                logger.error(f"Delivery failed: {e}")
                if subscription.max_retries > 0:
                    subscription.max_retries -= 1
                    await asyncio.sleep(1)
                    try:
                        await subscription.callback(message)
                    except Exception:
                        self._dead_letter_queue.append(message)

    def subscribe(self, subscription: Subscription):
        self._subscribers[subscription.topic].append(subscription)

    def unsubscribe(self, subscriber_id: str, topic: str):
        self._subscribers[topic] = [
            s for s in self._subscribers[topic]
            if s.subscriber_id != subscriber_id
        ]

    def get_metrics(self) -> dict:
        return self._metrics.copy()

Publish/Subscribe System

import asyncio
import uuid
import time
import logging
from dataclasses import dataclass, field
from typing import Any, Callable, Coroutine
from collections import defaultdict
from enum import Enum

logger = logging.getLogger(__name__)


class EventType(Enum):
    AGENT_STARTED = "agent.started"
    AGENT_COMPLETED = "agent.completed"
    AGENT_FAILED = "agent.failed"
    TASK_ASSIGNED = "task.assigned"
    TASK_COMPLETED = "task.completed"
    SYSTEM_ERROR = "system.error"


@dataclass
class Event:
    event_type: EventType
    data: dict = field(default_factory=dict)
    source: str = ""
    event_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    timestamp: float = field(default_factory=time.time)


class EventBus:
    def __init__(self):
        self._handlers: dict[EventType, list[Callable]] = defaultdict(list)
        self._wildcard_handlers: list[Callable] = []
        self._event_log: list[Event] = []
        self._max_log_size = 10000

    def on(self, event_type: EventType, handler: Callable[[Event], Coroutine]):
        self._handlers[event_type].append(handler)

    def on_all(self, handler: Callable[[Event], Coroutine]):
        self._wildcard_handlers.append(handler)

    async def emit(self, event: Event):
        self._event_log.append(event)
        if len(self._event_log) > self._max_log_size:
            self._event_log = self._event_log[-self._max_log_size:]

        handlers = self._handlers.get(event.event_type, []) + self._wildcard_handlers
        
        tasks = [handler(event) for handler in handlers]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                logger.error(f"Handler error for {event.event_type}: {result}")

    def get_event_history(self, event_type: EventType = None, limit: int = 100) -> list[Event]:
        events = self._event_log
        if event_type:
            events = [e for e in events if e.event_type == event_type]
        return events[-limit:]

Performance Considerations

Security Considerations

• Message Encryption: Encrypt sensitive payloads in transit and at rest
• Authentication: Verify sender identity before processing messages
• Authorization: Implement topic-level access control for pub/sub
• Input Validation: Sanitize message payloads to prevent injection attacks
• Audit Logging: Track all message flows for security monitoring

Mathematical Foundation

Message Delivery Latency:

Throughput Under Load:

Message Loss Probability (with retries):

Interview Questions

1. What are the tradeoffs between message passing and shared memory?

Answer: Message passing provides loose coupling, easier debugging, and natural distributed scaling but adds serialization overhead and latency. Shared memory offers minimal latency and high throughput but requires synchronization primitives, creates tight coupling, and is harder to distribute. Message passing is preferred for distributed systems and when traceability matters; shared memory suits high-performance single-node systems where latency is critical.

2. How does the pub/sub pattern handle subscriber failures?

Answer: Implement dead letter queues (DLQ) for messages that exceed retry limits. Each subscriber should have: 1) Configurable retry count with exponential backoff, 2) Message acknowledgment to prevent redelivery, 3) TTL on messages to prevent stale processing, 4) Circuit breaker to stop delivery during sustained failures. The publisher should be decoupled from subscriber health. Monitor DLQ size and alert on growth.

3. What is the difference between at-least-once, at-most-once, and exactly-once delivery?

Answer: At-least-once: Message is delivered one or more times; requires idempotent handlers. At-most-once: Message is delivered zero or one time; may lose messages. Exactly-once: Message is delivered exactly once; requires distributed transactions or two-phase commit which adds significant overhead. Most systems use at-least-once with idempotent processing as the practical choice.

4. How do you implement backpressure in a message-driven system?

Answer: Backpressure prevents producers from overwhelming consumers. Techniques: 1) Bounded queues that block producers when full, 2) Rate limiting at the producer, 3) Adaptive batching based on consumer capacity, 4) Flow control protocols (e.g., TCP-style windowing), 5) Load shedding when queues exceed thresholds. Monitor queue depth as a key metric.

5. What are the CAP theorem implications for agent communication?

Answer: The CAP theorem states a distributed system can guarantee only two of: Consistency, Availability, Partition tolerance. For agent communication: Choose CP for financial transactions requiring strong consistency. Choose AP for event systems where eventual consistency is acceptable. Most agent systems choose AP with eventual consistency since brief inconsistencies are acceptable.

6. How would you implement message ordering guarantees?

Answer: Per-partition ordering: Assign messages to partitions using a consistent key (e.g., agent ID). Within a partition, messages are ordered. Across partitions, ordering is not guaranteed. Techniques: 1) Single partition for strict ordering (limits throughput), 2) Sequence numbers for detection of out-of-order, 3) Consumer-side reordering with bounded buffers, 4) Causal ordering using vector clocks.

7. What is the observer pattern and how does it relate to pub/sub?

Answer: Observer pattern is a design pattern where subjects maintain a list of observers and notify them of state changes. Pub/sub is a messaging pattern where publishers send messages to topics and subscribers receive them. Key difference: Observer has direct reference to subjects (tight coupling); pub/sub uses a message broker (loose coupling). Pub/sub adds topic-based routing, message persistence, and replay capability.

8. How do you handle message schema evolution?

Answer: Use schema versioning and backward/forward compatibility: 1) Include version field in messages, 2) Use additive-only schema changes (new fields with defaults), 3) Never remove or rename fields (deprecate instead), 4) Use schema registries for validation, 5) Implement consumer-side schema migration. Tools like Avro or Protobuf enforce compatibility rules. Deploy schema changes before code changes.

Common Pitfalls

Summary with Key Takeaways

• Message passing provides loose coupling and natural distributed scaling with serialization overhead
• Pub/sub enables many-to-many communication with topic-based routing and event-driven processing
• Shared memory offers minimal latency but requires synchronization and creates tight coupling
• Backpressure prevents producers from overwhelming consumers; use bounded queues and rate limiting
• Delivery guarantees trade off between performance and reliability; at-least-once with idempotency is practical
• Schema evolution requires versioning and backward compatibility to prevent breaking changes
• Dead letter queues capture failed messages for debugging and reprocessing
• Monitoring queue depth, delivery latency, and failure rates is critical for operational health

KnowledgeCheck

1. Which communication pattern provides the loosest coupling between agents?

   • a) Shared memory
   • b) Direct message passing
   • c) Publish/subscribe
   • d) REST API calls

2. What is the primary purpose of a dead letter queue?

   • a) Store high-priority messages
   • b) Buffer messages during peak load
   • c) Capture messages that failed processing
   • d) Cache frequently accessed messages

3. In the CAP theorem, which two properties can a distributed system guarantee simultaneously?

   • a) Consistency and Availability
   • b) Consistency and Partition tolerance
   • c) Availability and Partition tolerance
   • d) Any two of the three

4. What technique prevents producers from overwhelming consumers?

   • a) Message serialization
   • b) Backpressure
   • c) Topic partitioning
   • d) Schema validation

5. Which delivery guarantee requires idempotent message handlers?

   • a) At-most-once
   • b) At-least-once
   • c) Exactly-once
   • d) Best-effort

6. How does a read-write lock improve shared memory performance?

   • a) Allows multiple concurrent readers
   • b) Eliminates all synchronization overhead
   • c) Provides faster writes than mutexes
   • d) Reduces memory usage

Answers: 1-c, 2-c, 3-c, 4-b, 5-b, 6-a