$catMANUAL||~40 min

A2A 协议实战：当 AI Agent 开始互相说话，事情变得有趣了

advertisement

A2A 协议实战：当 AI Agent 开始互相说话，事情变得有趣了

上个月写了篇 MCP 到底是什么？一个全栈开发者的实战理解，讲的是 AI Agent 怎么调用外部工具。当时就有人在评论区问：那 Agent 之间能不能互相调用？比如我有一个写代码的 Agent，还有一个做测试的 Agent，能不能让它们自己协作，不用我在中间当传话筒？

说实话，当时我对这块了解也不深。后来翻了一圈资料，发现 Google 在 2025 年 4 月搞了个 A2A（Agent-to-Agent）协议，专门解决这个问题。折腾了一阵子，踩了不少坑，今天来聊聊这个东西。

先搞清楚一个问题：A2A 和 MCP 到底啥关系？

这是最多人问的，也是最容易搞混的。

MCP（Model Context Protocol）解决的是 Agent 怎么调用工具。你的 Agent 想查数据库、想调 API、想读文件，这些都走 MCP。Agent 是主人，工具是仆人，关系很清楚。

A2A 解决的是 Agent 怎么跟 Agent 说话。你的 Agent 不是去调一个工具，而是去跟另一个同样有智能的 Agent 协作。两个 Agent 是平等的，互相不知道对方内部怎么实现的，只通过一个标准接口沟通。

用一个不太恰当但好理解的比喻：MCP 是你去餐厅点菜（你告诉服务员你要什么），A2A 是两个厨师在厨房商量怎么配合出一桌宴席（各自有各自的专长，互相协调）。

它们不是竞争关系，是互补的。 一个 Agent 可以同时用 MCP 调用工具，又用 A2A 跟其他 Agent 协作。

A2A 的核心概念

A2A 协议 1.0 版本在 2025 年正式发布，基于 JSON-RPC 2.0 over HTTP(S)，设计得很简洁。核心概念就几个：

Agent Card

这是 A2A 里最重要的概念之一。每个 A2A Server 都要发布一个 Agent Card，相当于自我介绍。里面写了：

  • 你是谁（名字、描述）
  • 你会干什么（skills 列表）
  • 你的服务地址在哪
  • 你需要什么认证方式

格式是 JSON，通常放在 /.well-known/agent.json 这个路径下。其他 Agent 想跟你协作，先去读你的 Agent Card，看看你会什么，再决定要不要找你。

json
1
{
2
  "name": "CodeReviewAgent",
3
  "description": "专门做代码审查的 Agent，支持 Python、JavaScript、Go",
4
  "url": "https://code-review-agent.example.com",
5
  "version": "1.0.0",
6
  "skills": [
7
    {
8
      "id": "review-python",
9
      "name": "Python Code Review",
10
      "description": "审查 Python 代码的质量、安全性和性能"
11
    },
12
    {
13
      "id": "review-js",
14
      "name": "JavaScript Code Review",
15
      "description": "审查 JavaScript/TypeScript 代码"
16
    }
17
  ],
18
  "capabilities": {
19
    "streaming": true,
20
    "pushNotifications": false
21
  }
22
}

Task（任务）

A2A 里的核心工作单元。你给另一个 Agent 发一个消息，它会创建一个 Task 来跟踪这个请求。Task 有自己的生命周期：

  • submitted → 刚提交
  • working → 正在处理
  • input-required → 需要更多信息
  • completed → 完成了
  • failed → 失败了
  • canceled → 取消了

这个设计很聪明。因为 Agent 之间的协作可能是长时间运行的，不像调 API 那样一个请求一个响应。Task 让你可以异步地跟踪进度。

Message 和 Part

Agent 之间的通信单元。一个 Message 有一个 role（user 或者 agent），里面包含一个或多个 Part。

Part 是最小的内容单元，有三种类型：

  • TextPart：纯文本
  • FilePart：文件（可以是 URL 引用或者内联的 base64）
  • DataPart：结构化 JSON 数据

