标签：#AI Agent #A2A协议 #MCP #多智能体 #开源协议
数据口径：综合自 A2A 官方文档、Linux 基金会公告与 2026 年年中公开报道

2025 年 4 月，Google 在 Cloud Next 上发布了 A2A（Agent2Agent）协议。说白了，它的目的就一个：让不同厂商、不同框架写的 AI Agent，能像网站之间互相访问那样通信、协作。2025 年 6 月，Google 把 A2A 捐给了 Linux 基金会托管（同月 Anthropic 的 MCP 也一并移交），到现在已有 150 多家企业/组织加入。

写过 MCP 的应该知道，它是解决"Agent 怎么调工具"（Agent ↔ 工具）的；A2A 补的是另一半——“Agent 怎么找同事”（Agent ↔ Agent）。只懂 MCP 不懂 A2A，差不多就是只会自己干活、不会拉人协作。

这篇文章从协议定位、核心概念、通信模式讲起，然后带你把一个能跑起来的 A2A 例子搭出来，最后聊聊 MCP + A2A 双协议在企业里怎么配合。看完你就能判断：自己手头的 Agent 系统，到底要不要上 A2A。

适合读者：写过 MCP、正在做 Agent 的开发者；想上手 A2A 实战的；以及要给团队设计多 Agent 系统的架构师。

---

太长不看版

• MCP 管工具：标准化"Agent 如何调用工具和数据"，解决的是 Agent 与外部系统的连接
• A2A 管 Agent：标准化"Agent 如何调用另一个 Agent"，解决的是 Agent 与 Agent 的协作
• A2A 是 Google 2025 年 4 月发布，2025 年 6 月捐赠给 Linux 基金会治理，150+ 合作伙伴加入
• 五个核心概念：Agent Card（身份证）、Task（工单）、Message（聊天记录）、Part（内容胶囊）、Artifact（最终成果）
• 三种通信模式：轮询（Polling）、SSE 流式、Webhook 推送
• 实战路径：10 行代码写一个 A2A Server → 20 行代码写 Client → Agent Card 发现 → 多 Agent 协作
• 一句话选型：接工具用 MCP，找同事用 A2A；两者互补，企业级落地两个都要

---

目录

• 一、引言：MCP 解决了一半问题，剩下的一半是 A2A
• 二、MCP 与 A2A：一张图看懂"管工具"和"管 Agent"
• 三、A2A 的五个核心概念
• 四、三种通信模式怎么选
• 五、实战：10 分钟跑通你的第一个 A2A
• 六、进阶实战：让 Agent 调 Agent
• 七、MCP + A2A：企业级双协议架构怎么搭
• 八、避坑指南
• 九、总结

---

一、引言：MCP 解决了一半问题，剩下的一半是 A2A

2025 年 MCP 火起来的时候，很多人以为"Agent 互操作"问题已经解决了。但真上手做多 Agent 系统的人会立刻撞上一堵墙：

我自己的 Agent 能调用工具了，但我怎么让它调用"别人家的 Agent"？

举个真实的业务场景。你要做一个"企业采购审批"系统，拆成三个 Agent：

1. 采购 Agent：对接 ERP，查库存、下采购单
2. 财务 Agent：对接财务系统，查预算、做付款
3. 风控 Agent：对接风控库，做供应商合规校验

用 MCP，每个 Agent 都能接到自己的系统上——但这三个 Agent 之间的"交接"怎么办？采购 Agent 查完库存，怎么把结果"交给"财务 Agent？财务 Agent 想请风控 Agent 复核，怎么发起？

回到只有 MCP 的时候，你大概率只能在主 Agent 里写死对子 Agent 的调用逻辑：每个子 Agent 的接口格式、鉴权方式、返回结构都各写各的，加一个子 Agent 就得适配一遍。说白了，就是 Function Calling 时代"工具写死"的痛，换了个地方又疼了一遍——从工具层挪到了 Agent 层。

A2A 就是冲着这堵墙来的。它把"Agent 调用 Agent"这件事标准化了：

• 每个 Agent 有一张 Agent Card（身份证），告诉别人"我是谁、能干啥、怎么找我"
• Agent 之间通过 Task（工单） 语义协作，而不是各家自定义的接口
• 任务结果统一返回 Artifact（成果物），文本、文件、结构化数据都能传

