☰
Agent通信协议A2A深度解析:从核心机制到落地实践
2026/10/17 5:12:04 网站建设 项目流程

最近和几个做 Agent 平台的朋友聊下来,大家最头疼的一件事出奇一致:Agent 各自跑得很欢,但彼此之间不会说话。你要把“查订单”的 Agent 和“开票”的 Agent 串起来,要么写死两边的接口,要么靠人肉拷贝数据,总之绕不开一个词——通信协议。A2A(Agent-to-Agent)协议就是冲这个痛点来的。2025 年 4 月它由 Google 端出来,6 月又捐给了 Linux 基金会,从一个公司项目变成了开放标准候选。这篇文章我会用实际落地过的经验,把 A2A 的核心机制、实操步骤、和 MCP 的分工关系以及常见坑一次讲透。不管你是刚开始接触 Agent 开发的新手,还是已经在搞 Agent 框架选型的工程负责人,都能从中找到可以直接参考的东西。

1. 为什么 Agent 需要自己的通信协议

1.1 Agent 协作的本质是“交接工作”

先想一个问题:你让两个人类同事协作,靠的是什么?靠的是双方都懂一套共同语言,比如中文,靠的是你交代任务时能把“上下文”说清楚,靠的是对方干完活后能给你一个明确的结果。Agent 和 Agent 之间也一样,但它们比人类更较真——没有统一的消息格式、没有统一的调用入口、没有统一的任务状态,两个 Agent 就是两个孤岛。

A2A 想做的,就是给 Agent 之间定义一套“共同语言”。它不关心你底层用的是哪个模型、跑在什么框架里,也不关心你是单体 Agent 还是多 Agent 编排,它只管在你需要调用另外一个 Agent 的时候,让你能够像调用一个本地函数一样自然地把任务发过去,再把结果收回来。

我见过太多团队在“多 Agent”上栽跟头:本地起了两三个 Agent,各自能力都不错,但一涉及跨 Agent 调用就开始写胶水代码。今天 A 要调 B 的接口,明天 C 也要调 B,后天 B 的返回格式变了一下,A 和 C 全挂。这种局面本质上不是代码问题,是缺少一层标准化的通信契约。

1.2 没有协议时的几种典型乱象

我总结了一下,没有统一通信协议时,多 Agent 项目通常会遇到下面这几类问题。

第一类,接口格式混乱。每个 Agent 的创始人会按照自己的喜好定义请求和响应,有的用 JSON,有的用 XML,有的甚至直接把文本塞进 query string。对接方拿到文档还得猜字段含义。第二类,任务没有生命周期。A 调用 B 之后,B 到底跑完了没有?是失败了还是在等输入?调用方完全没有概念,只能靠超时猜。第三类,能力发现靠人肉维护。你要知道“哪个 Agent 能干什么”,只能去翻同事的文档,或者直接问。Agent 数量一多,这种维护就成了灾难。

A2A 针对这三个问题,分别给出了对应方案:统一 JSON-RPC 风格的消息格式、标准化的 Task 生命周期、以及让 Agent 自己“介绍自己”的 Agent Card。这也是为什么我建议做 Agent 项目的人从一开始就考虑它,而不是等系统长成毛线团再重构。

1.3 A2A 到底解决什么问题,不解决什么问题

A2A 不是来替代大模型推理的,也不负责让 Agent 变聪明。它专注的只有一件事:不同 Agent 之间的互操作(interop)。你可以把它理解成 Agent 世界里的 HTTP 协议——HTTP 本身不产生网页内容,但它让全世界的信息系统能连在一起。

有一点要特别说清:A2A 解决的是“Agent 和 Agent”之间的对话,不是“Agent 和工具”之间的对话。后者通常用 MCP(Model Context Protocol),也就是让 Agent 能够访问数据库、API、文件系统这些“手脚”的协议。A2A 和 MCP 不是二选一,而是配合使用,后面我会专门展开讲这个关系。

2. A2A 协议核心机制拆解

2.1 Agent Card:先说清“我是谁、我能干什么”

A2A 里最容易被忽略、但也最重要的概念是 Agent Card。它本质是一个 JSON 文件,通常放在 Agent 服务根路径下的.well-known/agent-card.json,用来描述这个 Agent 的基本信息:叫什么、干什么、怎么调用、需要什么认证。