这种设计让 A2A 不只是传文字，还能传文件、传结构化数据，很灵活。

Artifact（产物）

Agent 完成任务后输出的东西。比如你让一个 Agent 写代码，它的 Artifact 可能是一个代码文件。让另一个 Agent 做分析，Artifact 可能是一个报告。

动手搞一个 A2A Server

说了这么多概念，不如直接上手。我用 Python SDK 来搭一个简单的 A2A Server。

环境准备

bash
1
# 创建虚拟环境
2
python3 -m venv a2a-env
3
source a2a-env/bin/activate
4
5
# 安装 SDK
6
pip install a2a-sdk

第一次装的时候我踩了个坑——a2a-sdk 需要 Python 3.10+，我服务器上默认是 3.9，直接报语法错误。后来用 pyenv 切到 3.11 才搞定。

写一个简单的 Agent Server

python
1
from a2a.server.apps import A2AStarletteApplication
2
from a2a.server.request_handlers import DefaultRequestHandler
3
from a2a.server.tasks import InMemoryTaskStore
4
from a2a.types import (
5
    AgentCard,
6
    AgentSkill,
7
    AgentCapabilities,
8
    TaskState,
9
    Message,
10
    TextPart,
11
    Artifact,
12
)
13
import uvicorn
14
15
# 定义 Agent 的能力
16
class MyAgent:
17
    """一个简单的翻译 Agent"""
18
19
    async def process_message(self, message: Message) -> dict:
20
        # 从 message 中提取文本
21
        text = ""
22
        for part in message.parts:
23
            if part.root.kind == "text":
24
                text = part.root.text
25
26
        # 简单的翻译逻辑（实际项目中会调用 LLM）
27
        if "hello" in text.lower():
28
            response_text = "你好！这是翻译 Agent 的回复。"
29
        elif "谢谢" in text:
30
            response_text = "You're welcome!"
31
        else:
32
            response_text = f"收到你的消息：{text}（翻译功能开发中...）"
33
34
        return {
35
            "status": TaskState.completed,
36
            "artifacts": [
37
                Artifact(
38
                    parts=[TextPart(text=response_text)],
39
                    name="translation",
40
                )
41
            ],
42
        }
43
44
# 创建 Agent Card
45
agent_card = AgentCard(
46
    name="TranslationAgent",
47
    description="一个简单的翻译 Agent，支持中英互译",
48
    url="http://localhost:8000",
49
    version="1.0.0",
50
    skills=[
51
        AgentSkill(
52
            id="translate-zh-en",
53
            name="中英翻译",
54
            description="将中文翻译成英文，或将英文翻译成中文",
55
        )
56
    ],
57
    capabilities=AgentCapabilities(streaming=False),
58
)
59
60
# 设置请求处理器
61
agent = MyAgent()
62
task_store = InMemoryTaskStore()
63
handler = DefaultRequestHandler(
64
    agent_card=agent_card,
65
    task_store=task_store,
66
)
67
68
# 创建 Starlette 应用
69
app = A2AStarletteApplication(
70
    agent_card=agent_card,
71
    request_handler=handler,
72
)
73
74
if __name__ == "__main__":
75
    uvicorn.run(app.build(), host="0.0.0.0", port=8000)

启动服务

bash
1
python server.py

正常的话你会看到 uvicorn 的启动日志，监听在 8000 端口。

这时候你可以用 curl 测试一下 Agent Card 是否正常：

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

应该返回你定义的那个 JSON。

写一个 A2A Client 来调用

Server 搭好了，现在写个 Client 来调用它：

python
1
import asyncio
2
from a2a.client import A2AClient
3
from a2a.types import (
4
    Message,
5
    TextPart,
6
    SendMessageRequest,
7
    MessageSendParams,
8
)
9
10
async def main():
11
    # 连接到 A2A Server
12
    client = A2AClient(url="http://localhost:8000")
13
14
    # 先看看对方的 Agent Card
15
    card = await client.get_agent_card()
16
    print(f"连接到: {card.name}")
17
    print(f"描述: {card.description}")
