Agent 间协议 A2A

MCP 解决"Agent 用工具"。A2A 解决"Agent 用 Agent"。Google 2025 年开源 A2A 协议，Anthropic、Microsoft 跟进。本篇讲清楚多 Agent 协作的标准化路径。

学前说明

5-6 第三章讲了 Supervisor-Worker 等多 Agent 架构，但都是"同一团队 / 同一系统内"的多 Agent。

A2A 解决更难的问题：不同厂商、不同部署、不同所有者的 Agent 怎么协作？

例：

• 你的客服 Agent 需要调用财务团队的"开发票 Agent"
• 你的研究 Agent 需要 OpenAI 的"图像分析 Agent"
• 用户的个人助理 Agent 需要购物平台的"比价 Agent"

类比：

• HTTP 之前：每个网站自己定协议，互不通
• HTTP 之后：浏览器能访问任何网站
• A2A 之前：每个 Agent 自己定协议
• A2A 之后：任何 Agent 能调用任何 Agent

学习目标

• 区分 A2A 与 MCP 的边界
• 理解 Google A2A 协议的核心机制（Agent Card、Task、Message）
• 实现简单的 A2A Server 和 Client
• 设计 Agent 发现和能力协商
• 评估 Agent Marketplace 生态的工程价值
• 处理多 Agent 协作的安全风险

与现有知识的衔接

• 5-6 多 Agent 系统：单系统内多 Agent（前置）
• 5-7 / 13 MCP 协议：Agent 用工具的协议（前置）
• 02 Skills 体系：Skills 是另一种"能力共享"
• 04 Lethal Trifecta：跨 Agent 调用的安全风险

---

第一章：A2A vs MCP

1.1 边界

很多人混淆 A2A 和 MCP。它们都是协议，但解决不同问题：

维度	MCP	A2A
调用方	LLM Agent	Agent
被调用方	Tools / Resources / Prompts	另一个 Agent
抽象级别	函数级（call_tool）	任务级（delegate_task）
状态	通常无状态（stateless）	有状态（task lifecycle）
协作	单次调用	长期协作（多轮、异步）
例子	调用 GitHub API	委派 "做一份市场调研"

1.2 类比

MCP 像调函数：
  result = github.create_issue(repo, title, body)
  
A2A 像找人：
  task_id = research_agent.delegate({
    goal: "调研 AI Agent Marketplace 现状",
    deadline: "24 小时",
    budget: "$10"
  })
  
  // 等结果
  result = await get_task_result(task_id)

调函数和找人的差别：

• 找人有目标，函数有参数
• 找人有时间，函数瞬间
• 找人有成本协商，函数固定
• 找人会回问澄清，函数不会
• 找人结果可能不达预期，函数确定

1.3 何时该 A2A 而不是 MCP

适合 A2A：

• 任务复杂，需要 Agent 自主判断
• 长期运行（小时-天）
• 跨组织（你的 Agent 调别人的 Agent）
• 能力是"完成一类任务"，不是"执行一个动作"

适合 MCP：

• 简单工具调用
• 短期（毫秒-秒）
• 同组织内部
• 能力是"执行明确动作"

1.4 实际架构常组合

你的主 Agent
  ↓ MCP 调本地工具
本地 Tools

  ↓ A2A 委派子任务
其他 Agent（可能是别家的）
  ↓ MCP 调它们自己的工具
它们的 Tools

A2A 是"高层"，MCP 是"底层"。

---

第二章：A2A 协议设计

Google 2025 年开源的 A2A 协议是当前最成熟的方案。

2.1 三个核心概念

1. Agent Card

Agent 的"名片"——告诉别人"我是谁、我能干什么"：

{
  "name": "research-agent",
  "version": "1.0.0",
  "description": "深度研究助手",
  "url": "https://research.example.com/a2a",
  "skills": [
    {
      "id": "deep-research",
      "name": "深度研究",
      "description": "对一个主题做多源调研",
      "inputs": ["topic", "depth", "deadline"],
      "outputs": ["report", "sources"]
    }
  ],
  "authentication": {
    "schemes": ["oauth2", "apikey"]
  },
  "capabilities": {
    "streaming": true,
    "pushNotifications": true
  }
}

发布在 well-known 路径：/.well-known/agent.json，让其他 Agent 能发现。

