Google A2A Agent协议深度解析：多Agent互操作、任务委派与MCP对比

AI与大数据

摘要

  • Google A2A（Agent-to-Agent）协议是2026年多Agent系统互操作的行业标准，解决「Agent之间怎么协作」的核心问题
  • A2A和MCP解决不同层面的问题：MCP管Agent与工具的连接，A2A管Agent与Agent的协作
  • Agent Card是A2A的「名片」：声明能力、端点、认证方式，让其他Agent发现和调用
  • 任务生命周期（Task Lifecycle）是A2A的核心抽象：提交→处理→完成/失败/需要人工介入
  • 本文提供从协议原理到Python实现的完整方案，含多Agent编排实战

目录

多Agent时代的互操作难题

2026年的Agent生态现状

2025年MCP协议解决了「Agent怎么连工具」的问题，2026年企业面临的新问题是：

我有客服Agent、订单Agent、物流Agent、退款Agent，它们怎么互相协作？

痛点 没有A2A时 有A2A后
Agent发现 硬编码URL，改一个挂一片 Agent Card自动注册发现
任务委派 自定义HTTP接口，每家不同 标准Task API
状态同步 轮询或自建WebSocket 标准事件流（SSE）
安全认证 各Agent各自实现 统一OAuth2/API Key
异构兼容 Python Agent调不了Go Agent 协议层统一，语言无关

一个真实的协作场景

用户问：「我上周买的耳机怎么还没到？」

没有A2A时，客服Agent需要：

  1. 自己调订单API查订单
  2. 自己调物流API查物流
  3. 自己判断是否需要退款
  4. 每个能力都要自己实现

有A2A后：

  1. 客服Agent将任务委派给订单Agent（查订单）
  2. 订单Agent返回结果后，委派给物流Agent（查物流）
  3. 如果物流异常，委派给退款Agent（发起退款）
  4. 每个Agent专注自己的领域，通过A2A标准协议协作

A2A协议核心概念

协议架构

┌──────────────────────────────────────────────────────────┐
│                    A2A Protocol Stack                     │
│                                                            │
│  ┌─────────────┐    ┌─────────────┐    ┌──────────────┐  │
│  │ Agent Card  │    │  Task API   │    │  Event Stream│  │
│  │ (发现/注册)  │    │ (任务管理)   │    │  (状态推送)   │  │
│  └─────────────┘    └─────────────┘    └──────────────┘  │
│                                                            │
│  ┌─────────────────────────────────────────────────────┐  │
│  │              Transport: HTTP + JSON-RPC / SSE         │  │
│  └─────────────────────────────────────────────────────┘  │
└──────────────────────────────────────────────────────────┘

核心实体

实体 说明 类比
Agent Card Agent的能力声明和端点信息 微服务注册中心的服务描述
Task 一次工作请求，有完整的生命周期 HTTP请求 + 异步Job
Message Task中的单条交互（用户输入/Agent响应） 聊天消息
Artifact Task产出的结果（文件、数据、报告） API响应体
Part Message或Artifact的内容片段（文本/文件/数据） MIME多部分内容

Agent Card：Agent的「名片」

Agent Card是一个JSON文档，描述Agent的能力和接入方式：

{
  "name": "order-agent",
  "description": "处理订单查询、创建和状态变更的Agent",
  "url": "https://agents.example.com/order",
  "version": "1.0.0",
  "capabilities": {
    "streaming": true,
    "pushNotifications": false
  },
  "skills": [
    {
      "id": "query-order",
      "name": "查询订单",
      "description": "根据订单号或用户ID查询订单详情",
      "inputModes": ["text"],
      "outputModes": ["text", "data"]
    },
    {
      "id": "create-order",
      "name": "创建订单",
      "description": "根据商品列表创建新订单",
      "inputModes": ["text", "data"],
      "outputModes": ["text", "data"]
    }
  ],
  "authentication": {
    "schemes": ["bearer"]
  }
}

Agent发现流程