打个比方：MCP 像 USB 接口——让所有工具"插上就能用"；A2A 像 HTTP 协议——让所有 Agent"连上就能聊"。USB 解决"设备怎么连电脑"，HTTP 解决"电脑之间怎么通信"，两者根本不冲突，是不同层次的事。

---

二、MCP 与 A2A：一张图看懂"管工具"和"管 Agent"

先用一张分层图理解两者的位置：

┌────────────────────────────────────────────────────────────┐
│                   Agent ↔ Agent（A2A 的地盘）                 │
│                                                            │
│     采购 Agent ── A2A ──▶ 财务 Agent                        │
│        │                    │                              │
│        │ A2A                │ A2A                          │
│        ▼                    ▼                              │
│     风控 Agent ── A2A ──▶ 审计 Agent                        │
└────────────────────────────────────────────────────────────┘
                         │
                         ▼
┌────────────────────────────────────────────────────────────┐
│                   Agent ↔ 工具（MCP 的地盘）                  │
│                                                            │
│     ERP 系统 MCP Server   财务系统 MCP Server   风控库 MCP    │
└────────────────────────────────────────────────────────────┘

• MCP：纵向连接，解决"Agent 怎么拿到能力"（数据、工具、系统）
• A2A：横向连接，解决"Agent 怎么找人协作"（任务委托、结果交接）

对比表：

对比维度	MCP	A2A
全称	Model Context Protocol	Agent2Agent Protocol
连接对象	Agent ↔ 工具 / 数据源	Agent ↔ Agent
定位	模型上下文与工具接入标准	智能体间任务协作标准
发起方	Anthropic（2024-11）	Google（2025-04）
治理	Linux 基金会（2025 移交）	Linux 基金会（2025-06 移交）
核心语义	Tools / Resources / Prompts	Agent Card / Task / Message / Part / Artifact
通信基础	stdio / Streamable HTTP	HTTP + JSON-RPC 2.0 + SSE / Webhook
类比	USB 接口（设备插上就能用）	HTTP 协议（主机之间能通信）
典型场景	Agent 查数据库、操作浏览器、调 GitHub	简历 Agent 把候选人交给面试 Agent

补充一句：A2A 不替代 MCP，也不依赖 MCP，两者互不干扰。一个 Agent 完全可以一边用 MCP 接工具，一边用 A2A 找同事。企业里比较顺手的搭法，是外层用 A2A 做协作、内层用 MCP 接系统。

---

三、A2A 的五个核心概念

A2A 的协议模型非常薄，一共就五个概念，记住它们协议就懂了一半。

3.1 Agent Card：Agent 的"身份证"

每个 A2A 兼容的 Agent 都在 /.well-known/agent.json 上挂一张"名片"，用 JSON 描述自己的身份和能力，别人通过它完成发现（Discovery）。

字段	作用
name / description	我是谁、我能干什么
url	怎么调用我（A2A 端点地址）
capabilities	支持流式（streaming）？支持推送（push）？
skills	技能清单：id、name、description、tags
defaultInputModes / defaultOutputModes	默认接受的输入/输出格式（text / file / data）

这个思路其实不新鲜：Web 服务早就习惯用 /.well-known/ 目录做自动发现，比如 OAuth 2.0 的 /.well-known/openid-configuration。A2A 只是把这个惯例搬了过来，约定每个 Agent 把名片放在 /.well-known/agent.json。调用方先读名片，再决定要不要把活派给它——发现和调用分开，设计上很省事。

3.2 Task：Agent 之间的"工单"

A2A 的协作语义不是"远程调用函数"，而是派发任务。技术上，A2A 的消息基于 JSON-RPC 2.0 规范在 HTTP 上传输；一次协作就是一个 Task，每个 Task 以 task_id 唯一标识，生命周期用状态机描述：

submitted（已提交）
   │
   ▼
working（执行中） ──────────▶ completed（完成）
   │        │                    │
   │        │                    ├── canceled（取消）
   │        │                    └── failed（失败）
   │        └──▶ input-required（等待补充输入）──▶ 回到 working
   └──（可选中断）──▶ 凭 task_id 恢复执行（resume）