2. Task

A2A 的核心抽象——委派的任务有完整生命周期：

created → working → input_required（如需澄清） → completed | failed | canceled

{
  "id": "task-abc123",
  "status": {
    "state": "working",
    "message": "已搜索 12 个来源，正在综合"
  },
  "messages": [
    { "role": "user", "content": "调研 AI Agent Marketplace 现状" },
    { "role": "agent", "content": "需要时间范围吗？" },
    { "role": "user", "content": "最近 6 个月" }
  ],
  "artifacts": [
    { "type": "document", "url": "..." }
  ]
}

3. Message

Task 内的对话单元。

{
  "role": "user" | "agent",
  "parts": [
    { "type": "text", "text": "..." },
    { "type": "file", "url": "..." },
    { "type": "data", "data": { ... } }
  ]
}

2.2 调用流程

2.3 与 MCP 的关键差异

机制	MCP	A2A
调用	call_tool	create_task
返回	立即返回	长期任务 ID
轮询	不需要	需要（或 webhook）
协商	不支持	支持（输入不够时 server 主动问）
流式	单次响应流	任务进度流

---

第三章：实现 A2A Server

3.1 暴露 Agent Card

import express from 'express';

const app = express();
app.use(express.json());

const agentCard = {
  name: 'my-research-agent',
  version: '1.0.0',
  url: 'https://my-server.com/a2a',
  skills: [
    {
      id: 'deep-research',
      name: 'Deep Research',
      description: 'Multi-source research with citations',
      inputs: [
        { name: 'topic', required: true },
        { name: 'depth', required: false, default: 'medium' },
        { name: 'sources', required: false, default: ['web'] }
      ]
    }
  ],
  authentication: { schemes: ['apikey'] },
  capabilities: { streaming: true, pushNotifications: true }
};

app.get('/.well-known/agent.json', (req, res) => {
  res.json(agentCard);
});

3.2 创建任务

const tasks = new Map<string, Task>();

app.post('/tasks', async (req, res) => {
  // 1. 认证
  const apiKey = req.headers['x-api-key'];
  if (!validApiKey(apiKey)) {
    return res.status(401).json({ error: 'unauthorized' });
  }
  
  // 2. 创建任务
  const task: Task = {
    id: crypto.randomUUID(),
    status: { state: 'created' },
    messages: [{ role: 'user', parts: req.body.parts }],
    artifacts: [],
    createdAt: new Date(),
    metadata: {
      clientAgent: req.body.from,
      skill: req.body.skill,
    }
  };
  
  tasks.set(task.id, task);
  
  // 3. 异步开始处理
  processTaskAsync(task.id);
  
  // 4. 立即返回任务 ID
  res.json(task);
});

async function processTaskAsync(taskId: string) {
  const task = tasks.get(taskId);
  if (!task) return;
  
  // 更新状态
  task.status = { state: 'working' };
  
  try {
    // 实际工作（Agent Loop）
    const result = await runResearchAgent(task);
    
    // 完成
    task.status = { state: 'completed' };
    task.artifacts = result.artifacts;
    
    // 通知 client（webhook）
    if (task.metadata.webhookUrl) {
      await notifyWebhook(task);
    }
  } catch (err) {
    task.status = { state: 'failed', message: err.message };
  }
}

3.3 任务查询

app.get('/tasks/:id', async (req, res) => {
  const task = tasks.get(req.params.id);
  if (!task) return res.status(404).json({ error: 'not_found' });
  
  // 检查权限：只有创建者能查
  if (!isAuthorized(req, task)) {
    return res.status(403).json({ error: 'forbidden' });
  }
  
  res.json(task);
});

// 流式订阅（SSE）
app.get('/tasks/:id/stream', async (req, res) => {
  const task = tasks.get(req.params.id);
  
  res.setHeader('Content-Type', 'text/event-stream');
  
  const sendUpdate = () => {
    res.write(`data: ${JSON.stringify(task)}\n\n`);
  };
  
  // 监听任务变化
  taskEmitter.on(`task:${task.id}`, sendUpdate);
  
  req.on('close', () => {
    taskEmitter.off(`task:${task.id}`, sendUpdate);
  });
});

3.4 多轮对话（input_required）

