In the ever-evolving landscape of AI agents, seamless communication and interoperability are paramount. Imagine a scenario where one specialized agent needs to leverage the output of another to fulfill a user’s request. Without a standardized way for these agents to interact, we often find ourselves building intricate and brittle integration layers.
Consider this:
Agent 1: A service connected to a local Spring Boot application that provides the current temperature of a specific location.
Agent 2: An OpenAI-powered agent that takes the temperature as input and suggests activities suitable for that weather.
The challenge arises because the raw output of Agent 1 isn’t directly consumable by Agent 2. We’d typically need to manually transform the data into a format OpenAI understands. This becomes a significant headache when you need to swap out services or when underlying models are updated, forcing you to rewrite substantial portions of your communication logic. Remember the pain when OpenAI released a new model or when you wanted to experiment with Anthropic’s Claude? It often meant significant code rewrites in the communication layer.
To address this very problem, Google introduced the Agent-to-Agent (A2A) protocol. This protocol aims to standardize the way AI agents communicate, fostering a more flexible and robust ecosystem. Let’s dive into how we can use the python-a2a library to implement this.
Step 1: Installation
First things first, let’s install the necessary Python library:
Bash
1 | pip install python-a2a |
Step 2: Creating an Agent Server (Addition Example)
Let’s build a simple agent that exposes an “add” method. This agent will receive a text message containing two comma-separated numbers and return their sum.
Python
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 | # --- Modified AddAgent Server Code ---from python_a2a import A2AServer, Message, TextContent, MessageRole, run_serverimport logginglogging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')class AddAgent(A2AServer): """ An agent that expects a single text message containing two numbers separated by a comma (e.g., "5,2") and returns their sum. """ # MODIFIED: handle_message now accepts ONE message argument def handle_message(self, message: Message): logging.info(f"Received message: {message}") # Check if it's a text message if not (hasattr(message, 'content') and message.content.type == "text"): warning_msg = "Error: Input must be a text message." logging.warning(warning_msg) return Message( content=TextContent(text=warning_msg), role=MessageRole.AGENT, parent_message_id=getattr(message, 'message_id', None), conversation_id=getattr(message, 'conversation_id', None) ) # Get the text and try to parse it text_input = message.content.text logging.info(f"Attempting to parse text: '{text_input}'") try: # Split the text by a comma (or another delimiter you choose) parts = text_input.split(',') if len(parts) != 2: raise ValueError("Input text must contain exactly two numbers separated by a comma.") # Attempt conversion to float num1 = float(parts[0].strip()) # Use strip() to remove leading/trailing whitespace num2 = float(parts[1].strip()) # Perform the addition result = num1 + num2 response_text = f"The sum of {num1} and {num2} is: {result}" logging.info(f"Calculation successful: {response_text}") # Return the result message return Message( content=TextContent(text=response_text), role=MessageRole.AGENT, parent_message_id=message.message_id, conversation_id=message.conversation_id ) except ValueError as e: # Handle cases where splitting fails or text cannot be converted error_msg = f"Error processing input '{text_input}': {e}" logging.error(error_msg) return Message( content=TextContent(text=error_msg), role=MessageRole.AGENT, parent_message_id=message.message_id, conversation_id=message.conversation_id ) except Exception as e: error_msg = f"An unexpected error occurred: {e}" logging.exception(error_msg) # Log full traceback return Message( content=TextContent(text="An internal server error occurred."), role=MessageRole.AGENT, parent_message_id=message.message_id, conversation_id=message.conversation_id )# Run the server (no changes needed here)if __name__ == "__main__": agent = AddAgent() print("Starting AddAgent server...") # Now the handle_message signature matches what run_server expects run_server(agent, host="localhost", port=5000) |
Save this code as a2a_add_server.py and run it. You should see the server starting on http://localhost:5000/a2a.
Step 3: Creating Another Agent Server (Subtraction Example)
Let’s create another agent, this time for subtraction, running on a different port.
Python
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 | # --- Modified AddAgent Server Code ---from python_a2a import A2AServer, Message, TextContent, MessageRole, run_serverimport logginglogging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')class SubstractAgent(A2AServer): """ An agent that expects a single text message containing two numbers separated by a comma (e.g., "5,2") and returns their Substraction. """ # MODIFIED: handle_message now accepts ONE message argument def handle_message(self, message: Message): logging.info(f"Received message: {message}") # Check if it's a text message if not (hasattr(message, 'content') and message.content.type == "text"): warning_msg = "Error: Input must be a text message." logging.warning(warning_msg) return Message( content=TextContent(text=warning_msg), role=MessageRole.AGENT, parent_message_id=getattr(message, 'message_id', None), conversation_id=getattr(message, 'conversation_id', None) ) # Get the text and try to parse it text_input = message.content.text logging.info(f"Attempting to parse text: '{text_input}'") try: # Split the text by a comma (or another delimiter you choose) parts = text_input.split(',') if len(parts) != 2: raise ValueError("Input text must contain exactly two numbers separated by a comma.") # Attempt conversion to float num1 = float(parts[0].strip()) # Use strip() to remove leading/trailing whitespace num2 = float(parts[1].strip()) # Perform the addition result = num1 - num2 response_text = f"The Substraction of {num1} and {num2} is: {result}" logging.info(f"Calculation successful: {response_text}") # Return the result message return Message( content=TextContent(text=response_text), role=MessageRole.AGENT, parent_message_id=message.message_id, conversation_id=message.conversation_id ) except ValueError as e: # Handle cases where splitting fails or text cannot be converted error_msg = f"Error processing input '{text_input}': {e}" logging.error(error_msg) return Message( content=TextContent(text=error_msg), role=MessageRole.AGENT, parent_message_id=message.message_id, conversation_id=message.conversation_id ) except Exception as e: error_msg = f"An unexpected error occurred: {e}" logging.exception(error_msg) # Log full traceback return Message( content=TextContent(text="An internal server error occurred."), role=MessageRole.AGENT, parent_message_id=message.message_id, conversation_id=message.conversation_id )# Run the server (no changes needed here)if __name__ == "__main__": agent = SubstractAgent() print("Starting SubstractAgent server...") # Now the handle_message signature matches what run_server expects run_server(agent, host="localhost", port=5001) |
Save this as a2a_substract_server.py and run it in a separate terminal. This server will start on http://localhost:5001/a2a.
Step 4: Creating an A2A Client
Now, let’s create a client that can communicate with both of these agents using the A2A protocol.
Python
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 | from python_a2a import A2AClient, Message, TextContent, MessageRole# Create a client to talk to our addition agentclient_add = A2AClient("http://localhost:5000/a2a")# Send a messagemessage_add = Message( content=TextContent(text="5,2"), role=MessageRole.USER)response_add = client_add.send_message(message_add)# Print the responseprint(f"Add Agent says: {response_add.content.text}")# Create a client to talk to our subtraction agentclient_sub = A2AClient("http://localhost:5001/a2a")# Send a messagemessage_sub = Message( content=TextContent(text="5,2"), role=MessageRole.USER)response_sub = client_sub.send_message(message_sub)# Print the responseprint(f"Sunstract Agent says: {response_sub.content.text}") |
Save this code as a2a_client_agent_add_substract.py and execute it. You should see the following output:
1 2 | Add Agent says: The sum of 5.0 and 2.0 is: 7.0Sunstract Agent says: The Substraction of 5.0 and 2.0 is: 3.0 |
Conclusion
This simple example demonstrates the fundamental principles of the A2A protocol using the python-a2a library. By standardizing the message format and communication layer, A2A paves the way for more modular, maintainable, and interoperable AI agent systems. When you need to swap out an underlying service or integrate with a new model, the A2A protocol helps abstract away the low-level communication details, reducing the need for extensive code modifications.
You can find the complete code for this example on GitHub: https://github.com/shdhumale/A2AServerClient.git
As the field of AI agents continues to mature, protocols like A2A will be crucial in building complex and collaborative AI ecosystems. By embracing these standards, we can move towards a future where AI agents can seamlessly work together to solve intricate problems.
No comments:
Post a Comment