Message Schema Design and Versioning in Multi-Agent Systems

A message schema is the contract between agents. It defines what information flows between them, in what format, and with what guarantees. Like all contracts, it needs to be designed carefully upfront and managed carefully over time. Teams that treat their agent message schemas as implementation details rather than API contracts discover the cost of that decision when they try to deploy a schema change and find that half their running workflows break.

This post covers the principles and practices for designing agent message schemas that are clear, extensible, and safe to evolve. The patterns here apply whether you are using A2A, MCP, or a custom protocol.

The Message Schema Is a Contract

In a multi-agent system, messages are the primary API surface. Every field you add is a new capability. Every field you remove is a breaking change. Every type change is a potential source of deserialization errors. The discipline required to manage agent message schemas correctly is the same discipline required to manage public REST APIs. Because that is what they are.

Core Fields Every Agent Message Needs

Regardless of the specific protocol or task type, every message in a multi-agent system should include a common set of fields that enable routing, tracing, debugging, and security verification.

from pydantic import BaseModel, Field
from typing import Any, Optional, Literal
from datetime import datetime
import uuid

class AgentRef(BaseModel):
    id: str       # stable agent identifier
    role: str     # "orchestrator" | "worker" | "specialist"
    version: str  # "1.4.2", for debugging which code sent this

class AgentMessage(BaseModel):
    # Identity and routing
    message_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    schema_version: str = "2.0"          # for receiver validation
    type: str                             # "task" | "result" | "status" | "error"

    # Participants
    from_agent: AgentRef
    to_agent: Optional[AgentRef] = None  # None = broadcast

    # Tracing
    task_id: str                          # groups related messages
    parent_task_id: Optional[str] = None # for sub-task relationships
    trace_id: str                         # links to distributed trace
    correlation_id: Optional[str] = None  # caller's own reference

    # Content
    payload: dict[str, Any]               # type-specific body
    metadata: dict[str, str] = {}         # key-value annotations

    # Timing
    created_at: datetime = Field(default_factory=datetime.utcnow)
    expires_at: Optional[datetime] = None # message TTL

    # Security
    delegation_token: Optional[str] = None  # JWT from trust hierarchy
    signature: Optional[str] = None         # message signature

# Type-specific payload schemas
class TaskPayload(BaseModel):
    description: str
    context: dict[str, Any] = {}
    priority: Literal["low", "normal", "high", "critical"] = "normal"
    idempotency_key: Optional[str] = None  # for safe retries

class ResultPayload(BaseModel):
    status: Literal["success", "partial", "failed"]
    data: Any
    artifacts: list[dict] = []
    notes: Optional[str] = None  # human-readable summary

Semantic Versioning for Schemas

Agent message schemas should follow semantic versioning. The major version increments on breaking changes. The minor version increments on backward-compatible additions. The patch version increments on backward-compatible fixes like documentation or constraint updates.

The most common mistake is treating the schema_version field as cosmetic. Agents should actively check the version of messages they receive and either reject or adapt messages from incompatible versions. A receiver that silently ignores unknown fields and proceeds has accidental backward compatibility. A receiver that explicitly validates the version and invokes the appropriate handler has intentional backward compatibility. One of those is a design choice. The other is a bug waiting to happen.

Backward Compatibility Rules

Two rules govern backward-compatible schema changes. New optional fields can be added in any minor version. Existing required fields cannot be removed or have their types changed in a minor version. Any removal or type change is a breaking change that requires a major version bump.

• Safe: adding a new optional field with a default value. Old senders will not send it; new receivers will use the default.
• Safe: adding a new optional enum value to an existing enum field. Old receivers should handle unknown enum values gracefully with a default behavior.
• Safe: relaxing a constraint (changing a field from required to optional).
• Breaking: removing a field. Old senders may still send it; old receivers may still expect it.
• Breaking: changing a field's type (string to object is the most common example).
• Breaking: adding a new required field without a default. Old senders will not populate it.
• Breaking: renaming a field.

Version Negotiation at Handshake

When two agents establish a connection, they should negotiate a schema version before exchanging task messages. The callee advertises the schema versions it supports. The caller selects the highest version it also supports. All subsequent messages use that negotiated version.

