深入解析 Google A2A 协议：多智能体协作的开放标准与实战指南

什么是 Agent2Agent（A2A）协议？

2025年4月，Google联合50余家AI厂商正式发布了Agent2Agent（A2A）协议，这是一个专为多智能体协作设计的开放标准。与此前各家"闭门造车"的私有接口不同，A2A定义了一套统一的通信规范，让基于LangChain、AutoGen、CrewAI甚至自研框架的智能体，都能像HTTP协议让不同服务器互通一样自由交互。

简单来说，A2A解决的核心问题是：我的智能体怎么调用你的智能体？ 不是通过硬编码API，而是通过标准化的消息协议，实现动态发现、任务委托、状态同步。

• 开放标准：基于JSON-RPC 2.0，任何语言都可实现

• 厂商中立：Google、Atlassian、SAP、MongoDB等50+厂商联合背书

• 安全优先：内置身份验证和权限控制机制

• 异步支持：支持长时间运行任务的状态追踪

A2A 与 MCP 的本质区别

很多开发者会把A2A与Anthropic的Model Context Protocol（MCP）混淆，但两者解决的是完全不同的问题：

• MCP：解决"智能体如何调用工具/资源"，是Agent↔工具的垂直连接协议

• A2A：解决"智能体如何与其他智能体协作"，是Agent↔Agent的水平协作协议

用一个比喻：MCP是"员工如何使用工作软件"，A2A是"不同部门的员工如何协同完成项目"。两者不是竞争关系，而是互补的技术栈。在一个完整的多智能体系统中，你可能同时使用MCP来让智能体访问数据库和API，用A2A来协调多个专家智能体分工合作。

# 技术栈对比
MCP:   Agent  ──→  Tools/Resources (数据库、API、文件系统)
A2A:   Agent  ──→  Agent (另一个具有独立能力的智能体)

# 消息格式对比
MCP:   {"method": "tools/call", "params": {"name": "query_db"}}
A2A:   {"jsonrpc": "2.0", "method": "tasks/send", "params": {"message": {...}}}

A2A 核心概念：Agent Card 与 Task

理解A2A需要掌握两个核心概念：

1. Agent Card（智能体名片）

每个A2A智能体都需要在 /.well-known/agent.json 发布自己的"名片"，描述自己的能力、支持的输入输出类型、认证方式等。这类似于OpenAPI的Swagger文档，让其他智能体能够自动发现并了解如何与你通信。

{
  "name": "数据分析智能体",
  "description": "专注于结构化数据分析和可视化",
  "url": "https://agent.example.com",
  "version": "1.0.0",
  "capabilities": {
    "streaming": true,
    "pushNotifications": false
  },
  "skills": [
    {
      "id": "data_analysis",
      "name": "数据分析",
      "description": "分析CSV/JSON数据并生成报告",
      "inputModes": ["text", "data"],
      "outputModes": ["text", "data", "file"]
    }
  ]
}

2. Task（任务）

A2A的工作单元是Task，每个Task有唯一ID，包含消息历史，并跟踪状态（submitted → working → completed/failed）。Task支持同步响应和流式响应两种模式。

用 Python 搭建 A2A 智能体服务

下面通过实战代码展示如何快速构建一个支持A2A协议的Python智能体服务。Google官方提供了 a2a-sdk Python包，大幅简化了实现工作。

# 安装依赖
pip install a2a-sdk fastapi uvicorn httpx

from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.types import AgentCard, AgentSkill, AgentCapabilities
from a2a.utils import new_agent_text_message

# 定义智能体业务逻辑
class MyAgentExecutor(AgentExecutor):
    async def execute(self, context: RequestContext, event_queue):
        # 获取用户最新消息
        user_msg = context.get_user_input()
        print(f"收到任务: {user_msg}")
        
        # 你的核心业务逻辑（调用LLM、查询数据库等）
        result = f"已处理您的请求：{user_msg}"
        
        # 返回结果
        await event_queue.enqueue_event(
            new_agent_text_message(result)
        )
    
    async def cancel(self, context, event_queue):
        pass

# 定义 Agent Card
agent_card = AgentCard(
    name="示例A2A智能体",
    description="一个演示A2A协议的智能体",
    url="http://localhost:9999",
    version="1.0.0",
    capabilities=AgentCapabilities(streaming=True),
    skills=[
        AgentSkill(
            id="general",
            name="通用对话",
            description="处理通用问答任务",
            inputModes=["text"],
            outputModes=["text"]
        )
    ]
)

# 启动服务
request_handler = DefaultRequestHandler(
    agent_executor=MyAgentExecutor(),
    task_store=None  # 使用内存存储
)

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

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app.build(), host="0.0.0.0", port=9999)

以上代码启动后，访问 http://localhost:9999/.well-known/agent.json 即可看到Agent Card，其他A2A智能体就可以自动发现并调用你的服务了。

多智能体协作实战：任务分发与结果聚合

A2A协议真正的威力在于多智能体的动态协作。以下示例展示一个"编排者智能体"如何将复杂任务分发给多个专家智能体：

import asyncio
from a2a.client import A2AClient
import httpx

async def orchestrate_task(user_request: str):
    """编排者：将任务分发给多个专家智能体"""
    
    async with httpx.AsyncClient() as http_client:
        # 连接搜索智能体
        search_agent = await A2AClient.get_client_from_agent_card_url(
            http_client, "http://search-agent:8001"
        )
        # 连接分析智能体
        analysis_agent = await A2AClient.get_client_from_agent_card_url(
            http_client, "http://analysis-agent:8002"
        )
        
        # 并行发送任务（提高效率）
        search_task, analysis_task = await asyncio.gather(
            search_agent.send_message(
                message=f"搜索相关资料：{user_request}"
            ),
            analysis_agent.send_message(
                message=f"分析需求背景：{user_request}"
            )
        )
        
        # 聚合结果
        search_result = search_task.result.parts[0].text
        analysis_result = analysis_task.result.parts[0].text
        
        return f"综合分析结果：\n搜索发现：{search_result}\n需求分析：{analysis_result}"

# 运行
result = asyncio.run(orchestrate_task("分析2025年AI智能体技术趋势"))
print(result)

通过并行调用多个智能体，整个流程的延迟大幅降低。这种模式是构建生产级多智能体系统的基础范式。

总结与展望

A2A协议的出现标志着AI智能体开发进入了"互联互通"的新阶段。它不仅仅是一个技术规范，更代表着整个行业向开放生态演进的共识。对开发者而言，现在开始学习A2A是一个绝佳时机：

• 协议本身基于成熟的JSON-RPC标准，学习成本低

• 官方Python SDK已相当完善，可快速上手

• 主流AI框架（LangGraph、AutoGen等）正在陆续添加A2A支持

• 企业级多智能体场景需求旺盛，掌握此技能将极具竞争优势

下一步建议：在GitHub上查看 google/a2a-samples 仓库，里面有完整的多语言示例，包括一个展示不同框架智能体互操作的完整Demo，是入门A2A协议最好的实战素材。