当前位置：首页 > ds >正文

浅试A2A

ds 2025/8/12 23:19:47

Agent2Agent (A2A) 协议学习文档

1. 简介

Agent2Agent (A2A) 协议是由谷歌发起的开放标准，旨在实现不同AI Agent系统之间的通信和互操作性。该协议为AI Agent之间提供了统一的通信方式，使得基于不同框架或由不同供应商构建的Agent能够发现彼此的能力、协商交互模式并在任务上协作。

1.1 核心概念

Agent 发现：Agent可以通过Agent卡片发现彼此的能力
任务管理：标准化的任务生命周期，具有明确的状态和转换
多种内容类型：支持Agent之间的文本、文件和结构化数据交换
实时更新：长时间运行任务的流式更新和实时协作

1.2 A2A与MCP的关系

A2A和MCP（模型上下文协议）在AI Agent生态系统中有不同但互补的目的：

A2A协议：实现不同AI Agent之间的通信，使它们能够协作、共享信息并协调行动
MCP协议：为单个Agent提供工具和上下文，通过给予它们访问外部工具、数据源和API的能力来增强它们的功能

2. A2A协议示例详解

2.1 Agent实现

假定有三个数学Agent：Sine Agent、Cosine Agent和Tangent Agent。

2.1.1 Sine Agent (agent1.py)

from python_a2a import A2AServer, skill, agent, run_server, TaskStatus, TaskState
import math
import re@agent(name="Sine Agent",description="Provides the sine of a number",version="1.0.0"
)
class SineAgent(A2AServer):@skill(name="Get Sine",description="Get the sine of a number",tags=["sine", "sin"])def get_sine(self, number):"""Get the sine of a number."""# Mock implementationreturn f"The sine of {number} is {math.sin(number)}"def handle_task(self, task):# Extract location from messageinput_message = task.message["content"]["text"]# regex to extract the number from the textmatch = re.search(r"([-+]?[0-9]*\.?[0-9]+)", input_message)number = float(match.group(1))print("number", number)# Get weather and create responsesine_output = self.get_sine(number)task.artifacts = [{"parts": [{"type": "text", "text": sine_output}]}]task.status = TaskStatus(state=TaskState.COMPLETED)return task# Run the server
if __name__ == "__main__":agent = SineAgent()run_server(agent, port=4737)

2.1.2 Cosine Agent (agent2.py)

from python_a2a import A2AServer, skill, agent, run_server, TaskStatus, TaskState
import math
import re@agent(name="Cosine Agent",description="Provides the cosine of a number",version="1.0.0"
)
class CosineAgent(A2AServer):@skill(name="Get Cos",description="Get the cosine of a number",tags=["cos", "cosine"])def get_cosine(self, number):"""Get the cosine of a number."""# Mock implementationreturn f"The cosine of {number} is {math.cos(number)}"def handle_task(self, task):# Extract location from messageinput_message = task.message["content"]["text"]# regex to extract the number from the textmatch = re.search(r"([-+]?[0-9]*\.?[0-9]+)", input_message)number = float(match.group(1))print("number", number)# Get weather and create responsecosine_output = self.get_cosine(number)task.artifacts = [{"parts": [{"type": "text", "text": cosine_output}]}]task.status = TaskStatus(state=TaskState.COMPLETED)return task# Run the server
if __name__ == "__main__":agent = CosineAgent()run_server(agent, port=4738)

2.1.3 Tangent Agent (agent3.py)

from python_a2a import A2AServer, skill, agent, run_server, TaskStatus, TaskState
import math
import re@agent(name="Tangent Agent",description="Provides the tangent of a number",version="1.0.0"
)
class TangentAgent(A2AServer):@skill(name="Get Tangent",description="Get the tangent of a number",tags=["tangent", "tan"])def get_tangent(self, number):"""Get the tangent of a number."""# Mock implementationreturn f"The tangent of {number} is {math.tan(number)}"def handle_task(self, task):# Extract location from messageinput_message = task.message["content"]["text"]# regex to extract the number from the textmatch = re.search(r"([-+]?[0-9]*\.?[0-9]+)", input_message)number = float(match.group(1))print("number", number)# Get weather and create responsetangent_output = self.get_tangent(number)task.artifacts = [{"parts": [{"type": "text", "text": tangent_output}]}]task.status = TaskStatus(state=TaskState.COMPLETED)return task# Run the server
if __name__ == "__main__":agent = TangentAgent()run_server(agent, port=4739)

当运行访问后可以看到Agent信息，下面以Sine Agent为例

在这里插入图片描述

2.2 客户端实现

2.2.1 Client (client.py)

from python_a2a import AgentNetwork, A2AClient, AIAgentRouter# Create an agent network
network = AgentNetwork(name="Math Assistant Network")# Add agents to the network
network.add("Sine", "http://localhost:4737")
network.add("Cosine", "http://localhost:4738")
network.add("Tangent", "http://localhost:4739")# Create a router to intelligently direct queries to the best agent
router = AIAgentRouter(llm_client=A2AClient("http://localhost:5000/openai"),  # LLM for making routing decisionsagent_network=network
)# Route a query to the appropriate agent
query = "Tan of 0.78545"
agent_name, confidence = router.route_query(query)
print(f"Routing to {agent_name} with {confidence:.2f} confidence")# Get the selected agent and ask the question
agent = network.get_agent(agent_name)
response = agent.ask(query)
print(f"Response: {response}")# Route a query to the appropriate agent
query = "Sine of 3.14159"
agent_name, confidence = router.route_query(query)
print(f"Routing to {agent_name} with {confidence:.2f} confidence")# Get the selected agent and ask the question
agent = network.get_agent(agent_name)
response = agent.ask(query)
print(f"Response: {response}")

3. 运行示例

要运行这个A2A协议示例，需要按以下步骤操作：

在不同的终端中分别启动三个Agent服务器：

# 终端1
python agent1.py# 终端2
python agent2.py# 终端3
python agent3.py

运行客户端程序：
```
python client.py
```

# 运行结果
Routing to Tangent with 0.50 confidence
Response: The tangent of 0.78545 is 1.0001036785795414
Routing to Sine with 0.80 confidence
Response: The sine of 3.14159 is 2.65358979335273e-06