为什么要设计这个玩意儿?因为你不能指望调用方提前知道每个 Agent 的接口细节。有了 Agent Card,调用方可以像查黄页一样发现一个 Agent,然后动态地决定怎么和它通信。这和微服务架构里的服务注册中心思路有点类似,只不过 A2A 把它做成了静态文件,实现更轻。

一个典型的中文 Agent Card 长这样:

{ "name": "order-agent", "description": "根据订单号查询订单状态、物流信息,支持导出订单明细", "url": "https://agent.example.com", "version": "1.0.0", "capabilities": { "streaming": true, "pushNotifications": false }, "skills": [ { "id": "order-lookup", "name": "订单查询", "description": "输入订单号,返回订单状态和物流轨迹" }, { "id": "order-export", "name": "订单导出", "description": "按时间范围导出订单明细表" } ], "authentication": { "schemes": [ { "type": "bearer", "bearerFormat": "JWT" } ] }, "defaultInputModes": ["text"], "defaultOutputModes": ["text"] }

注意,Agent Card 里的skills字段特别值得用起来。它不只是给人看的说明书,也是给调用方做“路由决策”的依据。A 要找一个“能导出订单”的 Agent,扫一圈 Agent Card 就能锁定目标。我在实际项目里把 Agent Card 做了动态生成,Agent 启动时根据自身注册的技能实时生成 JSON,这样技能更新了,对外的“名片”也自动更新。

2.2 Task 机制:把一次协作变成一个可追踪的任务

A2A 里,一次“让另一个 Agent 帮忙干活”的过程被抽象成了 Task。Task 的生命周期是 A2A 最核心的设计之一,它让调用方随时都知道“活儿干到哪一步了”。

官方定义的状态大约有六个:submitted(已提交)、working(执行中)、input-required(需要更多输入)、completed(完成)、failed(失败)、canceled(取消)。

我拿一个实际场景给你走一遍。你的主 Agent 把订单查询任务发给订单 Agent,这时候 Task 进入submitted;订单 Agent 开始查数据库,Task 变成working;如果订单号缺失,Task 变成input-required,主 Agent 需要再补一条消息;一切顺利的话,最终到completed,返回订单信息。

这个状态机最大的价值在于:它让“异步”变得可控。你可以轮询 Task 状态,也可以让 Agent 主动通知你。没有这套机制的话,两个 Agent 一旦有一个执行超过 5 秒,调用方就只能干瞪眼。

2.3 消息模型:协作是通过消息推动的

Task 是骨架,消息是血肉。A2A 里客户端和服务端通过互发消息来推进 Task,消息主要分为用户消息(UserMessage)和 Agent 消息(AgentMessage),每条消息有唯一的messageId、发送方角色(role)和具体内容(content)。

content本身是数组结构,可以被拆成多个部分。比如一条消息里并行发两段文本、一份文件引用,这些都是允许的。有人问我要不要自己设计一个“带元数据的消息头”,我通常说先把 A2A 原生的消息结构用好再说。它的metadata字段足够你塞业务信息了。

消息模型和 Task 模型注意区分:Task 是整个生命周期,消息是生命周期内的每一次往返。你可以把 Task 看成一场会议,消息就是会议上的一次次发言。

2.4 API 方法与传输层

A2A 在传输层复用 HTTP,在消息层采用 JSON-RPC 风格的方法调用。官方定义的几个核心方法,我用表格整理出来。

方法作用适用场景
task/send创建一个 Task 并发送第一条消息最常见的调用方式
message/send在既有 Task 上继续发消息补充信息、回答input-required
task/get查询 Task 当前状态和结果轮询状态时使用
task/cancel取消一个执行中的 Task发现任务不对时及时止损
task/pushNotificationConfig配置推送通知不想轮询时使用 webhook
task/resubscribe重新订阅事件流接收流式更新时配合使用

这些方法走的是标准的 JSON-RPC 2.0 格式。比如一次最简单的task/send请求是这样的:

{ "jsonrpc": "2.0", "id": 1, "method": "task/send", "params": { "taskId": "task-001", "message": { "role": "user", "messageId": "msg-001", "content": [ { "type": "text", "text": "请查询订单 ORD-2024-0001 的状态" } ] } } }

这个设计的好处是:任何语言、任何框架,只要你能起一个 HTTP 服务,就能实现一个 A2A 端点。你不需要引入重型的消息中间件,也不需要依赖某个特定厂商的 SDK。

