Agent Communication Protocols: How Agents Talk to Each Other

Why Communication Patterns Matter

In a multi-agent system, the architecture of how agents exchange information determines the system's reliability, debuggability, and scalability. Choose the wrong communication pattern and you get agents that duplicate work, lose context, or block each other. Choose the right one and the system feels like a well-coordinated team.

There are four main patterns for inter-agent communication: handoff-based message passing, event-driven communication, shared memory, and the blackboard pattern. Each fits different scenarios, and production systems often combine multiple patterns.

Pattern 1: Handoff-Based Message Passing

This is the default pattern in the OpenAI Agents SDK. One agent transfers the conversation — including the full message history — to another agent. The conversation itself is the message.

from agents import Agent, Runner, handoff

researcher = Agent(
    name="Researcher",
    instructions="Research the topic thoroughly, then hand off to the Writer to draft the article.",
    handoffs=[],  # Set below
)

writer = Agent(
    name="Writer",
    instructions="Using the research provided in the conversation, write a concise article.",
)

researcher.handoffs = [handoff(writer)]

result = Runner.run_sync(researcher, "Write an article about quantum computing trends")
print(result.final_output)

When to use: Linear workflows where one agent produces output that the next agent consumes. The conversation history serves as a natural message channel.

Limitations: The conversation history grows with every exchange, consuming context window tokens. Not suitable for high-frequency back-and-forth between agents.

Pattern 2: Event-Driven Communication

In event-driven systems, agents publish events to a message bus and other agents subscribe to relevant event types. This decouples agents — the publisher does not know who consumes the event, and subscribers do not know who published it.

While the Agents SDK does not include a built-in event bus, you can implement this pattern with a simple event system that agents interact with through tools:

from dataclasses import dataclass, field
from agents import Agent, Runner, RunContextWrapper, function_tool

@dataclass
class EventBus:
    events: list[dict] = field(default_factory=list)

    def publish(self, event_type: str, data: str, source: str):
        self.events.append({
            "type": event_type,
            "data": data,
            "source": source,
        })

    def get_events(self, event_type: str = None) -> list[dict]:
        if event_type:
            return [e for e in self.events if e["type"] == event_type]
        return self.events

@dataclass
class AppContext:
    event_bus: EventBus = field(default_factory=EventBus)

@function_tool
def publish_event(
    ctx: RunContextWrapper[AppContext],
    event_type: str,
    data: str,
) -> str:
    """Publish an event to the event bus."""
    ctx.context.event_bus.publish(event_type, data, source="agent")
    return f"Published event: {event_type}"

@function_tool
def read_events(
    ctx: RunContextWrapper[AppContext],
    event_type: str,
) -> str:
    """Read all events of a specific type from the event bus."""
    events = ctx.context.event_bus.get_events(event_type)
    if not events:
        return f"No events of type '{event_type}' found"
    return "\n".join(f"- {e['data']}" for e in events)

When to use: Systems where multiple agents need to react to the same events, or where you want loose coupling between agents. Good for logging, monitoring, and notification workflows.

Limitations: Harder to trace causality. When something goes wrong, you must reconstruct which events triggered which agent actions.

Pattern 3: Shared Memory via RunContext

As covered in the previous post, RunContext provides a shared memory space that all agents can read from and write to. This is the most direct form of inter-agent communication — agents communicate by modifying shared state.

from dataclasses import dataclass, field
from agents import Agent, RunContextWrapper, function_tool

@dataclass
class ResearchState:
    findings: dict[str, str] = field(default_factory=dict)
    synthesis: str = ""

@function_tool
def store_finding(
    ctx: RunContextWrapper[ResearchState],
    topic: str,
    finding: str,
) -> str:
    """Store a research finding in shared state."""
    ctx.context.findings[topic] = finding
    return f"Stored finding for topic: {topic}"

@function_tool
def read_all_findings(
    ctx: RunContextWrapper[ResearchState],
) -> str:
    """Read all research findings from shared state."""
    if not ctx.context.findings:
        return "No findings stored yet"
    return "\n".join(
        f"**{topic}**: {finding}"
        for topic, finding in ctx.context.findings.items()
    )

