KEY INSIGHT
Protocol design determines whether agents interoperate or become isolated silos; clean abstractions survive model changes while brittle interfaces crumble.
Agent communication requires explicit protocols governing message formats, turn-taking, error handling, and state management. Without standardized protocols, each agent pair requires custom integration code. Adding a new agent means updating every existing peer. Protocols prevent this N×M integration problem.
Communication protocols define three layers: syntactic (message structure and serialization), semantic (meaning of message contents and valid transitions), and pragmatic (conversational patterns and context management). Syntactic choices (JSON vs protobuf vs custom formats) matter less than semantic clarity and pragmatic reliability.
Messages carry intents: requests, responses, notifications, errors. Each intent type carries different expectations about acknowledgment, retry behavior, and timeout handling. A request expects a response; a notification does not. Protocols must distinguish these patterns to enable appropriate handling.
State management during conversations requires careful design. Does each agent maintain conversation state? Is there a shared conversation tracker? Does each message carry full context? Stateful protocols support richer interactions but complicate implementation. Stateless protocols scale better but require clients to track context.
Error handling deserves explicit protocol attention. Network failures, agent crashes, malformed messages, and timeout scenarios must all produce deterministic behavior. Protocols should specify retry budgets, backoff strategies, and dead-letter handling.
Protocol versioning enables evolution. Agents supporting multiple protocol versions interoperate with both old and new peers. The protocol should define negotiation mechanisms so agents discover mutual capabilities.
```python
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Optional
from enum import Enum
import json
class MessageType(Enum):
REQUEST = "request"
RESPONSE = "response"
NOTIFICATION = "notification"
ERROR = "error"
@dataclass
class Message:
id: str
type: MessageType
sender: str
receiver: Optional[str] # None for broadcast
action: str
payload: dict[str, Any]
correlation_id: Optional[str] = None # Links request/response
ttl_seconds: int = 30
metadata: dict[str, Any] = field(default_factory=dict)
class AgentProtocol(ABC):
@abstractmethod
async def send(self, message: Message) -> None:
pass
@abstractmethod
async def receive(self) -> Optional[Message]:
pass
@abstractmethod
async def acknowledge(self, message_id: str) -> None:
pass
class JSONAgentProtocol(AgentProtocol):
def __init__(self, transport):
self.transport = transport
self.pending = {}
async def send(self, message: Message) -> None:
serialized = json.dumps({
"id": message.id,
"type": message.type.value,
"sender": message.sender,
"receiver": message.receiver,
"action": message.action,
"payload": message.payload,
"correlation_id": message.correlation_id,
"ttl": message.ttl_seconds,
"metadata": message.metadata
})
if message.type == MessageType.REQUEST:
self.pending[message.id] = message
await self.transport.send(serialized)
async def receive(self) -> Optional[Message]:
raw = await self.transport.receive()
if not raw:
return None
data = json.loads(raw)
return Message(
id=data["id"],
type=MessageType(data["type"]),
sender=data["sender"],
receiver=data["receiver"],
action=data["action"],
payload=data["payload"],
correlation_id=data.get("correlation_id"),
ttl_seconds=data.get("ttl", 30),
metadata=data.get("metadata", {})
)
async def acknowledge(self, message_id: str) -> None:
ack = Message(
id=f"ack_{message_id}",
type=MessageType.NOTIFICATION,
sender="local",
receiver=None,
action="ack",
payload={"original_id": message_id}
)
await self.send(ack)
```