Multi-Agent Coordination: Communication Protocols

A single AI agent operating in isolation is powerful. A coordinated system of agents is transformative — and considerably harder to build correctly. Multi-agent coordination introduces all the classic distributed systems problems (partial failure, inconsistent state, network partitions) combined with the non-determinism inherent to language model inference. The protocols you choose for agent communication determine whether your system compounds intelligence or amplifies chaos.

Why Coordination Is Hard

When a human orchestrator assigns tasks to multiple AI agents, several failure modes emerge that don't exist with single agents:

Divergent state. Agent A and Agent B both read the same shared document and begin editing concurrently. Their outputs conflict in ways neither can detect.

Lost context. Agent B is handed a subtask without enough context to complete it correctly. It succeeds at the wrong thing.

Cascading failures. Agent C waits for Agent B's output, which depends on Agent A. Agent A times out. The entire pipeline stalls.

Hallucinated handoffs. An orchestrating agent fabricates a completion signal it never received from a subordinate, causing downstream agents to proceed on incorrect premises.

Sound protocols address each failure mode explicitly rather than hoping the language model handles it gracefully.

Communication Models

Shared Memory (Blackboard Architecture)

The simplest coordination model: all agents read from and write to a shared workspace. The workspace acts as a blackboard — any agent can inspect any other agent's work.

class AgentWorkspace:
    def __init__(self):
        self._store: dict[str, WorkspaceEntry] = {}
        self._lock = asyncio.Lock()
    
    async def write(self, key: str, value: Any, agent_id: str) -> None:
        async with self._lock:
            self._store[key] = WorkspaceEntry(
                value=value,
                agent_id=agent_id,
                timestamp=datetime.utcnow(),
                version=self._store.get(key, WorkspaceEntry()).version + 1
            )
    
    async def read(self, key: str) -> WorkspaceEntry | None:
        return self._store.get(key)
    
    async def read_all_by_agent(self, agent_id: str) -> list[WorkspaceEntry]:
        return [e for e in self._store.values() if e.agent_id == agent_id]

Blackboard architecture works well for small agent systems where agents need to observe each other's intermediate work (e.g., a research agent whose findings inform a synthesis agent). It breaks down at scale because every agent must understand the entire workspace schema and conflicts require explicit resolution logic.

Message Passing (Actor Model)

More structured: each agent is an actor with its own mailbox. Agents communicate exclusively through typed messages. No shared state.

@dataclass
class ResearchRequest:
    task_id: str
    query: str
    max_sources: int
    requester_id: str

@dataclass
class ResearchResult:
    task_id: str
    findings: list[Finding]
    confidence: float
    citations: list[str]
    agent_id: str

class ResearchAgent:
    async def handle_message(self, msg: ResearchRequest) -> ResearchResult:
        findings = await self.search_and_synthesize(msg.query, msg.max_sources)
        return ResearchResult(
            task_id=msg.task_id,
            findings=findings,
            confidence=self.assess_confidence(findings),
            citations=self.extract_citations(findings),
            agent_id=self.agent_id
        )

Typed messages create an explicit contract between agents. If the ResearchAgent changes its output schema, all consumers get a type error at integration time, not a runtime failure in production. This mirrors microservice API contracts but for agent-to-agent communication.

Orchestration vs. Choreography

Two architectural patterns emerge for multi-agent workflows:

Orchestration — a central orchestrator agent explicitly directs other agents, waits for their results, and makes routing decisions:

Orchestrator → assigns research task → ResearchAgent
Orchestrator ← receives findings ← ResearchAgent
Orchestrator → assigns synthesis task (with findings) → SynthesisAgent
Orchestrator ← receives draft ← SynthesisAgent
Orchestrator → assigns critique task (with draft) → CritiqueAgent

Advantages: easy to reason about, natural error handling at the orchestrator level, clear audit trail. Disadvantages: orchestrator becomes a bottleneck and single point of failure.

Choreography — agents react to events and emit events without central coordination:

ResearchAgent emits: research-completed event
SynthesisAgent subscribes to: research-completed → begins synthesis
CritiqueAgent subscribes to: synthesis-completed → begins critique

Advantages: scales horizontally, no single bottleneck, agents can be added or replaced without modifying others. Disadvantages: harder to debug, requires robust event infrastructure, error handling is distributed across agents.

Most production systems use a hybrid: choreography for independent parallel tasks, orchestration for sequential tasks with dependencies.

Handling Partial Failures

When one agent in a pipeline fails, the system must decide: retry, skip, escalate to human, or abort.

Timeout and Retry with Circuit Breaking