另外提一句,A2A 支持streaming模式。调用方可以实时看到被调用 Agent 的中间输出,而不是死等一个最终结果。在长耗时任务里,这个体验非常关键。

3. 实操:从零跑通一个 A2A 调用

3.1 环境准备与目录结构

纸上谈兵到这里为止,下面我们直接动手。我选用 Python + FastAPI 来做演示,因为这套组合上手成本低,所见即所得。即使你平时用 Node 或 Go,思路也一样。

你需要准备的东西不多:

  • Python 3.10 以上
  • FastAPI 和 uvicorn:pip install fastapi uvicorn
  • 一个趁手的 HTTP 测试工具,比如 curl

完整目录结构可以做成这样:

a2a-demo/ ├── agent_card.json ├── server.py └── client.py

我多说一句为什么用这个结构:agent_card.json是给外部看的“名片”,server.py实现 Agent 的服务端逻辑,client.py模拟另一个 Agent 来调用它。三个文件各司其职,方便你后续扩展成真正的多 Agent 系统。

3.2 编写 Agent Card

在项目根目录下创建agent_card.json,这里我写一个简化版本。

{ "name": "weather-agent", "description": "提供城市天气查询服务,支持根据城市名获取当前温度、空气质量", "url": "http://localhost:8000", "version": "1.0.0", "capabilities": { "streaming": false, "pushNotifications": false }, "skills": [ { "id": "weather-query", "name": "天气查询", "description": "输入城市名,返回当前天气数据" } ], "defaultInputModes": ["text"], "defaultOutputModes": ["text"] }

这里的url字段要写你的 Agent 实际运行的地址,我演示时用http://localhost:8000。如果你的 Agent 部署在服务器上并配了域名,这里就要写可公网访问的地址。

3.3 实现 Agent 服务端

接下来是服务端。为了让你看清楚 A2A 的本质,我不引入任何重量级的官方 SDK,直接用 FastAPI 实现一个兼容 JSON-RPC 的端点。实际生产里你可以用官方发布的a2a-sdk来减少重复工作,但理解了下面这段代码,你就能看懂官方 SDK 背后在做什么。

# server.py import json import uuid from fastapi import FastAPI, Request from fastapi.responses import JSONResponse app = FastAPI() # 模拟一个天气数据源 def get_weather(city: str) -> dict: data = { "北京": {"temp": 18, "aqi": 55, "condition": "晴"}, "上海": {"temp": 22, "aqi": 42, "condition": "多云"}, "深圳": {"temp": 27, "aqi": 30, "condition": "阵雨"}, } return data.get(city, {"temp": None, "aqi": None, "condition": "未知"}) # 这个 Agent 支持的方法 AGENT_METHODS = { "task/send": "TASK_SEND", "message/send": "MESSAGE_SEND", "task/get": "TASK_GET", "task/cancel": "TASK_CANCEL", } @app.get("/.well-known/agent-card.json") async def agent_card(): with open("agent_card.json", "r", encoding="utf-8") as f: card = json.load(f) return JSONResponse(content=card) @app.post("/") async def handle_rpc(request: Request): payload = await request.json() method = payload.get("method", "") params = payload.get("params", {}) request_id = payload.get("id", 0) task_id = params.get("taskId", str(uuid.uuid4())) if method == "task/send": messages = params.get("message", {}) text = "".join( seg.get("text", "") for seg in messages.get("content", []) if seg.get("type") == "text" ) # 简单解析:从输入里扣出城市名 city = None for c in ["北京", "上海", "深圳", "广州"]: if c in text: city = c break if city is None: return { "jsonrpc": "2.0", "id": request_id, "result": { "taskId": task_id, "status": { "state": "input-required", "message": {"role": "agent", "messageId": str(uuid.uuid4()), "content": [{"type": "text", "text": "请提供城市名"}]} } } } weather = get_weather(city) if weather["temp"] is None: return { "jsonrpc": "2.0", "id": request_id, "result": { "taskId": task_id, "status": { "state": "failed", "message": {"role": "agent", "messageId": str(uuid.uuid4()), "content": [{"type": "text", "text": f"暂不支持查询 {city} 的天气"}]} } } } return { "jsonrpc": "2.0", "id": request_id, "result": { "taskId": task_id, "status": { "state": "completed", "message": { "role": "agent", "messageId": str(uuid.uuid4()), "content": [{"type": "text", "text": json.dumps(weather, ensure_ascii=False)}] } } } } # 其他方法可以按需实现,这里先返回未实现 return { "jsonrpc": "2.0", "id": request_id, "error": {"code": -32601, "message": f"Method {method} not implemented"} }

