Skip to main content

Agent Message Protocols: Designing Communication Standards

An agent message protocol is the agreed-upon format and rules for how agents communicate. Without a protocol, agent A might send free-form text while agent B expects JSON, or agent C uses different field names. Protocols eliminate ambiguity, enable parsing, allow versioning, and make it easy to add new agents without breaking existing ones. In 2026, most agentic systems use JSON or structured text with clear schemas, but the protocol design principles apply to any format.

A good protocol is simple enough that agents can follow it consistently, expressive enough to convey necessary information, and forward-compatible so new fields do not break old agents. Think of it as the API contract between agents.

Core Elements of a Message Protocol​

Message structure​

A protocol typically specifies:

  1. Envelope: Metadata about the message (sender, receiver, timestamp, correlation ID).
  2. Body: The substantive content (request, response, or notification).
  3. Metadata: Additional context (message version, encoding, urgency).
{
  "envelope": {
    "message_id": "msg_001",
    "correlation_id": "cor_abc",
    "sender": "research_agent",
    "receiver": "validation_agent",
    "timestamp": "2026-06-02T10:15:30Z",
    "version": "1.0"
  },
  "body": {
    "type": "research_result",
    "content": "...",
    "metadata": {...}
  },
  "error": null
}

Semantic meaning​

Each message type (request, response, notification, error) must have a clear meaning:

  • Request: Agent A asks Agent B to perform work.
  • Response: Agent B returns a result.
  • Notification: An agent broadcasts information without expecting a response (e.g., "new best solution found").
  • Error: An agent reports failure and why.

Schema validation​

Define JSON schemas or similar constraints so agents can validate incoming messages before processing:

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "type": "object",
  "properties": {
    "envelope": {
      "type": "object",
      "properties": {
        "message_id": {"type": "string"},
        "sender": {"type": "string"},
        "receiver": {"type": "string"},
        "timestamp": {"type": "string", "format": "date-time"}
      },
      "required": ["message_id", "sender", "receiver", "timestamp"]
    },
    "body": {
      "type": "object",
      "properties": {
        "type": {"type": "string"},
        "content": {}
      },
      "required": ["type"]
    }
  },
  "required": ["envelope", "body"]
}

Designing a Message Protocol​

Step 1: Identify message types​

List all communication patterns your agents need:

  • Research agent → Validation agent: "Validate this finding"
  • Validation agent → Research agent: "Validation result"
  • Orchestrator → Worker: "Execute task X"
  • Worker → Orchestrator: "Task result"
  • Any agent → Logging service: "Log this event"

Step 2: Define content for each type​

For each message type, specify required and optional fields:

Message Type: ValidationRequest
From: research_agent
To: validation_agent
Fields:
  - claim (string, required): The claim to validate
  - sources (array of strings, optional): Sources supporting the claim
  - urgency (enum: low|medium|high, optional): Priority
Timeout: 60 seconds
Expected Response Type: ValidationResult

Step 3: Add versioning​

As your system evolves, agents may need to handle multiple protocol versions. Include a version in the envelope:

{
  "envelope": {
    "version": "2.0"
  }
}

Agents should be able to handle version 1.0 and 2.0 messages. Use the version to branch logic if needed.

Step 4: Define error handling​

Specify how agents report and handle errors:

{
  "envelope": {...},
  "body": {...},
  "error": {
    "code": "VALIDATION_FAILED",
    "message": "Claim contradicts existing evidence",
    "details": {"contradiction": "..."}
  }
}

Building a Protocol-Compliant Agent System​

Here is a complete example with a formal message protocol:

import anthropic
import json
import uuid
from datetime import datetime
from typing import Optional
from enum import Enum

client = anthropic.Anthropic()

class MessageType(Enum):
    RESEARCH_REQUEST = "research_request"
    RESEARCH_RESULT = "research_result"
    VALIDATION_REQUEST = "validation_request"
    VALIDATION_RESULT = "validation_result"
    ERROR = "error"

