Python A2A 是 Google 代理到代理 （A2A） 协议的实现，旨在标准化 AI 代理之间的通信。该协议解决了 AI 生态系统中的一个主要挑战：使不同的 AI 服务能够在没有自定义翻译层的情况下无缝通信。
随着 AI 环境分裂为专门的服务，每个服务都有自己的 API 格式和参数，开发人员花费了过多的时间来构建通信基础设施，而不是专注于 AI 逻辑。Python A2A 通过为 AI 代理提供一种标准化的相互通信方式来解决这个问题，而不管其底层实现如何。
Python A2A 入门
安装
# Basic installation
pip install python-a2a
# For OpenAI integration
pip install "python-a2a[openai]"
# For Anthropic Claude integration
pip install "python-a2a[anthropic]"
# For all optional dependencies
pip install "python-a2a[all]"
核心概念
Python A2A 实现了 A2A 协议中的几个关键概念：
- 消息结构：为文本、函数调用和响应定义的格式
- 对话线程：支持在交互中维护上下文
- 函数调用：Agent 暴露和调用函数的标准化方式
- 错误处理：一致的错误格式
构建您的第一个 A2A 代理
让我们从一个响应消息的简单 echo 代理开始：
from python_a2a import A2AServer, Message, TextContent, MessageRole, run_server
class EchoAgent(A2AServer):
"""A simple agent that echoes back messages with a prefix."""
def handle_message(self, message):
if message.content.type == "text":
return Message(
content=TextContent(text=f"Echo: {message.content.text}"),
role=MessageRole.AGENT,
parent_message_id=message.message_id,
conversation_id=message.conversation_id
)
# Run the server
if __name__ == "__main__":
agent = EchoAgent()
run_server(agent, host="0.0.0.0", port=5000)
现在，让我们创建一个客户端来与我们的代理对话：
from python_a2a import A2AClient, Message, TextContent, MessageRole
# Create a client to talk to our agent
client = A2AClient("http://localhost:5000/a2a")
# Send a message
message = Message(
content=TextContent(text="Hello, is this thing on?"),
role=MessageRole.USER
)
response = client.send_message(message)
# Print the response
print(f"Agent says: {response.content.text}")
代理之间的函数调用
A2A 的强大功能之一是标准化的函数调用。下面是一个公开数学函数的计算器代理：
import math
from python_a2a import (
A2AServer, Message, TextContent, FunctionCallContent,
FunctionResponseContent, FunctionParameter, MessageRole, run_server
)
class CalculatorAgent(A2AServer):
"""An agent that provides mathematical calculation functions."""
def handle_message(self, message):
if message.content.type == "text":
return Message(
content=TextContent(
text="I'm a calculator agent. You can call my functions:\n"
"- calculate: Basic arithmetic (operation, a, b)\n"
"- sqrt: Square root (value)"
),
role=MessageRole.AGENT,
parent_message_id=message.message_id,
conversation_id=message.conversation_id
)
elif message.content.type == "function_call":
function_name = message.content.name
params = {p.name: p.value for p in message.content.parameters}
try:
if function_name == "calculate":
operation = params.get("operation", "add")
a = float(params.get("a", 0))
b = float(params.get("b", 0))
if operation == "add":
result = a + b
elif operation == "subtract":
result = a - b
elif operation == "multiply":
result = a * b
elif operation == "divide":
if b == 0:
raise ValueError("Cannot divide by zero")
result = a / b
else:
raise ValueError(f"Unknown operation: {operation}")
return Message(
content=FunctionResponseContent(
name="calculate",
response={"result": result}
),
role=MessageRole.AGENT,
parent_message_id=message.message_id,
conversation_id=message.conversation_id
)
elif function_name == "sqrt":
value = float(params.get("value", 0))
if value < 0:
raise ValueError("Cannot calculate square root of negative number")
result = math.sqrt(value)
return Message(
content=FunctionResponseContent(
name="sqrt",
response={"result": result}
),
role=MessageRole.AGENT,
parent_message_id=message.message_id,
conversation_id=message.conversation_id
)
except Exception as e:
return Message(
content=FunctionResponseContent(
name=function_name,
response={"error": str(e)}
),
role=MessageRole.AGENT,
parent_message_id=message.message_id,
conversation_id=message.conversation_id
)
if __name__ == "__main__":
agent = CalculatorAgent()
run_server(agent, host="0.0.0.0", port=5001)
以下是调用 calculator 代理函数的方法：
from python_a2a import (
A2AClient, Message, FunctionCallContent,
FunctionParameter, MessageRole
)
client = A2AClient("http://localhost:5001/a2a")
# Create a function call message
function_call = Message(
content=FunctionCallContent(
name="calculate",
parameters=[
FunctionParameter(name="operation", value="add"),
FunctionParameter(name="a", value=5),
FunctionParameter(name="b", value=3)
]
),
role=MessageRole.USER
)
response = client.send_message(function_call)
if response.content.type == "function_response":
result = response.content.response.get("result")
if result is not None:
print(f"Result: {result}") # Output: Result: 8
LLM 支持的代理
Python A2A 包括与流行的 LLM 提供商的即用型集成。这是一个 OpenAI 支持的代理：
import os
from python_a2a import OpenAIA2AServer, run_server
# Create an agent powered by OpenAI
agent = OpenAIA2AServer(
api_key=os.environ["OPENAI_API_KEY"],
model="gpt-4",
system_prompt="You are a helpful AI assistant."
)
# Run the server
if __name__ == "__main__":
run_server(agent, host="0.0.0.0", port=5002)
同样，您可以创建由 Anthropic Claude 提供支持的代理：
import os
from python_a2a import ClaudeA2AServer, run_server
# Create an agent powered by Anthropic Claude
agent = ClaudeA2AServer(
api_key=os.environ["ANTHROPIC_API_KEY"],
model="claude-3-opus-20240229",
system_prompt="You are a helpful AI assistant."
)
# Run the server
if __name__ == "__main__":
run_server(agent, host="0.0.0.0", port=5003)
多代理工作流
A2A 的真正强大之处在于您连接多个代理。让我们构建一个研究助理工作流程：
from python_a2a import (
A2AClient, Message, TextContent, MessageRole, Conversation
)
def research_workflow(query):
# Connect to the specialized agents
llm_client = A2AClient("http://localhost:5002/a2a") # LLM agent
search_client = A2AClient("http://localhost:5003/a2a") # Search agent
summarize_client = A2AClient("http://localhost:5004/a2a") # Summarize agent
# Track the entire workflow in a conversation
conversation = Conversation()
conversation.create_text_message(
text=f"Research question: {query}",
role=MessageRole.USER
)
# Step 1: Generate search queries
print("Generating search queries...")
search_request = Message(
content=TextContent(
text=f"Based on this research question: '{query}', "
f"generate 3 specific search queries that would help find relevant information."
),
role=MessageRole.USER
)
search_queries_response = llm_client.send_message(search_request)
conversation.add_message(search_queries_response)
# Step 2: Retrieve information
print("Retrieving information...")
search_message = Message(
content=TextContent(
text=f"Search for information to answer: {query}\n\n"
f"Using these queries:\n{search_queries_response.content.text}"
),
role=MessageRole.USER
)
search_results = search_client.send_message(search_message)
conversation.add_message(search_results)
# Step 3: Synthesize information
print("Synthesizing information...")
summarize_message = Message(
content=TextContent(
text=f"Synthesize this information to answer the question: '{query}'\n\n"
f"Information:\n{search_results.content.text}"
),
role=MessageRole.USER
)
summary_response = summarize_client.send_message(summarize_message)
conversation.add_message(summary_response)
# Add final answer to the conversation
conversation.create_text_message(
text=f"Answer to your research question:\n\n{summary_response.content.text}",
role=MessageRole.AGENT
)
return conversation
# Example usage
if __name__ == "__main__":
query = input("What's your research question? ")
result = research_workflow(query)
print("\nResearch Complete!")
print("=" * 50)
print(result.messages[-1].content.text)
高级示例：使用 A2A 进行天气和旅行规划
以下示例演示了 A2A 如何简化代理之间的通信：
from python_a2a import A2AClient, Message, TextContent, MessageRole
# With A2A: Any agent -> Any other agent
def plan_trip(location):
# Connect to specialized agents - all using the same protocol
weather_client = A2AClient("http://localhost:5001/a2a")
openai_client = A2AClient("http://localhost:5002/a2a")
# Ask weather agent
weather_message = Message(
content=TextContent(text=f"What's the weather forecast for {location}?"),
role=MessageRole.USER
)
weather_response = weather_client.send_message(weather_message)
# Ask OpenAI agent, including weather info
planning_message = Message(
content=TextContent(
text=f"I'm planning a trip to {location}. Weather forecast: {weather_response.content.text}"
f"Please suggest activities."
),
role=MessageRole.USER
)
planning_response = openai_client.send_message(planning_message)
return planning_response.content.text
# This works with ANY A2A-compatible agents - no custom adapters needed!
管理对话
Python A2A 提供了一个类来管理多轮交互：Conversation
from python_a2a import Conversation, MessageRole, A2AClient
# Create a new conversation
conversation = Conversation()
# Add a user message
conversation.create_text_message(
text="What's the weather like today?",
role=MessageRole.USER
)
# Connect to an agent
weather_agent = A2AClient("http://localhost:5001/a2a")
# Send the last message in the conversation
last_message = conversation.messages[-1]
response = weather_agent.send_message(last_message)
# Add the response to the conversation
conversation.add_message(response)
# Continue the conversation
conversation.create_text_message(
text="Thanks! Will I need an umbrella?",
role=MessageRole.USER
)
# Send the new message, including conversation history
response = weather_agent.send_message(
conversation.messages[-1],
conversation_id=conversation.conversation_id
)
# Add this response too
conversation.add_message(response)
# Print the entire conversation
for msg in conversation.messages:
role = "User" if msg.role == MessageRole.USER else "Agent"
print(f"{role}: {msg.content.text}")
错误处理
Python A2A 提供标准化的错误处理：
from python_a2a import A2AClient, Message, TextContent, MessageRole
client = A2AClient("http://localhost:5001/a2a")
try:
message = Message(
content=TextContent(text="Hello"),
role=MessageRole.USER
)
response = client.send_message(message)
except ConnectionError as e:
print(f"Connection error: {e}")
except TimeoutError as e:
print(f"Timeout error: {e}")
except Exception as e:
print(f"Unexpected error: {e}")
创建自定义 A2A 代理
以下是创建您自己的 A2A 兼容代理的模板：
from python_a2a import A2AServer, Message, TextContent, MessageRole, run_server
class MyCustomAgent(A2AServer):
def __init__(self, **kwargs):
super().__init__(**kwargs)
# Initialize your agent-specific components here
def handle_message(self, message):
"""Process incoming messages and return responses.
This is the main entry point for handling messages in your agent.
"""
if message.content.type == "text":
# Process text messages
user_text = message.content.text
# Generate your response (replace with your logic)
response_text = f"You said: {user_text}"
return Message(
content=TextContent(text=response_text),
role=MessageRole.AGENT,
parent_message_id=message.message_id,
conversation_id=message.conversation_id
)
elif message.content.type == "function_call":
# Handle function calls (if your agent supports them)
# ...
pass
# Return a default response if no other conditions matched
return Message(
content=TextContent(text="I don't know how to handle this message type."),
role=MessageRole.AGENT,
parent_message_id=message.message_id,
conversation_id=message.conversation_id
)
# Run your agent
if __name__ == "__main__":
my_agent = MyCustomAgent()
run_server(my_agent, host="0.0.0.0", port=5005)
结论
Python A2A 为 AI 代理提供了一种标准化的通信方式，从而可以更轻松地构建多代理系统。通过采用此协议，您可以：
- 减少在通信基础设施上花费的开发时间
- 轻松交换和升级代理实施
- 构建模块化、可扩展的 AI 系统
- 利用专用代理执行不同的任务
A2A 协议及其 Python 实现正在帮助构建一个未来，让 AI 系统可以像乐高积木一样组成，并且专门的代理可以无缝地协同工作。
发表评论 取消回复