Server 发现信息不够时，主动要更多输入：

async function runResearchAgent(task: Task) {
  const initialMessage = task.messages[0];
  
  // 检查信息完整性
  if (!hasEnoughContext(initialMessage)) {
    // 进入 input_required 状态
    task.status = { 
      state: 'input_required',
      message: '需要更多信息：研究的时间范围？地区？'
    };
    
    // 等 client 补充
    await waitForUserInput(task.id);
  }
  
  // 继续工作
  // ...
}

// Client 补充信息
app.post('/tasks/:id/messages', async (req, res) => {
  const task = tasks.get(req.params.id);
  
  if (task.status.state !== 'input_required') {
    return res.status(400).json({ error: 'task_not_waiting_for_input' });
  }
  
  task.messages.push({
    role: 'user',
    parts: req.body.parts
  });
  
  // 触发 Server 继续
  taskEmitter.emit(`task:${task.id}:input`, req.body);
  
  res.json(task);
});

---

第四章：实现 A2A Client

4.1 发现 + 调用

class A2AClient {
  async discover(agentUrl: string): Promise<AgentCard> {
    const wellKnown = `${agentUrl}/.well-known/agent.json`;
    const response = await fetch(wellKnown);
    return await response.json();
  }
  
  async createTask(
    agentUrl: string,
    skill: string,
    parts: MessagePart[]
  ): Promise<Task> {
    const response = await fetch(`${agentUrl}/tasks`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'X-API-Key': this.apiKey
      },
      body: JSON.stringify({
        skill,
        parts,
        from: this.myAgentId,
      })
    });
    
    return await response.json();
  }
  
  async waitForCompletion(
    agentUrl: string,
    taskId: string,
    options: { onProgress?: (task: Task) => void; timeout?: number } = {}
  ): Promise<Task> {
    // 用 SSE 流式订阅
    const eventSource = new EventSource(`${agentUrl}/tasks/${taskId}/stream`);
    
    return new Promise((resolve, reject) => {
      const timeout = setTimeout(() => {
        eventSource.close();
        reject(new Error('timeout'));
      }, options.timeout ?? 24 * 3600 * 1000);
      
      eventSource.onmessage = (e) => {
        const task: Task = JSON.parse(e.data);
        options.onProgress?.(task);
        
        if (task.status.state === 'completed') {
          clearTimeout(timeout);
          eventSource.close();
          resolve(task);
        }
        
        if (task.status.state === 'input_required') {
          // 处理需要补充输入
          this.handleInputRequired(agentUrl, task);
        }
        
        if (task.status.state === 'failed') {
          clearTimeout(timeout);
          eventSource.close();
          reject(new Error(task.status.message));
        }
      };
    });
  }
  
  async sendInput(agentUrl: string, taskId: string, parts: MessagePart[]) {
    return await fetch(`${agentUrl}/tasks/${taskId}/messages`, {
      method: 'POST',
      headers: { /* ... */ },
      body: JSON.stringify({ parts })
    });
  }
}

4.2 你的 Agent 用 A2A

// 例：你的主 Agent 调用研究 Agent
class MainAgent {
  private a2a = new A2AClient();
  
  async handleUserRequest(request: string) {
    if (await this.needsResearch(request)) {
      // 调用研究 Agent
      const card = await this.a2a.discover('https://research.example.com');
      
      const task = await this.a2a.createTask(
        card.url,
        'deep-research',
        [{ type: 'text', text: request }]
      );
      
      // 用户实时看到进度
      const result = await this.a2a.waitForCompletion(card.url, task.id, {
        onProgress: (task) => {
          this.showProgressToUser(task.status.message);
        }
      });
      
      return this.synthesize(result.artifacts);
    }
    
    // 否则直接处理
    return await this.localProcess(request);
  }
}

---

第五章：Agent 发现机制

5.1 静态发现

最简单：你已经知道目标 Agent 的 URL。

const knownAgents = {
  research: 'https://research.example.com',
  finance: 'https://finance.company.com',
  scheduling: 'https://schedule.example.com',
};

适合：

• 内部 Agent
• 数量少
• 你掌控

5.2 注册表发现

像 npm registry：

// 集中注册表
const registry = await fetch('https://a2a-registry.org/agents?capability=research');

