Communication Between Agents

With more than one agent, the architecture stops being about prompts and starts being about how information moves. There are three paradigms, and most production systems blend them — each gets a step below:

The deeper question this lesson keeps returning to is what you put on the wire. An agent's message is tokens that cost money and fill a finite context window — passing the wrong thing is the most common cause of expensive, slow, confused multi-agent systems.

Before any paradigm, agree on a message envelope — metadata wrapping every payload regardless of content. It is what makes a system debuggable and recoverable. Treat it like an HTTP header set.

JSON is the pragmatic default for serialization: readable in logs, universally supported, and already what LLM tool calls produce. Reach for Protobuf or Avro only for a strict cross-language contract or a measured throughput problem. And validate at every boundary — a malformed message from a hallucinating agent must fail loudly at the edge, not corrupt downstream state.

from pydantic import BaseModel, Field
from datetime import datetime, timezone
from uuid import uuid4

class Envelope(BaseModel):
    message_id: str = Field(default_factory=lambda: str(uuid4()))
    correlation_id: str
    causation_id: str | None = None
    sender: str
    recipient: str | None = None
    schema_version: int = 1
    idempotency_key: str | None = None
    timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

Message passing suits the supervisor pattern: a coordinator hands a typed task to a named worker and awaits a typed result. Define request and response as schemas, then carry them in the envelope.

from typing import Literal

class ResearchRequest(BaseModel):
    type: Literal["research_request"] = "research_request"
    query: str
    deadline_seconds: int = 30

class ResearchResult(BaseModel):
    type: Literal["research_result"] = "research_result"
    findings: list[str]
    confidence: float

class Message(BaseModel):
    envelope: Envelope
    body: ResearchRequest | ResearchResult

A worker runs its agent loop and returns a Message whose reply envelope copies the request's correlation_id (preserving the task thread) and sets causation_id to the request's message_id — so you can reconstruct which message produced which.

The supervisor must await with a timeout — never assume a worker responds, since an agent can stall on a slow tool call or burn its budget. Wrap the call in asyncio.wait_for(..., timeout=req.deadline_seconds) and retry, fall back, or escalate on TimeoutError. Synchronous request/response is easiest to reason about but couples the supervisor's liveness to the slowest worker; for long-running work, go async — fire the request, return a handle, let the worker post back when ready.

In the blackboard pattern, agents don't address each other. They read from and write to a shared workspace and a controller wakes the agent whose preconditions are met — ideal for agents collaborating on one evolving artifact (a document, a plan, a codebase).

The hard part is concurrency: if two agents read v3, both edit, and both write, the second silently clobbers the first — a lost update. Use optimistic concurrency control — every write declares the version it was based on, and the store rejects stale writes.

from dataclasses import dataclass, field

class StaleWriteError(Exception): ...

@dataclass
class Blackboard:
    state: dict = field(default_factory=dict)
    version: int = 0

    def read(self) -> tuple[dict, int]:
        return dict(self.state), self.version

    def write(self, patch: dict, base_version: int) -> int:
        if base_version != self.version:
            raise StaleWriteError(f"based on v{base_version}, store at v{self.version}")
        self.state.update(patch)
        self.version += 1
        return self.version

Agents do read-modify-write with a bounded retry budget — a compare-and-swap loop. Read state and its version, compute a patch (possibly via the model), then write(patch, base_version=ver); on StaleWriteError, re-read and retry, capping attempts so contention can't loop forever. In production the blackboard is Redis, Postgres, or DynamoDB and the version becomes a row version or conditional-write expression — same pattern.

Event-driven communication is the loosest coupling: an agent publishes a fact (source_fetched, draft_ready) with no idea who consumes it. New agents subscribe without editing the publisher — this scales a system's number of behaviors without scaling its interconnections.

Use a durable log, not fire-and-forget pub/sub, so a crashed consumer can resume. Redis Streams gives consumer groups and acknowledgements; Kafka and NATS JetStream offer the same semantics at scale.

import json, redis.asyncio as redis

r = redis.Redis(decode_responses=True)

async def publish(event_type: str, env: Envelope, payload: dict):
    await r.xadd("agent.events", {
        "type": event_type,
        "envelope": env.model_dump_json(),
        "payload": json.dumps(payload),
    })

async def subscribe(group: str, consumer: str, handler):
    while True:
        resp = await r.xreadgroup(group, consumer, {"agent.events": ">"},
                                  count=10, block=5000)
        for _stream, messages in resp or []:
            for msg_id, fields in messages:
                await handler(fields)                        # dedup on idempotency_key
                await r.xack("agent.events", group, msg_id)  # ack ONLY on success

Acknowledge only after the handler succeeds, so an unacked event is redelivered after a crash; dedup on the envelope's idempotency key to make redelivery safe. Create the group once at startup with xgroup_create(..., mkstream=True).

This decision determines whether your system is fast and cheap or slow and broke. Every token passed between agents is paid for and competes for the receiver's context window. There are three options; mature systems use all three deliberately.

The reference approach mirrors REST: pass a handle, dereference only if needed. An Artifact schema carries an artifact_id, a short summary for routing, a uri to the full body, and a token_estimate so the receiver can budget before fetching. A coordinator routes on the cheap summary; only the downstream agent that needs detail pays to dereference the artifact. That is the difference between cost that grows linearly with agents and cost that explodes quadratically because everyone forwards everything.

Now apply all three paradigms to a coordinator / search / synthesis research system — the workflow you'll formalize in Lesson 5 with the Anthropic Agent SDK. The coordinator uses message passing to delegate; the search agent writes artifacts to a blackboard; a findings_ready event triggers synthesis without polling:

coordinator ──ResearchRequest (msg)──▶ search_agent
                                          │ writes artifacts
                                          ▼
                                      blackboard ──findings_ready (event)──▶ synthesis
coordinator ◀────────── Report (summary + refs) ──────────────────────────┘

Every edge passes the minimum fidelity its receiver needs: the event carries only a blackboard version and artifact IDs, synthesis fetches full artifacts on demand, and the report returns a summary plus citations rather than the raw transcript.