18
    print(f"技能: {[s.name for s in card.skills]}")
19
20
    # 发送一条消息
21
    message = Message(
22
        role="user",
23
        parts=[TextPart(text="Hello, how are you?")],
24
    )
25
26
    request = SendMessageRequest(
27
        params=MessageSendParams(message=message)
28
    )
29
30
    response = await client.send_message(request)
31
32
    # 处理响应
33
    if hasattr(response.root.result, 'artifacts'):
34
        for artifact in response.root.result.artifacts:
35
            for part in artifact.parts:
36
                if part.root.kind == "text":
37
                    print(f"翻译结果: {part.root.text}")
38
39
if __name__ == "__main__":
40
    asyncio.run(main())

跑起来之后，Client 会先去读 Server 的 Agent Card，然后发一条消息过去，Server 处理后返回翻译结果。

第一次跑的时候我翻车了。 报了个 ConnectionRefusedError，查了半天发现是 Server 那边启动失败了——端口被另一个进程占了。换了个端口就好了。这种小坑在折腾新工具的时候太常见了。

来点高级的：流式响应

上面的例子是同步的——发一个请求，等完整响应。但很多时候 Agent 处理任务需要很长时间，你不可能一直等着。A2A 支持流式响应（SSE），可以实时看到进度。

Server 端改一下，启用 streaming：

python
1
from a2a.server.apps import A2AStarletteApplication
2
from a2a.server.request_handlers import DefaultRequestHandler
3
from a2a.server.tasks import InMemoryTaskStore
4
from a2a.types import (
5
    AgentCard,
6
    AgentSkill,
7
    AgentCapabilities,
8
    TaskState,
9
    Message,
10
    TextPart,
11
    Artifact,
12
    TaskStatusUpdateEvent,
13
    TaskArtifactUpdateEvent,
14
)
15
import uvicorn
16
import asyncio
17
18
class StreamingAgent:
19
    """支持流式响应的 Agent"""
20
21
    async def process_message_stream(self, message, task_id):
22
        """流式处理消息"""
23
        text = ""
24
        for part in message.parts:
25
            if part.root.kind == "text":
26
                text = part.root.text
27
28
        # 模拟分步处理
29
        steps = [
30
            f"正在分析输入: {text[:20]}...",
31
            "正在调用翻译引擎...",
32
            "正在校对结果...",
33
            "翻译完成！",
34
        ]
35
36
        for i, step in enumerate(steps):
37
            yield TaskStatusUpdateEvent(
38
                task_id=task_id,
39
                status=TaskState.working,
40
                message=Message(
41
                    role="agent",
42
                    parts=[TextPart(text=step)],
43
                ),
44
            )
45
            await asyncio.sleep(0.5)  # 模拟耗时
46
47
        # 最终结果
48
        yield TaskArtifactUpdateEvent(
49
            task_id=task_id,
50
            artifact=Artifact(
51
                parts=[TextPart(text=f"翻译结果: {text} -> [翻译后的文本]")],
52
                name="translation",
53
            ),
54
        )
55
56
        yield TaskStatusUpdateEvent(
57
            task_id=task_id,
58
            status=TaskState.completed,
59
        )

Client 端用流式接收：

python
1
async def stream_example():
2
    client = A2AClient(url="http://localhost:8000")
3
4
    message = Message(
5
        role="user",
6
        parts=[TextPart(text="这是一段很长的文本需要翻译...")],
7
    )
8
9
    request = SendMessageRequest(
10
        params=MessageSendParams(message=message)
11
    )
12
13
    # 流式接收
14
    async for event in client.send_message_stream(request):
15
        if hasattr(event.root, 'status'):
16
            print(f"状态: {event.root.status}")
17
        if hasattr(event.root, 'artifact'):
18
            for part in event.root.artifact.parts:
19
                if part.root.kind == "text":
20
                    print(f"结果: {part.root.text}")

流式的好处是用户体验好，你能实时看到 Agent 在干什么。缺点是实现起来比同步复杂不少，而且调试的时候 SSE 的日志不太好追踪。