// 返回
[
  { url: 'https://research-agent-1.com', rating: 4.5, price: '...' },
  { url: 'https://research-agent-2.com', rating: 4.2, price: '...' },
  ...
]

// 选一个
const chosen = pickBest(results);
const card = await a2a.discover(chosen.url);

适合：

• Agent Marketplace
• 公开 Agent 生态

5.3 联邦发现（Federation）

类似 ActivityPub。每个 Agent 知道几个"邻居"，递归发现：

async function discover(capability: string, depth = 3) {
  const visited = new Set();
  const queue = [...this.knownPeers];
  
  while (queue.length > 0 && depth > 0) {
    const peer = queue.shift();
    if (visited.has(peer)) continue;
    visited.add(peer);
    
    const card = await a2a.discover(peer);
    
    if (cardMatches(card, capability)) {
      return card;
    }
    
    // 问邻居"你认识谁会做 X"
    const recommendations = await a2a.askForReferral(peer, capability);
    queue.push(...recommendations);
    
    depth--;
  }
}

适合：

• 去中心化生态
• 隐私（不依赖中心注册表）

5.4 现实的中间方案

2026 年大多数 A2A 用混合：

应用层硬编码已知 Agent（你信任的）
  +
Agent Marketplace（找新的）
  +
未来：联邦发现（实验中）

---

第六章：能力协商

不同 Agent 能力不同。客户端怎么知道某个 Agent 真的能做你要的事？

6.1 Skill 描述

Agent Card 里的 skill 应该详细：

{
  "skills": [
    {
      "id": "deep-research",
      "name": "Deep Research",
      "description": "Multi-source web research with citations",
      
      "inputs": {
        "topic": { "type": "string", "required": true, "max_length": 500 },
        "depth": { 
          "type": "enum", 
          "values": ["quick", "medium", "thorough"],
          "default": "medium" 
        },
        "sources": {
          "type": "array",
          "items": { "enum": ["web", "academic", "news"] }
        }
      },
      
      "outputs": {
        "report": { "type": "markdown" },
        "sources": { "type": "array", "items": { "type": "url" } }
      },
      
      "constraints": {
        "max_duration": "PT24H",  // ISO 8601: 24小时
        "max_sources": 50
      },
      
      "pricing": {
        "type": "per_task",
        "amount": 5.00,
        "currency": "USD"
      },
      
      "examples": [
        {
          "input": { "topic": "AI Agent Marketplace 2025", "depth": "thorough" },
          "output_preview": "..."
        }
      ]
    }
  ]
}

6.2 协商场景

场景 1：能力不够

Client 请求：研究 50 个来源
Server Agent Card 说：max_sources: 30
→ Client 知道协商：要么减少要求，要么找别的 Agent

场景 2：需要确认

Client：分析这个文档
Server：我支持但需要授权读你的私有 GitHub
→ Client 决定是否授权

场景 3：成本协商

Client：做一个市场调研
Server：根据深度收费 $1-$50
→ Client 选预算

6.3 实现协商

class A2AClient {
  async negotiateAndStart(
    agentUrl: string,
    skill: string,
    request: any
  ) {
    const card = await this.discover(agentUrl);
    
    // 检查能力
    const skillCard = card.skills.find(s => s.id === skill);
    if (!skillCard) throw new Error('skill_not_supported');
    
    // 验证输入
    const violations = validateAgainstSchema(request, skillCard.inputs);
    if (violations.length > 0) {
      // 协商：要么降级要求
      throw new NegotiationError(violations);
    }
    
    // 检查成本
    if (skillCard.pricing.amount > this.budget) {
      throw new BudgetExceeded();
    }
    
    // OK，正式发起
    return await this.createTask(agentUrl, skill, request);
  }
}

---

第七章：Agent Marketplace

类似 MCP Marketplace 但是 Agent 级别。

7.1 现状（2026）

主流方案：

• Google A2A Directory（Google 主导）
• Hugging Face Agent Hub
• 第三方目录（类似 Smithery 之于 MCP）

7.2 评估第三方 Agent

跟评估 MCP 一样小心：

维度	检查
来源	知名公司 / 开源项目 / 个人
评分	用户反馈
安全审查	是否有合规认证
数据处理	你的数据会被它怎么用
SLA	可用性、响应时间承诺
定价	透明、可预测

