A2A 代理间协作怎么配置:从代理卡到任务调用的完整流程
如果你想让两个 Agent 互相发送任务并拿到结果,核心配置只有两件事:每个代理提供一份描述自身能力的代理卡(Agent Card),调用方按 A2A 协议向对方暴露的端点发送任务请求。
下面从零开始,用 Python 演示最小可用的协作配置。
先分清谁提供能力、谁发起调用
A2A 协作里有两个角色:服务端代理负责公开端点并处理任务,客户端代理负责发现对方并发送请求。
配置前先确定:你的 Agent 是提供能力的一方,还是调用别人的一方。
两者可以同时存在,但第一次联调建议分开测试。
需要准备的环境:
- Python 3.9 或以上
- 安装
a2a-sdk和uvicorn(示例使用 pip 安装) - 两个可互通的端口,例如服务端用
8001,客户端用8002 - 服务端代理卡中填写的
url必须是客户端能访问到的地址,本机测试用http://127.0.0.1:8001,跨机测试改成实际 IP 或域名
编写服务端代理卡并启动服务
代理卡是一个 JSON 文件,放在服务端能访问的路径下,例如 agent_card.json。
最小内容如下:
{
"name": "echo_agent",
"description": "回显收到的文本",
"url": "http://127.0.0.1:8001",
"version": "1.0.0",
"capabilities": {
"streaming": false,
"pushNotifications": false
},
"skills": [
{
"id": "echo",
"name": "回显",
"description": "返回输入文本"
}
]
}
关键字段说明:url 是客户端发送任务的地址,skills 列出代理能处理的任务类型。
如果你的代理支持流式响应,把 streaming 设为 true,否则客户端不要期待流式返回。
启动服务端代码示例(保存为 server.py):
from a2a.server import A2AServer, AgentCard
from a2a.server.tasks import TaskStore
agent_card = AgentCard.from_json_file("agent_card.json")
async def handle_task(task):
text = task.messages[-1].parts[0].text
return {"text": f"收到:{text}"}
server = A2AServer(agent_card=agent_card, task_handler=handle_task, task_store=TaskStore())
if __name__ == "__main__":
import uvicorn
uvicorn.run(server.app, host="0.0.0.0", port=8001)
运行 python server.py,服务端会监听 8001 端口。
此时用浏览器访问 http://127.0.0.1:8001/.well-known/agent-card.json 应该能看到代理卡内容,这是客户端发现代理的标准路径。
客户端如何发现代理并发送任务
客户端需要先获取服务端的代理卡,然后按卡里的 url 发送任务请求。
下面是一个最小客户端示例(保存为 client.py):
import asyncio
from a2a.client import A2AClient
async def main():
client = await A2AClient.from_agent_card_url("http://127.0.0.1:8001/.well-known/agent-card.json")
task = await client.send_task({
"messages": [
{"role": "user", "parts": [{"text": "你好,协作测试"}]}
]
})
print("任务ID:", task.id)
print("响应:", task.messages[-1].parts[0].text)
asyncio.run(main())
运行 python client.py,如果输出类似 响应: 收到:你好,协作测试,说明两个 Agent 已经完成一次协作调用。
注意 from_agent_card_url 会自动拉取代理卡并构造客户端,你不需要手动拼接请求体。
联调时最容易卡住的三个地方
第一,代理卡里的 url 写成了 localhost 或 127.0.0.1,但客户端在另一台机器上。
跨机时改成服务端实际可访问的 IP 或域名,并确认防火墙放行端口。
第二,代理卡路径不对。
A2A 客户端默认从 /.well-known/agent-card.json 获取代理卡,如果你把文件放在别的路径,需要在客户端代码里显式指定完整 URL,或者把服务端路由改到标准路径。
第三,任务处理函数返回格式与客户端预期不一致。
示例中返回 {"text": ...},客户端从 task.messages[-1].parts[0].text 读取。
如果服务端返回纯字符串,客户端解析会失败。
建议先用 curl 直接发一个请求看原始返回:
curl -X POST http://127.0.0.1:8001/ \
-H "Content-Type: application/json" \
-d '{"messages":[{"role":"user","parts":[{"text":"测试"}]}]}'
如果 curl 返回 404,检查服务端路由是否挂在 / 上;
返回 422 通常是请求体缺少 messages 字段。
验证协作配置是否完整
一次完整的验证应该覆盖三个动作:客户端能拉取代理卡、能发送任务、能收到符合格式的响应。
可以在客户端代码里加两行日志:
print("代理名称:", client.agent_card.name)
print("可用技能:", [s.id for s in client.agent_card.skills])
如果这两行能正常输出,说明发现阶段没问题。
然后确认任务 ID 和响应文本都非空。
对于需要长时间处理的任务,检查 task.status 是否从 working 变为 completed,不要只看到连接成功就认为协作完成。
关于代理间协作的两个常见疑问
两个代理都需要写代理卡吗? 如果双方互相调用,各自都需要提供代理卡。
如果只是单向调用,只有被调用的服务端需要公开代理卡,客户端可以只作为发起方,不暴露自己的端点。
代理卡里的 url 可以带路径吗? 可以,但客户端发送任务时会直接向该 URL 发 POST 请求,所以服务端路由必须匹配。
建议初次配置时使用根路径,减少路径拼接错误。
配置完成后,建议把代理卡文件和启动命令纳入版本管理,并在修改 url 或 skills 后重新启动服务端,避免客户端读到旧卡。
如果协作仍不稳定,优先检查网络连通性和代理卡字段是否与 SDK 版本兼容。