多 Agent 协作：这才是 A2A 的真正威力

单个 Agent 的例子其实用 REST API 也能搞，A2A 的真正价值在于多 Agent 协作。

想象一个场景：你有一个 需求分析 Agent、一个 代码生成 Agent、一个 测试 Agent。用户提一个需求，三个 Agent 自动协作完成。

python
1
import asyncio
2
from a2a.client import A2AClient
3
from a2a.types import Message, TextPart, SendMessageRequest, MessageSendParams
4
5
class Orchestrator:
6
    """多 Agent 编排器"""
7
8
    def __init__(self):
9
        self.agents = {
10
            "analyst": "http://localhost:8001",   # 需求分析 Agent
11
            "coder": "http://localhost:8002",     # 代码生成 Agent
12
            "tester": "http://localhost:8003",    # 测试 Agent
13
        }
14
15
    async def run_workflow(self, requirement: str):
16
        """执行完整的开发工作流"""
17
18
        # 第一步：需求分析
19
        print("📋 Step 1: 分析需求...")
20
        analyst = A2AClient(url=self.agents["analyst"])
21
        analysis = await analyst.send_message(
22
            SendMessageRequest(
23
                params=MessageSendParams(
24
                    message=Message(
25
                        role="user",
26
                        parts=[TextPart(text=f"分析这个需求: {requirement}")],
27
                    )
28
                )
29
            )
30
        )
31
        analysis_text = self._extract_text(analysis)
32
        print(f"分析结果: {analysis_text}")
33
34
        # 第二步：代码生成
35
        print("\n💻 Step 2: 生成代码...")
36
        coder = A2AClient(url=self.agents["coder"])
37
        code = await coder.send_message(
38
            SendMessageRequest(
39
                params=MessageSendParams(
40
                    message=Message(
41
                        role="user",
42
                        parts=[
43
                            TextPart(
44
                                text=f"根据以下分析生成代码:\n{analysis_text}"
45
                            )
46
                        ],
47
                    )
48
                )
49
            )
50
        )
51
        code_text = self._extract_text(code)
52
        print(f"生成的代码:\n{code_text}")
53
54
        # 第三步：测试
55
        print("\n🧪 Step 3: 运行测试...")
56
        tester = A2AClient(url=self.agents["tester"])
57
        test_result = await tester.send_message(
58
            SendMessageRequest(
59
                params=MessageSendParams(
60
                    message=Message(
61
                        role="user",
62
                        parts=[
63
                            TextPart(
64
                                text=f"测试以下代码:\n{code_text}"
65
                            )
66
                        ],
67
                    )
68
                )
69
            )
70
        )
71
        test_text = self._extract_text(test_result)
72
        print(f"测试结果: {test_text}")
73
74
        return {
75
            "analysis": analysis_text,
76
            "code": code_text,
77
            "test": test_text,
78
        }
79
80
    def _extract_text(self, response):
81
        """从响应中提取文本"""
82
        result = response.root.result
83
        if hasattr(result, 'artifacts') and result.artifacts:
84
            for artifact in result.artifacts:
85
                for part in artifact.parts:
86
                    if part.root.kind == "text":
87
                        return part.root.text
88
        return "No text in response"
89
90
# 使用
91
async def main():
92
    orchestrator = Orchestrator()
93
    result = await orchestrator.run_workflow(
94
        "写一个 Python 函数，计算斐波那契数列的第 n 项"
95
    )
96
97
if __name__ == "__main__":
98
    asyncio.run(main())

这个编排器会依次调用三个 Agent，把前一个的输出作为后一个的输入。当然，实际项目中你可能需要更复杂的逻辑——比如错误处理、重试、并行执行等。

A2A vs MCP：到底该用哪个？

既然网站上已经有 MCP 的文章了，这里做个对比，帮你理清思路。

用 MCP 的场景：

  • Agent 调用数据库
  • Agent 调用外部 API
  • Agent 读写文件
  • Agent 使用搜索引擎
  • 工具是"无状态"的，调一次出一次结果