协调者Agent                    Agent Registry              目标Agent
     │                              │                          │
     │  1. 查找能"查询订单"的Agent    │                          │
     ├─────────────────────────────→│                          │
     │                              │  2. 返回匹配的Agent Card  │
     │←─────────────────────────────┤                          │
     │                              │                          │
     │  3. 向目标Agent发送Task                                   │
     ├────────────────────────────────────────────────────────→│
     │                              │                          │
     │  4. 接收Task状态更新（SSE）                                │
     │←────────────────────────────────────────────────────────┤

Agent Card通常托管在 /.well-known/agent.json 路径，类似OAuth的 .well-known/openid-configuration。

任务生命周期与状态机

Task状态机

         ┌──────────┐
         │ submitted │  任务已提交
         └────┬─────┘
              │
              ▼
         ┌──────────┐
    ┌────│ working  │────┐
    │    └──────────┘    │
    │         │          │
    ▼         ▼          ▼
┌────────┐ ┌────────┐ ┌──────────────────┐
│ input- │ │completed│ │ failed           │
│required│ └────────┘ └──────────────────┘
└────────┘
    │
    ▼
 需要更多输入
（如缺少订单号）

Task API核心方法

方法 说明 使用场景
tasks/send 发送新任务 委派工作给其他Agent
tasks/get 查询任务状态 轮询等待结果
tasks/cancel 取消任务 用户放弃或超时
tasks/subscribe 订阅任务事件流 实时接收进度更新

发送任务示例

{
  "jsonrpc": "2.0",
  "method": "tasks/send",
  "id": "req-001",
  "params": {
    "id": "task-uuid-001",
    "message": {
      "role": "user",
      "parts": [
        {
          "type": "text",
          "text": "查询订单号 ORD-20260703-8842 的物流状态"
        }
      ]
    }
  }
}

任务结果

{
  "jsonrpc": "2.0",
  "id": "req-001",
  "result": {
    "id": "task-uuid-001",
    "status": "completed",
    "artifacts": [
      {
        "name": "物流信息",
        "parts": [
          {
            "type": "data",
            "data": {
              "order_id": "ORD-20260703-8842",
              "carrier": "顺丰速运",
              "tracking_no": "SF1234567890",
              "status": "运输中",
              "estimated_delivery": "2026-07-05"
            }
          }
        ]
      }
    ]
  }
}

A2A vs MCP：协议层对比

这是面试和架构选型的核心问题：

维度 MCP (Model Context Protocol) A2A (Agent-to-Agent)
解决问题 Agent如何连接和使用工具/数据 Agent之间如何协作和委派任务
通信模式 Client-Server（Agent是Client） Peer-to-Peer（Agent互为Client/Server）
核心抽象 Tool、Resource、Prompt Task、Message、Artifact
发现机制 MCP Server声明工具列表 Agent Card声明技能
状态管理 无状态（单次工具调用） 有状态（Task生命周期）
传输协议 stdio / HTTP SSE HTTP JSON-RPC / SSE
典型场景 读文件、查数据库、调API 任务委派、多Agent编排
关系 互补，不是竞争 互补，不是竞争

一句话总结：MCP让Agent能用手（工具），A2A让Agent能组队（协作）。

组合使用架构

用户
 │
 ▼
┌──────────────┐
│ 协调者Agent   │
│ (Orchestrator)│
└──┬───┬───┬──┘
   │   │   │        A2A协议
   ▼   ▼   ▼
┌─────┐┌─────┐┌─────┐
│客服  ││订单  ││物流  │    Agent层
│Agent││Agent││Agent│
└──┬──┘└──┬──┘└──┬──┘
   │      │      │         MCP协议
   ▼      ▼      ▼
┌─────┐┌─────┐┌─────┐
│知识库││订单DB││物流API│   工具/数据层
│MCP  ││MCP  ││MCP  │
└─────┘└─────┘└─────┘

Python A2A Server实现

from dataclasses import dataclass, field
from typing import Optional
from enum import Enum
import uuid

