How AI Agents Communicate: Protocols and Implementation

Photo by Pavan Trikutam on Unsplash

Hey there! 👋 Welcome back to our AI agents series. Today, we’re diving into something super cool — making agents talk to each other. If you haven’t checked out the previous articles, you might want to give them a quick read first.

Previous Articles: ( Day 1, Day 2)

What We’re Building

We’re creating a multi-agent system where agents can:

• Send and receive messages
• Coordinate on tasks
• Share information
• Handle communication failures
• Make collective decisions

Let’s see how it works!

The Basics: Message Passing

First, let’s build a simple messaging system:

from dataclasses import dataclass
from typing import List, Dict, Any
from enum import Enum
import uuid
import time
from queue import Queue

class MessageType(Enum):
    REQUEST = "request"
    RESPONSE = "response"
    BROADCAST = "broadcast"
    ERROR = "error"

@dataclass
class Message:
    sender_id: str
    receiver_id: str
    message_type: MessageType
    content: Any
    conversation_id: str = None
    timestamp: float = None
    
    def __post_init__(self):
        if not self.conversation_id:
            self.conversation_id = str(uuid.uuid4())
        if not self.timestamp:
            self.timestamp = time.time()

class MessageBus:
    def __init__(self):
        self.queues: Dict[str, Queue] = {}
        self.subscribers: Dict[str, List[str]] = {}
    
    def register_agent(self, agent_id: str):
        """Register a new agent with the message bus"""
        if agent_id not in self.queues:
            self.queues[agent_id] = Queue()
            self.subscribers[agent_id] = []
    
    def send_message(self, message: Message):
        """Send a message to a specific agent"""
        if message.receiver_id in self.queues:
            self.queues[message.receiver_id].put(message)
    
    def broadcast(self, sender_id: str, content: Any):
        """Send a message to all subscribers"""
        message = Message(
            sender_id=sender_id,
            receiver_id="broadcast",
            message_type=MessageType.BROADCAST,
            content=content…