用 A2A 的场景：

  • 多个 Agent 协作完成复杂任务
  • Agent 需要长时间运行（几分钟甚至几小时）
  • 需要 Agent 之间交换上下文
  • 不同团队/公司开发的 Agent 需要互操作
  • 需要保持 Agent 的"不透明性"（不暴露内部实现）

一个实际的例子：

你在做一个自动化代码审查系统。这个系统里：

  • 用 MCP 连接 GitHub API（读 PR、写评论）
  • 用 MCP 连接 SonarQube（查代码质量）
  • 用 A2A 连接一个独立的"安全审查 Agent"（它有自己的知识库和推理逻辑）
  • 用 A2A 连接一个独立的"性能分析 Agent"（它有自己的 benchmark 工具）

MCP 管工具调用，A2A 管 Agent 协作，各司其职。

踩坑记录

折腾 A2A 的过程中踩了不少坑，记录一下给大家避雷。

坑 1：Agent Card 的 URL 必须是绝对地址

我一开始把 Agent Card 里的 URL 写成了相对路径 /api/agent，结果 Client 端解析的时候直接报错。A2A 规范要求 URL 必须是完整的绝对地址，带协议和域名。

json
1
// ❌ 错误
2
{ "url": "/api/agent" }
3
4
// ✅ 正确
5
{ "url": "https://my-agent.example.com" }

坑 2：Task 状态流转不能跳级

Task 的状态有严格的流转顺序，不能从 submitted 直接跳到 completed，中间必须经过 working。我有一次图省事直接返回 completed，结果 Client 端解析出错。

正确的状态流转：

  • submitted → working → completed
  • submitted → working → failed
  • submitted → working → input-required → working → completed
  • 任何非终态都可以 → canceled

坑 3：JSON-RPC 的 id 字段必须匹配

A2A 基于 JSON-RPC 2.0，请求和响应的 id 必须一致。我用 SDK 的时候没注意这个，自己手写 HTTP 请求的时候踩了坑。SDK 帮你处理了，但如果用 curl 测试的话要注意。

坑 4：SSE 流式连接的超时问题

流式响应用的是 Server-Sent Events（SSE），如果你的 Agent 处理时间很长（超过 30 秒），中间件（nginx、负载均衡器）可能会把连接断掉。解决办法：

  • 在 Agent Card 里声明 streaming 能力
  • 中间件配置更长的超时时间
  • 定期发送心跳事件（比如每 10 秒发一个状态更新）
python
1
# nginx 配置
2
location /a2a/ {
3
    proxy_pass http://backend;
4
    proxy_read_timeout 300s;
5
    proxy_send_timeout 300s;
6
    proxy_set_header Connection '';
7
    proxy_http_version 1.1;
8
    chunked_transfer_encoding off;
9
}

坑 5：认证和授权别忘了

A2A 规范支持标准的 HTTP 认证（Bearer Token、OAuth2 等），但很多人搭 demo 的时候跳过了这一步。在生产环境里，Agent 之间的通信必须加密（HTTPS）+ 认证。

我在测试环境里用 HTTP 搞了一周都没问题，切到生产环境发现所有请求都被 nginx 的安全模块拦了——因为它检测到没有认证头。加上 Bearer Token 之后就好了。

与现有框架的集成

A2A 不是要替代现有的 Agent 框架，而是给它们一个通用的通信标准。目前已经有多个框架支持或正在支持 A2A：

Google ADK（Agent Development Kit）

Google 自己的 Agent 开发框架，原生支持 A2A。如果你用 ADK 开发 Agent，暴露为 A2A Server 几乎零成本。

LangGraph

LangChain 的图编排框架，可以通过适配器接入 A2A。把 LangGraph 的 Agent 包装成 A2A Server，就能被其他 A2A Client 调用。

CrewAI

多 Agent 协作框架，天然适合 A2A 的场景。每个 Crew 可以暴露为一个 A2A Agent，多个 Crew 之间通过 A2A 协作。

自己造轮子