任务可以异步执行——发起方提交 Task 后可以去干别的，再凭 task_id 回来查状态、甚至在中断后恢复。这跟"函数调用必须同步等结果"完全不同，是 A2A 能支撑长任务的关键。

3.3 Message 与 Part：聊天记录与内容胶囊

• Message：一条消息，带角色（user / agent / system），是 Task 流转过程中的对话记录
• Part：消息内容的"胶囊"，有 TextPart（文本）、FilePart（文件）、DataPart（结构化 JSON 数据）三种类型，支持多模态

3.4 Artifact：任务成果物

Task 干完活的最终产出，是一个或多个 Part 的集合——可以是文本报告、生成的文件，也可以是结构化数据。调用方拿到 Artifact 就等于拿到了"交付物"。

把这五个概念串一遍，就是一次完整的协作：调用方通过 Agent Card 找到对方 → 发起 Task → 过程中用 Message（含 Part）交换信息 → 最后拿到 Artifact 作为成果。

---

四、三种通信模式怎么选

A2A 的"结果获取"有两种独立机制：响应方式（同步等待 / 轮询）和推送机制（SSE 流式 / Webhook 推送通知），组合起来覆盖三种常见模式：

模式	工作方式	适用场景	优点	缺点
同步等待 / 轮询	send_message 可带 awaitResponse 同步等结果；或先提交 Task，再隔段时间 get_task() 查状态	短任务、实现最简单	兼容性最好，任何 HTTP 客户端都能做	同步等待会占连接；轮询有延迟、空轮询浪费资源
SSE 流式	Server 通过 Server-Sent Events 持续推送 TaskUpdateEvent，实时下发状态与进度	长任务、要实时进度（如"正在生成报告 50%"）	实时、单向简单、基于标准 HTTP	单向，客户端无法中途插话
Webhook 推送通知	Server 在任务状态变化时，主动 POST 到客户端注册的 callback_url	超长任务、客户端不在线等场景	客户端无需挂在线，最省资源	需要公网回调地址，要处理鉴权、重试与幂等

有个细节容易搞混：SSE 其实出现在两个地方。一个是任务执行过程中的"流式响应"，实时把状态和进度推给调用方；另一个是"推送通知"的其中一种模式（取值 webhook / sse / none）。Agent 支持哪些能力，都写在它的 capabilities 里，调用方读名片就知道了。

实战建议：从轮询开始，跑通再说。需要进度展示时升级 SSE；任务长到客户端等不住时再上 Webhook。不要一上来就三件套。

---

五、实战：10 分钟跑通你的第一个 A2A

5.1 环境准备

# 官方 A2A Python SDK + FastAPI + Uvicorn
pip install a2a-sdk "fastapi[standard]" uvicorn

5.2 写一个 A2A Server（Agent 服务端）

新建 weather_agent.py——一个"天气查询 Agent"，对外暴露 A2A 标准接口：

# weather_agent.py —— 用官方 a2a-sdk 写的 A2A Server
from fastapi import FastAPI
from a2a.server import A2AServerFramework, InMemorySessionManager
from a2a.types import (
    AgentCard,
    AgentCapabilities,
    AgentRequest,
    AgentSkill,
    Artifact,
    Message,
    TextPart,
    Task,
    TaskState,
    TaskStatus,
)

app = FastAPI()


class WeatherAgent(A2AServerFramework):
    """一个"天气查询 Agent"，对外提供 A2A 标准接口"""

    def __init__(self):
        super().__init__(
            session_manager=InMemorySessionManager(),
            agent_card=AgentCard(
                name="weather-agent",
                description="查询指定城市实时天气的 Agent",
                url="http://localhost:8000/",
                version="1.0.0",
                capabilities=AgentCapabilities(
                    streaming=True,
                    push_notifications=False,
                ),
                skills=[
                    AgentSkill(
                        id="weather",
                        name="天气查询",
                        description="输入城市名，返回该城市今天的天气",
                    ),
                ],
            ),
        )

    async def process_agent_request(self, request: AgentRequest) -> Task:
        # 1. 从请求消息里取出文本内容（Part 是内容胶囊）
        texts = [p.text for p in request.message.parts if isinstance(p, TextPart)]
        city = texts[0].strip() if texts else "北京"

        # 2. 执行业务逻辑（真实项目里这里调天气 API / 数据库）
        weather_map = {
            "北京": "晴，气温 26℃",
            "上海": "多云，气温 28℃",
            "深圳": "阵雨，气温 30℃",
        }
        result = f"{city} 今天{weather_map.get(city, '晴，气温 25℃')}，微风"

        # 3. 返回任务结果：Task（状态） + Artifact（成果，一个或多个 Part 的集合）
        return Task(
            id=request.task_id,
            status=TaskStatus(
                state=TaskState.COMPLETED,
                message=Message(role="agent", parts=[TextPart(text=result)]),
            ),
            artifacts=[Artifact(parts=[TextPart(text=result)])],
        )


