Communication Patterns for Multi-Agent Systems

Multi agent communication is the set of rules that decide how autonomous agents exchange information, hand off tasks, and stay in sync without stepping on each other. Get it wrong and you end up with agents overwriting each other's work, looping forever, or silently dropping context between steps. Get it right and you can add a fifth agent to a four-agent pipeline without rewriting the other four. This article walks through five concrete communication patterns, when to reach for each one, and working Python you can run today.

Why Multi Agent Communication Is Different From Function Calls

A single agent calling tools is a straight line: prompt in, tool call out, result back in. Multi agent systems break that line into a graph. Agent A might need to wait on Agent B, Agent B might fan out work to Agent C and Agent D in parallel, and any of them might fail mid-task. The communication layer is what turns a pile of independent LLM calls into a system that behaves predictably.

Three problems show up in almost every multi agent build:

• State drift: two agents holding different views of "what's true right now."
• Message ambiguity: one agent sends free text, the receiving agent guesses at structure.
• Silent failure: an agent times out or returns garbage and nothing downstream notices.

Every pattern below is really just a different trade-off between coupling, latency, and failure visibility. There is no single "correct" pattern. Pick based on how many agents you have, whether they run in parallel, and how much you're willing to pay in coordination overhead.

Pattern 1: Direct Message Passing

The simplest pattern: agents call each other directly, synchronously, passing structured payloads. This works well for linear pipelines of two to four agents where each step depends on the previous one finishing.

from dataclasses import dataclass, field
from typing import Any

@dataclass
class AgentMessage:
    sender: str
    recipient: str
    content: str
    metadata: dict = field(default_factory=dict)

class ResearchAgent:
    def run(self, topic: str) -> AgentMessage:
        findings = f"Three key facts about {topic}: ..."
        return AgentMessage(
            sender="researcher",
            recipient="writer",
            content=findings,
            metadata={"topic": topic, "confidence": 0.82},
        )

class WriterAgent:
    def run(self, message: AgentMessage) -> AgentMessage:
        draft = f"Draft based on: {message.content}"
        return AgentMessage(
            sender="writer",
            recipient="editor",
            content=draft,
            metadata=message.metadata,
        )

researcher = ResearchAgent()
writer = WriterAgent()

research_msg = researcher.run("battery recycling")
draft_msg = writer.run(research_msg)
print(draft_msg.content)

This is easy to debug because the call stack IS the communication log. The downside is tight coupling: the writer agent's signature has to match what the researcher emits. If you add a fact-checker agent between them, you're editing code in two places instead of one. Direct passing scales fine up to a handful of agents; past that, the wiring becomes the bottleneck.

Pattern 2: Shared Memory (Blackboard) Pattern

Instead of agents talking to each other, they all read and write to a shared state object, often called a blackboard. Each agent watches the blackboard, does its part when its preconditions are met, and writes results back. No agent needs to know who else exists.

class Blackboard:
    def __init__(self):
        self.state: dict[str, Any] = {}
        self.log: list[str] = []

    def write(self, key: str, value: Any, author: str):
        self.state[key] = value
        self.log.append(f"{author} wrote '{key}'")

    def read(self, key: str, default=None):
        return self.state.get(key, default)

    def has(self, key: str) -> bool:
        return key in self.state


class PricingAgent:
    def step(self, board: Blackboard):
        if board.has("product") and not board.has("price"):
            product = board.read("product")
            board.write("price", f"${len(product) * 4}.99", "pricing_agent")


class CopyAgent:
    def step(self, board: Blackboard):
        if board.has("price") and not board.has("listing_copy"):
            price = board.read("price")
            product = board.read("product")
            copy = f"{product} now available for {price}!"
            board.write("listing_copy", copy, "copy_agent")


board = Blackboard()
board.write("product", "wireless charger", "intake_agent")

agents = [PricingAgent(), CopyAgent()]
for _ in range(3):
    for agent in agents:
        agent.step(board)

print(board.state["listing_copy"])
print(board.log)

The blackboard decouples agents completely, you can add or remove agents without touching the others as long as they agree on key names. This is the pattern behind most "swarm" style designs. The cost is that debugging gets harder: instead of a call stack, you have a shared mutable object and you need the log to reconstruct who did what. In production, back the blackboard with something durable (Redis, Postgres, or a simple SQLite table) instead of a Python dict so a crashed agent doesn't lose state.

Pattern 3: Publish/Subscribe Event Bus

Pub/sub sits between direct calls and shared memory. Agents publish events to named topics; other agents subscribe to the topics they care about. Nobody calls anybody directly, and nobody has to poll shared state either.

from collections import defaultdict
from typing import Callable

class EventBus:
    def __init__(self):
        self.subscribers: dict[str, list[Callable]] = defaultdict(list)

    def subscribe(self, topic: str, handler: Callable):
        self.subscribers[topic].append(handler)

    def publish(self, topic: str, payload: dict):
        for handler in self.subscribers[topic]:
            handler(payload)


bus = EventBus()

