智能体间通信（A2A）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

RAG：让 Agent 能查外部知识；
HITL：让 Agent 必要时找人类介入；
A2A：让 Agent 必要时找另一个 Agent 协作。
```，Google 的 **A2A 协议（Agent2Agent Protocol）** 是一项开放标准，目标是让不同框架构建的智能体，比如 LangGraph、CrewAI、Google ADK 等，能够跨平台协同工作。

---

## 1. 为什么需要 A2A？

假设你要做一个复杂任务：

```text
帮我分析最近一周的销售数据，找出异常原因，然后生成一份汇报，并安排明天下午开会讨论。

1
2
3
4
5
6
7
8

读取数据；
清洗数据；
分析异常；
解释业务原因；
生成报告；
查日历；
安排会议；
通知相关人员。

1
2
3
4
5

数据 Agent：负责取数和清洗；
分析 Agent：负责指标分析和异常检测；
报告 Agent：负责生成汇报；
日历 Agent：负责查时间和安排会议；
通知 Agent：负责发邮件或消息。

1
2
3
4
5
6

它规定了 Agent 怎么介绍自己；
怎么被其他 Agent 发现；
怎么接收任务；
怎么返回结果；
怎么处理长任务；
怎么做安全认证。

1
2

MCP：规定 Agent / LLM 怎么连接工具和外部资源；
A2A：规定 Agent 怎么连接另一个 Agent。

1
2
3

用户 User
A2A 客户端 A2A Client
A2A 服务器 A2A Server

1

帮我看看明天上午 10 点到 11 点有没有空。

1
2
3

用户：提出需求
客户端 Agent：决定要找谁帮忙
服务器 Agent：真正执行某个专业任务

1
2
3
4
5
6
7

这个 Agent 叫什么；
它能做什么；
它的服务地址是什么；
它支持什么输入输出格式；
它有哪些技能；
它是否支持流式输出；
它需要什么认证方式。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

agent_card = {
    "name": "WeatherBot",
    "description": "提供准确的天气预报和历史数据。",
    "url": "http://weather-service.example.com/a2a",
    "version": "1.0.0",
    "capabilities": {
        "streaming": True,
        "pushNotifications": False,
        "stateTransitionHistory": True
    },
    "authentication": {
        "schemes": ["apiKey"]
    },
    "skills": [...]
}

1
2
3
4
5
6

它是天气 Agent；
它能查当前天气；
它能查天气预报；
它支持文本输入和文本输出；
它需要 API Key；
它的调用地址是什么。

1

/.well-known/agent.json

1

所有可用 Agent 都把 Agent Card 注册进去。

1
2
3
4

谁会查天气？
谁会生成报告？
谁会查日历？
谁会做财务分析？

1

主控 Agent 预先知道日历 Agent 的地址和能力。

1

检查明天上午 10 点到 11 点是否有空。

1
2
3
4
5

submitted：已提交；
working：处理中；
input-required：需要补充输入；
completed：已完成；
failed：失败。

1
2
3
4
5

生成一份长报告；
跑一次数据分析；
处理视频；
查多个系统；
等待人工审批。

1
2
3
4
5
6

文本；
文件；
结构化 JSON；
音频；
视频；
元数据。

1
2
3
4
5
6
7
8
9

{
  "role": "user",
  "parts": [
    {
      "type": "text",
      "text": "美元兑欧元汇率是多少？"
    }
  ]
}

1
2
3
4
5
6

报告文件；
分析结果；
生成的代码；
图片；
表格；
最终答案。

1

HTTP(S) + JSON-RPC 2.0

1

http://calendar-agent.example.com/

1
2
3
4
5
6
7
8

{
  "jsonrpc": "2.0",
  "id": "1",
  "method": "sendTask",
  "params": {
    ...
  }
}

1

我要调用远程服务的某个方法 method，并传入 params。

1
2
3
4

查当前天气；
查一个汇率；
判断一句话情感；
查询某个日历时间段是否空闲。

1
2
3
4
5

客户端发送请求
↓
服务端立即处理
↓
一次性返回完整结果

1

"method": "sendTask"

1
2
3
4
5
6
7

客户端提交任务
↓
服务器返回 task_id 和处理中状态
↓
客户端隔一段时间查一次状态
↓
直到任务完成或失败

1
2
3

生成一份 30 页报告；
跑一次大规模数据分析；
处理一批文件。

1
2
3
4

长文本生成；
实时分析进度；
逐步返回搜索结果；
代码生成过程。

1
2
3
4
5

客户端建立连接
↓
服务端持续推送状态或部分结果
↓
客户端实时接收

1

"method": "sendTaskSubscribe"

1
2
3
4
5

客户端提交任务，并留下 webhook 地址
↓
服务端慢慢处理
↓
状态变化时主动通知客户端

1
2
3

视频转写完成后通知；
大型报告生成完成后通知；
人工审批完成后通知。

1

Source Agent → A2A Protocol → Target Agent

1

Source Agent → MCP → Tools

1
2

A2A：Agent 找 Agent
MCP：Agent 找工具/资源

1

用户：帮我查明天上午是否有空，并安排会议。

1

主控 Agent → MCP → Google Calendar 工具

1

主控 Agent → A2A → Calendar Agent → Google Calendar

1
2
3
4
5
6