agent = WeatherAgent()
app.include_router(agent.get_router())

注意，这里没有自定义鉴权、没有自定义参数格式、没有自定义返回结构。函数签名、技能描述、调用方式，全凭 Agent Card 和协议自己声明。对比一下在代码里写死对接逻辑的做法——换一家 Agent 就得重写一遍，这就是"协议化"和"写死接口"的差别。

启动：

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

5.3 写一个 A2A Client（调用方）

新建 weather_client.py：

# weather_client.py —— A2A Client（调用方）
import asyncio
from a2a.client import A2AClient
from a2a.types import Message, TaskState, TextPart


async def main():
    # 1. 连接 A2A Server（注意：SDK 构造参数是 url）
    client = A2AClient(url="http://localhost:8000/")

    # 2. 给 Agent 发任务（A2A 的语义是"派发 Task"，不是"调用函数"）
    response = await client.send_message(
        session_id="session-001",
        message=Message(role="user", parts=[TextPart(text="北京今天天气怎么样？")]),
    )
    task = response.result

    # 3. 读取任务状态与成果
    print("任务状态:", task.status.state)
    if task.status.state == TaskState.COMPLETED:
        for artifact in task.artifacts:
            for part in artifact.parts:
                if isinstance(part, TextPart):
                    print("Agent 回复:", part.text)


# 长任务场景：提交后轮询，而不是同步等
async def poll_example():
    client = A2AClient(url="http://localhost:8000/")
    response = await client.send_message(
        session_id="session-002",
        message=Message(role="user", parts=[TextPart(text="上海天气")]),
    )
    task_id = response.result.id

    for _ in range(10):  # 最多轮询 10 次
        task = await client.get_task(task_id=task_id, session_id="session-002")
        state = task.status.state
        print("当前状态:", state)
        if state in (TaskState.COMPLETED, TaskState.FAILED, TaskState.CANCELED):
            break
        await asyncio.sleep(1)


if __name__ == "__main__":
    asyncio.run(main())

运行：

python weather_client.py

预期输出：

任务状态: TaskState.COMPLETED
Agent 回复: 北京 今天晴，气温 26℃，微风

5.4 Agent Card 发现机制

A2A 的"找 Agent"靠标准路径 /.well-known/agent.json。启动 Server 后直接访问：

curl http://localhost:8000/.well-known/agent.json

返回的就是这张"身份证"（字段与 3.1 节对应）：

{
  "name": "weather-agent",
  "description": "查询指定城市实时天气的 Agent",
  "url": "http://localhost:8000/",
  "version": "1.0.0",
  "capabilities": {"streaming": true, "pushNotifications": false},
  "skills": [
    {"id": "weather", "name": "天气查询", "description": "输入城市名，返回该城市今天的天气"}
  ]
}

命名说明：协议线格式统一用驼峰命名（pushNotifications）；Python SDK 内部用 snake_case（push_notifications），由 SDK 自动完成转换。

翻译成大白话：只要是支持 A2A 的客户端，拿到这个 URL 就能自动摸清这个 Agent 的本事，然后直接给它派活，接口文档都可以省了。这也是这两年 Agent 生态能"即插即用"的原因之一。

---

六、进阶实战：让 Agent 调 Agent

第五节的 Client 是人写的，现在让 Agent 调 Agent。场景：一个"简历初筛 Agent"收到候选人简历后，通过 A2A 把候选人派给"面试 Agent"生成面试题。