class TaskStatus(Enum):
    SUBMITTED = "submitted"
    WORKING = "working"
    INPUT_REQUIRED = "input-required"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class Task:
    id: str
    status: TaskStatus
    messages: list[dict] = field(default_factory=list)
    artifacts: list[dict] = field(default_factory=list)

class OrderAgent:
  agent_card = {
      "name": "order-agent",
      "url": "https://agents.example.com/order",
      "skills": [
          {"id": "query-order", "name": "查询订单"},
          {"id": "create-order", "name": "创建订单"},
      ],
  }

  def __init__(self, order_service):
      self.order_service = order_service
      self.tasks: dict[str, Task] = {}

  async def handle_task_send(self, params: dict) -> dict:
      task_id = params.get("id", str(uuid.uuid4()))
      message = params["message"]
      user_text = message["parts"][0]["text"]

      task = Task(id=task_id, status=TaskStatus.WORKING)
      task.messages.append(message)
      self.tasks[task_id] = task

      try:
          if "查询" in user_text or "query" in user_text.lower():
              order_id = self._extract_order_id(user_text)
              if not order_id:
                  task.status = TaskStatus.INPUT_REQUIRED
                  task.messages.append({
                      "role": "agent",
                      "parts": [{"type": "text", "text": "请提供订单号"}],
                  })
                  return self._task_to_response(task)

              order = await self.order_service.get_order(order_id)
              task.status = TaskStatus.COMPLETED
              task.artifacts.append({
                  "name": "订单详情",
                  "parts": [{"type": "data", "data": order}],
              })
          else:
              task.status = TaskStatus.FAILED
              task.messages.append({
                  "role": "agent",
                  "parts": [{"type": "text", "text": "不支持的操作"}],
              })
      except Exception as e:
          task.status = TaskStatus.FAILED
          task.messages.append({
              "role": "agent",
              "parts": [{"type": "text", "text": f"处理失败: {e}"}],
          })

      return self._task_to_response(task)

  def _extract_order_id(self, text: str) -> Optional[str]:
      import re
      match = re.search(r'ORD-\d{8}-\d+', text)
      return match.group(0) if match else None

  def _task_to_response(self, task: Task) -> dict:
      return {
          "id": task.id,
          "status": task.status.value,
          "messages": task.messages,
          "artifacts": task.artifacts,
      }

多Agent协作实战：客服分流系统

架构设计

class CustomerServiceOrchestrator:
    def __init__(self, agents: dict[str, str]):
        self.agents = agents  # skill_id -> agent_url

    async def handle_user_query(self, query: str) -> str:
        intent = await self._classify_intent(query)

        agent_url = self.agents.get(intent)
        if not agent_url:
            return "抱歉，暂时无法处理您的问题"

        task_result = await self._send_a2a_task(agent_url, query)

        if task_result["status"] == "input-required":
            return task_result["messages"][-1]["parts"][0]["text"]

        if task_result["status"] == "completed":
            if task_result.get("artifacts"):
                return self._format_artifacts(task_result["artifacts"])
            return task_result["messages"][-1]["parts"][0]["text"]

        return "处理出现问题，请稍后重试"

    async def _classify_intent(self, query: str) -> str:
        if any(kw in query for kw in ["订单", "下单", "购买"]):
            return "order"
        if any(kw in query for kw in ["物流", "快递", "发货", "到哪"]):
            return "logistics"
        if any(kw in query for kw in ["退款", "退货", "取消"]):
            return "refund"
        return "general"

    async def _send_a2a_task(self, agent_url: str, query: str) -> dict:
        import httpx
        async with httpx.AsyncClient() as client:
            resp = await client.post(f"{agent_url}/a2a", json={
                "jsonrpc": "2.0",
                "method": "tasks/send",
                "id": str(uuid.uuid4()),
                "params": {
                    "message": {
                        "role": "user",
                        "parts": [{"type": "text", "text": query}],
                    }
                },
            })
            return resp.json()["result"]

安全模型与生产部署

A2A安全要求