def on_order_placed(payload: dict):
    print(f"[inventory_agent] reserving stock for {payload['item']}")
    bus.publish("stock_reserved", {"item": payload["item"], "qty": payload["qty"]})

def on_stock_reserved(payload: dict):
    print(f"[shipping_agent] scheduling shipment for {payload['item']} x{payload['qty']}")

def on_order_placed_notify(payload: dict):
    print(f"[notification_agent] emailing customer about {payload['item']}")

bus.subscribe("order_placed", on_order_placed)
bus.subscribe("order_placed", on_order_placed_notify)
bus.subscribe("stock_reserved", on_stock_reserved)

bus.publish("order_placed", {"item": "desk lamp", "qty": 2})

This is the toy in-process version. For a real system, swap the in-memory EventBus for a message broker like Redis Streams, RabbitMQ, or Kafka, so agents can run as separate processes or separate machines and still react to the same events. Pub/sub shines when one event needs to fan out to multiple independent agents (notify, log, bill, ship) that don't need to know about each other. The trade-off is that tracing a single request across five topic handlers requires correlation IDs, otherwise your logs turn into soup.

Pattern 4: Orchestrator-Worker (Hub and Spoke)

Here a single orchestrator agent owns the plan and dispatches subtasks to worker agents, then collects and merges their results. Workers never talk to each other directly, which keeps the system easy to reason about even as you add more workers. This is the pattern behind most agent frameworks you'll see in 2026, including LangGraph's graph-based orchestration and CrewAI's crew/task model, though you don't need a framework to build it.

from concurrent.futures import ThreadPoolExecutor

class Orchestrator:
    def __init__(self, workers: dict[str, Callable[[str], str]]):
        self.workers = workers

    def run(self, task: str, subtasks: dict[str, str]) -> dict[str, str]:
        results = {}
        with ThreadPoolExecutor(max_workers=len(subtasks)) as pool:
            futures = {
                pool.submit(self.workers[worker_name], instruction): worker_name
                for worker_name, instruction in subtasks.items()
            }
            for future in futures:
                worker_name = futures[future]
                try:
                    results[worker_name] = future.result(timeout=30)
                except Exception as exc:
                    results[worker_name] = f"ERROR: {exc}"
        return self.merge(results)

    def merge(self, results: dict[str, str]) -> dict[str, str]:
        results["summary"] = " | ".join(f"{k}: {v}" for k, v in results.items())
        return results


def seo_worker(instruction: str) -> str:
    return f"keyword analysis for: {instruction}"

def competitor_worker(instruction: str) -> str:
    return f"competitor scan for: {instruction}"

orchestrator = Orchestrator({
    "seo": seo_worker,
    "competitor": competitor_worker,
})

output = orchestrator.run(
    task="launch plan",
    subtasks={"seo": "wireless chargers", "competitor": "wireless chargers"},
)
print(output["summary"])

The orchestrator pattern gives you a single place to enforce timeouts, retries, and ordering, which is exactly why it dominates production agent systems. Model Context Protocol (MCP) servers fit naturally here too: the orchestrator treats each MCP-exposed tool or sub-agent as a worker with a well-defined schema, so you get the hub-and-spoke topology plus a standard contract for what each worker accepts and returns. The failure mode to watch for is the orchestrator becoming a bottleneck or single point of failure, if it crashes, everything downstream stalls, so it needs its own retry and checkpoint logic.

Pattern 5: Contract-Based Message Schemas

Whichever topology you pick, the actual message format matters as much as the wiring. Free-text messages between agents ("hey, here's what I found...") force every receiving agent to re-parse natural language, which is slow and fragile. Define a schema instead, and validate against it at the boundary.

from pydantic import BaseModel, ValidationError

class TaskResult(BaseModel):
    agent_id: str
    task_id: str
    status: str
    payload: dict
    confidence: float

def receive(raw_message: dict) -> TaskResult:
    try:
        result = TaskResult(**raw_message)
    except ValidationError as exc:
        raise ValueError(f"malformed message from agent: {exc}") from exc

    if result.status not in {"success", "partial", "failed"}:
        raise ValueError(f"unknown status: {result.status}")

    return result


incoming = {
    "agent_id": "researcher-01",
    "task_id": "task-882",
    "status": "success",
    "payload": {"facts": ["fact one", "fact two"]},
    "confidence": 0.91,
}

validated = receive(incoming)
print(validated.status, validated.confidence)

This single change, using pydantic (or a plain dataclass with manual checks if you want zero dependencies) instead of raw dicts or strings, eliminates an entire category of bugs: the downstream agent that silently reads result["confidance"] (typo) and gets None forever. Treat every agent-to-agent message like an API contract, because that's what it is.

Handling Failures and Retries in Multi Agent Communication

Agents fail differently than normal services. An LLM call can "succeed" with a 200-equivalent response that's still wrong, hallucinated, or off-topic. Build failure handling around three checks, not just exceptions:

