Message Protocol: Multi-Agent Communication Standards

Overview

When agents talk to each other, they need a common message format.

Message Protocol standardizes this communication.

---

Message Format

Standard Structure

class Message:
    """Standard agent message"""

    id: str              # Unique message ID
    sender_id: str       # Who sent this
    recipient_id: str    # Who should receive
    timestamp: float     # When sent
    type: str            # "task" | "result" | "error" | "status"
    content: dict        # Message content
    metadata: dict       # Optional metadata
    reply_to: str        # References previous message


class TaskMessage(Message):
    type = "task"
    content = {
        "description": "What to do",
        "priority": "high" | "normal" | "low",
        "deadline": "ISO-8601 timestamp",
        "required_format": "json" | "text"
    }


class ResultMessage(Message):
    type = "result"
    content = {
        "result": "The actual result",
        "status": "success" | "partial" | "failed",
        "execution_time_ms": 1234,
        "error": None  # If failed
    }


class ErrorMessage(Message):
    type = "error"
    content = {
        "error_code": "TIMEOUT" | "INVALID_TASK",
        "message": "What went wrong",
        "retry_after_ms": 5000  # If applicable
    }

---

Message Queue

Async Message Passing

from typing import AsyncIterator
import asyncio

class MessageQueue:
    """Queue for agent messages"""

    def __init__(self):
        self.queues = {}  # agent_id → queue

    async def send_message(self, message: Message):
        """Send message to agent"""

        recipient = message.recipient_id

        if recipient not in self.queues:
            self.queues[recipient] = asyncio.Queue()

        await self.queues[recipient].put(message)

    async def receive_messages(self, agent_id: str) -> AsyncIterator[Message]:
        """Receive messages for this agent"""

        if agent_id not in self.queues:
            self.queues[agent_id] = asyncio.Queue()

        queue = self.queues[agent_id]

        while True:
            message = await queue.get()
            yield message

---

Request/Reply Pattern

Correlated Messages

class RequestReplyProtocol:
    """Request with guaranteed reply"""

    async def send_and_wait(self, task_message: Message, timeout=30):
        """Send task, wait for reply"""

        task_id = task_message.id

        # Send task
        await self.message_queue.send_message(task_message)

        # Wait for reply
        try:
            async for message in self.receive_messages():
                if message.reply_to == task_id:
                    return message
        except asyncio.TimeoutError:
            return ErrorMessage(
                error_code="TIMEOUT",
                message=f"No response within {timeout}s"
            )

---

3 Warnings ⚠️

Warning 1: Message Loss

# ❌ WRONG
# No message persistence
await send_message(msg)
# Process crashes, message lost!

# ✅ RIGHT
# Persist before sending
persist_to_db(msg)
await send_message(msg)
mark_sent(msg)

Warning 2: Infinite Loops

# ❌ WRONG
# Agent A sends to B sends to A
# Infinite loop!

# ✅ RIGHT
# Track message chain
max_hops = 5
if message.hops >= max_hops:
    return error("max hops exceeded")

Warning 3: Protocol Drift

# ❌ WRONG
# Each agent uses different format
# Agent 1: {"task": ...}
# Agent 2: {"action": ...}

# ✅ RIGHT
# Enforce standard format
validate_schema(message)

---

Last Updated: August 9, 2026