层级 措施 说明
传输层 HTTPS/TLS 1.3 所有Agent间通信必须加密
认证层 OAuth2 Bearer Token Agent Card声明支持的认证方式
授权层 Skill级别权限 协调者Agent只能调用被授权的Skill
审计层 Task ID全链路追踪 每个Task有唯一ID，可审计

生产部署清单

  • Agent Card托管在 /.well-known/agent.json
  • 每个Agent独立部署，通过K8s Service暴露
  • 使用API Gateway统一入口，做认证和限流
  • Task超时设置（默认30秒，长任务用subscribe模式）
  • 监控：Task成功率、平均处理时间、Agent可用性

面试考点与选型指南

Q1：A2A和MCP是什么关系？

互补关系。MCP解决Agent与工具/数据的连接（垂直集成），A2A解决Agent与Agent的协作（水平集成）。生产系统通常两者同时使用。

Q2：A2A和直接调HTTP API有什么区别？

A2A提供了标准化的Task生命周期、Agent发现、状态推送和Artifact传递。直接调HTTP需要自己实现这些，且不同团队的Agent无法互操作。

Q3：什么时候不需要A2A？

单Agent + MCP工具就够用时不需A2A。A2A的价值在多个专业Agent需要协作的场景，如客服分流、多步骤审批、跨域任务编排。

Q4：A2A的任务超时怎么处理？

发送方设置超时，超时后调用 tasks/cancel。对于长任务，使用 tasks/subscribe 订阅SSE事件流，而不是轮询 tasks/get。

A2A协议演进与生态现状（2026）

时间线

时间 里程碑
2025 Q2 Google发布A2A协议草案
2025 Q4 50+企业签署互操作承诺（Salesforce、SAP、ServiceNow等）
2026 Q1 A2A v0.2，增加流式Artifact和Push Notification
2026 Q2 Linux Foundation旗下开放治理，与MCP并列成「AI协议双子星」

生态参与者

  • 协议层：Google A2A、Anthropic MCP、IBM ACP（Agent Communication Protocol）
  • 框架层：LangGraph、CrewAI、AutoGen、Google ADK
  • 平台层：Vertex AI Agent Engine、Azure AI Agent Service
  • 工具层：各类MCP Server（数据库、API、文件系统）

2026年的共识是：MCP + A2A 组合使用，而非二选一。

错误处理与重试策略

Task级别的错误分类

错误类型 HTTP状态 处理策略
临时故障 503 指数退避重试，最多3次
认证失败 401 刷新Token后重试1次
参数错误 400 不重试，返回input-required
目标Agent过载 429 切换备用Agent或排队

幂等Task发送

async def send_task_idempotent(agent_url: str, task_id: str, message: dict) -> dict:
    for attempt in range(3):
        try:
            resp = await client.post(f"{agent_url}/a2a", json={
                "jsonrpc": "2.0",
                "method": "tasks/send",
                "id": str(uuid.uuid4()),
                "params": {"id": task_id, "message": message},
            })
            if resp.status_code == 200:
                return resp.json()["result"]
            if resp.status_code in (503, 429):
                await asyncio.sleep(2 ** attempt)
                continue
            raise AgentError(resp.text)
        except httpx.TimeoutException:
            await asyncio.sleep(2 ** attempt)
    raise AgentError(f"Task {task_id} failed after 3 retries")

同一个 task_id 重复发送，目标Agent应返回已有Task状态而非创建新Task——这是A2A的幂等保障。

流式响应与SSE实战

长任务（如生成报告、数据分析）不能等完成后一次性返回，需要 tasks/subscribe 订阅SSE事件流。

SSE事件类型

event: task-status-update
data: {"id": "task-001", "status": "working", "progress": 30}

event: task-artifact-update
data: {"id": "task-001", "artifact": {"name": "初步分析", "parts": [...]}}

event: task-status-update
data: {"id": "task-001", "status": "completed"}

Python SSE客户端