class AgentExecutor:
    def __init__(self, agent: BaseAgent, max_retries: int = 3):
        self.agent = agent
        self.max_retries = max_retries
        self._failure_count = 0
        self._circuit_open = False
        self._circuit_opened_at: datetime | None = None
    
    async def execute(self, task: AgentTask) -> AgentResult:
        if self._circuit_open:
            if datetime.utcnow() - self._circuit_opened_at < timedelta(minutes=5):
                raise CircuitOpenError(f"Agent {self.agent.id} circuit is open")
            self._circuit_open = False  # try half-open
        
        for attempt in range(self.max_retries):
            try:
                result = await asyncio.wait_for(
                    self.agent.run(task),
                    timeout=30.0
                )
                self._failure_count = 0
                return result
            except asyncio.TimeoutError:
                if attempt == self.max_retries - 1:
                    self._failure_count += 1
                    if self._failure_count >= 5:
                        self._circuit_open = True
                        self._circuit_opened_at = datetime.utcnow()
                    raise AgentTimeoutError(f"Agent {self.agent.id} timed out after {self.max_retries} attempts")
                await asyncio.sleep(2 ** attempt)  # exponential backoff

Result Validation Before Handoff

Before passing one agent's output to the next, validate it structurally. A language model may produce output that looks correct but violates schema invariants:

class ValidatedHandoff:
    def __init__(self, schema: type[BaseModel]):
        self.schema = schema
    
    async def pass_to_next_agent(
        self, 
        result: Any, 
        next_agent: BaseAgent,
        task_context: TaskContext
    ) -> AgentResult:
        try:
            validated = self.schema.model_validate(result)
        except ValidationError as e:
            # Don't propagate invalid data — escalate
            raise HandoffValidationError(
                f"Agent output failed validation before handoff: {e}",
                raw_output=result,
                validation_errors=e.errors()
            )
        
        return await next_agent.run(
            task=Task(input=validated, context=task_context)
        )

Consensus for High-Stakes Decisions

For decisions where errors are costly (code that will be deployed, emails that will be sent, purchases that will be made), use a multi-agent voting mechanism:

async def consensus_decision(
    agents: list[BaseAgent],
    task: DecisionTask,
    threshold: float = 0.67
) -> ConsensusResult:
    # Run all agents in parallel
    results = await asyncio.gather(
        *[agent.decide(task) for agent in agents],
        return_exceptions=True
    )
    
    # Filter out failures
    valid_results = [r for r in results if not isinstance(r, Exception)]
    
    if len(valid_results) < len(agents) * 0.5:
        raise InsufficientConsensusError("Too many agents failed to produce a result")
    
    # Tally votes
    votes: dict[str, list[AgentDecision]] = defaultdict(list)
    for result in valid_results:
        votes[result.decision].append(result)
    
    # Find majority
    for decision, supporters in sorted(votes.items(), key=lambda x: -len(x[1])):
        agreement_ratio = len(supporters) / len(valid_results)
        if agreement_ratio >= threshold:
            return ConsensusResult(
                decision=decision,
                confidence=agreement_ratio,
                supporting_agents=[r.agent_id for r in supporters],
                dissenting_agents=[r.agent_id for r in valid_results if r.decision != decision]
            )
    
    raise NoConsensusError("Agents could not reach consensus above threshold")

Running three agents and requiring two-thirds agreement reduces single-model hallucination risk substantially. The cost is three times the inference time and tokens, which is appropriate only for high-stakes decisions.

Observability for Multi-Agent Systems

Distributed tracing is non-negotiable. Every agent execution should propagate a trace context so you can reconstruct the full causal chain of a complex multi-agent workflow:

from opentelemetry import trace

tracer = trace.get_tracer("agent-system")

async def traced_agent_run(agent: BaseAgent, task: AgentTask, parent_context) -> AgentResult:
    with tracer.start_as_current_span(
        f"agent.{agent.agent_id}.run",
        context=parent_context,
        attributes={
            "agent.id": agent.agent_id,
            "agent.type": agent.__class__.__name__,
            "task.id": task.task_id,
            "task.type": task.task_type,
        }
    ) as span:
        try:
            result = await agent.run(task)
            span.set_attribute("result.confidence", result.confidence)
            span.set_status(trace.StatusCode.OK)
            return result
        except Exception as e:
            span.set_status(trace.StatusCode.ERROR, str(e))
            span.record_exception(e)
            raise

With distributed tracing, a single failed multi-agent workflow produces a flame graph showing exactly which agent failed, at what point in the pipeline, after how many retries, and what its inputs were. Without it, debugging is archaeology.

The field of multi-agent coordination is evolving rapidly. Standards like MCP (Model Context Protocol) and A2A (Agent-to-Agent) are emerging to formalize these communication patterns. Building on typed message contracts, circuit breakers, result validation, and distributed tracing gives you a foundation that can adopt those standards as they mature.