教程：逐步构建两个使用谷歌A2A协议的Python代理
谷歌的代理到代理（A2A）协议是一个开放标准，它使不同的AI代理能够相互通信并协同工作，无论它们来自哪个框架或供应商（GitHub - google/A2A (opens in a new tab)）。简单来说，A2A为你的代理提供了一种通用的交互语言。本教程将指导你创建两个简单的Python代理——一个作为A2A 服务器，另一个作为A2A 客户端——它们通过A2A协议相互通信。我们将涵盖设置，解释A2A的基本工作原理，并为两种代理提供带注释的代码示例。不假设任何A2A的先前经验，我们将避免复杂工具链或多代理协调等高级概念，以保持友好性。
什么是A2A协议？
从本质上讲，A2A是一种标准化代理交换消息、请求和数据的协议（基于HTTP）。以下是A2A中的一些关键概念：
- 代理卡片: 一个公共元数据文件（通常托管在
/.well-known/agent.json），描述代理的能力、技能、终端URL和认证需求。其他代理或客户端获取此卡片以发现代理能做什么（GitHub - google/A2A (opens in a new tab)）。这本质上是A2A网络中代理的“名片”。 - A2A服务器: 暴露HTTP API端点并实现A2A方法的代理（例如，用于发送任务的端点）。它接收请求并代表其他代理执行任务（GitHub - google/A2A (opens in a new tab)）。
- A2A客户端: 发送请求到A2A服务器的URL以发起任务或对话的应用程序或代理（GitHub - google/A2A (opens in a new tab)）。在代理到代理的场景中，一个代理的客户端组件会通过A2A调用另一个代理的服务器组件。
- 任务: 在A2A中工作的基本单位。客户端开始一个任务时，通过（使用
tasks/send请求）向代理发送消息。每个任务都有一个唯一的ID并经历一个状态生命周期（例如submitted、working、input-required、completed等） (GitHub - google/A2A (opens in a new tab))。 - 消息: 客户端（角色为
"user"）和代理（角色为"agent"）之间通信的一个回合。客户端的请求是一个消息（来自“用户”），而代理的回答是一个消息（来自“代理”）(GitHub - google/A2A (opens in a new tab)）。 - 部分: 消息中的一个内容片段。消息由一个或多个部分组成。部分可以是文本、文件或结构化数据。例如，一个文本部分（
TextPart）携带纯文本，而一个文件部分可能携带图像或其他二进制数据（GitHub - google/A2A (opens in a new tab)）。在本教程中，为了清晰起见，我们将坚持使用简单的文本部分。
A2A通信工作原理（基本流程）： 当一个代理希望通过A2A与另一个代理通信时，交互通常遵循标准化的流程：
- 发现： 客户端代理首先通过获取其代理卡片（通常来自服务器的
/.well-known/agent.jsonURL）来发现另一个代理（GitHub - google/A2A (opens in a new tab)）。此卡片告知客户端代理的名称、功能以及发送请求的位置。 - 启动： 随后，客户端通过发送请求到服务器代理的A2A端点来启动任务——通常是通过
POST到tasks/send端点。这个初始请求包括一个唯一的任务ID和第一条用户消息（例如，问题或命令）（GitHub - google/A2A (opens in a new tab)）。 - 处理： 服务器代理接收任务请求并进行处理。如果代理支持流响应，它可能会在工作时流中间更新回去（使用服务器发送事件），虽然在我们的简单示例中我们不会使用流。否则，代理将处理任务并在完成时发送最终响应（GitHub - google/A2A (opens in a new tab)）。响应将包含任务的结果——通常是来自代理的一条或多条消息（如果这是任务的一部分，还可以选择性地包含文件等工件）。对于基本的文本交互，最终响应将只包括代理的回答消息。
总的来说，A2A定义了一个清晰的请求-响应周期：客户端找到代理的能力（代理卡片），发送一个任务（带有初始用户消息），并以一致的JSON格式接收代理的响应（作为消息和工件）。现在，让我们设置环境并创建一个最小化示例来看看这在行动中如何运作。
安装和设置
在编写任何代码之前，请确保安装了所需的工具：
- Python 3.12或更高版本 – A2A样例建议使用Python 3.12+（甚至是Python 3.13）以确保兼容性（A2A/samples/python/hosts/cli at main · google/A2A · GitHub (opens in a new tab)）。
- UV（Python包管理工具） – 这是A2A项目推荐的管理依赖并运行样本的可选但推荐工具（A2A/samples/python/hosts/cli at main · google/A2A · GitHub (opens in a new tab)）。UV是一个现代包管理工具；然而，如果你愿意，也可以使用pip/venv手动安装依赖。在本教程中，我们将保持简单并使用标准的Python工具。
- Google A2A代码库 – 通过克隆GitHub仓库获取官方A2A代码：
这将下载项目到你的机器上。我们的教程代码将引用该仓库的部分内容（特别是样本A2A客户端工具）。
git clone https://github.com/google/A2A.git - 依赖项 – 如果使用 UV，可以让它自动处理依赖项。例如，从仓库的
samples/python目录中，可以运行uv sync（来安装依赖）或按照官方README中的示例运行命令uv run ...。如果偏爱使用pip，确保安装我们示例代码所需的任何库：Flask 库将用于为我们的代理创建一个简单的HTTP服务器，requests 将用于客户端调用代理。（官方A2A示例代码使用FastAPI/UVICorn和一个内部客户端库，但为了初学者友好性，我们使用Flask和requests进行说明。）pip install flask requests
创建一个简单的A2A服务器代理
首先，我们将编写服务器端：一个侦听传入任务并回应的代理。我们的代理将非常简单——它将接受一个文本提示并用友好的消息将其回显。尽管它很简单，但它将遵循A2A协议结构，以便任何A2A客户端都可以与之通信。
代理能力和代理卡片： 为了让我们的代理可被发现，它需要提供一个代理卡片。在实际部署中，我们会在服务器上托管一个JSON文件在 /.well-known/agent.json。对于我们的Flask应用程序，我们可以通过一个端点提供这些数据。代理卡片通常包括如代理名称、描述、其A2A终端的URL、支持的功能（能力）等字段（GitHub - google/A2A (opens in a new tab)）。我们将构建一个带有几个关键字段的最小代理卡片。
tasks/send 端点： A2A期望代理实现某些API端点。最重要的是 tasks/send，客户端将调用（HTTP POST）以发送新任务（带有用户消息）给代理（GitHub - google/A2A (opens in a new tab)）。我们的Flask应用将定义/tasks/send的路由来处理这些请求。它将解析传入的JSON（包含任务ID和用户的消息），进行处理（在我们的例子中，只是生成一个回波响应），并遵循A2A格式返回JSON响应。
让我们用Flask编写代码创建我们的简单A2A服务器代理：
from flask import Flask, request, jsonify
app = Flask(__name__)
# 定义代理卡片数据（关于此代理的元数据）
AGENT_CARD = {
"name": "EchoAgent",
"description": "一个简单的代理，它会回显用户消息。",
"url": "http://localhost:5000", # 此代理托管的基本URL
"version": "1.0",
"capabilities": {
"streaming": False, # 此代理不支持流响应
"pushNotifications": False # 此简单示例中没有推送通知
}
# 在完整的代理卡片中，可能会有更多字段如认证信息等。
}
# 在众所周知的URL提供代理卡片。
@app.get("/.well-known/agent.json")
def get_agent_card():
"""端点提供此代理的元数据（代理卡片）。"""
return jsonify(AGENT_CARD)
# 在A2A端点处理传入的任务请求。
@app.post("/tasks/send")
def handle_task():
"""端点供A2A客户端发送新任务（含初始用户消息）。"""
task_request = request.get_json() # 解析传入的JSON请求
# 从请求提取任务ID和用户的消息文本。
task_id = task_request.get("id")
user_message = ""
try:
# 根据A2A规范，用户消息位于 task_request["message"]["parts"][0]["text"]
user_message = task_request["message"]["parts"][0]["text"]
except Exception as e:
return jsonify({"error": "Invalid request format"}), 400
# 对于这个简单的代理，“处理”只是回显消息。
agent_reply_text = f"Hello! You said: '{user_message}'"
# 以A2A任务格式制定响应。
# 我们将返回一个状态为“已完成”的任务对象和代理的消息。
response_task = {
"id": task_id,
"status": {"state": "completed"},
"messages": [
task_request.get("message", {}), # 在历史中包含原用户消息
{
"role": "agent", # 代理的回复
"parts": [{"text": agent_reply_text}] # 代理消息内容作为TextPart
}
]
# 如果代理返回文件或其他数据，还可以包括一个“工件”字段。
}
return jsonify(response_task)
# 如果直接执行此脚本，则运行Flask应用程序（A2A服务器）。
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000)理解服务器代码： 让我们分解上面的代码所发生的事情：
- 我们创建了一个Flask应用并定义了一个全局的
AGENT_CARD字典。这包含了我们代理的基本信息。值得注意的是，它有一个名称、描述、一个指向自身的URL（在localhost和端口5000上），以及一个能力部分。我们设置streaming: False因为我们的代理只会做简单的同步回复（如果一个代理支持流， 它可能会处理tasks/sendSubscribe并流部分结果，我们在这里不会讨论）。 - 我们设置了一个路由
/.well-known/agent.json返回代理卡片JSON。这模仿了标准的位置，在该位置预计可以找到代理的卡片进行发现（GitHub - google/A2A (opens in a new tab)）。 - 我们为
/tasks/send定义了一个POST路由。当A2A客户端发送一个任务时，Flask会调用handle_task()。在此函数内部：- 我们将请求的JSON体解析为
task_request。根据A2A的模式，这将包含"id"（任务ID）和一个"message"对象表示用户的初始消息。 - 我们提取用户消息的文本。在A2A中，一个消息可以有多个部分（例如，文本、文件）。这里我们就假设第一部分是文本，并获取
task_request["message"]["parts"][0]["text"]。在一个更健壮的实现中，我们会验证部分类型并处理错误（我们的代码只做一个简单的try/except，如果格式不如预期则返回400错误）。 - 然后，我们生成代理的响应。我们的逻辑很简单：我们接收用户文本并加上
"Hello! You said: …"。在一个真正的代理中，这就是核心人工智能逻辑所在的位置（调用语言模型、执行计算等）。 - 接下来，我们构建响应的任务对象。我们必须回显
id（以便客户端知道这是哪个任务），将status.state设置为"completed"（因为我们完成了任务），并提供一条messages列表。我们包含原始的用户消息（用于上下文/历史），然后添加我们的代理回复作为另一个消息。每个消息有一个role（"user"或"agent"）和一个parts列表。我们为代理的消息使用一个包含响应文本的TextPart。 - 最后，我们将这个转换为JSON。Flask将为我们序列化Python字典为JSON。
- 我们将请求的JSON体解析为
有了这个服务器运行，我们的代理就有效地“在线”于 http://localhost:5000。它可以提供其代理卡片并处理任务请求。
创建一个简单的A2A客户端代理
现在我们有了一个代理服务器，我们需要一个客户端与之对话。这个客户端可以是另一个代理或只是一个面向用户的应用程序。在A2A术语中，它是一个A2A客户端因为它将使用我们的服务器代理的A2A API（GitHub - google/A2A (opens in a new tab)）。对于我们的教程，我们将编写一个小的Python脚本来扮演第二个代理（或用户）的角色，向我们的EchoAgent发送一个问题并接收答案。
客户端步骤：
- 发现代理： 客户端应该获取服务器代理的代理卡片以了解其端点和功能。我们知道我们的代理卡片位于
http://localhost:5000/.well-known/agent.json。我们可以使用Pythonrequests库GET这个URL。 - 发送任务： 接下来，客户端将发送一个POST请求到代理的
tasks/send端点，附带一个JSON有效载荷，表示任务。此负载应包括：- 一个唯一的
"id"为任务（任何唯一字符串或数字；例如，"task1"）， - 一个
"message"对象带role: "user"和一个parts列表，包含用户查询作为TextPart。
- 一个唯一的
- 接收响应： 代理将以JSON格式响应任务（包括代理的回复消息）。客户端需要读取这一点并提取代理的回复。
Let's write the client code to perform these steps:
import requests
import uuid
# 1. 通过获取其代理卡片发现代理
AGENT_BASE_URL = "http://localhost:5000"
agent_card_url = f"{AGENT_BASE_URL}/.well-known/agent.json"
response = requests.get(agent_card_url)
if response.status_code != 200:
raise RuntimeError(f"Failed to get agent card: {response.status_code}")
agent_card = response.json()
print("Discovered Agent:", agent_card["name"], "-", agent_card.get("description", ""))
# 2. 为代理准备任务请求
task_id = str(uuid.uuid4()) # 生成一个随机唯一任务ID
user_text = "生命的意义是什么？"
task_payload = {
"id": task_id,
"message": {
"role": "user",
"parts": [
{"text": user_text}
]
}
}
print(f"Sending task {task_id} to agent with message: '{user_text}'")
# 3. 将任务发送到代理的 tasks/send 端点
tasks_send_url = f"{AGENT_BASE_URL}/tasks/send"
result = requests.post(tasks_send_url, json=task_payload)
if result.status_code != 200:
raise RuntimeError(f"Task request failed: {result.status_code}, {result.text}")
task_response = result.json()
# 4. 处理代理的响应
# 响应应包含任务ID、状态和消息（包括代理的回复）。
if task_response.get("status", {}).get("state") == "completed":
# 列表中的最后一条消息应该是代理的答案（因为我们的代理在消息中包含历史）。
messages = task_response.get("messages", [])
if messages:
agent_message = messages[-1] # 最后一条消息（来自代理）
agent_reply_text = ""
# 从代理的消息部分中提取文本
for part in agent_message.get("parts", []):
if "text" in part:
agent_reply_text += part["text"]
print("Agent's reply:", agent_reply_text)
else:
print("No messages in response!")
else:
print("Task did not complete. Status:", task_response.get("status"))理解客户端代码： 以下是客户端的工作原理：
- 使用
requests.get从http://localhost:5000/.well-known/agent.json获取代理卡片。成功后，我们解析JSON并打印代理的名称和描述以验证我们找到了正确的代理。 - 使用
uuid.uuid4()生成一个唯一的任务ID。这只是为了确保我们不重复使用ID（在实际场景中，客户端必须确保其发送的每个任务具有唯一的ID）。 - 我们创建一个Python字典形式的任务有效载荷。该有效载荷遵循A2A模式：它有一个
"id"（我们的生成ID）和一个"message"表示用户的查询。该消息具有role: "user"和一个包含文本"生命的意义是什么？"（可以更改此文本为任何内容 —— EchoAgent将只是回显它）的部分。在本例中。 - 我们使用
requests.post将这个有效载荷发送到代理的tasks/send端点。注意我们使用json=...参数，这指示requests库以JSON格式发送字典。 - 我们检查响应代码是否为200 OK。如果状态不是200，我们引发错误。（在更宽容的脚本中，您可能会更宽松地处理非200响应。）
- 如果调用成功，我们将JSON响应解析为
task_response。这应该是我们代理返回的任务对象。然后我们检查它：- 我们检查
status.state是否为"completed"– 在我们的设计中，代理在一次回复后将其标为完成。例如，如果代理需要更多输入（"input-required"）或失败，我们将处理这些情况。为了简便起见，我们假设完成。 - 然后我们从响应中获取
messages列表。我们期望这个列表中的最后一条信息是代理的回复（因为我们的服务器代码在用户的后面添加了代理的信息）。我们从代理的消息部分提取文本。在这种情况下，它应该是我们构建的回声响应。 - 最后，我们打印出代理的回复。
- 我们检查
此时，我们的客户端将已打印代理的响应。在下一节中，我们将同时运行服务器和客户端来查看整个交互。
运行和测试代理
要测试我们的两个代理，请按照以下步骤进行：
-
启动服务器代理： 运行Flask服务器脚本（我们将其称为
echo_agent.py）。例如，在你的终端中：python echo_agent.py这应该在
http://localhost:5000上启动Flask开发服务器。你应该看到类似“运行在 http://0.0.0.0:5000/”的消息。代理现在正在侦听请求。（确保没有其他东西正在使用端口5000，或者在代码中更改端口以防需要。） (opens in a new tab) -
运行客户端脚本： 打开另一个终端并运行客户端脚本（如
client_agent.py）：python client_agent.py客户端应打印出如下内容：
发现代理: EchoAgent – 一个简单的代理，用来回显用户消息。 已向代理发送任务 3f8ae8ac-...，消息为：'生命的意义是什么？' 代理回复: Hello! You said: '生命的意义是什么？'这表示客户端成功发现了代理，发送了问题并接收到了答案。代理的回复是我们预期的回显信息。
-
验证交互： 你也可以检查服务器的控制台输出。Flask会记录它处理的请求。你应该看到对
/.well-known/agent.json的GET请求和对/tasks/send的POST请求，以及每个请求的200状态。这确认了协议流程：- 发现（客户端GET代理卡片） → 启动（客户端POST任务） → 处理/完成（服务器回复答案）。
恭喜！您刚刚使用A2A协议实现了基本的代理通信。一个代理充当服务（EchoAgent），另一个作为进行查询的客户端。虽然这个例子很简单（“AI”只是重复输入的东西），但它展示了任何A2A交互所需的基本架构。
使用A2A客户端库（可选）
在我们的客户端代码中，我们使用 requests 手动创建了HTTP请求。官方的A2A GitHub仓库提供了帮助类来简化这个过程。例如，有一个 A2AClient 类可以自动处理获取代理卡片和发送任务，还有一个 A2ACardResolver 用于发现代理的卡片。下面是如何使用这些（对于那些对更惯用的方法感兴趣的读者）：
import asyncio
from samples.python.common.client import A2AClient, A2ACardResolver
from samples.python.common.types import TaskSendParams, Message, TextPart
async def query_agent(agent_url, user_text):
# 自动获取代理卡片
card_resolver = A2ACardResolver(agent_url)
agent_card = card_resolver.get_agent_card()
print("发现代理:", agent_card.name)
# 使用代理的卡片创建A2A客户端
client = A2AClient(agent_card=agent_card)
# 准备任务参数（使用A2A类型类）
payload = TaskSendParams(
id=str(uuid.uuid4()),
message=Message(role="user", parts=[TextPart(text=user_text)])
)
# 发送任务并等待完成
result_task = await client.send_task(payload) # send_task 是一个异步方法
# 从result_task中提取代理的回复
if result_task.status.state.value == "completed":
# A2A任务对象可以用来检查信息和工件
for msg in result_task.messages:
if msg.role == "agent":
# 打印代理信息的文本部分
print("代理的回复:", " ".join(part.text for part in msg.parts if hasattr(part, "text")))在上述代码段中，A2ACardResolver 和 A2AClient 来自A2A样本库（位于仓库的samples/python/common目录）。TaskSendParams、Message 和 TextPart 是与A2A JSON模式相对应的数据类（可能基于Pydantic模型）。使用这些，我们不需要手动构建字典；我们创建Python对象，库将处理JSON序列化和HTTP调用。客户端的send_task方法是异步的（因此我们使用await），并返回一个Task对象。我们的示例展示了如何从该对象中获取代理的回复。
注意： 上面的代码是为了说明，要求A2A仓库的代码在你的Python路径中可访问。如果你已经克隆了仓库并安装了其要求（通过UV或pip），你可以将其集成到你的客户端中。官方CLI工具基本上执行这些步骤——它读取代理卡片并进入一个循环，发送任务并打印响应（A2A/samples/python/hosts/cli at main · google/A2A · GitHub (opens in a new tab)）。
结论
在本教程中，我们介绍了谷歌A2A协议的基础知识，并展示了简单的两个代理通信。我们设置了一个最小的A2A服务器（EchoAgent）和一个与之交互的客户端。在此过程中，我们了解了代理卡片（用于发现）以及任务和消息在A2A中的结构。
虽然我们的示例是一个简单的回声代理，但你现在已经具备了构建更复杂代理的基础。比如，你可以用调用语言模型或API替换回声逻辑，来让代理真正解决问题或回答问题。A2A的美妙之处在于，只要你的代理遵循协议（提供代理卡片并实现任务端点），任何其他A2A兼容的代理或工具都可以与之集成。
下一步: 你可能会探索官方A2A代码库样本以获得更高级的模式。例如，谷歌的样本包括一个代理从文本生成图像并以图像工件的形式返回它们（A2A/samples/python/agents/crewai/README.md at main · google/A2A · GitHub (opens in a new tab)），以及另一个处理多轮表单填写对话的代理。你还可以尝试提供的命令行界面（CLI）或网络演示来与代理聊天。要在我们的EchoAgent上使用CLI，你可以运行：uv run hosts/cli --agent http://localhost:5000（如果你有UV和仓库设置）以启动与代理的REPL式聊天（A2A/samples/python/hosts/cli at main · google/A2A · GitHub (opens in a new tab)）。
通过标准化代理通信，A2A为丰富的可互操作AI代理生态系统打开了大门。在这里我们只是初步认识了它。祝在A2A上实验愉快——愿你的代理能有效地协作！