如果你的 Agent 是自己写的（比如用 OpenAI API 直接调），也可以手动实现 A2A 协议。核心就是：

  1. 实现一个 HTTP Server，暴露 /.well-known/agent.json 和 JSON-RPC 端点
  2. 处理 message/send 请求
  3. 返回符合 A2A 规范的 Task 或 Message

安全性考量

Agent 之间互相通信，安全是个大问题。A2A 在设计的时候考虑了几个关键点：

不透明性（Opacity）

这是 A2A 最重要的安全原则之一。Agent 之间协作时，不需要暴露自己的内部实现。你的 Agent 用的什么模型、有什么工具、记忆了什么信息，对方都不知道。

这跟 API 调用不一样。API 通常需要知道对方的接口定义，但 A2A 只需要知道 Agent Card 里的公开信息。

认证和授权

A2A 支持标准的 HTTP 认证机制：

  • Bearer Token：最简单，适合内部系统
  • OAuth 2.0：适合跨组织的 Agent 协作
  • API Key：适合简单的场景

Agent Card 里可以声明需要的认证方式，Client 在发起请求时带上对应的凭证。

输入验证

永远不要信任其他 Agent 的输出。即使你通过 A2A 调用了一个"可信"的 Agent，它的输出也可能包含恶意内容（提示注入攻击）。在处理 A2A 响应时，一定要做输入验证和清洗。

性能和扩展性

A2A 基于 HTTP，性能瓶颈主要在网络延迟和 Agent 处理时间。几个优化建议：

连接复用

如果你要频繁调用同一个 Agent，保持 HTTP 连接复用（Keep-Alive）。Python 的 httpx 库默认支持，A2A SDK 底层用的就是它。

并行调用

如果有多个 Agent 需要调用，而且它们之间没有依赖关系，用 asyncio.gather 并行调用：

python
1
import asyncio
2
3
async def parallel_call():
4
    results = await asyncio.gather(
5
        agent_a.send_message(request_a),
6
        agent_b.send_message(request_b),
7
        agent_c.send_message(request_c),
8
    )
9
    return results

我测试过，三个 Agent 并行调用的总耗时约等于最慢的那个 Agent 的耗时，比串行调用快了将近 3 倍。

缓存 Agent Card

Agent Card 不会经常变化，Client 端可以缓存它，不用每次都去拉。A2A 规范建议 Agent Card 的 URL 支持 HTTP 缓存头（ETag、Last-Modified）。

目前的生态和现状

说实话，A2A 现在还处于早期阶段。2025 年 4 月发布，到现在一年多，生态还在建设中。

好的方面：

  • Google 主推，背后有大厂支持
  • DeepLearning.AI 出了专门的课程
  • 多个主流 Agent 框架正在接入
  • 规范已经到 1.0 版本，基本稳定

不足的方面：

  • 实际生产案例还不多
  • SDK 的文档和示例还不够完善
  • 社区规模跟 MCP 比还有差距
  • 调试工具和可观测性支持有限

我个人判断，A2A 和 MCP 会成为 AI Agent 生态的两个基础协议。MCP 解决工具调用，A2A 解决 Agent 协作。就像 HTTP 和 WebSocket 一样，各有各的场景。

下一步打算

后面打算在 Hermes Agent 上试试接入 A2A，让它能跟其他 Agent 协作。具体来说：

  1. 把 Hermes Agent 包装成 A2A Server，暴露它的技能
  2. 搭一个简单的编排器，让多个 A2A Agent 协作完成复杂任务
  3. 研究一下 A2A 的 Push Notification 机制，看看怎么用于长时间运行的任务

到时候再写一篇实战文章。有啥问题评论区聊。

参考资料：

  • A2A 协议官方文档：https://a2a-protocol.org/
  • A2A GitHub 仓库：https://github.com/a2aproject/A2A
  • A2A Python SDK：https://github.com/a2aproject/a2a-python
  • DeepLearning.AI A2A 课程：https://www.deeplearning.ai/

advertisement

A2A 协议实战：当 AI Agent 开始互相说话，事情变得有趣了 — AI Hub