# recruiter_agent.py —— 主 Agent 内部通过 A2AClient 调用子 Agent
import asyncio
from fastapi import FastAPI
from a2a.server import A2AServerFramework, InMemorySessionManager
from a2a.client import A2AClient
from a2a.types import (
    AgentCard, AgentCapabilities, AgentRequest, AgentSkill,
    Artifact, Message, TextPart, Task, TaskState, TaskStatus,
)

app = FastAPI()

# 面试 Agent 的 A2A 地址（另一个独立部署的服务）
INTERVIEW_AGENT_URL = "http://localhost:8001/"


class RecruiterAgent(A2AServerFramework):
    """简历初筛 Agent：先自评，再把候选人派给面试 Agent"""

    def __init__(self):
        super().__init__(
            session_manager=InMemorySessionManager(),
            agent_card=AgentCard(
                name="recruiter-agent",
                description="简历初筛与面试派发 Agent",
                url="http://localhost:9000/",
                version="1.0.0",
                capabilities=AgentCapabilities(streaming=False, push_notifications=False),
                skills=[AgentSkill(id="screen", name="简历筛选", description="评估简历并派发给面试官")],
            ),
        )

    async def process_agent_request(self, request: AgentRequest) -> Task:
        # 1. 解析候选人信息
        texts = [p.text for p in request.message.parts if isinstance(p, TextPart)]
        candidate = texts[0] if texts else ""

        # 2. 本 Agent 先做初筛判断（简化演示）
        if "算法" not in candidate:
            return Task(
                id=request.task_id,
                status=TaskStatus(
                    state=TaskState.COMPLETED,
                    message=Message(role="agent", parts=[TextPart(text="候选人无算法背景，直接淘汰")]),
                ),
                artifacts=[Artifact(parts=[TextPart(text="淘汰")])],
            )

        # 3. 通过 A2A 调用"面试 Agent"，把候选人派过去 —— Agent 调 Agent 的关键一步
        interview_client = A2AClient(url=INTERVIEW_AGENT_URL)
        resp = await interview_client.send_message(
            session_id=f"interview-{request.task_id}",
            message=Message(role="user", parts=[TextPart(text=candidate)]),
        )
        interview_task = resp.result

        # 4. 从子 Agent 的成果（Artifact）里提取回复文本
        interview_feedback = "（无返回）"
        if interview_task.artifacts:
            first_part = interview_task.artifacts[0].parts[0]
            if isinstance(first_part, TextPart):
                interview_feedback = first_part.text

        # 5. 汇总结果返回给最上层调用方
        final = f"初筛通过，面试 Agent 反馈：{interview_feedback}"
        return Task(
            id=request.task_id,
            status=TaskStatus(
                state=TaskState.COMPLETED,
                message=Message(role="agent", parts=[TextPart(text=final)]),
            ),
            artifacts=[Artifact(parts=[TextPart(text=final)])],
        )


agent = RecruiterAgent()
app.include_router(agent.get_router())

跑通这条链的关键点：

1. 子 Agent 是独立服务（:8001），有自己的 Agent Card，随时可以被替换成别的团队/厂商的 Agent
2. 主 Agent 只做编排：A2AClient.send_message() 一行完成"派活"，不关心子 Agent 内部实现
3. 接口是协议定的，不是代码定的：换掉面试 Agent，主 Agent 一行代码不用改

说到底，A2A 想做的事情，就是让多 Agent 之间的配合不再靠写死代码，而是靠一套大家都认的协议。

---

七、MCP + A2A：企业级双协议架构怎么搭

放到企业里落地，比较顺的搭法是分两层：

┌────────────────────────────────────────────────────────────┐
│                 用户 / 业务系统（门户、审批流、IM）             │
└───────────────────────────┬────────────────────────────────┘
                            ▼
┌────────────────────────────────────────────────────────────┐
│                 编排层：主 Agent（规划 + 路由）                │
│         LangGraph / Google ADK / OpenAI Agents SDK          │
│         —— 负责拆任务、定路由、管会话状态                      │
└───┬───────────────┬───────────────┬────────────────────────┘
    │ A2A           │ A2A           │ A2A       （Agent ↔ Agent）
    ▼               ▼               ▼