class MessageProtocol:
    """Handles message creation and validation according to the protocol."""
    
    VERSION = "1.0"
    
    @staticmethod
    def create_message(
        sender: str,
        receiver: str,
        message_type: MessageType,
        content: dict,
        correlation_id: Optional[str] = None
    ) -> dict:
        """Create a compliant message."""
        return {
            "envelope": {
                "message_id": str(uuid.uuid4()),
                "correlation_id": correlation_id or str(uuid.uuid4()),
                "sender": sender,
                "receiver": receiver,
                "timestamp": datetime.utcnow().isoformat() + "Z",
                "version": MessageProtocol.VERSION
            },
            "body": {
                "type": message_type.value,
                "content": content
            },
            "error": None
        }
    
    @staticmethod
    def create_error_message(
        sender: str,
        receiver: str,
        code: str,
        message: str,
        correlation_id: str
    ) -> dict:
        """Create an error message."""
        return {
            "envelope": {
                "message_id": str(uuid.uuid4()),
                "correlation_id": correlation_id,
                "sender": sender,
                "receiver": receiver,
                "timestamp": datetime.utcnow().isoformat() + "Z",
                "version": MessageProtocol.VERSION
            },
            "body": {"type": "error"},
            "error": {
                "code": code,
                "message": message
            }
        }
    
    @staticmethod
    def validate_message(msg: dict) -> tuple[bool, str]:
        """Validate message structure."""
        try:
            assert "envelope" in msg, "Missing envelope"
            assert "body" in msg, "Missing body"
            assert "message_id" in msg["envelope"], "Missing message_id"
            assert "sender" in msg["envelope"], "Missing sender"
            assert "receiver" in msg["envelope"], "Missing receiver"
            assert "type" in msg["body"], "Missing body type"
            return True, "Valid"
        except AssertionError as e:
            return False, str(e)

class ProtocolCompliantAgentSystem:
    def __init__(self):
        self.model = "claude-3-5-sonnet-20241022"
        self.inbox = {}  # For simulating async message delivery
    
    def research_agent(self, query: str) -> dict:
        """Research agent that produces a research_result message."""
        response = client.messages.create(
            model=self.model,
            max_tokens=600,
            system="""You are a research agent. Given a query, produce a research finding.
            Output JSON: {"claim": "...", "sources": [...], "confidence": 0.0-1.0}""",
            messages=[{"role": "user", "content": query}]
        )
        
        try:
            content = json.loads(response.content[0].text)
        except:
            content = {"claim": response.content[0].text, "sources": [], "confidence": 0.5}
        
        msg = MessageProtocol.create_message(
            sender="research_agent",
            receiver="validation_agent",
            message_type=MessageType.RESEARCH_RESULT,
            content=content
        )
        return msg
    
    def validation_agent(self, research_message: dict) -> dict:
        """Validation agent that consumes a research_result and produces validation_result."""
        is_valid, err = MessageProtocol.validate_message(research_message)
        if not is_valid:
            return MessageProtocol.create_error_message(
                sender="validation_agent",
                receiver="research_agent",
                code="INVALID_MESSAGE",
                message=err,
                correlation_id=research_message["envelope"]["correlation_id"]
            )
        
        research_content = research_message["body"]["content"]
        
        response = client.messages.create(
            model=self.model,
            max_tokens=400,
            system="""You are a validation agent. Given a claim and sources, assess validity.
            Output JSON: {"valid": true/false, "reasoning": "...", "issues": [...]}""",
            messages=[{
                "role": "user",
                "content": f"Claim: {research_content.get('claim')}\nSources: {research_content.get('sources')}"
            }]
        )
        
        try:
            content = json.loads(response.content[0].text)
        except:
            content = {"valid": True, "reasoning": "Could not validate", "issues": []}
        
        msg = MessageProtocol.create_message(
            sender="validation_agent",
            receiver="orchestrator",
            message_type=MessageType.VALIDATION_RESULT,
            content=content,
            correlation_id=research_message["envelope"]["correlation_id"]
        )
        return msg
    
    def orchestrate(self, user_query: str) -> dict:
        """End-to-end orchestration using the message protocol."""
        print(f"User query: {user_query}\n")
        
        # Research phase
        print("=== Phase 1: Research ===")
        research_msg = self.research_agent(user_query)
        print(f"Research message (ID: {research_msg['envelope']['message_id']}):")
        print(json.dumps(research_msg, indent=2)[:300] + "...\n")
        
        # Validation phase
        print("=== Phase 2: Validation ===")
        validation_msg = self.validation_agent(research_msg)
        print(f"Validation message (ID: {validation_msg['envelope']['message_id']}):")
        print(json.dumps(validation_msg, indent=2)[:300] + "...\n")
        
        return validation_msg

