A2A 协议完全指南：2026 年 Google 开源 AI 代理通信标准

发布时间：2026 年 3 月 · 版本：v0.3.0 · 许可证：Apache 2.0 · 维护者：Google + Linux Foundation

2025 年 4 月，Google 正式开源了 A2A（Agent2Agent）协议 —— 一个旨在解决 AI 代理互操作性问题的开放标准。随着 AI 代理（AI Agent）在 2026 年的爆发式增长，不同框架、不同公司开发的代理之间如何通信协作成为了关键挑战。A2A 协议应运而生，被誉为"AI 代理时代的 HTTP"。

A2A 协议的核心目标是让基于不同框架（如 Google ADK、LangGraph、BeeAI 等）构建的 AI 代理能够安全、高效地通信和协作，而无需暴露各自的内部状态、记忆或工具实现。这一设计理念与 MCP（Model Context Protocol）形成互补：MCP 让代理连接外部工具，A2A 让代理互相连接。

为什么需要 A2A 协议？

AI 代理的"巴别塔"问题

2026 年的 AI 开发生态中，存在着众多代理框架和平台：

• Google ADK - Google 官方的代理开发框架
• LangGraph - LangChain 推出的图式代理框架
• CrewAI - 面向任务编排的多代理框架
• AutoGen - Microsoft 开源的多代理对话框架
• Goose - Block 公司开源的本机代理
• OpenClaw - 快速增长的开源代理平台

这些框架各自为政，代理之间无法直接通信。想象一下：你有一个擅长数据分析的 LangGraph 代理，另一个擅长生成报告的 CrewAI 代理，但它们无法协作完成"分析数据并生成报告"的任务。

A2A 的解决方案

A2A 协议通过标准化通信接口，让代理能够：

1. 发现彼此的能力 - 通过"代理卡片"（Agent Cards）声明功能
2. 协商交互方式 - 支持文本、表单、媒体等多种模态
3. 安全协作长任务 - 支持流式和异步通信
4. 保持内部隐私 - 无需暴露记忆、工具或专有逻辑

A2A vs MCP：有什么区别？

特性	A2A Protocol	MCP (Model Context Protocol)
目标	代理 ↔ 代理通信	代理 ↔ 工具/数据源连接
场景	多代理协作编排	单个代理扩展能力
通信方式	JSON-RPC 2.0 over HTTP	JSON-RPC 2.0 over stdio/HTTP
发现机制	Agent Cards	MCP Server Registry
典型用例	数据分析代理 + 报告生成代理	代理 + GitHub/数据库/文件系统
互补关系	代理间的"HTTP"	代理与工具的"USB"

最佳实践：在现代 AI 系统中，同时使用 A2A 和 MCP。例如：一个 A2A 协调代理通过 MCP 连接本地工具，同时通过 A2A 调用其他专业代理。

核心概念

1. Agent Card（代理卡片）

每个 A2A 兼容的代理都必须提供一个"代理卡片"，声明其能力：

{
  "name": "数据分析代理",
  "description": "专门处理 CSV/Excel 数据分析任务",
  "url": "https://agents.example.com/data-analyzer",
  "version": "1.0.0",
  "capabilities": {
    "inputFormats": ["text/csv", "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"],
    "outputFormats": ["application/json", "text/markdown"],
    "skills": ["统计分析", "数据可视化", "异常检测"]
  },
  "authentication": {
    "type": "bearer",
    "required": true
  }
}

2. 通信模式

A2A 支持三种通信模式：

• 同步请求/响应 - 适用于快速任务
• 流式响应（SSE） - 适用于长任务，实时返回进度
• 异步推送通知 - 适用于耗时任务，完成后回调

3. 任务状态机

[新建] → [进行中] → [完成/失败/取消]
           ↓
      [需要输入] → [继续]

快速开始

系统要求

• Python：3.10+
• Node.js：18+（可选，用于 JS SDK）
• Go：1.21+（可选，用于 Go SDK）
• Java：17+（可选，用于 Java SDK）
• .NET：8.0+（可选，用于 .NET SDK）

安装 SDK

Python SDK（推荐）

# 安装 A2A Python SDK
pip install a2a-sdk

# 验证安装
python -c "import a2a; print(a2a.__version__)"

Node.js SDK

npm install @a2a-js/sdk

Go SDK

go get github.com/a2aproject/a2a-go

Java SDK（Maven）

<dependency>
  <groupId>org.a2a</groupId>
  <artifactId>a2a-sdk</artifactId>
  <version>0.3.0</version>
</dependency>

.NET SDK

