This tutorial demonstrates how to implement Google’s Gemini API, which is used for natural language processing, in order to build a messaging system that is ACP compliant. This tutorial starts with installing and configuring the google-generativeai libraries. Next, it introduces message types, performatives and the ACPMessage Data class which standardizes communication between agents. The guide defines the ACPAgent class and ACPMessageBroker to demonstrate how structured messages can be created, sent, routed, and processed among autonomous agents. By using code examples that are easy to understand, the guide teaches users how they can implement queries, requests for actions, broadcasting of information and more, all while keeping conversation threads and handling errors.
import google.generativeai as genai
Download json
import time
import uuid
Import Enum
Typing import Dict List Any Optional
Dataclass import from asdict
GEMINI_API_KEY = "Use Your Gemini API Key"
genai.configure(api_key=GEMINI_API_KEY)
Importing essential Python modules is necessary to implement a structured ACP. These include JSON handling, timing, unique identifiers generation, and type annotations. This retrieves user’s Gemini API Key placeholder, and then configures Google-generativeai for future calls to Gemini Language Model.
class ACPMessageType(Enum):
"""Standard ACP message types"""
? "request"
RESPONSE "response"
ATTENTION = "inform"
Question = "query"
SUBSCRIBE "subscribe"
Unsubscribe = "unsubscribe"
ARROR "error"
A = "acknowledge"This enumeration describes the main message types used by the Agent Communication Protocol. These include requests, answers, informational broadcasts and queries as well as control actions such subscribe management, error signals and acknowledgements. The protocol centralizes these message types to ensure consistent routing and handling of inter-agent communication throughout the system.
class ACPPerformative(Enum):
"""ACP speech acts (performatives)"""
Tell = "tell"
ASK = "ask"
RESPONSE = "reply"
REQUEST_ACTION = "request-action"
ASSENT = "agree"
REVERSE = "refuse"
PROPOSITION = "propose"
Accept = "accept"
REJECT = "reject"ACPPerformative is a list of the different speech acts an agent can perform when they interact with each other under the ACP framework. This standardized labeling system maps the high-level intention, such as asking for something, posing a question, giving commands or negotiating agreements onto standard labels. The clear taxonomy allows agents to respond and interpret messages contextually, which ensures robust and semantically-rich communication.
@dataclass
Class ACP Message
"""Agent Communication Protocol Message Structure"""
message_id : str
sender: str
Receivers:
Performative: str
Content:[str, Any]
protocol: str "ACP-1.0"
No conversation_id
reply_to: str= None
language: str "english"
Encoding: str "json"
timestamp: float = None
def __post_init__(self):
If self.timestamp equals None:
self.timestamp = time.time()
If self.conversation_id = None
self.conversation_id = str(uuid.uuid4())
def to_acp_format(self) -> 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(acp_msg, indent=2)
@classmethod
def from_acp_format(cls, acp_string: str) -> 'ACPMessage':
"""Parse ACP message from string format"""
data = json.loads(acp_string)
Return to cls
message_id=data["message-id"],
sender=data["sender"],
receiver=data["receiver"],
performative=data["performative"],
content=data["content"],
protocol=data.get("protocol", "ACP-1.0"),
conversation_id=data.get("conversation-id"),
reply_to=data.get("reply-to"),
language=data.get("language", "english"),
encoding=data.get("encoding", "json"),
timestamp=data.get("timestamp", time.time())
)
ACPMessage is a data class that contains all fields needed for an ACP structured exchange. This includes identifiers and participants as well as performatives and payloads. It also has metadata like protocol versions, languages, and timestamps. This class’s __post_init__ methods automatically populate missing conversation_ids and timestamps, making it possible to track each message. Utility methods from_acp_format & to_acp_format serialize data to and from a standardized JSON format for easy transmission & parsing.
Class: ACPAgent
"""Agent implementing Agent Communication Protocol"""
def __init__(self, agent_id: str, name: str, capabilities: List[str]):
self.agent_id = agent_id
Name = self-name
self.capabilities = capabilities
self.model = genai.GenerativeModel("gemini-1.5-flash")
self.message_queue: List[ACPMessage] = []
Subscribe to Dict[str, List[str]] = {}
"Dict": self-conversations[str, List[ACPMessage]] = {}
def create_message(self, receiver: str, performative: str,
Content:[str, Any], conversation_id: str = None,
reply_to: str = None) -> ACPMessage:
"""Create a new ACP-compliant message"""
Return ACPMessageContent =Content =Content =content
message_id=str(uuid.uuid4()),
sender=self.agent_id,
receiver=receiver,
performative=performative,
content=content,
conversation_id=conversation_id,
reply_to=reply_to
)
def send_inform(self, receiver: str, fact: str, data: Any = None) -> ACPMessage:
"""Send an INFORM message (telling someone a fact)"""
content = {"fact": fact, "data": data}
Self.create_message()receiver, ACPPerformative.TELL.value, content)
def send_query(self, receiver: str, question: str, query_type: str = "yes-no") -> ACPMessage:
"""Send a QUERY message (asking for information)"""
content = {"question": question, "query-type": query_type}
return self.create_message(receiver, ACPPerformative.ASK.value, content)
def send_request(self, receiver: str, action: str, parameters: Dict = None) -> ACPMessage:
"""Send a REQUEST message (asking someone to perform an action)"""
content = {"action": action, "parameters": parameters or {}}
return self.create_message(receiver, ACPPerformative.REQUEST_ACTION.value, content)
def send_reply(self, original_msg: ACPMessage, response_data: Any) -> ACPMessage:
"""Send a REPLY message in response to another message"""
content = {"response": response_data, "original-question": original_msg.content}
return self.create_message(
original_msg.sender,
ACPPerformative.REPLY.value,
content,
conversation_id=original_msg.conversation_id,
reply_to=original_msg.message_id
)
def process_message(self, message: ACPMessage) -> Optional[ACPMessage]:
"""Process incoming ACP message and generate appropriate response"""
self.message_queue.append(message)
conv_id = message.conversation_id
If conv_id is not present in the self.conversations, then:
self.conversations[conv_id] = []
self.conversations[conv_id].append(message)
if message.performative == ACPPerformative.ASK.value:
return self._handle_query(message)
elif message.performative == ACPPerformative.REQUEST_ACTION.value:
return self._handle_request(message)
elif message.performative == ACPPerformative.TELL.value:
return self._handle_inform(message)
Return No
def _handle_query(self, message: ACPMessage) -> ACPMessage:
"""Handle incoming query messages"""
question = message.content.get("question", "")
prompt = f"As agent {self.name} with capabilities {self.capabilities}, answer: {question}"
try:
response = self.model.generate_content(prompt)
Answer = Response.text.strip()
except:
Question = "Unable to process query at this time"
return self.send_reply(message, {"answer": answer, "confidence": 0.8})
def _handle_request(self, message: ACPMessage) -> ACPMessage:
"""Handle incoming action requests"""
"action = message.content.get""action", "")
Get the message content by using parameters."parameters", {})
If any (capability to act.lower() "capabilities" means:
result = f"Executing {action} with parameters {parameters}"
Status "agreed"
else:
"result = f""Cannot perform {action} - not in my capabilities"
Status "refused"
return self.send_reply(message, {"status": status, "result": result})
def _handle_inform(self, message: ACPMessage) -> Optional[ACPMessage]:
"""Handle incoming information messages"""
Get the content of a message by using this formula: fact = message.content.get"fact", "")
print(f"[{self.name}] Received information: {fact}")
ack_content = {"status": "received", "fact": fact}
return self.create_message(message.sender, "acknowledge", ack_content,
conversation_id=message.conversation_id)
The ACPAgent class is an autonomous entity that can send, receive, and process ACP-compliant message using Gemini’s language model. The class manages its message queue, subscriptions and conversation history. It also provides helper methods to create ACPMessage instances in the correct format. Process_message routes incoming messages to the appropriate handlers.
Class ACPMessageBroker
"""Message broker implementing ACP routing and delivery"""
def __init__(self):
Self-agents:[str, ACPAgent] = {}
Self.message_log : List[ACPMessage] = []
Self-routing_table:[str, str] = {}
def register_agent(self, agent: ACPAgent):
"""Register an agent with the message broker"""
self.agents[agent.agent_id] Agent
self.routing_table[agent.agent_id] = "local"
print(f"✓ Registered agent: {agent.name} ({agent.agent_id})")
def route_message(self, message: ACPMessage) -> bool:
"""Route ACP message to appropriate recipient"""
When message.receiver does not exist in the self.agent:
print(f"✗ Receiver {message.receiver} not found")
False
print(f"n📨 ACP MESSAGE ROUTING:")
print(f"From: {message.sender} → To: {message.receiver}")
print(f"Performative: {message.performative}")
print(f"Content: {json.dumps(message.content, indent=2)}")
receiver_agent = self.agents[message.receiver]
response = receiver_agent.process_message(message)
self.message_log.append(message)
If response:
print(f"n📤 GENERATED RESPONSE:")
print(f"From: {response.sender} → To: {response.receiver}")
print(f"Content: {json.dumps(response.content, indent=2)}")
If response.receiver is self.agents
self.agents[response.receiver].process_message(response)
self.message_log.append(response)
Return True
def broadcast_message(self, message: ACPMessage, recipients: List[str]):
"""Broadcast message to multiple recipients"""
For recipients:
msg_copy = ACPMessage(
message_id=str(uuid.uuid4()),
sender=message.sender,
receiver=recipient,
performative=message.performative,
content=message.content.copy(),
conversation_id=message.conversation_id
)
self.route_message(msg_copy)ACPMessageBroker acts as the central router for ACP, keeping a registry and message log. It allows users to register their agents, send individual messages with route_message (which handles log-in, lookup and response chaining), and broadcast a message to many recipients.
def demonstrate_acp():
"""Comprehensive demonstration of Agent Communication Protocol"""
print("🤖 AGENT COMMUNICATION PROTOCOL (ACP) DEMONSTRATION")
print("=" * 60)
Broker = ACPMessageBroker()
ACPAgent (= researcher)"agent-001", "Dr. Research", ["analysis", "research", "data-processing"])
assistant = ACPAgent("agent-002", "AI Assistant", ["information", "scheduling", "communication"])
Calculator = ACPAgent"agent-003", "MathBot", ["calculation", "mathematics", "computation"])
broker.register_agent(researcher)
broker.register_agent(assistant)
broker.register_agent(calculator)
print(f"n📋 REGISTERED AGENTS:")
For agent_id in broker.agents.items():
print(f" • {agent.name} ({agent_id}): {', '.join(agent.capabilities)}")
print(f"n🔬 SCENARIO 1: Information Query (ASK performative)")
query_msg = assistant.send_query("agent-001", "What are the key factors in AI research?")
broker.route_message(query_msg)
print(f"n🔢 SCENARIO 2: Action Request (REQUEST-ACTION performative)")
calc_request = researcher.send_request("agent-003", "calculate", {"expression": "sqrt(144) + 10"})
broker.route_message(calc_request)
print(f"n📢 SCENARIO 3: Information Sharing (TELL performative)")
info_msg = researcher.send_inform("agent-002", "New research paper published on quantum computing")
broker.route_message(info_msg)
print(f"n📊 PROTOCOL STATISTICS:")
print(f" • Total messages processed: {len(broker.message_log)}")
print(f" • Active conversations: {len(set(msg.conversation_id for msg in broker.message_log))}")
print(f" • Message types used: {len(set(msg.performative for msg in broker.message_log))}")
print(f"n📋 SAMPLE ACP MESSAGE FORMAT:")
sample_msg = assistant.send_query("agent-001", "Sample question for format demonstration")
print(sample_msg.to_acp_format())The demo_acp functions orchestrates an interactive walkthrough of ACP: It registers three different agents, (Researcher AI Assistant, MathBot,) initializes the broker, queries for information and requests a computation. It prints summary statistics about the flow of messages after routing and responding to each message. This message provides a complete end-to-end demonstration of the ACP protocol.
Def Setup Guide():
print("""
🚀 GOOGLE COLAB SETUP GUIDE:
1. Get Gemini API Key: https://makersuite.google.com/app/apikey
2. Replace: GEMINI_API_KEY = "YOUR_ACTUAL_API_KEY"
3. Run: demonstrate_acp()
🔧 ACP PROTOCOL FEATURES:
• Standardized message format with required fields
• Speech act performatives (TELL, ASK, REQUEST-ACTION, etc.)
• Conversation tracking and message threading
• Error handling and acknowledgments
• Message routing and delivery confirmation
📝 EXTEND THE PROTOCOL:
```python
# Create custom agent
my_agent = ACPAgent("my-001", "CustomBot", ["custom-capability"])
broker.register_agent(my_agent)
You can send a custom message
My_agent.send_query ("agent-001", "Your question here")
broker.route_message(msg)
```
""")
if name == ___ "__main__":
setup_guide()
demonstrate_acp()
Finaly, the setup_guide functions provides an easy-to-follow guide for running Google Colab’s ACP demonstration. This function explains how to configure and obtain your Gemini API keys and run the demonstrate_acp procedure. This function summarizes the key features in protocol such as standard message formats, performatives or message routing. This code example shows how to send customized messages and register custom agents.
The tutorial ends with the introduction of ACP multi-agents capable to perform research, computation and collaboration. These sample scenarios demonstrate common uses, such as information requests, computational needs, and the sharing of factual data. Meanwhile, the broker provides reliable message delivery, logging, and logging. The framework can be extended by the readers adding more agent capabilities, domain-specific action, and/or sophisticated notification mechanisms.
Download this page Notebook on GitHub. The researchers are the sole owners of all credit. Also, feel free to follow us on Twitter Don’t forget about our 95k+ ML SubReddit Subscribe Now our Newsletter.
Asif Razzaq serves as the CEO at Marktechpost Media Inc. As an entrepreneur, Asif has a passion for harnessing Artificial Intelligence’s potential to benefit society. Marktechpost is his latest venture, a media platform that focuses on Artificial Intelligence. It is known for providing in-depth news coverage about machine learning, deep learning, and other topics. The content is technically accurate and easy to understand by an audience of all backgrounds. This platform has over 2,000,000 monthly views which shows its popularity.