from typing import Optional

class SchemaVersionNegotiator:
    def __init__(self, supported_versions: list[str]):
        self.supported = set(supported_versions)

    def negotiate(self, peer_supported: list[str]) -> Optional[str]:
        """
        Select the highest mutually supported major.minor version.
        Returns None if no compatible version exists.
        """
        from packaging.version import Version
        compatible = [v for v in peer_supported if v in self.supported]
        if not compatible:
            return None
        return str(max(Version(v) for v in compatible))

    def get_handler(self, version: str):
        """Return the appropriate message handler for this version."""
        major = version.split(".")[0]
        handler = self.handlers.get(f"v{major}")
        if not handler:
            raise ValueError(f"No handler for schema version {version}")
        return handler

# Usage during A2A handshake
negotiator = SchemaVersionNegotiator(["2.0", "2.1", "1.2"])

# Peer only supports v1
negotiated = negotiator.negotiate(["1.0", "1.2"])
assert negotiated == "1.2"  # highest compatible version

# Peer supports v2
negotiated = negotiator.negotiate(["2.0", "2.1", "3.0"])
assert negotiated == "2.1"  # our highest supported, not peer's highest

# No compatible version
negotiated = negotiator.negotiate(["3.0", "3.1"])
assert negotiated is None  # must reject connection

Message Routing and Content-Based Filtering

In multi-agent systems with many message producers and consumers, a message router becomes a critical infrastructure component. The router receives messages from all agents and delivers them to the appropriate recipients based on message type, topic, or content. Content-based filtering lets agents declare subscriptions using predicates rather than explicit addresses.

Using a Schema Registry

As your multi-agent system grows, maintaining schema definitions in individual agent codebases becomes unsustainable. A schema registry provides a central store for all schema versions, enables compatibility checking before deployment, and gives you a single source of truth for what each version of each message type looks like.

# Schema registry client
import httpx
from functools import lru_cache

class SchemaRegistry:
    def __init__(self, registry_url: str):
        self.url = registry_url

    @lru_cache(maxsize=256)
    async def get_schema(self, message_type: str, version: str) -> dict:
        """Fetch the JSON Schema for a given message type and version."""
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                f"{self.url}/schemas/{message_type}/{version}"
            )
            resp.raise_for_status()
            return resp.json()

    async def validate(self, message: dict) -> tuple[bool, list[str]]:
        """Validate a message against its declared schema version."""
        msg_type = message.get("type")
        schema_ver = message.get("schema_version")
        if not msg_type or not schema_ver:
            return False, ["Missing type or schema_version"]
        schema = await self.get_schema(msg_type, schema_ver)
        return jsonschema_validate(message, schema)

    async def check_compatibility(
        self,
        message_type: str,
        old_version: str,
        new_version: str,
    ) -> dict:
        """Check if new_version is backward compatible with old_version."""
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{self.url}/compatibility/{message_type}",
                json={"old": old_version, "new": new_version},
            )
            return resp.json()  # {"compatible": bool, "breaking_changes": []}

Schema Migration Patterns

When a breaking schema change is unavoidable, the migration needs to be managed carefully. The standard approach is to run both the old and new schema versions in parallel for a transition period. Agents that have been upgraded to the new version can handle both. Agents still running the old version continue to receive the old schema. Once all agents are upgraded, the old version is deprecated.

• Expand-contract: first expand the schema by adding new optional fields alongside old ones. Agents start writing to the new fields. Once all agents are updated, contract the schema by removing the old fields.
• Parallel fields: send both old and new field formats in the same message during the transition. Old receivers use the old field. New receivers use the new field. Remove the old field once migration is complete.
• Schema transformation layer: add a translation layer in the message router that converts old-format messages to new-format before delivery. This lets you migrate the schema without requiring simultaneous updates to all agents.

Conclusion

Message schema design is one of the most consequential decisions in a multi-agent system and one of the most underinvested. Teams that invest in clear contracts, proper versioning, backward compatibility rules, and schema registries ship systems that evolve cleanly over time. Teams that treat schemas as implementation details spend disproportionate time on migration debugging that could have been avoided.