# Example
if __name__ == "__main__":
    system = ProtocolCompliantAgentSystem()
    result = system.orchestrate("What is the capital of France?")
    print("Final result:")
    print(json.dumps(result, indent=2))

Protocol Design Best Practices​

Be explicit about optional fields​

Mark fields as required or optional in your schema. Optional fields can be added without breaking old agents.

Use standardized data types​

Use ISO 8601 for dates, standard number formats, and well-defined enums. This reduces parsing errors.

Include correlation IDs​

Correlation IDs link related messages (request → response, error feedback). Essential for debugging multi-hop workflows.

Document timeouts​

Each message type should have an expected response time. Agents should not wait indefinitely.

Backward compatibility​

If you introduce a new field in version 2.0, old agents (version 1.0) should be able to ignore it and still function.

Protocol Comparison Table​

AspectJSONMessagePackProtocol Buffers
ReadabilityExcellentPoorMedium
EfficiencyMediumExcellentExcellent
Schema validationVia separate schema filesManualBuilt-in
LLM-friendlinessExcellent (text)Poor (binary)Poor (binary)
Best forLLM-based agentsHigh-throughput systemsCompiled services

For LLM-based agents, JSON is almost always the right choice because LLMs can easily parse and generate JSON.

Key Takeaways​

  • A message protocol defines the format, content, and semantics of inter-agent communication.
  • Include envelope (metadata), body (content), and error fields in each message.
  • Use versioning, correlation IDs, and schema validation to make protocols robust.
  • JSON is ideal for LLM-based agents; use structured schemas to validate messages before processing.
  • Design protocols for forward compatibility so new agents do not break existing ones.

Frequently Asked Questions​

What if an agent receives a message with an unrecognized type?​

Log the error, validate the message structure, and return an error message indicating the unrecognized type. Do not crash or silently ignore. This allows graceful degradation if a new agent type is introduced before all agents are updated.

Should I use the same protocol for all agents or different protocols per pair?​

Use a single standardized protocol across all agents. Different per-pair protocols create complexity and make it hard to add new agents. Protocol flexibility comes from optional fields and message types, not from different per-pair schemas.

How do I handle large messages (e.g., research results with thousands of sources)?​

Implement pagination or chunking: split large content across multiple messages, each with a sequence number. The receiving agent reassembles them. Alternatively, store large payloads in an external database and send only a reference (a URI or ID) in the message.

Can agents send messages asynchronously or must they wait for responses?​

Both are possible. Synchronous (agent A sends, waits for response) is simpler but blocks. Asynchronous (agent A sends, continues, receives response later via callback or polling) is faster but requires correlation IDs and state tracking. Choose based on your latency requirements.

How do I debug message protocol violations in production?​

Log all messages (envelope + body) to a centralized log. Use correlation IDs to trace a request through the system. Set up alerting for errors or validation failures. Periodically sample messages to detect protocol drift (agents violating the agreed format).

Further Reading​