When to use: When agents need to incrementally build up a shared data structure — a report, a plan, a customer profile. Ideal when the order of contributions does not matter.

Limitations: No built-in notification. Agent B does not know that Agent A wrote new data unless it explicitly checks. This works in sequential workflows but requires polling in concurrent ones.

Pattern 4: The Blackboard Pattern

The blackboard pattern is a specialized form of shared memory where the shared state is structured as a workspace that agents contribute to iteratively. Each agent examines the current state of the blackboard, determines if it can contribute, adds its contribution, and signals that the blackboard has been updated.

This pattern is particularly powerful for problems that require iterative refinement:

from dataclasses import dataclass, field
from agents import Agent, Runner, RunContextWrapper, function_tool, handoff

@dataclass
class Blackboard:
    problem_statement: str = ""
    proposed_solutions: list[dict] = field(default_factory=list)
    critiques: list[dict] = field(default_factory=list)
    final_recommendation: str = ""
    iteration: int = 0

@function_tool
def read_blackboard(ctx: RunContextWrapper[Blackboard]) -> str:
    """Read the current state of the blackboard."""
    bb = ctx.context
    solutions = "\n".join(
        f"  - [{s['author']}]: {s['solution']}"
        for s in bb.proposed_solutions
    ) or "  None yet"
    critiques = "\n".join(
        f"  - [{c['author']}]: {c['critique']}"
        for c in bb.critiques
    ) or "  None yet"
    return f"""Problem: {bb.problem_statement}
Iteration: {bb.iteration}
Solutions:\n{solutions}
Critiques:\n{critiques}"""

@function_tool
def propose_solution(
    ctx: RunContextWrapper[Blackboard],
    solution: str,
) -> str:
    """Add a proposed solution to the blackboard."""
    ctx.context.proposed_solutions.append({
        "author": "solver",
        "solution": solution,
    })
    return "Solution recorded on blackboard"

@function_tool
def add_critique(
    ctx: RunContextWrapper[Blackboard],
    critique: str,
) -> str:
    """Add a critique of existing solutions to the blackboard."""
    ctx.context.critiques.append({
        "author": "critic",
        "critique": critique,
    })
    ctx.context.iteration += 1
    return "Critique recorded on blackboard"

An orchestrator reads the blackboard, sends a solver agent to propose solutions, then sends a critic agent to critique them, then loops back for refinement.

When to use: Iterative problem-solving, document drafting with review cycles, and any workflow where multiple perspectives must converge on a solution.

Choosing the Right Pattern

Pattern	Best For	Coupling	Complexity
Handoff messaging	Linear workflows	Tight	Low
Event-driven	Reactive, notification-heavy systems	Loose	Medium
Shared memory	Incremental data building	Medium	Low
Blackboard	Iterative refinement, multi-perspective	Medium	Medium

Most production systems combine patterns. A customer support system might use handoffs for routing (Pattern 1), shared RunContext for customer data (Pattern 3), and events for logging interactions (Pattern 2).

FAQ

Can agents communicate directly without going through the orchestrator?

Yes. With handoffs, agents can transfer directly to each other without returning to the orchestrator first. Agent A can hand off to Agent B, which can hand off to Agent C. However, this creates tighter coupling and can make the workflow harder to trace. An orchestrator-mediated approach is easier to debug and modify.

How do I prevent agents from overwriting each other's state?

Use append-only data structures (lists, not strings) for fields that multiple agents write to. Each agent appends its contribution rather than replacing the existing value. For cases where replacement is necessary, track the author and timestamp of each write so you can audit conflicts.

What is the performance cost of shared state vs. message passing?

Shared state via RunContext has essentially zero overhead — it is an in-memory Python object. Message passing through handoffs has the cost of an additional LLM call for each handoff. Event-driven communication has the cost of the event bus implementation, which is negligible for in-process buses but significant if using external message queues.

---

#AgentCommunication #MultiAgentSystems #MessagePassing #BlackboardPattern #OpenAIAgentsSDK #AgenticAI #LearnAI #AIEngineering