async def subscribe_task(agent_url: str, task_id: str):
    async with httpx.AsyncClient() as client:
        async with client.stream("POST", f"{agent_url}/a2a/subscribe", json={
            "jsonrpc": "2.0",
            "method": "tasks/subscribe",
            "params": {"id": task_id},
        }) as response:
            async for line in response.aiter_lines():
                if line.startswith("data: "):
                    event = json.loads(line[6:])
                    yield event
                    if event.get("status") in ("completed", "failed"):
                        break

协调者Agent可将中间进度转发给用户，避免长时间无响应。

与LangGraph/CrewAI框架的关系

维度 A2A协议 LangGraph/CrewAI
层级 传输协议（类似HTTP） 应用框架（类似Spring）
互操作 跨团队、跨语言 框架内组件协作
状态管理 Task生命周期 Graph State / Crew Memory
适用场景 企业间Agent互调 单团队内部编排

推荐组合：团队内部用LangGraph编排，跨团队/跨组织用A2A协议暴露Agent能力。LangGraph节点可以是一个A2A Client，调用外部Agent作为工具。

多租户与企业级治理

Agent Registry设计

class AgentRegistry:
    def register(self, tenant_id: str, agent_card: dict) -> str:
        agent_id = f"{tenant_id}/{agent_card['name']}"
        self.store.put(agent_id, {
            **agent_card,
            "tenant_id": tenant_id,
            "registered_at": datetime.utcnow().isoformat(),
            "rate_limit": 100,  # requests/minute
        })
        return agent_id

    def discover(self, tenant_id: str, skill: str) -> list[dict]:
        agents = self.store.list(prefix=f"{tenant_id}/")
        return [a for a in agents if any(s["id"] == skill for s in a.get("skills", []))]

企业治理清单

  • 租户隔离：每个租户的Agent只能发现同租户的Agent（或显式授权的跨租户Agent）
  • 限流：按Agent/租户维度限流，防止协调者拖垮专业Agent
  • 审计：所有Task记录持久化，含请求方、目标方、耗时、结果
  • 版本管理：Agent Card带版本号，Breaking Change需新版本号

从0搭建A2A Agent Registry

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI(title="A2A Agent Registry")

class AgentRegistration(BaseModel):
    tenant_id: str
    agent_card: dict

registry: dict[str, dict] = {}

@app.post("/agents/register")
async def register(reg: AgentRegistration):
    agent_id = f"{reg.tenant_id}/{reg.agent_card['name']}"
    registry[agent_id] = reg.agent_card
    return {"agent_id": agent_id, "url": reg.agent_card["url"]}

@app.get("/agents/discover")
async def discover(tenant_id: str, skill: str = None):
    results = []
    for agent_id, card in registry.items():
        if not agent_id.startswith(f"{tenant_id}/"):
            continue
        if skill and not any(s["id"] == skill for s in card.get("skills", [])):
            continue
        results.append({"agent_id": agent_id, **card})
    return results

@app.get("/agents/{agent_id}/.well-known/agent.json")
async def get_agent_card(agent_id: str):
    if agent_id not in registry:
        raise HTTPException(404, "Agent not found")
    return registry[agent_id]

部署在K8s上，配合Ingress暴露，即成为企业内部的Agent服务发现中心。

面试考点与选型指南（进阶篇）

Q5：A2A和gRPC/Webhook比有什么优势？

标准化Task生命周期、Agent发现、Artifact格式。gRPC每个团队自定义proto，无法跨组织互操作。A2A是AI Agent领域的「HTTP」——统一协议降低集成成本。

Q6：协调者Agent挂了怎么办？

Task状态在目标Agent侧持久化。协调者恢复后通过 tasks/get 查询未完成任务状态，继续编排。关键Task状态不要只放协调者内存。

Q7：A2A适合实时对话场景吗？

适合。用 tasks/subscribe SSE推送中间结果，用户体验接近流式聊天。但延迟比直接LLM调用多一跳（协调者→专业Agent），高实时场景需评估。

Q8：如何测试A2A Agent？

三层测试：(1) 单元测试——Mock Task API验证状态机；(2) 契约测试——验证Agent Card格式和Task响应Schema；(3) 集成测试——启动真实Agent端到端委派。

A2A生产环境Checklist

