AI

Implement Agent Communication Protocol

Medium
Agents

Agent Communication Protocols

Multi-agent systems require structured communication channels.

Task

Implement a MessageBus with:

  1. Per-agent priority queues for message routing.
  2. Priority-based message retrieval.
  3. Broadcast capability.
  4. Full message log for auditability.

Constraints

  • Priority 1 is highest; always deliver highest priority first.
  • broadcast excludes the sender.
  • Unregistered agents cannot send or receive.
  • Messages must be immutable after sending.

Examples

Example 1:
Input: bus = MessageBus() bus.register('agent_a') bus.register('agent_b') bus.send(AgentMessage('agent_a', 'agent_b', 'hello', 'task')) bus.receive('agent_b').content
Output: 'hello'
Explanation: Message from a to b is delivered via the bus.

Starter Code

from typing import Any, Optional
from dataclasses import dataclass, field
import uuid
from datetime import datetime

@dataclass
class AgentMessage:
    sender: str
    receiver: str
    content: Any
    msg_type: str  # 'task' | 'result' | 'error' | 'status' | 'broadcast'
    msg_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    reply_to: Optional[str] = None
    timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat())
    priority: int = 5  # 1 (highest) to 10 (lowest)

class MessageBus:
    def __init__(self):
        self.queues: dict = {}
        self.message_log: list = []

    def register(self, agent_name: str) -> None:
        pass

    def send(self, message: AgentMessage) -> bool:
        pass

    def receive(self, agent_name: str, block: bool = False) -> Optional[AgentMessage]:
        # Return highest priority message (lowest number)
        pass

    def broadcast(self, sender: str, content: Any) -> int:
        # Send to all agents except sender; return count
        pass
Lines: 1Characters: 0
Ready
The AI Interview - Master AI/ML Interviews