agent-communication

Agent Communication

Patterns for communication between AI agents. Covers message passing, shared state, event-driven design, pub/sub, inbox/outbox, and structured message formats.

When to Use

• User is building multi-agent systems that need inter-agent communication
• User needs shared state between agents
• User wants event-driven agent coordination
• User is designing message formats for agent-to-agent data exchange
• User asks about pub/sub or inbox/outbox patterns for agents

Core Patterns

Direct Message Passing

Agents communicate through explicit function calls with typed messages.

from dataclasses import dataclass
from typing import Any

@dataclass(frozen=True)
class AgentMessage:
    sender: str
    recipient: str
    msg_type: str  # "request", "response", "notification"
    payload: dict
    correlation_id: str  # Links requests to responses

class MessageBus:
    def __init__(self):
        self._handlers: dict[str, list] = {}
        self._inbox: dict[str, list[AgentMessage]] = {}

    def register(self, agent_id: str, handler):
        self._handlers[agent_id] = handler
        self._inbox[agent_id] = []

    def send(self, message: AgentMessage):
        self._inbox[message.recipient].append(message)

    async def deliver(self, agent_id: str) -> list[AgentMessage]:
        messages = self._inbox[agent_id]
        self._inbox[agent_id] = []
        return messages

# Usage
bus = MessageBus()
bus.send(AgentMessage(
    sender="orchestrator",
    recipient="researcher",
    msg_type="request",
    payload={"task": "Find recent papers on RAG optimization"},
    correlation_id="task-001"
))

Shared State Store

Agents read and write to a shared state store for coordination.

import asyncio
from dataclasses import dataclass, field

@dataclass(frozen=True)
class StateEntry:
    value: Any
    updated_by: str
    version: int

class SharedState:
    def __init__(self):
        self._state: dict[str, StateEntry] = {}
        self._lock = asyncio.Lock()
        self._watchers: dict[str, list] = {}

    async def get(self, key: str) -> StateEntry | None:
        return self._state.get(key)

    async def put(self, key: str, value: Any, agent_id: str) -> StateEntry:
        async with self._lock:
            current = self._state.get(key)
            version = (current.version + 1) if current else 1
            entry = StateEntry(value=value, updated_by=agent_id, version=version)
            self._state = {**self._state, key: entry}  # Immutable update

            # Notify watchers
            for callback in self._watchers.get(key, []):
                await callback(key, entry)
            return entry

    def watch(self, key: str, callback):
        watchers = self._watchers.get(key, [])
        self._watchers = {**self._watchers, key: [*watchers, callback]}

# Usage
state = SharedState()
await state.put("research_findings", {"papers": [...]}, agent_id="researcher")
await state.put("code_review", {"issues": [...]}, agent_id="reviewer")

# Another agent reads the state
findings = await state.get("research_findings")