上线前逐项确认：

  • Agent Card托管在 /.well-known/agent.json，HTTPS访问
  • 所有Task API实现幂等（相同task_id返回已有状态）
  • 认证：OAuth2 Bearer Token或mTLS
  • 超时：默认30秒，长任务用subscribe模式
  • 限流：按调用方Agent限流（100 req/min起步）
  • 日志：每个Task记录完整审计日志
  • 监控：Task成功率、P99延迟、Agent可用性
  • 降级：目标Agent不可用时返回友好错误，协调者可切换备用Agent
  • 测试：契约测试 + 端到端集成测试
  • 文档：每个Skill有清晰的输入输出说明和示例

未来展望：A2A + MCP + 策略层

2026下半年的技术趋势是三层协议栈趋于成熟：

策略层（Policy）  → 谁能调用谁、数据出境限制、合规审计
协作层（A2A）     → Agent间任务委派与状态同步
工具层（MCP）     → Agent访问数据和API

对于国内企业，还需关注：

  • 数据主权：Agent Card中声明数据驻留区域
  • 合规：金融/医疗Agent需额外资质认证
  • 国产化：支持国产LLM作为Agent后端（通义、文心、DeepSeek等）

A2A协议的语言无关性在这里是优势——后端LLM可以替换，协议层不变。

多Agent系统的性能与成本考量

延迟分析

用户 → 协调者Agent(200ms) → 专业Agent(500ms) → MCP工具(300ms) → 返回
总延迟 ≈ 1000-2000ms（vs 单Agent直接处理 500-800ms）

每多一个A2A跳转，延迟增加200-500ms。对于实时对话场景：

策略 延迟 适用
单Agent + 多MCP工具 低 简单问答
协调者 + 1个专业Agent 中 大多数客服场景
协调者 + 链式多Agent 高 复杂审批/跨域编排

Token成本

每个Agent有独立的系统提示词和工具定义。3个Agent协作的Token成本不是1个Agent的3倍（因为有上下文隔离），但通常是 1.5-2.5倍。协调者应尽可能传递结构化状态而非完整对话，控制成本。

A2A安全攻防：常见攻击面

攻击 描述 防御
Agent冒充 伪造Agent Card指向恶意端点 Registry白名单 + mTLS
Task注入 恶意协调者发送超大Task耗尽资源 请求体大小限制 + 超时
权限提升 协调者调用未授权的Skill Skill级别RBAC
数据渗出 专业Agent返回超范围数据 输出过滤 + 审计

生产环境必须在API Gateway层做统一认证，不要依赖各Agent自行校验。

动手实战：30分钟搭建双Agent协作Demo

架构

用户 → CustomerServiceOrchestrator → OrderAgent (A2A)
                                  → LogisticsAgent (A2A)

Step 1：启动OrderAgent（FastAPI）

from fastapi import FastAPI
app = FastAPI()

@app.get("/.well-known/agent.json")
async def agent_card():
    return OrderAgent.agent_card

@app.post("/a2a")
async def handle_a2a(request: dict):
    method = request.get("method")
    if method == "tasks/send":
        return {"jsonrpc": "2.0", "id": request["id"],
                "result": await order_agent.handle_task_send(request["params"])}

Step 2：启动LogisticsAgent（同样结构，不同Skill）

Step 3：启动协调者

orchestrator = CustomerServiceOrchestrator(agents={
    "order": "http://localhost:8001",
    "logistics": "http://localhost:8002",
})
result = await orchestrator.handle_user_query("我上周买的耳机怎么还没到？")
print(result)

Step 4：验证

  • 输入包含「订单」关键词 → 路由到OrderAgent
  • 输入包含「物流」关键词 → 路由到LogisticsAgent
  • 检查返回的Task状态是否为 completed
  • 用curl直接调用 /.well-known/agent.json 验证Agent Card格式

扩展方向

  1. 加入Agent Registry，协调者从Registry发现Agent而非硬编码URL
  2. 实现 tasks/subscribe SSE，实时推送处理进度
  3. 给每个Agent接入MCP工具（订单DB、物流API）