┌──────────┐   ┌──────────┐   ┌──────────┐
│ 采购 Agent│   │ 财务 Agent│   │ 风控 Agent│
└─────┬────┘   └─────┬────┘   └─────┬────┘
      │ MCP         │ MCP          │ MCP      （Agent ↔ 工具）
      ▼             ▼              ▼
 ERP 系统 MCP   财务系统 MCP    风控库 MCP
（查库存/下单）  （查预算/付款）  （合规校验）

说白了：主 Agent 用 A2A 给专业 Agent 派活、收成果；每个专业 Agent 再用 MCP 接自己那套系统。一个管横向协作，一个管纵向接入，各干各的。

场景	用哪个	原因
Agent 查数据库、调浏览器、操作业务系统	MCP	接工具/数据是 MCP 的本职
Agent 之间交接任务、跨团队协作	A2A	找"同事"是 A2A 的本职
单 Agent + 多个工具（一个系统内的 Agent）	只用 MCP	没有协作需求就不引 A2A
多 Agent 跨部门/跨厂商（采购→财务→风控）	MCP + A2A 都要	内层接工具、外层找人，缺一不可
想把 Agent 能力开放给外部合作伙伴	A2A	Agent Card 发现 + OAuth，天然适合对外开放

现在的趋势也挺明显：Google 的 ADK 直接原生支持 A2A；OpenAI 的 Agents SDK 走的是功能对等的"手递手（handoffs）"、把 Agent 当工具用；LangChain/LangGraph 一边做 MCP 适配器，一边也在跟进 A2A。路线虽有不同，"协作要协议化"是共识。对开发者来说，先会 MCP，再补 A2A，基本就够用了。

---

八、避坑指南

1. 别把 A2A 当 MCP 的替代品：一个管"Agent 与工具"，一个管"Agent 与 Agent"，硬要二选一会把架构做残
2. Task 是异步的，别在同步请求里等长任务：提交 Task 后要按"状态机"处理——短任务同步等、中长任务轮询、要进度的用 SSE 流式、等不住的上 Webhook，别所有任务都指望一次返回
3. Webhook 别裸奔：启用推送通知前，需在 Agent Card 的 capabilities 中声明 push_notifications 并注册 callback_url；回调接口要鉴权（OAuth 2.0 / 签名），并处理重试与幂等，否则一次失败就丢任务
4. Agent Card 的 URL 必须可发现：内网 IP、临时地址会导致别人永远"找不到"你的 Agent，生产环境用稳定的域名 + HTTPS
5. 调用链变长，故障面变大：Agent 调 Agent 再调 Agent，任何一环超时都会放大。给每一层都设超时和重试，并保留 Task 状态日志方便排查
6. 别把 A2A 当 RPC 用：它是"任务委托"语义，适合"把活派给别人"，不适合高频低延迟的细粒度调用——那种场景走内部函数调用更合适
7. 先用轮询跑通：三分钟能跑通的东西，别为了"高级"一上来就上 Webhook，那是自找麻烦

---

九、总结

1. Agent 互操作是三件套：Function Calling 管应用内部、MCP 接工具、A2A 接 Agent。三层职责不一样，串起来就是从"自己干活"到"找人协作"
2. A2A 的底子很薄：Agent Card / Task / Message / Part / Artifact 五个概念，把 Agent 之间的协作从"写死代码"变成"认协议"，这才是生态级互操作该有的样子
3. 生态在实打实推进：Linux 基金会托管、150+ 合作伙伴、主流框架陆续支持，现在学不算晚
4. 上手不复杂：10 行代码起一个 Server，20 行写个 Client，接着搞定 Agent Card 发现，再让 Agent 调 Agent，最后用 LangGraph/ADK 编排成完整架构
5. 学习顺序建议先 MCP 再 A2A：连工具都接不好，找人协作也是白搭；两个通了，Agent 开发的基本功就齐了

---

参考资料

• A2A 协议官方站点
• A2A 协议规范（Agent2Agent Protocol Specification）
• Google 发布 A2A 协议公告（2025-04，Google Cloud Next）
• Linux 基金会成立 Agent2Agent 协议项目公告（2025-06）
• A2A 官方 GitHub 仓库（含 Python/JS SDK 与示例）
• A2A Python SDK 文档
• Google Agent 开发套件 ADK（与 A2A 深度集成）