dotnet add package A2A

实战案例：构建多代理协作系统

场景描述

假设我们需要构建一个"内容创作工作流"，涉及三个专业代理：

1. 研究代理 - 搜索和整理信息
2. 写作代理 - 根据研究结果撰写文章
3. 审校代理 - 检查语法和事实准确性

步骤 1：创建研究代理服务器

# research_agent.py
from a2a.server import A2AServer
from a2a.types import AgentCard, Task, TaskStatus

class ResearchAgent:
    def __init__(self):
        self.card = AgentCard(
            name="研究代理",
            description="搜索和整理网络信息",
            version="1.0.0",
            capabilities={
                "skills": ["网络搜索", "信息提取", "摘要生成"],
                "inputFormats": ["text/plain"],
                "outputFormats": ["application/json"]
            }
        )

    async def execute(self, task: Task):
        """执行研究任务"""
        query = task.input.text
        # 调用搜索工具（实际项目中集成 web_search 或 Serper API）
        results = await self.search_web(query)

        return Task(
            id=task.id,
            status=TaskStatus.COMPLETED,
            output={
                "summary": results.summary,
                "sources": results.sources,
                "key_points": results.key_points
            }
        )

    async def search_web(self, query: str):
        # 实现搜索逻辑
        pass

# 启动服务器
server = A2AServer(
    agent=ResearchAgent(),
    host="0.0.0.0",
    port=8080
)
server.run()

步骤 2：创建写作代理客户端

# writing_agent.py
from a2a.client import A2AClient
from a2a.types import Task, TaskRequest

class WritingAgent:
    def __init__(self):
        self.research_client = A2AClient(
            agent_url="http://localhost:8080",
            api_key="your-api-key"
        )

    async def write_article(self, topic: str):
        # 步骤 1：请求研究代理收集信息
        research_task = TaskRequest(
            input={"text": f"研究主题：{topic}，需要提供关键事实、数据来源和相关引用"},
            mode="streaming"  # 使用流式模式获取实时进度
        )

        research_result = await self.research_client.execute(research_task)

        # 步骤 2：基于研究结果撰写文章
        article = await self.generate_article(
            topic=topic,
            research_data=research_result.output
        )

        return article

    async def generate_article(self, topic: str, research_data: dict):
        # 调用 LLM 生成文章
        pass

# 使用示例
async def main():
    writer = WritingAgent()
    article = await writer.write_article("A2A 协议详解")
    print(article)

import asyncio
asyncio.run(main())

步骤 3：编排完整工作流

# workflow_orchestrator.py
from a2a.client import A2AClient
from a2a.types import Task, TaskRequest

class ContentWorkflow:
    def __init__(self):
        self.research_agent = A2AClient("http://localhost:8080")
        self.writing_agent = A2AClient("http://localhost:8081")
        self.review_agent = A2AClient("http://localhost:8082")

    async def execute_workflow(self, topic: str):
        print(f"📝 开始内容创作工作流：{topic}")

        # 阶段 1：研究
        print("🔍 阶段 1：信息收集...")
        research_task = TaskRequest(
            input={"text": topic},
            mode="sync"
        )
        research_result = await self.research_agent.execute(research_task)
        print(f"✅ 研究完成，找到 {len(research_result.output['sources'])} 个来源")

        # 阶段 2：写作
        print("✍️  阶段 2：文章撰写...")
        writing_task = TaskRequest(
            input={
                "topic": topic,
                "research_data": research_result.output
            },
            mode="streaming"
        )
        async for chunk in self.writing_agent.execute_stream(writing_task):
            print(f"📝 写作进度：{chunk.progress}%")

        draft = chunk.output
        print(f"✅ 初稿完成，{len(draft['content'])} 字")

        # 阶段 3：审校
        print("🔎 阶段 3：质量审校...")
        review_task = TaskRequest(
            input={
                "content": draft['content'],
                "check_types": ["grammar", "facts", "citations"]
            },
            mode="sync"
        )
        review_result = await self.review_agent.execute(review_task)

        print("✅ 工作流完成！")
        return {
            "final_content": review_result.output['revised_content'],
            "quality_score": review_result.output['quality_score'],
            "suggestions": review_result.output['suggestions']
        }

# 运行工作流
async def main():
    workflow = ContentWorkflow()
    result = await workflow.execute_workflow("2026 年 AI 代理发展趋势")
    print("\n📄 最终文章：")
    print(result['final_content'])

asyncio.run(main())

高级特性

1. 流式响应处理