深度辨析：A2A不是银弹的5个场景

场景1：单Agent + MCP就够 — 如果你的AI助手只需要查数据库、调API、读文件，MCP完全够用。引入A2A反而增加一跳延迟和调试复杂度。

场景2：团队内部协作 — 同一个团队内的Agent编排，用LangGraph/AutoGen等框架更高效。A2A的价值在跨团队、跨组织、跨技术栈的互操作。

场景3：实时性要求极高 — 每多一个A2A跳转增加200-500ms。语音对话、实时翻译等场景不适合多层Agent委派。

场景4：简单问答 — 「今天天气怎么样」不需要路由到天气Agent、再路由到定位Agent。过度编排是反模式。

场景5：数据安全要求极高 — Agent间传递的数据越多，泄露面越大。如果每个Agent都需要完整用户上下文，A2A的上下文隔离优势就不存在了。此时应重新评估Agent拆分粒度。

正确的A2A使用信号：多个团队独立开发专业Agent、需要动态发现能力、任务需要跨域委派、异构技术栈（Python Agent调Go Agent）。

2026年A2A与多Agent生态趋势

趋势1：协议标准化加速。 A2A和MCP均进入Linux Foundation治理，2026年下半年预计发布1.0正式版，企业采纳门槛大幅降低。

趋势2：Agent Marketplace。 类似App Store的Agent市场出现，企业可订阅第三方专业Agent（法务、财务、HR），通过A2A协议即插即用。

趋势3：与低代码平台融合。 Dify、Coze等平台支持将工作流导出为A2A兼容的Agent，降低Agent开发门槛。

趋势4：Agent安全合规框架。 OWASP发布AI Agent Top 10安全风险，A2A的安全模型（认证、授权、审计）成为企业采购Agent服务的评估标准。

趋势5：边缘Agent协作。 物联网边缘设备上的轻量Agent通过A2A与云端Agent协作，处理本地实时决策+云端复杂推理的混合模式。

A2A协议术语速查表

术语 含义 面试怎么说
Agent Card Agent能力声明JSON 「托管在/.well-known/agent.json」
Task 有生命周期的工请求 「A2A的核心抽象，非无状态调用」
Skill Agent Card中声明的具体能力 「按Skill发现和路由Agent」
Artifact Task产出的结果 「支持文本、数据、文件多种格式」
Part Message/Artifact的内容片段 「类似MIME多部分」
SSE Server-Sent Events 「tasks/subscribe实时推送进度」
JSON-RPC A2A的传输协议 「标准方法：tasks/send, tasks/get」
Orchestrator 协调者Agent 「按意图路由到专业Agent」
Registry Agent注册中心 「企业级Agent发现服务」
input-required Task状态之一 「缺少信息时请求用户补充」

推荐学习路线（2周）

第1周：协议理解+单Agent — 阅读A2A官方规范 → 理解Agent Card和Task状态机 → 实现一个OrderAgent → 测试tasks/send和tasks/get。

第2周：多Agent协作+生产化 — 实现协调者Orchestrator → 搭建Agent Registry → 对比A2A和MCP的关系 → 实现SSE订阅 → 准备面试8题。

最后送一句话：MCP让Agent能用手，A2A让Agent能组队。2026年的AI应用，两手都要硬。

总结与延伸阅读

Google A2A协议填补了多Agent互操作的标准化空白，与MCP形成「工具层 + 协作层」的完整协议栈。

设计要点回顾：

  1. A2A解决Agent间协作，MCP解决Agent与工具的连接
  2. Agent Card是发现机制，声明在 /.well-known/agent.json
  3. Task是核心抽象，有完整的生命周期状态机
  4. 多Agent编排用协调者模式，按意图路由到专业Agent
  5. 生产必须HTTPS + OAuth2 + Task全链路追踪

相关阅读：

权威参考：

本站提供浏览器本地工具，免注册即可试用 →

#A2A协议#Agent互操作#Google A2A#多Agent协作#MCP对比#2026