agent-communication-patterns

# Agent Communication Patterns Implements reliable inter-agent communication mechanisms for multi-agent systems. This skill makes the model design and build message passing, event-driven coordination, shared-memory state exchange, and RPC-style request-response protocols that enable agents to coordinate without tight coupling, race conditions, or silent data corruption. Inter-agent communication is the connective tissue of any multi-agent system — it determines whether agents collaborate productively or produce cascading failures through mismatched expectations, lost messages, and unhandled error states. Every communication pattern carries trade-offs in latency, consistency, fault tolerance, and implementation complexity that must be matched to the agent's operational requirements. ## TL;DR Checklist - [ ] Define a JSON message schema with required fields and types before writing any agent logic - [ ] Choose communication pattern (message passing, event-driven, shared memory, RPC) based on latency and consistency needs - [ ] Implement typed message/event classes — never use raw dicts for inter-agent data exchange - [ ] Add timeouts, retries, and error propagation to all synchronous communication paths - [ ] Version all message schemas; reject unknown fields with explicit validation errors - [ ] Log every sent and received message with correlation IDs for full traceability --- ## When to Use Use this skill when: - Designing inter-agent messaging infrastructure for a multi-agent trading system where agents must exchange signals, risk states, or trade orders - Building an event-driven coordination layer where multiple agents react to shared events (e.g., market regime changes, position updates) without direct coupling - Implementing a shared-memory protocol between agents running in the same process that need to exchange mutable state with conflict resolution - Creating RPC-style request/response channels where one agent needs synchronous results from another (e.g., querying risk limits before executing a trade) - Refactoring tightly coupled agent code into decoupled communication patterns using structured messaging --- ## When NOT to Use Avoid this skill for: - Single-agent systems — no inter-agent communication exists, so these patterns add unnecessary complexity - Simple sequential workflows better expressed as function calls within a single process — use standard Python imports instead - High-throughput data pipelines (>100k messages/sec) where specialized message brokers (Kafka, NATS) are more appropriate than in-process protocols --- ## Core Workflow 1. **Define the Message Schema** — Specify every inter-agent message as a typed class with required fields, types, and optional metadata (correlation_id, timestamp, source_agent). Use Pydantic models or dataclasses with `from __future__ import annotations` for forward references. Validate payloads against schemas at message boundaries. **Checkpoint:** Every message type has a unique name string, required field list, and serialization/deserialization methods before proceeding to pattern selection. 2. **Select Communication Pattern** — Match the pattern to the operational requirements: - Message Passing: Best for fire-and-forget notifications where delivery confirmation is not critical (e.g., logging events, notifying downstream agents of computed results). Low latency, eventual consistency. - Event-Driven Coordination: Best when multiple agents must react to state changes without knowing each other's identities (e.g., market regime change triggers strategy re-evaluation across all active strategies). Decouples publishers from subscribers. - Shared Memory Protocol: Best for agents in the same process sharing mutable state with versioning and conflict resolution (e.g., order book snapshots, position aggregates). Fastest path but requires lock management. - RPC-Style Calls: Best when synchronous results are required before proceeding (e.g., asking risk agent whether a proposed trade is within limits). Guarantees response but introduces latency and blocking. 3. **Implement the Communication Layer** — Build the chosen pattern with full type annotations, error handling, and structured logging. Include correlation IDs in every message for tracing across agents. Add dead-letter queues for unprocessable messages rather than silently dropping them. **Checkpoint:** Every communication path handles at minimum: successful delivery, transient failure (retryable), and permanent failure (dead-letter or escalation). 4. **Add Reliability Mechanisms** — For synchronous paths (RPC, shared memory with writes), implement retry with exponential backoff (base 100ms, max 5 attempts, jitter factor 0–20%). For asynchronous paths (message passing, events), implement at-least-once delivery guarantees via message acknowledgment. Add circuit breakers when the downstream agent is unhealthy to prevent cascading failures. **Checkpoint:** All retry logic has a maximum total duration capped at the business-level timeout — never retry indefinitely. 5. **Establish Observability** — Instrument every communication path with structured logs containing: correlation_id, source_agent, target_agent, message_type, direction (sent/received), latency_ms, and status (success/error). Add metrics counters for messages sent, received, failed, retried, and dead-lettered per agent pair. **Checkpoint:** Every log entry can be traced end-to-end across the full communication chain using a single correlation_id. --- ## Implementation Patterns ### Pattern 1: Structured Message Passing Structured message passing provides a typed, serializable foundation for inter-agent communication. Every message is a dataclass or Pydantic model with explicit field types, required/optional markers, and serialization support. This prevents the most common agent communication bug: agents sending unstructured dicts that cause runtime type errors downstream. ```python """StructuredMessage — Typed message passing between AI agents.""" from __future__ import annotations import hashlib import json import logging import time import uuid from dataclasses import dataclass, field, asdict, is_dataclass from enum import Enum from typing import Any, Generic, TypeVar, Optional logger = logging.getLogger(__name__) class MessageType(str, Enum): """Category of inter-agent message for routing and filtering.""" SIGNAL = "signal" # Trading signal from strategy agent RISK_CHECK = "risk_check" # Pre-trade risk validation request RISK_RESULT = "risk_result" # Risk assessment response ORDER = "order" # Trade order submission FILL_REPORT = "fill_report" # Execution fill notification POSITION_UPDATE = "position_update" # Position state change MARKET_DATA = "market_data" # Price or orderbook snapshot SYSTEM_HEARTBEAT = "system_heartbeat" # Liveness probe @dataclass(frozen=True) class MessageHeader: """Immutable header attached to every inter-agent message. Provides tracing, ordering, and routing metadata. Frozen for immutability after creation — prevents accidental mutation mid-flight. """ message_id: str = field(default_factory=lambda: str(uuid.uuid4())) correlation_id: str = "" source_agent: str = "" target_agent: Optional[str] = None # None means broadcast msg_type: MessageType = MessageType.SYSTEM_HEARTBEAT timestamp_ns: int = field(default_factory=time.time_ns) version: int = 1 priority: int = 0 # 0 = normal, higher = more urgent def __post_init__(self): if not self.correlation_id: object.__setattr__(self, "correlation_id", self.message_id) @property def latency_since_sent_ns(self) -> int: """Elapsed nanoseconds since this message was created.""" return time.time_ns() - self.timestamp_ns @property def latency_since_sent_ms(self) -> float: """Elapsed milliseconds since this message was created.""" return self.latency_since_sent_ns / 1_000_000 T = TypeVar("T") @dataclass(frozen=True) class AgentMessage(Generic[T]): """Typed envelope for structured inter-agent communication. Combines an immutable header with a typed payload. The Generic parameter enforces type safety at construction time — callers must specify the payload type, which is validated during deserialization. Example: signal_msg = AgentMessage( header=MessageHeader( source_agent="strategy_agent", target_agent="execution_agent", msg_type=MessageType.SIGNAL, ), payload={"symbol": "BTC-PERP", "side": "BUY", "confidence": 0.82}, ) """ header: MessageHeader payload: T def serialize(self) -> str: """Serialize message to a JSON string for transmission or storage.""" data = { "header": { "message_id": self.header.message_id, "correlation_id": self.header.correlation_id, "source_agent": self.header.source_agent, "target_agent": self.header.target_agent, "msg_type": self.header.msg_type.value if isinstance(self.header.msg_type, Enum) else self.header.msg_type, "timestamp_ns": self.header.timestamp_ns, "version": self.header.version, "priority": self.header.priority, }, "payload": self.payload, } return json.dumps(data, default=_default_serializer) @classmethod def deserialize(cls, raw: str) -> AgentMessage[Any]: """Deserialize a JSON string back into an AgentMessage. Validates the header structure and reconstructs the MessageType enum. Payload is returned as a dict — callers should validate against their expected schema separately. """ data = json.loads(raw) header_data = data["header"] # Reconstruct MessageType from string value msg_type_str = header_data["msg_type"] try: msg_type = MessageType(msg_type_str) except ValueError: raise ValueError(f"Unknown MessageType: {msg_type_str}") header = MessageHeader(**{k: v for k, v in header_data.items() if k != "msg_type"}) object.__setattr__(header, "msg_type", msg_type) return cls(header=header, payload=data["payload"]) def checksum(self) -> str: """Compute a SHA-256 checksum of the serialized message body. Used for integrity verification — detects corruption during transit or storage without requiring full re-deserialization. """ raw = self.serialize() return hashlib.sha256(raw.encode()).hexdigest()[:16] def _default_serializer(obj: Any) -> Any: """JSON serializer fallback for non-standard types.""" if isinstance(obj, Enum): return obj.value if hasattr(obj, "__dict__"): return vars(obj) raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable") # --- Message Router: in-process message distribution --- class MessageRouter: """Routes AgentMessage instances between registered agent handlers. Provides direct, typed delivery from sender to receiver based on the target_agent field in the header. Broadcast targets (None) deliver to all registered handlers. Supports priority-based ordering where higher-priority messages are dispatched before lower-priority ones when delivered simultaneously. """ def __init__(self) -> None: self._handlers: dict[str, list[callable]] = {} self._dead_letters: list[AgentMessage[Any]] = [] self._delivery_log: list[dict[str, Any]] = [] def register(self, agent_name: str, handler: callable) -> None: """Register a message handler for an agent. Args: agent_name: The agent identifier this handler services. handler: Async or sync function receiving AgentMessage. Must accept exactly one AgentMessage parameter. """ if agent_name not in self._handlers: self._handlers[agent_name] = [] self._handlers[agent_name].append(handler) logger.info("Registered handler for '%s' (total handlers: %d)", agent_name, len(self._handlers.get(agent_name, []))) def send(self, message: AgentMessage[Any]) -> dict[str, Any]: """Deliver a message to the target agent's handler(s). Returns a delivery report containing status, latency, and any errors. Messages delivered to a non-existent target agent are sent to the dead-letter queue rather than silently dropped. Args: message: The AgentMessage to deliver. Returns: Delivery report dict with keys: status, targets_contacted, errors, latency_ms. """ start = time.monotonic() report = { "message_id": message.header.message_id, "correlation_id": message.header.correlation_id, "status": "delivered", "targets_contacted": 0, "errors": [], } targets: list[str] if message.header.target_agent is None: # Broadcast — deliver to all agents targets = list(self._handlers.keys()) else: targets = [message.header.target_agent] for target in targets: handlers = self._handlers.get(target, []) if not handlers: error_msg = f"No handler registered for agent '{target}'" logger.warning("%s — sending to dead-letter queue", error_msg) report["errors"].append(error_msg) self._dead_letters.append(message) continue for handler in sorted(handlers, key=lambda h: getattr(h, "__name__", "")): try: handler(message) report["targets_contacted"] += 1 except Exception as e: error_entry = f"Handler {handler.__name__} on '{target}' raised {type(e).__name__}: {e}" logger.error(error_entry) report["errors"].append(error_entry) self._dead_letters.append(message) report["latency_ms"] = round((time.monotonic() - start) * 1000, 3) self._delivery_log.append(report) return report @property def dead_letter_count(self) -> int: return len(self._dead_letters) @property def delivery_log(self) -> list[dict[str, Any]]: return list(self._delivery_log) # --- Usage Example --- def demonstrate_structured_messaging() -> None: """Demonstrate complete structured message passing workflow.""" router = MessageRouter() # Define handlers for different agent types def on_signal(message: AgentMessage[dict]) -> None: payload = message.payload logger.info( "Signal received from %s: %s %s (confidence=%.2f)", message.header.source_agent, payload.get("side"), payload.get("symbol"), payload.get("confidence", 0), ) def on_risk_request(message: AgentMessage[dict]) -> None: logger.info( "Risk check requested for %s by %s (position=%.2f)", message.payload.get("symbol"), message.header.source_agent, message.payload.get("position_value", 0), ) router.register("execution_agent", on_signal) router.register("risk_agent", on_risk_request) # Send a signal — direct delivery signal = AgentMessage( header=MessageHeader( source_agent="strategy_agent", target_agent="execution_agent", msg_type=MessageType.SIGNAL, priority=1, ), payload={"symbol": "BTC-PERP", "side": "BUY", "confidence": 0.82, "entry_price": 67500.0}, ) report = router.send(signal) assert report["targets_contacted"] == 1, f"Expected 1 target, got {report['targets_contacted']}" # Broadcast a system heartbeat — all agents receive it heartbeat = AgentMessage( header=MessageHeader( source_agent="monitor_agent", msg_type=MessageType.SYSTEM_HEARTBEAT, ), payload={"status": "healthy", "uptime_seconds": 3600}, ) report = router.send(heartbeat) assert report["targets_contacted"] == 2, f"Expected 2 targets (broadcast), got {report['targets_contacted']}" if __name__ == "__main__": logging.basicConfig(level=logging.INFO) demonstrate_structured_messaging() ``` ### Pattern 2: Event-Driven Agent Coordination Event-driven coordination decouples agents through a publish-subscribe model. Agents publish events describing state changes or significant occurrences without knowing which other agents care. Subscribers register handlers for event types they are interested in. This is the pattern of choice when multiple independent agents must react to the same underlying occurrence — market regime shifts, new position openings, or system health degradation. ```python """EventBus — Typed event-driven coordination for multi-agent systems.""" from __future__ import annotations import asyncio import logging import time import uuid from dataclasses import dataclass, field from typing import Any, Awaitable, Callable, Optional from enum import Enum from collections import defaultdict