⚠️ 风险：你委派任务给一个 Agent，它可能：

• 看你的数据
• 调用别的服务（你不知道）
• 泄露你的信息
• 给错误结果

参考 04 Lethal Trifecta：调用外部 Agent 等于把"私有数据"交给"不可信内容"+ "外部通信"。完整 Trifecta。

7.3 企业内部 A2A 网关

企业级方案：所有 A2A 调用通过内部网关：

你的 Agent → 内部 A2A Gateway → 外部 Agent

Gateway 做：
- 数据脱敏
- 调用审计
- 成本控制
- 安全策略
- 故障重试

类似 17 Multi-Model Routing 的 LLM Gateway，但是 Agent 级别。

---

第八章：安全与隐私

A2A 是 04 Lethal Trifecta 的极端场景。

8.1 委派 = 信任

调用别人的 Agent 等于：

• 把你的需求告诉它（信息）
• 让它访问它的工具（行动）
• 等它给你结果（信任）

每个环节都是信任 chain。

8.2 信息泄露

你调用研究 Agent 时：
- 你的研究主题暴露
- 你的研究方向反映商业意图
- 你提供的数据被它看到

→ 谁能保证它不存？
→ 它的提供者不会用来训练？
→ 不会卖给竞品？

防御：

• 不传敏感数据（研究主题脱敏）
• 签 DPA（Data Processing Agreement）
• 只用可信 Agent

8.3 行动滥用

你委派 Agent 做"分析这份文档"：
- Agent 可能调用搜索（合理）
- Agent 可能调用邮件（？）
- Agent 可能调用任何它的工具

→ 你怎么限制它"只能做你要的事"？

A2A 协议里 client 不直接控制 server 的工具。Server 自己决定怎么完成任务。

防御：

• 选成熟的 Agent（有 SLA、有审计）
• Sandbox 提供的数据（什么都不给除了任务描述）
• 监控不寻常的成本/时长

8.4 凭证与认证

// API Key 简单但风险高
const apiKey = 'sk-xxx';  // 泄露 = 灾难

// OAuth 2.1 + scope
const token = await oauth.getToken({
  scopes: ['research:read', 'research:create_task'],
  // 不给 'admin:*'
});

参考 13 MCP 的 OAuth 章。A2A 也用同样规范。

8.5 跨 Agent 的攻击传染

危险场景：

你的 Agent A 调用别人的 Agent B
B 已被 prompt injection
B 返回的结果含有恶意指令
A 看到 B 的结果，被注入
A 干坏事

防御：

• 把外部 Agent 的输出当作"不可信内容"（参考 01 Context Engineering）
• 用 XML 标签隔离
• 不让外部 Agent 的输出直接触发你的工具调用

---

第九章：实战例子

9.1 案例：客服 + 财务 Agent 协作

// 用户："给我开张发票"
// 客服 Agent 识别后委派给财务 Agent

class CustomerServiceAgent {
  async handleInvoiceRequest(user: User, orderId: string) {
    // 用 A2A 调用财务 Agent
    const financeAgentUrl = 'https://finance.internal.company.com';
    
    const task = await this.a2a.createTask(financeAgentUrl, 'create-invoice', [
      { type: 'data', data: { orderId, userId: user.id } }
    ]);
    
    const result = await this.a2a.waitForCompletion(financeAgentUrl, task.id, {
      onProgress: (t) => this.notifyUser(`财务系统：${t.status.message}`)
    });
    
    return result.artifacts.find(a => a.type === 'invoice_pdf');
  }
}

9.2 案例：研究 Agent 的多 Agent 编排

复杂研究任务，主 Agent 协调多个子 Agent：

class ResearchOrchestrator {
  async deepResearch(topic: string) {
    // 1. 委派"网络调研"
    const webTask = await this.a2a.createTask(
      'https://web-research.example.com',
      'web-search',
      [{ type: 'text', text: topic }]
    );
    
    // 2. 同时委派"学术调研"
    const academicTask = await this.a2a.createTask(
      'https://academic.example.com',
      'paper-search',
      [{ type: 'text', text: topic }]
    );
    
    // 3. 同时委派"专家访谈"（可能要排队几小时）
    const expertTask = await this.a2a.createTask(
      'https://expert-network.example.com',
      'find-and-interview-expert',
      [{ type: 'text', text: topic }]
    );
    
    // 4. 等所有完成
    const [web, academic, expert] = await Promise.all([
      this.a2a.waitForCompletion(/* ... */),
      this.a2a.waitForCompletion(/* ... */),
      this.a2a.waitForCompletion(/* ... */),
    ]);
    
    // 5. 综合
    return await this.synthesize([web, academic, expert]);
  }
}