启动服务:

uvicorn server:app --host 0.0.0.0 --port 8000

这里我做了几处常见设计:先读 Agent Card,再用一个字典来路由方法。你如果想增加新能力,只需要在AGENT_METHODS里注册新方法,并且补充对应分支即可。实际项目中,分支里的逻辑通常不是这么简单的if-else,而是会连接大模型、调用外部 API 等,但外层这个协议壳是一样的。

3.4 客户端调用与验证

再写一个极简客户端脚本,模拟另一个 Agent 来调用天气 Agent。

# client.py import httpx BASE_URL = "http://localhost:8000" def send_task(task_text: str, task_id: str): payload = { "jsonrpc": "2.0", "id": 1, "method": "task/send", "params": { "taskId": task_id, "message": { "role": "user", "messageId": "msg-abc-123", "content": [{"type": "text", "text": task_text}] } } } resp = httpx.post(BASE_URL + "/", json=payload, timeout=10) return resp.json() if __name__ == "__main__": result = send_task("帮我查一下上海的天气", "task-001") print(result)

运行:

python client.py

正常的话,你会看到返回的state是completed,并且数据里带着上海的天气信息。如果你把输入改成不包含城市名的文本,会看到state变成input-required,服务端会提示你补城市名。

我把这个例子跑完后的心得是:A2A 的接入门槛真的不高,核心就是把 JSON-RPC 的壳子写好。你甚至可以把现有的内部 API 包一层这个协议,立刻就能被其他 Agent 发现和调用。这也是我觉得 A2A 能被广泛落地的原因——学习成本低,改造成本也可控。

4. A2A 和 MCP 怎么配合使用

4.1 MCP 管“工具”,A2A 管“同事”

很多第一次接触 A2A 的人都会问:它和 MCP 到底什么关系?是不是有 MCP 就够了?

MCP 是让 Agent 能调用工具的协议。你的 Agent 要读数据库,要调第三方 API,要在本地执行命令,这些都属于 MCP 的范畴。它解决的是“Agent 的手脚”问题。A2A 解决的是“Agent 的社交”问题——当你需要另一个 Agent 帮你干活,而不是直接调一个工具时,就需要 A2A。

我用一句话总结:MCP 是 Agent 访问世界资源的接口,A2A 是 Agent 与 Agent 之间沟通的接口。可以类比成 MCP 是“肢体”,A2A 是“语言”。

4.2 一条现实中的调用链

举一个实际的电商客服场景。用户问“我那个订单咋还没发货”,客服主 Agent 收到问题后,需要查订单系统、查物流系统、查售后规则。如果这些系统都是 Agent 形态,A2A 和 MCP 的配合是这样的:

主 Agent 先用 MCP 调用内部订单查询工具,拿到订单号;然后通过 A2A 把“查询物流轨迹”这个任务发给物流 Agent;物流 Agent 自己内部可能又用 MCP 去访问物流公司的 API;拿到结果后,通过 A2A 把结构化数据返回给主 Agent;主 Agent 组织语言,回复用户。

在这条链路里,每一个 Agent 内部的工具调用走 MCP,Agent 之间的协作走 A2A。层与层之间不互相干扰,哪个环节出了问题,也能通过日志快速定位。

我更推荐在项目初期就同时规划这两层。只上 MCP 不做 A2A,你的 Agent 可以很强,但彼此之间还是孤岛;只上 A2A 不用 MCP,你的 Agent 之间有话说,但真正办事时没手没脚。

4.3 A2A 在实际架构中的位置

从系统架构视角看,A2A 应该被放在 Agent 编排层之上。无论你用的是 LangGraph、AutoGen、CrewAI 还是自己手写的流程引擎,A2A 都可以作为一个统一的“对外出口”存在。

团队内部一个常见的做法是:每个业务 Agent 单独部署一个 HTTP 服务,暴露 A2A 端点,然后在编排层做一个 Agent 路由,根据任务类型自动选择调哪个 Agent。这个路由角色本身也可以是一个 Agent,它的判断依据有一部分就来自各 Agent 的 Agent Card。

