A Coding Guide to Building a Scalable Multi-Agent Communication Systems Using Agent Communication Protocol (ACP)
In this tutorial, we implement the Agent Communication Protocolthrough building a flexible, ACP-compliant messaging system in Python, leveraging Google’s Gemini API for natural language processing. Beginning with the installation and configuration of the google-generativeai library, the tutorial introduces core abstractions, message types, performatives, and the ACPMessage data class, which standardizes inter-agent communication. By defining ACPAgent and ACPMessageBroker classes, the guide demonstrates how to create, send, route, and process structured messages among multiple autonomous agents. Through clear code examples, users learn to implement querying, requesting actions, and broadcasting information, while maintaining conversation threads, acknowledgments, and error handling.
import google.generativeai as genai
import json
import time
import uuid
from enum import Enum
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, asdict
GEMINI_API_KEY = "Use Your Gemini API Key"
genai.configureWe import essential Python modules, ranging from JSON handling and timing to unique identifier generation and type annotations, to support a structured ACP implementation. It then retrieves the user’s Gemini API key placeholder and configures the google-generativeai client for subsequent calls to the Gemini language model.
class ACPMessageType:
"""Standard ACP message types"""
REQUEST = "request"
RESPONSE = "response"
INFORM = "inform"
QUERY = "query"
SUBSCRIBE = "subscribe"
UNSUBSCRIBE = "unsubscribe"
ERROR = "error"
ACK = "acknowledge"
The ACPMessageType enumeration defines the core message categories used in the Agent Communication Protocol, including requests, responses, informational broadcasts, queries, and control actions like subscription management, error signaling, and acknowledgments. By centralizing these message types, the protocol ensures consistent handling and routing of inter-agent communications throughout the system.
class ACPPerformative:
"""ACP speech acts"""
TELL = "tell"
ASK = "ask"
REPLY = "reply"
REQUEST_ACTION = "request-action"
AGREE = "agree"
REFUSE = "refuse"
PROPOSE = "propose"
ACCEPT = "accept"
REJECT = "reject"
The ACPPerformative enumeration captures the variety of speech acts agents can use when interacting under the ACP framework, mapping high-level intentions, such as making requests, posing questions, giving commands, or negotiating agreements, onto standardized labels. This clear taxonomy enables agents to interpret and respond to messages in contextually appropriate ways, ensuring robust and semantically rich communication.
@dataclass
class ACPMessage:
"""Agent Communication Protocol Message Structure"""
message_id: str
sender: str
receiver: str
performative: str
content: Dictprotocol: str = "ACP-1.0"
conversation_id: str = None
reply_to: str = None
language: str = "english"
encoding: str = "json"
timestamp: float = None
def __post_init__:
if self.timestamp is None:
self.timestamp = time.timeif self.conversation_id is None:
self.conversation_id = str)
def to_acp_format-> str:
"""Convert to standard ACP message format"""
acp_msg = {
"message-id": self.message_id,
"sender": self.sender,
"receiver": self.receiver,
"performative": self.performative,
"content": self.content,
"protocol": self.protocol,
"conversation-id": self.conversation_id,
"reply-to": self.reply_to,
"language": self.language,
"encoding": self.encoding,
"timestamp": self.timestamp
}
return json.dumps@classmethod
def from_acp_format-> 'ACPMessage':
"""Parse ACP message from string format"""
data = json.loadsreturn cls,
conversation_id=data.get,
reply_to=data.get,
language=data.get,
encoding=data.get,
timestamp=data.get)
)
The ACPMessage data class encapsulates all the fields required for a structured ACP exchange, including identifiers, participants, performative, payload, and metadata such as protocol version, language, and timestamps. Its __post_init__ method auto-populates missing timestamp and conversation_id values, ensuring every message is uniquely tracked. Utility methods to_acp_format and from_acp_format handle serialization to and from the standardized JSON representation for seamless transmission and parsing.
class ACPAgent:
"""Agent implementing Agent Communication Protocol"""
def __init__:
self.agent_id = agent_id
self.name = name
self.capabilities = capabilities
self.model = genai.GenerativeModelself.message_queue: List=self.subscriptions: Dict] = {}
self.conversations: Dict] = {}
def create_message-> ACPMessage:
"""Create a new ACP-compliant message"""
return ACPMessage),
sender=self.agent_id,
receiver=receiver,
performative=performative,
content=content,
conversation_id=conversation_id,
reply_to=reply_to
)
def send_inform-> ACPMessage:
"""Send an INFORM message"""
content = {"fact": fact, "data": data}
return self.create_messagedef send_query-> ACPMessage:
"""Send a QUERY message"""
content = {"question": question, "query-type": query_type}
return self.create_messagedef send_request-> ACPMessage:
"""Send a REQUEST message"""
content = {"action": action, "parameters": parameters or {}}
return self.create_messagedef send_reply-> ACPMessage:
"""Send a REPLY message in response to another message"""
content = {"response": response_data, "original-question": original_msg.content}
return self.create_messagedef process_message-> Optional:
"""Process incoming ACP message and generate appropriate response"""
self.message_queue.appendconv_id = message.conversation_id
if conv_id not in self.conversations:
self.conversations=self.conversations.appendif message.performative == ACPPerformative.ASK.value:
return self._handle_queryelif message.performative == ACPPerformative.REQUEST_ACTION.value:
return self._handle_requestelif message.performative == ACPPerformative.TELL.value:
return self._handle_informreturn None
def _handle_query-> ACPMessage:
"""Handle incoming query messages"""
question = message.content.getprompt = f"As agent {self.name} with capabilities {self.capabilities}, answer: {question}"
try:
response = self.model.generate_contentanswer = response.text.stripexcept:
answer = "Unable to process query at this time"
return self.send_replydef _handle_request-> ACPMessage:
"""Handle incoming action requests"""
action = message.content.getparameters = message.content.getif anyfor capability in self.capabilities):
result = f"Executing {action} with parameters {parameters}"
status = "agreed"
else:
result = f"Cannot perform {action} - not in my capabilities"
status = "refused"
return self.send_replydef _handle_inform-> Optional:
"""Handle incoming information messages"""
fact = message.content.getprintack_content = {"status": "received", "fact": fact}
return self.create_messageThe ACPAgent class encapsulates an autonomous entity capable of sending, receiving, and processing ACP-compliant messages using Gemini’s language model. It manages its own message queue, conversation history, and subscriptions, and provides helper methodsto construct correctly formatted ACPMessage instances. Incoming messages are routed through process_message, which delegates to specialized handlers for queries, action requests, and informational messages.
class ACPMessageBroker:
"""Message broker implementing ACP routing and delivery"""
def __init__:
self.agents: Dict= {}
self.message_log: List=self.routing_table: Dict= {}
def register_agent:
"""Register an agent with the message broker"""
self.agents= agent
self.routing_table= "local"
print")
def route_message-> bool:
"""Route ACP message to appropriate recipient"""
if message.receiver not in self.agents:
printreturn False
printprintprintprint}")
receiver_agent = self.agentsresponse = receiver_agent.process_messageself.message_log.appendif response:
printprintprint}")
if response.receiver in self.agents:
self.agents.process_messageself.message_log.appendreturn True
def broadcast_message:
"""Broadcast message to multiple recipients"""
for recipient in recipients:
msg_copy = ACPMessage),
sender=message.sender,
receiver=recipient,
performative=message.performative,
content=message.content.copy,
conversation_id=message.conversation_id
)
self.route_messageThe ACPMessageBroker serves as the central router for ACP messages, maintaining a registry of agents and a message log. It provides methods to register agents, deliver individual messages via route_message, which handles lookup, logging, and response chaining, and to send the same message to multiple recipients with broadcast_message.
def demonstrate_acp:
"""Comprehensive demonstration of Agent Communication Protocol"""
printDEMONSTRATION")
printbroker = ACPMessageBrokerresearcher = ACPAgentassistant = ACPAgentcalculator = ACPAgentbroker.register_agentbroker.register_agentbroker.register_agentprintfor agent_id, agent in broker.agents.items:
print: {', '.join}")
print")
query_msg = assistant.send_querybroker.route_messageprint")
calc_request = researcher.send_request+ 10"})
broker.route_messageprint")
info_msg = researcher.send_informbroker.route_messageprintprint}")
print)}")
print)}")
printsample_msg = assistant.send_queryprint)
The demonstrate_acp function orchestrates a hands-on walkthrough of the entire ACP framework: it initializes a broker and three distinct agents, registers them, and illustrates three key interaction scenarios, querying for information, requesting a computation, and sharing an update. After routing each message and handling responses, it prints summary statistics on the message flow. It showcases a formatted ACP message, providing users with a clear, end-to-end example of how agents communicate under the protocol.
def setup_guide:
print
ACP PROTOCOL FEATURES:
• Standardized message format with required fields
• Speech act performatives• Conversation tracking and message threading
• Error handling and acknowledgments
• Message routing and delivery confirmation
EXTEND THE PROTOCOL:
```python
# Create custom agent
my_agent = ACPAgentbroker.register_agent# Send custom message
msg = my_agent.send_querybroker.route_message```
""")
if __name__ == "__main__":
setup_guidedemonstrate_acpFinally, the setup_guide function provides a quick-start reference for running the ACP demo in Google Colab, outlining how to obtain and configure your Gemini API key and invoke the demonstrate_acp routine. It also summarizes key protocol features, such as standardized message formats, performatives, and message routing. It provides a concise code snippet illustrating how to register custom agents and send tailored messages.
In conclusion, this tutorial implements ACP-based multi-agent systems capable of research, computation, and collaboration tasks. The provided sample scenarios illustrate common use cases, information queries, computational requests, and fact sharing, while the broker ensures reliable message delivery and logging. Readers are encouraged to extend the framework by adding new agent capabilities, integrating domain-specific actions, or incorporating more sophisticated subscription and notification mechanisms.
Download the Notebook on GitHub. All credit for this research goes to the researchers of this project. Also, feel free to follow us on Twitter and don’t forget to join our 95k+ ML SubReddit and Subscribe to our Newsletter.
Asif RazzaqWebsite | + postsBioAsif Razzaq is the CEO of Marktechpost Media Inc.. As a visionary entrepreneur and engineer, Asif is committed to harnessing the potential of Artificial Intelligence for social good. His most recent endeavor is the launch of an Artificial Intelligence Media Platform, Marktechpost, which stands out for its in-depth coverage of machine learning and deep learning news that is both technically sound and easily understandable by a wide audience. The platform boasts of over 2 million monthly views, illustrating its popularity among audiences.Asif Razzaq
https://www.marktechpost.com/author/6flvq/Yandex Releases Yambda: The World’s Largest Event Dataset to Accelerate Recommender SystemsAsif Razzaq
https://www.marktechpost.com/author/6flvq/Stanford Researchers Introduced Biomni: A Biomedical AI Agent for Automation Across Diverse Tasks and Data TypesAsif Razzaq
https://www.marktechpost.com/author/6flvq/DeepSeek Releases R1-0528: An Open-Source Reasoning AI Model Delivering Enhanced Math and Code Performance with Single-GPU EfficiencyAsif Razzaq
https://www.marktechpost.com/author/6flvq/A Coding Guide for Building a Self-Improving AI Agent Using Google’s Gemini API with Intelligent Adaptation Features
#coding #guide #building #scalable #multiagentA Coding Guide to Building a Scalable Multi-Agent Communication Systems Using Agent Communication Protocol (ACP)
In this tutorial, we implement the Agent Communication Protocolthrough building a flexible, ACP-compliant messaging system in Python, leveraging Google’s Gemini API for natural language processing. Beginning with the installation and configuration of the google-generativeai library, the tutorial introduces core abstractions, message types, performatives, and the ACPMessage data class, which standardizes inter-agent communication. By defining ACPAgent and ACPMessageBroker classes, the guide demonstrates how to create, send, route, and process structured messages among multiple autonomous agents. Through clear code examples, users learn to implement querying, requesting actions, and broadcasting information, while maintaining conversation threads, acknowledgments, and error handling.
import google.generativeai as genai
import json
import time
import uuid
from enum import Enum
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, asdict
GEMINI_API_KEY = "Use Your Gemini API Key"
genai.configureWe import essential Python modules, ranging from JSON handling and timing to unique identifier generation and type annotations, to support a structured ACP implementation. It then retrieves the user’s Gemini API key placeholder and configures the google-generativeai client for subsequent calls to the Gemini language model.
class ACPMessageType:
"""Standard ACP message types"""
REQUEST = "request"
RESPONSE = "response"
INFORM = "inform"
QUERY = "query"
SUBSCRIBE = "subscribe"
UNSUBSCRIBE = "unsubscribe"
ERROR = "error"
ACK = "acknowledge"
The ACPMessageType enumeration defines the core message categories used in the Agent Communication Protocol, including requests, responses, informational broadcasts, queries, and control actions like subscription management, error signaling, and acknowledgments. By centralizing these message types, the protocol ensures consistent handling and routing of inter-agent communications throughout the system.
class ACPPerformative:
"""ACP speech acts"""
TELL = "tell"
ASK = "ask"
REPLY = "reply"
REQUEST_ACTION = "request-action"
AGREE = "agree"
REFUSE = "refuse"
PROPOSE = "propose"
ACCEPT = "accept"
REJECT = "reject"
The ACPPerformative enumeration captures the variety of speech acts agents can use when interacting under the ACP framework, mapping high-level intentions, such as making requests, posing questions, giving commands, or negotiating agreements, onto standardized labels. This clear taxonomy enables agents to interpret and respond to messages in contextually appropriate ways, ensuring robust and semantically rich communication.
@dataclass
class ACPMessage:
"""Agent Communication Protocol Message Structure"""
message_id: str
sender: str
receiver: str
performative: str
content: Dictprotocol: str = "ACP-1.0"
conversation_id: str = None
reply_to: str = None
language: str = "english"
encoding: str = "json"
timestamp: float = None
def __post_init__:
if self.timestamp is None:
self.timestamp = time.timeif self.conversation_id is None:
self.conversation_id = str)
def to_acp_format-> str:
"""Convert to standard ACP message format"""
acp_msg = {
"message-id": self.message_id,
"sender": self.sender,
"receiver": self.receiver,
"performative": self.performative,
"content": self.content,
"protocol": self.protocol,
"conversation-id": self.conversation_id,
"reply-to": self.reply_to,
"language": self.language,
"encoding": self.encoding,
"timestamp": self.timestamp
}
return json.dumps@classmethod
def from_acp_format-> 'ACPMessage':
"""Parse ACP message from string format"""
data = json.loadsreturn cls,
conversation_id=data.get,
reply_to=data.get,
language=data.get,
encoding=data.get,
timestamp=data.get)
)
The ACPMessage data class encapsulates all the fields required for a structured ACP exchange, including identifiers, participants, performative, payload, and metadata such as protocol version, language, and timestamps. Its __post_init__ method auto-populates missing timestamp and conversation_id values, ensuring every message is uniquely tracked. Utility methods to_acp_format and from_acp_format handle serialization to and from the standardized JSON representation for seamless transmission and parsing.
class ACPAgent:
"""Agent implementing Agent Communication Protocol"""
def __init__:
self.agent_id = agent_id
self.name = name
self.capabilities = capabilities
self.model = genai.GenerativeModelself.message_queue: List=self.subscriptions: Dict] = {}
self.conversations: Dict] = {}
def create_message-> ACPMessage:
"""Create a new ACP-compliant message"""
return ACPMessage),
sender=self.agent_id,
receiver=receiver,
performative=performative,
content=content,
conversation_id=conversation_id,
reply_to=reply_to
)
def send_inform-> ACPMessage:
"""Send an INFORM message"""
content = {"fact": fact, "data": data}
return self.create_messagedef send_query-> ACPMessage:
"""Send a QUERY message"""
content = {"question": question, "query-type": query_type}
return self.create_messagedef send_request-> ACPMessage:
"""Send a REQUEST message"""
content = {"action": action, "parameters": parameters or {}}
return self.create_messagedef send_reply-> ACPMessage:
"""Send a REPLY message in response to another message"""
content = {"response": response_data, "original-question": original_msg.content}
return self.create_messagedef process_message-> Optional:
"""Process incoming ACP message and generate appropriate response"""
self.message_queue.appendconv_id = message.conversation_id
if conv_id not in self.conversations:
self.conversations=self.conversations.appendif message.performative == ACPPerformative.ASK.value:
return self._handle_queryelif message.performative == ACPPerformative.REQUEST_ACTION.value:
return self._handle_requestelif message.performative == ACPPerformative.TELL.value:
return self._handle_informreturn None
def _handle_query-> ACPMessage:
"""Handle incoming query messages"""
question = message.content.getprompt = f"As agent {self.name} with capabilities {self.capabilities}, answer: {question}"
try:
response = self.model.generate_contentanswer = response.text.stripexcept:
answer = "Unable to process query at this time"
return self.send_replydef _handle_request-> ACPMessage:
"""Handle incoming action requests"""
action = message.content.getparameters = message.content.getif anyfor capability in self.capabilities):
result = f"Executing {action} with parameters {parameters}"
status = "agreed"
else:
result = f"Cannot perform {action} - not in my capabilities"
status = "refused"
return self.send_replydef _handle_inform-> Optional:
"""Handle incoming information messages"""
fact = message.content.getprintack_content = {"status": "received", "fact": fact}
return self.create_messageThe ACPAgent class encapsulates an autonomous entity capable of sending, receiving, and processing ACP-compliant messages using Gemini’s language model. It manages its own message queue, conversation history, and subscriptions, and provides helper methodsto construct correctly formatted ACPMessage instances. Incoming messages are routed through process_message, which delegates to specialized handlers for queries, action requests, and informational messages.
class ACPMessageBroker:
"""Message broker implementing ACP routing and delivery"""
def __init__:
self.agents: Dict= {}
self.message_log: List=self.routing_table: Dict= {}
def register_agent:
"""Register an agent with the message broker"""
self.agents= agent
self.routing_table= "local"
print")
def route_message-> bool:
"""Route ACP message to appropriate recipient"""
if message.receiver not in self.agents:
printreturn False
printprintprintprint}")
receiver_agent = self.agentsresponse = receiver_agent.process_messageself.message_log.appendif response:
printprintprint}")
if response.receiver in self.agents:
self.agents.process_messageself.message_log.appendreturn True
def broadcast_message:
"""Broadcast message to multiple recipients"""
for recipient in recipients:
msg_copy = ACPMessage),
sender=message.sender,
receiver=recipient,
performative=message.performative,
content=message.content.copy,
conversation_id=message.conversation_id
)
self.route_messageThe ACPMessageBroker serves as the central router for ACP messages, maintaining a registry of agents and a message log. It provides methods to register agents, deliver individual messages via route_message, which handles lookup, logging, and response chaining, and to send the same message to multiple recipients with broadcast_message.
def demonstrate_acp:
"""Comprehensive demonstration of Agent Communication Protocol"""
printDEMONSTRATION")
printbroker = ACPMessageBrokerresearcher = ACPAgentassistant = ACPAgentcalculator = ACPAgentbroker.register_agentbroker.register_agentbroker.register_agentprintfor agent_id, agent in broker.agents.items:
print: {', '.join}")
print")
query_msg = assistant.send_querybroker.route_messageprint")
calc_request = researcher.send_request+ 10"})
broker.route_messageprint")
info_msg = researcher.send_informbroker.route_messageprintprint}")
print)}")
print)}")
printsample_msg = assistant.send_queryprint)
The demonstrate_acp function orchestrates a hands-on walkthrough of the entire ACP framework: it initializes a broker and three distinct agents, registers them, and illustrates three key interaction scenarios, querying for information, requesting a computation, and sharing an update. After routing each message and handling responses, it prints summary statistics on the message flow. It showcases a formatted ACP message, providing users with a clear, end-to-end example of how agents communicate under the protocol.
def setup_guide:
print🔧 ACP PROTOCOL FEATURES:
• Standardized message format with required fields
• Speech act performatives• Conversation tracking and message threading
• Error handling and acknowledgments
• Message routing and delivery confirmation
📝 EXTEND THE PROTOCOL:
```python
# Create custom agent
my_agent = ACPAgentbroker.register_agent# Send custom message
msg = my_agent.send_querybroker.route_message```
""")
if __name__ == "__main__":
setup_guidedemonstrate_acpFinally, the setup_guide function provides a quick-start reference for running the ACP demo in Google Colab, outlining how to obtain and configure your Gemini API key and invoke the demonstrate_acp routine. It also summarizes key protocol features, such as standardized message formats, performatives, and message routing. It provides a concise code snippet illustrating how to register custom agents and send tailored messages.
In conclusion, this tutorial implements ACP-based multi-agent systems capable of research, computation, and collaboration tasks. The provided sample scenarios illustrate common use cases, information queries, computational requests, and fact sharing, while the broker ensures reliable message delivery and logging. Readers are encouraged to extend the framework by adding new agent capabilities, integrating domain-specific actions, or incorporating more sophisticated subscription and notification mechanisms.
Download the Notebook on GitHub. All credit for this research goes to the researchers of this project. Also, feel free to follow us on Twitter and don’t forget to join our 95k+ ML SubReddit and Subscribe to our Newsletter.
Asif RazzaqWebsite | + postsBioAsif Razzaq is the CEO of Marktechpost Media Inc.. As a visionary entrepreneur and engineer, Asif is committed to harnessing the potential of Artificial Intelligence for social good. His most recent endeavor is the launch of an Artificial Intelligence Media Platform, Marktechpost, which stands out for its in-depth coverage of machine learning and deep learning news that is both technically sound and easily understandable by a wide audience. The platform boasts of over 2 million monthly views, illustrating its popularity among audiences.Asif Razzaqhttps://www.marktechpost.com/author/6flvq/Yandex Releases Yambda: The World’s Largest Event Dataset to Accelerate Recommender SystemsAsif Razzaqhttps://www.marktechpost.com/author/6flvq/Stanford Researchers Introduced Biomni: A Biomedical AI Agent for Automation Across Diverse Tasks and Data TypesAsif Razzaqhttps://www.marktechpost.com/author/6flvq/DeepSeek Releases R1-0528: An Open-Source Reasoning AI Model Delivering Enhanced Math and Code Performance with Single-GPU EfficiencyAsif Razzaqhttps://www.marktechpost.com/author/6flvq/A Coding Guide for Building a Self-Improving AI Agent Using Google’s Gemini API with Intelligent Adaptation Features
#coding #guide #building #scalable #multiagent
map