用户隐私；
企业内部数据；
日历信息；
财务数据；
工具调用权限；
身份凭证。

1
2
3
4
5

mTLS；
审计日志；
Agent Card 中声明认证要求；
OAuth 2.0 Token 或 API Key；
凭证放在 HTTP Header 里，不放 URL 或消息体里。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

async def create_agent(client_id, client_secret) -> LlmAgent:
    toolset = CalendarToolset(client_id=client_id, client_secret=client_secret)
    return LlmAgent(
        model='gemini-2.0-flash-001',
        name='calendar_agent',
        description="可帮助管理用户日历的 Agent",
        instruction=f"""
        你是一个可以帮助用户管理日历的 Agent。
        用户会请求日历状态信息或修改日历。请使用提供的工具与日历 API 交互。
        如未指定，默认使用 'primary' 日历。
        使用 Calendar API 工具时，请采用规范的 RFC3339 时间戳。
        今天是 {datetime.datetime.now()}。
        """,
        tools=await toolset.get_tools(),
    )

1

创建一个会使用 Google Calendar API 的 Agent。

1

CalendarToolset(client_id=client_id, client_secret=client_secret)

1
2
3
4

查日历；
检查空闲时间；
修改日历；
使用规范时间格式。

1
2
3
4
5
6
7

skill = AgentSkill(
    id='check_availability',
    name='检查可用性',
    description="使用 Google Calendar 检查用户某一时间段的空闲情况",
    tags=['calendar'],
    examples=['我明天上午 10 点到 11 点有空吗？'],
)

1

这个 Agent 有一个技能：检查日历空闲时间。

1

如果我需要查询用户有没有空，可以调用它。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

agent_card = AgentCard(
    name='Calendar Agent',
    description="可管理用户日历的 Agent",
    url=f'http://{host}:{port}/',
    version='1.0.0',
    defaultInputModes=['text'],
    defaultOutputModes=['text'],
    capabilities=AgentCapabilities(streaming=True),
    skills=[skill],
)

1
2
3
4
5
6
7

名字：Calendar Agent
能力：管理日历
地址：http://host:port/
输入：text
输出：text
支持：streaming
技能：check_availability

1
2
3
4
5
6
7
8
9

runner = Runner(
    app_name=agent_card.name,
    agent=adk_agent,
    artifact_service=InMemoryArtifactService(),
    session_service=InMemorySessionService(),
    memory_service=InMemoryMemoryService(),
)

agent_executor = ADKAgentExecutor(runner, agent_card)

1
2

Runner：负责运行 ADK Agent；
Executor：把 ADK Agent 包装成 A2A 可以调用的执行器。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

request_handler = DefaultRequestHandler(
    agent_executor=agent_executor,
    task_store=InMemoryTaskStore()
)

a2a_app = A2AStarletteApplication(
    agent_card=agent_card,
    http_handler=request_handler
)

app = Starlette(routes=routes)
uvicorn.run(app, host=host, port=port)

1
2

Calendar Agent 不再只是本地代码；
它变成了一个可通过 A2A 协议访问的远程 Agent 服务。

1
2
3
4
5
6
7
8

1. 一个 Agent 创建自己的 Agent Card
2. 它把 Agent Card 暴露出来
3. 另一个 Agent 读取 Agent Card
4. 发现它有某种技能
5. 发送任务请求
6. 远程 Agent 处理任务
7. 返回结果、流式结果或任务状态
8. 客户端 Agent 汇总结果并继续完成用户任务

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

主控 Agent 收到用户请求：
“我明天上午 10 点到 11 点有空吗？”

↓
主控 Agent 发现 Calendar Agent 会 check_availability

↓
通过 A2A 发送任务

↓
Calendar Agent 调用 Google Calendar API

↓
Calendar Agent 返回空闲状态

↓
主控 Agent 把结果告诉用户

1
2
3
4

LangGraph Agent 负责主流程；
CrewAI Agent 负责多角色协作；
ADK Agent 负责 Google 工具；
自研 Agent 负责企业内部系统。

1

数据采集 Agent → 分析 Agent → 报告 Agent → 邮件 Agent → 日历 Agent

1

主 Agent → 市场数据 Agent → 外部 API

1
2
3
4
5
6
7

模块化；
可扩展；
跨框架；
任务委托；
能力复用；
降低集成成本；
构建复杂多 Agent 系统。

1
2
3
4
5

每个 Agent 一套接口；
每个系统单独适配；
集成成本很高；
难以动态发现能力；
难以统一追踪任务。

1
2
3
4
5

每个 Agent 用 Agent Card 描述自己；
统一用 HTTP + JSON-RPC 通信；
统一任务格式；
统一认证方式；
统一支持同步、异步、流式等交互。

1
2
3
4
5

目标设定与监控：让 Agent 有目标、有进度、有纠偏；
异常处理与恢复：让 Agent 出错后能恢复；
HITL：让 Agent 必要时找人类；
RAG：让 Agent 必要时查知识；
A2A：让 Agent 必要时找其他 Agent。

1

Agent 不再孤立工作，而是可以组成网络。

1

用户 → Agent → 输出

1
2
3
4

用户 → 主控 Agent → 专业 Agent A
                 → 专业 Agent B
                 → 专业 Agent C
      → 汇总结果 → 输出

1
2

MCP = Agent 连接工具
A2A = Agent 连接 Agent