Agent Communication & Coordination

Once you have multiple agents, the challenge shifts from "what can each agent do?" to "how do agents work together effectively?" This lesson covers message passing patterns, shared blackboards, handoff protocols, conflict resolution, and structured inter-agent communication.

---

The Communication Challenge

In a multi-agent system, agents must:

1. Share information — Pass findings, results, and context between agents.
2. Coordinate actions — Avoid duplicating work or conflicting with each other.
3. Resolve disagreements — Handle cases where agents produce contradictory results.
4. Maintain coherence — Ensure the final output is consistent, not a disconnected patchwork.

---

Message Passing

The simplest coordination mechanism: agents send structured messages to each other.

Structured Message Format

from dataclasses import dataclass, field
from datetime import datetime
from typing import Any

@dataclass
class AgentMessage:
    sender: str
    recipient: str
    message_type: str   # "request", "response", "inform", "handoff"
    content: str
    metadata: dict[str, Any] = field(default_factory=dict)
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())

class MessageBus:
    """Central message bus for agent communication."""

    def __init__(self):
        self.queues: dict[str, list[AgentMessage]] = {}
        self.log: list[AgentMessage] = []

    def register_agent(self, agent_name: str):
        self.queues[agent_name] = []

    def send(self, message: AgentMessage):
        self.log.append(message)
        if message.recipient in self.queues:
            self.queues[message.recipient].append(message)
        elif message.recipient == "broadcast":
            for name, queue in self.queues.items():
                if name != message.sender:
                    queue.append(message)

    def receive(self, agent_name: str) -> list[AgentMessage]:
        messages = self.queues.get(agent_name, [])
        self.queues[agent_name] = []
        return messages

Usage

bus = MessageBus()
bus.register_agent("researcher")
bus.register_agent("analyst")

# Researcher sends findings to analyst
bus.send(AgentMessage(
    sender="researcher",
    recipient="analyst",
    message_type="inform",
    content="Found 5 relevant papers on AI in healthcare.",
    metadata={"papers": ["paper1.pdf", "paper2.pdf", "..."]},
))

# Analyst receives the message
messages = bus.receive("analyst")
for msg in messages:
    print(f"From {msg.sender}: {msg.content}")

---

Shared Blackboard Pattern

A shared blackboard is a central state store that all agents can read from and contribute to.

     ┌───────────┐     ┌───────────┐     ┌───────────┐
     │ Agent A   │     │ Agent B   │     │ Agent C   │
     └─────┬─────┘     └─────┬─────┘     └─────┬─────┘
           │                 │                   │
           └────────┬────────┴───────────────────┘
                    │
                    ▼
     ┌──────────────────────────────────────────────────┐
     │              BLACKBOARD (Shared State)           │
     │                                                  │
     │  research_findings: [...]                        │
     │  analysis_results: {...}                         │
     │  draft_report: "..."                             │
     │  status: {"research": "done", "analysis": "..."}│
     └──────────────────────────────────────────────────┘

from threading import Lock

class Blackboard:
    def __init__(self):
        self._state: dict[str, Any] = {}
        self._lock = Lock()
        self._history: list[dict] = []

    def write(self, agent_name: str, key: str, value: Any):
        with self._lock:
            self._state[key] = value
            self._history.append({
                "agent": agent_name,
                "key": key,
                "action": "write",
                "timestamp": datetime.now().isoformat(),
            })

    def read(self, key: str, default: Any = None) -> Any:
        with self._lock:
            return self._state.get(key, default)

    def read_all(self) -> dict:
        with self._lock:
            return dict(self._state)

    def append_to_list(self, agent_name: str, key: str, value: Any):
        with self._lock:
            if key not in self._state:
                self._state[key] = []
            self._state[key].append(value)
            self._history.append({
                "agent": agent_name,
                "key": key,
                "action": "append",
                "timestamp": datetime.now().isoformat(),
            })

---

Handoff Patterns

When one agent completes its part and passes control to another, a structured handoff ensures no context is lost.

Structured Handoff

@dataclass
class Handoff:
    from_agent: str
    to_agent: str
    task_description: str
    context: str
    completed_work: str
    remaining_instructions: str

def execute_handoff(
    handoff: Handoff,
    agents: dict[str, Agent],
) -> str:
    """Execute a structured handoff between agents."""
    target = agents.get(handoff.to_agent)
    if not target:
        raise ValueError(f"Unknown agent: {handoff.to_agent}")