1. Timeout: every agent call gets a hard deadline. An agent stuck in a retry loop against a flaky tool should not block the whole pipeline.
2. Schema validation: reject malformed output before it reaches the next agent, as shown above.
3. Semantic sanity check: a cheap secondary check (regex, length bounds, or a smaller model) that catches obviously wrong output, like an empty summary or a price of "$0.00" from a pricing agent.

import time

def call_with_retry(agent_fn, payload, retries=2, timeout_s=10):
    last_error = None
    for attempt in range(retries + 1):
        start = time.monotonic()
        try:
            result = agent_fn(payload)
            elapsed = time.monotonic() - start
            if elapsed > timeout_s:
                raise TimeoutError(f"agent exceeded {timeout_s}s")
            return result
        except Exception as exc:
            last_error = exc
            time.sleep(min(2 ** attempt, 5))
    raise RuntimeError(f"agent failed after {retries + 1} attempts: {last_error}")

Wrap every inter-agent call with something like this rather than trusting each agent implementation to handle its own retries consistently. Centralizing retry logic in the communication layer, not inside each agent, is what lets you change the policy (backoff curve, max attempts) in one place.

Observability: Tracing Messages Across Agents

Once you have more than two agents, "which agent said what, in what order" becomes the first question you ask when something breaks. Attach a correlation ID to every message at the entry point and propagate it through every hop.

import uuid
import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
logger = logging.getLogger("agent_trace")

def start_trace() -> str:
    return str(uuid.uuid4())[:8]

def log_hop(trace_id: str, agent: str, action: str, detail: str = ""):
    logger.info(f"trace={trace_id} agent={agent} action={action} {detail}")

trace_id = start_trace()
log_hop(trace_id, "researcher", "received_task", "topic=battery recycling")
log_hop(trace_id, "researcher", "emitted_result", "confidence=0.82")
log_hop(trace_id, "writer", "received_task", "from=researcher")
log_hop(trace_id, "writer", "emitted_result", "words=340")

This is deliberately low-tech: plain structured logs with a shared trace ID. It's enough to reconstruct any request's full path across a pub/sub bus, a blackboard, or an orchestrator, and it costs nothing to add on day one versus retrofitting it after a production incident. If you outgrow logs, the same trace ID slots straight into an OpenTelemetry span without changing the agent code.

Choosing the Right Pattern for Your System

A quick decision guide based on shape of the workflow:

• Linear, 2-4 steps, one owner: direct message passing. Simplest to write and debug.
• Many agents contributing partial facts to one evolving artifact: blackboard. Good for research and drafting pipelines.
• One event triggers several independent reactions: pub/sub event bus. Good for notification, logging, and billing side effects.
• A planner needs to fan work out and merge results: orchestrator-worker. The default choice for most production agent systems, and the pattern most frameworks (LangGraph, CrewAI, AutoGen) implement under the hood.
• Any of the above, at scale: add contract-based schemas and a correlation ID from the start. Retrofitting either one later is far more painful than building it in from message one.

Most real systems end up as a hybrid: an orchestrator-worker skeleton for the main flow, with a pub/sub bus for cross-cutting concerns like logging and alerts, and pydantic schemas enforced at every boundary. Start with the simplest pattern that fits your current agent count, and only add topology complexity when you actually hit its limits, not before.

FAQ

What is the difference between multi agent communication and simple tool calling? Tool calling is one agent invoking a deterministic function and getting a return value; the agent stays in control of the loop. Multi agent communication involves two or more autonomous agents, each capable of its own reasoning, exchanging messages, which means you need to handle ordering, partial failure, and conflicting state, problems that don't exist when a single agent calls a single tool.

Should agents communicate through natural language or structured data? Structured data at the boundary, natural language inside the reasoning. Let each agent think and generate in free text internally, but require it to emit a validated schema (JSON, a pydantic model, or similar) before that output crosses into another agent. This keeps the flexibility of LLM reasoning without pushing parsing ambiguity onto every downstream consumer.

How many agents is too many for direct message passing? Once you're wiring more than three or four agents together with direct calls, the coupling usually gets painful enough to switch to an event bus or orchestrator. Watch for the tell: if adding one new agent means editing the code of two or three existing agents, you've outgrown direct passing.

Do I need a message broker like Kafka or RabbitMQ for a small multi agent project? No. An in-process event bus (as shown above) or even a shared database table is enough for a single-process prototype or a small production job. Reach for a real broker only when agents run as separate services, need durability across restarts, or need to scale independently.

How do I stop two agents from overwriting the same shared state? Use optimistic locking (version numbers on writes) or namespace the blackboard keys so each agent only ever writes to keys it owns. For anything beyond a prototype, back the blackboard with a real datastore that supports transactions rather than a plain in-memory dict, so concurrent writes fail loudly instead of silently clobbering each other.

What is Model Context Protocol's role in multi agent communication? MCP standardizes how an agent discovers and calls tools and sub-agents, giving you a consistent schema for requests and responses across different servers. It's most useful in the orchestrator-worker pattern, where the orchestrator treats every MCP-exposed capability as an interchangeable worker with a predictable contract, rather than hand-rolling a different interface for each one.