每个子 Agent 是独立的服务、独立的团队维护、独立的扩缩。

9.3 案例：用户的个人助理调用商业 Agent

// 用户：帮我订张去东京的便宜机票

class PersonalAssistantAgent {
  async bookFlight(request: string) {
    // 1. 解析需求
    const intent = await this.parseFlight(request);
    
    // 2. 调用比价 Agent（A2A）
    const compareAgentUrl = 'https://flight-compare.example.com';
    const compareTask = await this.a2a.createTask(
      compareAgentUrl, 
      'compare-flights',
      [{ type: 'data', data: intent }]
    );
    const options = await this.a2a.waitForCompletion(/* ... */);
    
    // 3. 给用户选
    const chosen = await this.askUser(options);
    
    // 4. 调用预订 Agent（不同的）
    const bookingAgentUrl = 'https://booking.example.com';
    const bookingTask = await this.a2a.createTask(
      bookingAgentUrl,
      'book-flight',
      [{ type: 'data', data: chosen }]
    );
    
    // 但这一步要用户授权（涉及付款）
    return await this.a2a.waitForCompletion(/* ... */);
  }
}

注意：跨 Agent 调用涉及钱时，必须用户每步确认。

---

第十章：A2A vs 多 Agent 框架

10.1 框架（LangGraph、AutoGen 等）

5-6 第三章讲的多 Agent 框架（Supervisor-Worker 等）：

特点：

• 同一进程 / 同一团队
• 共享内存、共享状态
• 程序员控制全部
• 紧耦合

10.2 A2A 协议

特点：

• 跨进程 / 跨组织
• 网络通信
• 各自维护
• 松耦合

10.3 选型

场景	用框架	用 A2A
同团队多个 Agent	✅ 简单	浪费
跨团队调用	不能	✅
跨公司调用	不能	✅
需要严格状态共享	✅	困难
高频小调用	✅	网络开销
偶尔大任务	浪费内存	✅

可以共用：内部用框架，对外用 A2A。

---

第十一章：未来方向

11.1 协议成熟化

A2A 2025 才公开，2026 年仍在演进。预期：

• ISO/W3C 讨论标准化
• 跨厂商互操作测试
• 社区的最佳实践沉淀

11.2 Agent 信誉系统

类似 npm 的 weekly downloads + GitHub stars：

• Agent 评分机制
• 历史可靠性数据
• 用户评论

让选 Agent 像选 npm 包。

11.3 经济层

A2A 自然带出Agent 经济：

Agent A 调用 Agent B → 付钱
Agent B 自己用 Agent C → 也付钱
形成调用链

需要：

• 微支付
• 自动化合约（智能合约？）
• 防止套娃式高成本

11.4 Agent Identity

Agent 的"身份"：

• 谁创建的
• 哪些数据训练
• 经过什么审核
• 历史行为

类似 SSL 证书之于网站。

---

第十二章：行动建议

A2A 还相对早期，建议：

起步：

• 内部多 Agent 用 5-6 章的框架（LangGraph 等）
• 不急着用 A2A

观望：

• 关注 Google A2A 进展
• 关注 Anthropic Multi-Agent System
• 看竞争对手是否在用

采用：

• 真有跨组织 Agent 调用需求时
• 内部多团队，接口稳定后

警惕：

• 不要过度工程（小团队不需要 A2A）
• 不要轻信第三方 Agent（参考 04 Trifecta）

---

权威资料

• Google A2A 协议
• Anthropic Multi-Agent Research System
• agents.json
• Microsoft Magentic-One
• 5-6 多 Agent 系统（前置）
• 5-7 / 13 MCP 协议（前置）
• 04 Lethal Trifecta（必读）
• 17 Multi-Model Routing 与成本优化

核对日期：2026-06-12