Phase 4: Multi-Agent Systems · 55 min · Python · Pydantic · Anthropic SDK
Agent Communication Protocols
Treat agent-to-agent messages like API calls, not chat messages.
Real-world application: Protocol design is a systems engineering skill that most AI engineers from a data science background lack. Engineers who think about inter-agent communication like API design — with versioning, typed schemas, and backward compatibility — signal production engineering experience. This is tested in system design rounds at companies building multi-agent products.
What you will learn
- Design structured message schemas for agent-to-agent communication
- Implement a shared context object that flows through a multi-agent workflow
- Handle schema versioning when agent protocols evolve
The Problem
Free-text agent communication has three critical flaws: ambiguity (the receiving agent must infer meaning), cost (every status check requires an LLM call to parse), and fragility (natural language descriptions of results change unpredictably). Structured message schemas eliminate all three.
The principle is straightforward: treat inter-agent communication like an internal API. If you wouldn't send free-text JSON to a microservice, don't send it between agents.
Agent message schema design
Every message between agents should be a typed Pydantic object. Standard fields:
class AgentMessage(BaseModel):
message_id: str # Unique ID for deduplication and tracing
schema_version: str # "1.0.0" — for migration support
from_agent: str # Sender identity
to_agent: str # Intended recipient
task_id: str # The workflow task this belongs to
payload: dict # The actual task or question
status: str # "pending" | "complete" | "failed" | "partial"
confidence: float # 0.0–1.0: how certain is the sender?
metadata: dict # Timing, costs, token counts, source refs
timestamp: str # ISO 8601
The confidence field deserves special attention. It enables downstream agents to make decisions without calling an LLM:
if result.confidence < 0.6:
# Re-research with a different source
...
elif result.confidence < 0.8:
# Add a "low confidence — verify before use" caveat
...
else:
# Proceed normally
...
This is logic that would require a full LLM call if confidence were expressed in natural language ("I'm fairly sure but not completely certain...").
Agent A sends Agent B a free-text string: "The research on topic X is done. I found some interesting things." What's wrong with this?
Even though LLMs can parse natural language, inter-agent communication should be structured for three reasons: (1) Reliability — no parsing ambiguity. (2) Logic branching — Agent B can check result.status == "complete" without calling an LLM. (3) Cost — checking a field is free; parsing free text costs tokens. Treat agent-to-agent messages like API calls, not chat messages.
Shared context objects
Individual messages carry task-level data. But multi-agent workflows also need shared, persistent context: the original user request, accumulated results, decisions made, errors encountered.
A WorkflowContext object flows through the pipeline as an append-only log:
class WorkflowContext(BaseModel):
workflow_id: str
task: str # Original user request
results: list[AgentResult] = [] # Accumulated results (append-only)
errors: list[str] = [] # Any errors encountered
decisions: list[str] = [] # Routing decisions made
created_at: str
schema_version: str = "1.0.0"
Why append-only? Mutating previous results creates debugging nightmares. If Agent C overwrites Agent A's result, you lose the ability to reconstruct what happened. Append-only means the context is a complete audit trail of the workflow's history.
Why not just pass the conversation history? Conversation history is unstructured text. Searching it requires string parsing or LLM calls. A typed WorkflowContext lets you access context.results[-1].confidence in one line without parsing.
Protocol versioning
Agent communication protocols evolve. You add a confidence field in v1.2. You rename result to output in v2.0. Old agents running v1 still send messages. What happens?
Backward-compatible evolution (the right way):
# v1: original schema
class AgentResult_v1(BaseModel):
status: str
result: dict
# v2: adds confidence field with a default — v1 messages still valid
class AgentResult_v2(BaseModel):
status: str
result: dict
confidence: float = 1.0 # Optional with sensible default
schema_version: str = "2.0.0"
By making new fields Optional with defaults, v1 agents that don't send confidence get the default (1.0). v2 agents can start sending it. No big-bang migration required.
Rules for backward-compatible schema evolution:
- Add new fields as Optional with defaults — never required
- Never remove fields (mark as deprecated, remove in a future major version)
- Never change a field's type
- Use semantic versioning in the schema (
major.minor.patch)
You update your AgentResult schema from v1 (no confidence field) to v2 (confidence: float). Old agents still running v1. How do you handle this?
Making new fields optional with sensible defaults is the backward-compatible evolution pattern. v1 agents that don't send confidence get the default. v2 agents can start sending confidence. No big-bang upgrade required. This is the same principle as REST API versioning: don't break existing clients when you add fields.
Build
Build it
"""
lesson_04_05_protocols.py
Agent communication protocol design:
- AgentMessage, AgentResult, WorkflowContext Pydantic models
- Schema versioning with migration adapter
- AgentBus: simple in-memory message routing
- 3-agent pipeline demo using structured messages
"""
import os
import uuid
from datetime import datetime
from typing import Any, Optional
from pydantic import BaseModel, Field
import anthropic
ANTHROPIC_KEY = os.getenv("ANTHROPIC_API_KEY")
client = anthropic.Anthropic(api_key=ANTHROPIC_KEY) if ANTHROPIC_KEY else None
# --- Schema v1 ---
class AgentResult_v1(BaseModel):
schema_version: str = "1.0.0"
from_agent: str
task_id: str
status: str
result: Any
error: Optional[str] = None
# --- Schema v2: adds confidence + sources ---
class AgentResult(BaseModel):
schema_version: str = "2.0.0"
message_id: str = Field(default_factory=lambda: str(uuid.uuid4())[:8])
from_agent: str
task_id: str
status: str # "success" | "failed" | "partial"
result: Any
confidence: float = 1.0 # Optional with default — v1 compatible
sources: list[str] = [] # Optional with default — v1 compatible
error: Optional[str] = None
timestamp: str = Field(default_factory=lambda: datetime.utcnow().isoformat())
def migrate_v1_to_v2(v1: AgentResult_v1) -> AgentResult:
"""Migration adapter: upgrade a v1 result to v2."""
return AgentResult(
schema_version="2.0.0",
from_agent=v1.from_agent,
task_id=v1.task_id,
status=v1.status,
result=v1.result,
confidence=1.0, # Default when v1 didn't have this field
error=v1.error,
)
class WorkflowContext(BaseModel):
"""
Shared context flowing through a multi-agent pipeline.
Append-only: agents add results, never overwrite previous ones.
"""
workflow_id: str = Field(default_factory=lambda: str(uuid.uuid4())[:8])
task: str
schema_version: str = "2.0.0"
results: list[AgentResult] = []
errors: list[str] = []
decisions: list[str] = []
created_at: str = Field(default_factory=lambda: datetime.utcnow().isoformat())
def add_result(self, result: AgentResult) -> None:
self.results.append(result)
def add_decision(self, decision: str) -> None:
self.decisions.append(f"{datetime.utcnow().isoformat()}: {decision}")
def latest_result(self) -> Optional[AgentResult]:
return self.results[-1] if self.results else None
def summary(self) -> str:
lines = [
f"Workflow: {self.workflow_id} Task: {self.task[:50]}",
f"Results: {len(self.results)} Decisions: {len(self.decisions)} Errors: {len(self.errors)}",
]
for r in self.results:
lines.append(
f" [{r.from_agent}] status={r.status} confidence={r.confidence:.2f} "
f"result_type={type(r.result).__name__}"
)
for d in self.decisions:
lines.append(f" decision: {d}")
return "\n".join(lines)
class AgentBus:
"""
Simple in-memory message bus for agent-to-agent communication.
In production, replace with a proper message queue.
"""
def __init__(self):
self._handlers: dict[str, callable] = {}
self._log: list[AgentResult] = []
def register(self, agent_name: str, handler: callable):
self._handlers[agent_name] = handler
def send(self, result: AgentResult, to_agent: str) -> Any:
"""Route a result to the appropriate agent handler."""
self._log.append(result)
handler = self._handlers.get(to_agent)
if not handler:
raise ValueError(f"No handler registered for '{to_agent}'")
return handler(result)
def print_log(self):
print("\n--- Message Bus Log ---")
for r in self._log:
print(
f" [{r.message_id}] {r.from_agent} → status={r.status} "
f"confidence={r.confidence:.2f} version={r.schema_version}"
)
def _llm(prompt: str, system: str = "") -> str:
if not client:
return f"[Mock: {prompt[:60]}...]"
resp = client.messages.create(
model="claude-haiku-4-5",
max_tokens=350,
system=system or "You are a helpful assistant.",
messages=[{"role": "user", "content": prompt}],
)
return resp.content[0].text
def researcher_agent(task: str, task_id: str) -> AgentResult:
content = _llm(f"Research: {task}. Give 3–4 concise key facts.", "You are a researcher.")
return AgentResult(
from_agent="researcher",
task_id=task_id,
status="success",
result={"summary": content},
confidence=0.84,
sources=["llm_knowledge"],
)
def writer_agent(research_result: AgentResult) -> AgentResult:
if research_result.confidence < 0.5:
return AgentResult(
from_agent="writer",
task_id=research_result.task_id,
status="partial",
result={"draft": "Low confidence research — draft requires human review."},
confidence=0.3,
error="Upstream confidence too low to produce high-quality draft.",
)
summary = research_result.result.get("summary", "")
draft = _llm(f"Write a 150-word brief using: {summary}", "You are a technical writer.")
return AgentResult(
from_agent="writer",
task_id=research_result.task_id,
status="success",
result={"draft": draft},
confidence=0.88,
)
def reviewer_agent(write_result: AgentResult) -> AgentResult:
draft = write_result.result.get("draft", "")
review = _llm(
f"Review this brief. Start with APPROVED or NEEDS_REVISION:\n{draft}",
"You are an editorial reviewer.",
)
approved = "APPROVED" in review.upper()
return AgentResult(
from_agent="reviewer",
task_id=write_result.task_id,
status="success",
result={"verdict": "APPROVED" if approved else "NEEDS_REVISION", "review": review},
confidence=0.9,
)
if __name__ == "__main__":
task = "How transformer attention mechanisms work"
task_id = str(uuid.uuid4())[:8]
ctx = WorkflowContext(task=task)
print("=" * 60)
print("STRUCTURED PROTOCOL PIPELINE")
print("=" * 60)
# Step 1: Research
r1 = researcher_agent(task, task_id)
ctx.add_result(r1)
ctx.add_decision(f"research complete: confidence={r1.confidence:.2f}")
print(f"\n[researcher] status={r1.status} confidence={r1.confidence:.2f}")
# Confidence gate
if r1.confidence < 0.6:
ctx.errors.append("Research confidence too low")
ctx.add_decision("HALT: research confidence below threshold")
print("[pipeline] Halting — low confidence research.")
else:
r2 = writer_agent(r1)
ctx.add_result(r2)
ctx.add_decision(f"writing complete: status={r2.status}")
print(f"[writer] status={r2.status} confidence={r2.confidence:.2f}")
r3 = reviewer_agent(r2)
ctx.add_result(r3)
ctx.add_decision(f"review complete: verdict={r3.result.get('verdict')}")
print(f"[reviewer] status={r3.status} verdict={r3.result.get('verdict')}")
print("\n" + ctx.summary())
# Schema migration demo
print("\n" + "=" * 60)
print("SCHEMA MIGRATION DEMO: v1 → v2")
print("=" * 60)
v1_result = AgentResult_v1(
from_agent="legacy_agent",
task_id="legacy-001",
status="success",
result={"data": "some old format result"},
)
v2_result = migrate_v1_to_v2(v1_result)
print(f"v1: schema_version={v1_result.schema_version} has_confidence=False")
print(f"v2: schema_version={v2_result.schema_version} confidence={v2_result.confidence} (default applied)")
Unlock the full lesson
You've read the first 2 sections. The rest of this lesson covers Check Yourself — plus a hands-on lab, quiz, and project artifact.
Create a free account — the opening phases of 29 of 35 courses are free, no credit card.
Browse all courses · View pricing · DeVenture Academy