async def handle_streaming_task(client: A2AClient, task: TaskRequest):
    async for event in client.execute_stream(task):
        if event.type == "progress":
            print(f"进度：{event.data['progress']}%")
        elif event.type == "partial_output":
            print(f"部分结果：{event.data['content']}")
        elif event.type == "completed":
            print(f"任务完成！最终结果：{event.data['output']}")

2. 错误处理与重试

from a2a.exceptions import AgentUnavailable, TaskFailed

async def execute_with_retry(client: A2AClient, task: TaskRequest, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await client.execute(task)
        except AgentUnavailable:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)  # 指数退避
        except TaskFailed as e:
            print(f"任务失败：{e.message}")
            raise

3. 认证与安全

# 使用 Bearer Token 认证
client = A2AClient(
    agent_url="https://agents.example.com/analyzer",
    api_key="your-secret-key",
    auth_type="bearer"
)

# 使用 OAuth 2.0
from a2a.auth import OAuth2Provider

oauth = OAuth2Provider(
    client_id="your-client-id",
    client_secret="your-client-secret",
    token_url="https://auth.example.com/oauth/token"
)
client = A2AClient(
    agent_url="https://agents.example.com/analyzer",
    auth_provider=oauth
)

A2A 生态工具

官方资源

• GitHub 仓库：https://github.com/a2aproject/A2A
• 协议规范：https://a2a-protocol.org/latest/specification/
• 文档站点：https://a2a-protocol.org
• 示例代码：https://github.com/a2aproject/a2a-samples
• 免费课程：https://goo.gle/dlai-a2a（Google Cloud + IBM Research 联合出品）

社区工具

工具	描述	链接
A2A Inspector	调试和检查 Agent Cards	pip install a2a-inspector
A2A Gateway	代理网关和负载均衡	https://github.com/a2aproject/a2a-gateway
A2A Registry	代理发现和服务注册	https://github.com/a2aproject/a2a-registry

最佳实践

✅ 推荐做法

1. 明确声明能力 - 在 Agent Card 中详细描述代理的技能和限制
2. 使用流式通信 - 对于耗时超过 5 秒的任务，优先使用流式模式
3. 实现优雅降级 - 当依赖的代理不可用时，提供备选方案
4. 记录交互日志 - 便于调试和审计代理间的通信
5. 设置超时限制 - 避免无限等待，建议设置 30-60 秒超时

❌ 避免的陷阱

1. 过度依赖单个代理 - 设计冗余和备选方案
2. 忽略认证 - 生产环境必须启用认证
3. 暴露敏感数据 - 不要在 Agent Card 中泄露内部实现细节
4. 同步阻塞长任务 - 使用异步或流式模式处理长任务

与 OpenClaw 集成

如果你正在使用 OpenClaw 构建 AI 助手，可以通过以下方式集成 A2A：

# 在 OpenClaw 技能中调用 A2A 代理
from a2a.client import A2AClient

async def a2a_research_task(query: str):
    """调用外部研究代理"""
    client = A2AClient("http://research-agent:8080")
    result = await client.execute({
        "input": {"text": query},
        "mode": "sync"
    })
    return result.output

# 注册为 OpenClaw 技能
# 在 skills/a2a-integration/SKILL.md 中定义

未来展望

A2A 协议目前由 Linux Foundation 托管，Google 作为主要贡献者。2026 年的发展路线图包括：

• v0.4.0（2026 Q2） - 增加多模态支持（图像、音频）
• v0.5.0（2026 Q3） - 引入代理市场发现机制
• v1.0.0（2026 Q4） - 正式稳定版，向后兼容保证

随着更多公司和项目加入 A2A 生态，我们有望看到一个真正互联的 AI 代理网络，就像今天的 Web 一样开放和互操作。

总结

A2A 协议代表了 AI 代理发展的下一个阶段：从孤立的智能体到协作的代理网络。对于开发者而言，现在正是学习和采用 A2A 的好时机：

• ✅ 开源免费 - Apache 2.0 许可，无使用限制
• ✅ 多语言支持 - Python、JS、Go、Java、.NET
• ✅ 大厂背书 - Google 出品，Linux Foundation 托管
• ✅ 生态完善 - 丰富的 SDK、文档和示例
• ✅ 前瞻设计 - 与 MCP 互补，面向未来多代理系统

开始行动：访问 https://github.com/a2aproject/A2A 获取最新代码和文档，或者通过 Google 的免费课程系统学习 A2A 协议。

---

参考资料：

1. A2A Protocol 官方仓库
2. A2A 协议规范
3. Google 官方博客：A2A 协议发布
4. A2A 免费课程
5. A2A 示例代码库