Every agent boundary is a trust boundary. When Agent A hands off work to Agent B, the message carrying that work is the contract: and unlike an internal function call, there is no compiler to catch mismatches between what A sends and what B expects. Unvalidated, unversioned messages between agents are the leading cause of subtle cascade failures in enterprise multi-agent systems: a missing field propagates silently until it triggers a downstream exception hours or days later, deep in a batch job.
The Message Envelope Pattern
Every inter-agent message should be wrapped in an envelope that separates routing metadata (headers) from business payload. This lets infrastructure components, routers, loggers, monitors, process the envelope without deserializing the payload.
from pydantic import BaseModel, Field, ConfigDict
from typing import Literal, Annotated, Union
from uuid import uuid4
from datetime import datetime, UTC
import msgpack
class MessageHeader(BaseModel):
# WHY trace_id separate from correlation_id: trace_id tracks a single
# message's journey through the system (for distributed tracing spans).
# correlation_id links messages that belong to the same business transaction
#, e.g., all messages spawned by one user request share a correlation_id.
trace_id: str = Field(default_factory=lambda: str(uuid4()))
correlation_id: str
version: str = "1.0"
message_type: str
source_agent: str
target_agent: str | None = None # None means broadcast / topic routing
priority: int = Field(default=5, ge=1, le=10)
timestamp: datetime = Field(default_factory=lambda: datetime.now(UTC))
# WHY schema_hash: a hash of the schema version at publish time.
# Consumers can detect schema drift without parsing the full payload.
schema_hash: str | None = None
class AgentMessage(BaseModel):
model_config = ConfigDict(extra="ignore") # tolerate unknown fields
header: MessageHeader
payload: dict # typed further by discriminated union (see below)flowchart TD
A["Producing Agent"] --> B["AgentMessage\n(envelope + payload)"]
B --> C["Message Router\n(reads header only)"]
C -->|"message_type: ContractAnalysis"| D["ComplianceAgent"]
C -->|"message_type: RiskScore"| E["RiskScoringAgent"]
C -->|"message_type: AuditEvent"| F["ArchiverAgent"]
D --> G["Validate payload\nagainst Pydantic schema"]
E --> G
F --> G
G -->|"ValidationError"| H["Dead-Letter Queue"]
G -->|"Valid"| I["Process Task"]
style A fill:#EEF0F7,stroke:#6366F1,color:#0F172A
style C fill:#fff7ed,stroke:#f59e0b,color:#0F172A
style G fill:#fff7ed,stroke:#f59e0b,color:#0F172A
style H fill:#fff7ed,stroke:#f59e0b,color:#0F172A
style I fill:#f0fdf9,stroke:#0D9488,color:#0F172ATyped Payloads with Discriminated Unions
Discriminated unions let a single message bus carry multiple payload types while preserving type safety. The message_type field in the header acts as the discriminator, Pydantic selects the correct payload model automatically.
class ContractAnalysisPayload(BaseModel):
model_config = ConfigDict(extra="ignore")
message_type: Literal["ContractAnalysis"]
contract_id: str
document_text: str
jurisdiction: str | None = None # Optional: additive field, v1.1
clauses_flagged: list[str] = Field(default_factory=list)
class RiskScorePayload(BaseModel):
model_config = ConfigDict(extra="ignore")
message_type: Literal["RiskScore"]
entity_id: str
score: float = Field(ge=0.0, le=1.0)
contributing_factors: list[str]
confidence: float = Field(ge=0.0, le=1.0)
# WHY Annotated + Field(discriminator=...): Pydantic v2 uses the
# message_type literal to select the correct model at validation time.
# This eliminates if/elif chains in consumer code and makes type
# checking work correctly across the entire payload handling path.
AgentPayload = Annotated[
Union[ContractAnalysisPayload, RiskScorePayload],
Field(discriminator="message_type"),
]
def parse_message(raw: bytes, encoding: str = "msgpack") -> AgentMessage:
"""
WHY validate at the consumption boundary, not inside the handler:
raising ValidationError here means the queue consumer can catch it
and route to DLQ before any business logic runs. Validating inside
the handler means partial processing may have already happened.
"""
data = msgpack.unpackb(raw, raw=False) if encoding == "msgpack" else json.loads(raw)
return AgentMessage.model_validate(data)Schema Evolution Strategies
Schema evolution is the discipline of changing message contracts without breaking existing consumers. There are two safe strategies and one unsafe one to avoid.
flowchart TD
A["Schema Change Needed"] --> B{"Breaking or\nnon-breaking?"}
B -->|"New optional field\nor relaxed constraint"| C["Additive-only:\nAdd as Optional with default\nDeploy producer first\nConsumers update at leisure"]
B -->|"Rename existing field"| D["Field aliasing:\nAdd new name, keep old via alias\nBoth names valid during migration\nRemove old name after all consumers updated"]
B -->|"Remove required field\nor change type"| E["BREAKING CHANGE\nRequires versioned message type\nor coordinated deployment lockout"]
C --> F["Safe: zero-downtime deploy"]
D --> F
E --> G["Risky: plan carefully\nor redesign to avoid"]
style A fill:#EEF0F7,stroke:#6366F1,color:#0F172A
style B fill:#fff7ed,stroke:#f59e0b,color:#0F172A
style C fill:#f0fdf9,stroke:#0D9488,color:#0F172A
style D fill:#f0fdf9,stroke:#0D9488,color:#0F172A
style E fill:#fff7ed,stroke:#f59e0b,color:#0F172A
style F fill:#f0fdf9,stroke:#0D9488,color:#0F172Afrom pydantic import AliasChoices
class ContractAnalysisPayloadV2(BaseModel):
"""
WHY AliasChoices for field renaming: this model accepts BOTH
'jurisdiction_code' (new name) and 'jurisdiction' (old name).
Producers can migrate to the new name while old consumers continue
receiving their expected field name. Once all consumers accept the
new name, the alias can be dropped in a future cleanup release.
"""
model_config = ConfigDict(extra="ignore", populate_by_name=True)
message_type: Literal["ContractAnalysis"]
contract_id: str
document_text: str
jurisdiction_code: str | None = Field(
default=None,
validation_alias=AliasChoices("jurisdiction_code", "jurisdiction"),
)
clauses_flagged: list[str] = Field(default_factory=list)
# New field in v1.2: Optional with default, safe to add
regulatory_framework: str | None = NoneSerialization: JSON vs MessagePack
JSON is the default for its human readability and universal tooling support. Switch to MessagePack when payload size or deserialization throughput becomes a bottleneck. At 50,000 messages per second, MessagePack’s 20-40% size reduction and 2-3x faster deserialization can meaningfully reduce infrastructure costs. The tradeoff: binary messages require tooling (msgpack CLI, hex viewers) for debugging instead of jq. A practical middle ground: use JSON in development and staging environments, MessagePack in production, toggled via a MESSAGE_ENCODING environment variable that both producers and consumers read.
Contract testing, writing tests that assert both producer output and consumer expectations against a shared schema, is the engineering practice that keeps all these patterns from drifting. Run contract tests in CI on every change to a message schema, and fail the build if a producer would generate a message that any registered consumer cannot parse.
Reliable message contracts remove an entire class of runtime surprises from your agent network. In the next lesson, you’ll use these durable, well-typed messages as the substrate for implementing async coordination patterns: scatter-gather, fan-out/fan-in, and saga orchestration, that let hundreds of agents work together without blocking each other.