比如你把现有几个 Agent 接进来,A2A 的 Agent Card 帮你在不知道“谁有这能力”的情况下找到正确目标,这对快速搭原型很有价值。

5. 落地时最容易踩的坑

5.1 Agent Card 路径写错

我见到的第一个高频错误,是把 Agent Card 放在/agent-card.json而不是/.well-known/agent-card.json。按规定应该放在.well-known目录下,这不是随便定的,是为了和现有网络标准保持一致,方便工具自动发现。多个 Agent 平台集成时,先按/.well-known/agent-card.json去找。你的文件放错地方,对方装作看不见,排查起来还以为是网络问题。

5.2 任务卡在working不返回

这个问题在刚开始做 A2A 时特别常见。客户端发了task/send,服务端也确实在处理,但因为业务逻辑阻塞,迟迟没有返回最终状态,调用方就一直干等。

解决思路是:如果你的任务是耗时的,不要在单个 HTTP 请求里同步跑完,而是先返回submitted或working,再通过后台任务慢慢执行,并提供一个可查询的端点让客户端来轮询或订阅。如果你不打算实现异步,那就要明确超时策略,避免连接吊死。

5.3 状态变成input-required后没人处理

A2A 允许 Agent 在执行过程中发现自己缺少关键信息,然后进入input-required状态。有些 Agent 没有处理这个分支,收到input-required就直接抛异常,或者干脆无视。

正确做法是:调用方遇到input-required时,应该回到“人机交互层”补充信息,然后再通过message/send把信息发给对方。如果你的 Agent 是编排给另一个 Agent 用的,你要在协议层就约定好“缺少信息时的重试策略”,不要让它变成一条死路。

5.4 轮询频率过于激进

Task 的异步模型让很多人倾向于高频轮询来拿结果。我在项目里见过每 200 毫秒就轮询一次的调用方,这种行为不仅浪费带宽,还可能把服务端打到限流。

更合理的做法是优先使用 A2A 的推送通知能力,也就是task/pushNotificationConfig。如果你的部署环境不支持推送,那就把轮询间隔控制在 2 到 5 秒之间,并且在拿到终态后立即停止轮询。

5.5 安全认证只做了一半

Agent Card 里可以声明认证方式,比如 Bearer Token 或者 OAuth。但声明归声明,服务器端如果没校验,等于没穿裤子出门。A2A 协议本身不强制定认证方式,落地时你要自己保证每个 Agent 端点都做身份验证和权限校验。

跨团队或跨公司调用的时候,我建议至少用 JWT 做身份标记,并配合现有的网关做传输层加密。千万别把 A2A 理解成“内网自己人随便调”,它只是定义通信格式,安全边界仍然需要你自己画。

5.6 可观测性没跟上

多 Agent 协作一旦跑起来,调用链就会变得非常长:主 Agent → 子 Agent → 子 Agent → 工具。没有可观测性工具,出了问题你连锅都找不到。我习惯在 A2A 的参数里带上taskId和messageId,并把这些 ID 透传到日志系统。这样一条请求从发出到返回的完整链路都能串起来。

如果预算允许,建议接入 LLM 可观测性平台,比如 Langfuse 这类工具,跟踪 Agent 之间的每一次 message 流转,能省去大量的排查时间。这个习惯越早培养越好,等系统复杂了再补,成本翻倍。

6. 写到最后:A2A 能给你的项目带来什么

如果你现在手头正在做 Agent 项目,我个人的建议是:别等“标准稳定了”再动手。A2A 的设计足够简单,你完全可以把它当作一套“Agent 互操作规范”来慢慢引入。先让两个 Agent 用 A2A 打通一条链路,再把 Agent Card 做好,最后逐步接入更多的 Agent。

我在实际项目里最直观的感受是:A2A 把以前“两个 Agent 之间需要专门写适配器”的问题,变成了“两边都遵守同一个协议”的问题。前者的维护成本是线性的,Agent 一多就爆炸;后者的成本是可控的,每新增一个 Agent,只需要做一次协议接入。

如果你看完这些还是不知道从哪里下手,给你一个最小的下一步:把你现在系统里最常被其他模块调用的那个 Agent,暴露一个 A2A 端点,写好 Agent Card,然后试着用另一个 Agent 调它一次。不用管工具链、不用管框架,先让两个 Agent 说上话。说上话了,后面的事就顺了。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询