Communication Between Agents › Communication Patterns
Michael BrenndoerferJuly 27, 20254 min read
AI Agent HandbookAll chapters Part 10 of 15Multi-Agent Systems - Teamwork Among AIs
Communication Between Agents › Communication PatternsOne section · Sections in this chapter4 / 11 ↓
Sections in this chapter11
Now that we have a message format, let's explore different ways agents can communicate.
Request-Response
The simplest pattern: Agent A sends a request, Agent B sends a response.
Agent A: "What's the weather in Paris?"
Agent B: "Sunny, 22°C"This is synchronous. Agent A waits for Agent B's response before continuing. It's straightforward but can be inefficient if Agent B takes a long time.
Fire-and-Forget
Agent A sends a message but doesn't wait for a response. It continues with other work.
Agent A: "Log this event: user clicked button"
Logger Agent: [receives message, logs it]
Agent A: [already moved on to other tasks]This is asynchronous and efficient, but Agent A has no way to know if the message was received or processed successfully.
Publish-Subscribe
Multiple agents can subscribe to specific types of messages. When Agent A publishes a message, all subscribed agents receive it.
Agent A: [publishes "new_data_available"]
Agent B: [subscribed to "new_data_available", receives notification]
Agent C: [subscribed to "new_data_available", receives notification]
Agent D: [not subscribed, doesn't receive anything]This pattern works well for broadcasting information to multiple interested parties without Agent A needing to know who they are.
Here's a simple implementation:
In[6]:
Code## Using Claude Sonnet 4.5 for pub-sub coordination
class MessageBroker:
"""
Simple publish-subscribe message broker for agent communication.
"""
def __init__(self):
self.subscriptions = {} # topic -> list of agents
self.message_history = []
def subscribe(self, topic, agent_name):
"""
Agent subscribes to a topic.
"""
if topic not in self.subscriptions:
self.subscriptions[topic] = []
if agent_name not in self.subscriptions[topic]:
self.subscriptions[topic].append(agent_name)
print(f"{agent_name} subscribed to '{topic}'")
def publish(self, topic, message):
"""
Publish a message to all subscribers of a topic.
"""
self.message_history.append({
"topic": topic,
"message": message,
"timestamp": datetime.utcnow().isoformat()
})
if topic in self.subscriptions:
subscribers = self.subscriptions[topic]
print(f"\nPublishing to '{topic}' ({len(subscribers)} subscribers)")
return subscribers
else:
print(f"\nPublishing to '{topic}' (no subscribers)")
return []
def get_history(self, topic=None):
"""
Get message history, optionally filtered by topic.
"""
if topic:
return [msg for msg in self.message_history if msg["topic"] == topic]
return self.message_history
## Example: Multiple agents coordinating through pub-sub
broker = MessageBroker()
## Agents subscribe to topics they care about
broker.subscribe("market_data", "trading_agent")
broker.subscribe("market_data", "analysis_agent")
broker.subscribe("market_data", "reporting_agent")
broker.subscribe("user_actions", "logging_agent")
broker.subscribe("user_actions", "analytics_agent")
## When market data arrives, all interested agents are notified
market_subscribers = broker.publish("market_data", {
"symbol": "AAPL",
"price": 178.50,
"change": 2.3
})
print(f"Notified: {', '.join(market_subscribers)}")
## When a user action occurs, different agents are notified
action_subscribers = broker.publish("user_actions", {
"action": "purchase",
"item": "premium_subscription"
})
print(f"Notified: {', '.join(action_subscribers)}")Out[6]:
Consoletrading_agent subscribed to 'market_data' analysis_agent subscribed to 'market_data' reporting_agent subscribed to 'market_data' logging_agent subscribed to 'user_actions' analytics_agent subscribed to 'user_actions' Publishing to 'market_data' (3 subscribers) Notified: trading_agent, analysis_agent, reporting_agent Publishing to 'user_actions' (2 subscribers) Notified: logging_agent, analytics_agent
Conversation Threads
For complex coordination, agents might need back-and-forth conversations:
Agent A: "I need to schedule a meeting with the team."
Agent B: "What time works for you?"
Agent A: "Tuesday afternoon."
Agent B: "Tuesday at 2pm is available. Shall I book it?"
Agent A: "Yes, please."
Agent B: "Done. Meeting scheduled."This requires maintaining conversation context. Each message needs to reference the thread it belongs to:
In[7]:
Codeclass ConversationThread:
"""
Manages a conversation thread between agents.
"""
def __init__(self, thread_id, participants):
self.thread_id = thread_id
self.participants = participants
self.messages = []
def add_message(self, sender, content):
"""
Add a message to the thread.
"""
if sender not in self.participants:
raise ValueError(f"{sender} is not a participant in this thread")
message = {
"thread_id": self.thread_id,
"sender": sender,
"content": content,
"timestamp": datetime.utcnow().isoformat(),
"sequence": len(self.messages)
}
self.messages.append(message)
return message
def get_history(self):
"""
Get the full conversation history.
"""
return self.messages
def get_context_for_agent(self, agent_name):
"""
Get conversation context formatted for a specific agent.
"""
context = f"Conversation thread {self.thread_id}\n"
context += f"Participants: {', '.join(self.participants)}\n\n"
for msg in self.messages:
context += f"{msg['sender']}: {msg['content']}\n"
return context
## Example: Two agents having a conversation
thread = ConversationThread(
thread_id="meeting_planning_001",
participants=["coordinator_agent", "calendar_agent"]
)
thread.add_message("coordinator_agent", "I need to schedule a team meeting for next week.")
thread.add_message("calendar_agent", "I can see Tuesday 2pm or Thursday 3pm are available. Which works better?")
thread.add_message("coordinator_agent", "Tuesday 2pm works. Please book it.")
thread.add_message("calendar_agent", "Meeting scheduled for Tuesday at 2pm. I've sent invites to all team members.")
print("=== Conversation Thread ===")
print(thread.get_context_for_agent("coordinator_agent"))Out[7]:
Console=== Conversation Thread === Conversation thread meeting_planning_001 Participants: coordinator_agent, calendar_agent coordinator_agent: I need to schedule a team meeting for next week. calendar_agent: I can see Tuesday 2pm or Thursday 3pm are available. Which works better? coordinator_agent: Tuesday 2pm works. Please book it. calendar_agent: Meeting scheduled for Tuesday at 2pm. I've sent invites to all team members.