Message Protocol

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
# Agent 2

# RIGHT
# Enforce standard format
validate_schema(message)

-

Last Updated: August 9, 2026