Google A2A 协议实战：用 TypeScript 构建生产级多 Agent 协作系统

当 MCP（Model Context Protocol）解决了「AI 模型如何调用工具」的问题后，2026 年 AI 工程界面临一个更宏大的挑战：多个 AI Agent 之间如何互相发现、协商和协作？ Google 在 2025 年 4 月发布的 A2A（Agent-to-Agent）协议正是为了解决这个问题。截至目前，A2A 已获得 Salesforce、SAP、LangChain 等 50+ 企业级平台的支持，GitHub 上基于 A2A 的 Agent 仓库增长超过 300%。如果你已经掌握了 MCP，那 A2A 就是你需要补齐的下一块拼图——它将你的 Agent 从「工具使用者」升级为「协作者」。

🔗 一、A2A 协议核心原理与设计哲学

1.1 为什么 MCP 不够用？

MCP 解决了 N×M 的工具集成问题：N 个模型连接 M 个工具，只需 N+M 个适配器。但它有一个根本性限制——MCP 是 Client-Server 架构，工具是被动的。

在真实的业务场景中，你经常需要多个 Agent 协同完成一个复杂任务：

• 旅行规划场景：行程 Agent 调用酒店 Agent 查询空房，再调用航班 Agent 比价，最后由保险 Agent 生成保障方案
• 代码审查场景：安全 Agent 扫描漏洞，性能 Agent 分析瓶颈，架构 Agent 评估设计，最后由总结 Agent 汇总报告
• 客服场景：对话 Agent 处理用户问题，知识库 Agent 检索资料，工单 Agent 创建跟进，情绪 Agent 监控用户满意度

这些场景中，Agent 之间的关系是对等协商而非主从调用。每个 Agent 有自己的专业领域、决策能力和状态管理。MCP 的 Client-Server 模型无法表达这种对等关系。

💡 **提示：**MCP 和 A2A 不是竞争关系，而是互补的两层。MCP 是「Agent ↔ 工具」的通信协议，A2A 是「Agent ↔ Agent」的通信协议。一个完整的多 Agent 系统同时需要两者。

1.2 A2A 协议架构

A2A 基于 HTTP + JSON-RPC 2.0，定义了 Agent 之间交互的四个核心概念：

每个 A2A Agent 通过一个标准的 /.well-known/agent.json 端点暴露自己的 Agent Card，描述自身的能力、支持的输入输出格式、以及认证方式。这就像 Agent 世界的「名片」。

// agent-card.json — 一个典型的 Agent Card 定义
{
  "name": "CodeReviewAgent",
  "description": "专业代码审查 Agent，支持安全扫描、性能分析和架构评估",
  "url": "https://codereview.example.com/a2a",
  "version": "1.0.0",
  "capabilities": {
    "streaming": true,
    "pushNotifications": false,
    "stateTransitionHistory": true
  },
  "authentication": {
    "schemes": ["Bearer"]
  },
  "defaultInputModes": ["text", "file"],
  "defaultOutputModes": ["text", "file"],
  "skills": [
    {
      "id": "security-scan",
      "name": "安全漏洞扫描",
      "description": "扫描代码中的安全漏洞，包括 SQL 注入、XSS、SSRF 等",
      "tags": ["security", "vulnerability"],
      "examples": ["扫描这个 API 接口的安全风险"]
    },
    {
      "id": "performance-analysis",
      "name": "性能分析",
      "description": "分析代码性能瓶颈，提供优化建议",
      "tags": ["performance", "optimization"],
      "examples": ["分析这个函数的性能问题"]
    }
  ]
}

1.3 A2A vs MCP：核心差异对比

📌 **记住：**选择 MCP 还是 A2A 取决于你连接的目标类型。如果目标是一个「被动的工具」（接受输入、返回输出），用 MCP。如果目标是一个「主动的协作者」（有自己的决策逻辑、可能需要多轮对话），用 A2A。

🚀 二、TypeScript 实战：构建 A2A 多 Agent 系统

2.1 项目结构设计

一个生产级的 A2A 系统需要以下核心模块：

# 项目目录结构
a2a-agent-system/
├── src/
│   ├── core/
│   │   ├── agent-card.ts      # Agent Card 定义与注册
│   │   ├── task-manager.ts    # Task 生命周期管理
│   │   ├── message-bus.ts     # 消息路由与分发
│   │   └── discovery.ts       # Agent 发现服务
│   ├── agents/
│   │   ├── base-agent.ts      # Agent 基类
│   │   ├── code-review.ts     # 代码审查 Agent
│   │   ├── security-scan.ts   # 安全扫描 Agent
│   │   └── summary.ts         # 汇总 Agent
│   ├── transport/
│   │   ├── http-transport.ts  # HTTP 传输层
│   │   └── sse-transport.ts   # SSE 流式传输层
│   └── index.ts               # 入口文件
├── package.json
└── tsconfig.json

2.2 实现 A2A Agent 基类

下面是一个完整的 A2A Agent 基类实现，包含 Agent Card 注册、Task 处理和消息通信：

// src/core/agent-card.ts — Agent Card 类型定义
export interface AgentSkill {
  id: string
  name: string
  description: string
  tags: string[]
  examples?: string[]
}

export interface AgentCard {
  name: string
  description: string
  url: string
  version: string
  capabilities: {
    streaming: boolean
    pushNotifications: boolean
    stateTransitionHistory: boolean
  }
  authentication?: {
    schemes: string[]
  }
  defaultInputModes: string[]
  defaultOutputModes: string[]
  skills: AgentSkill[]
}

// src/agents/base-agent.ts — Agent 基类
import { randomUUID } from 'crypto'

export type TaskState = 'submitted' | 'working' | 'input-required' | 'completed' | 'failed' | 'canceled'

export interface Task {
  id: string
  state: TaskState
  messages: Message[]
  artifacts: Artifact[]
  metadata: Record<string, unknown>
  createdAt: Date
  updatedAt: Date
}

export interface Message {
  role: 'user' | 'agent'
  parts: MessagePart[]
  messageId: string
}

export type MessagePart = 
  | { type: 'text'; text: string }
  | { type: 'file'; file: { name: string; mimeType: string; data: string } }
  | { type: 'data'; data: unknown; mimeType: string }

export interface Artifact {
  name: string
  description?: string
  parts: MessagePart[]
  index: number
  append?: boolean
  lastChunk?: boolean
}

export abstract class BaseAgent {
  protected tasks: Map<string, Task> = new Map()
  
  constructor(
    protected readonly card: import('./agent-card').AgentCard
  ) {}

  // 获取 Agent Card（供服务发现）
  getCard() {
    return this.card
  }

  // 创建新 Task
  createTask(): Task {
    const task: Task = {
      id: randomUUID(),
      state: 'submitted',
      messages: [],
      artifacts: [],
      metadata: {},
      createdAt: new Date(),
      updatedAt: new Date()
    }
    this.tasks.set(task.id, task)
    return task
  }

  // 获取 Task 状态
  getTask(taskId: string): Task | undefined {
    return this.tasks.get(taskId)
  }

  // 处理消息（子类实现具体逻辑）
  async handleMessage(taskId: string, message: Message): Promise<Task> {
    const task = this.tasks.get(taskId)
    if (!task) throw new Error(`Task ${taskId} not found`)

    task.messages.push(message)
    task.state = 'working'
    task.updatedAt = new Date()

    try {
      const result = await this.process(task, message)
      task.state = result.state
      if (result.artifact) {
        task.artifacts.push(result.artifact)
      }
    } catch (error) {
      task.state = 'failed'
      task.metadata.error = (error as Error).message
    }

    task.updatedAt = new Date()
    return task
  }

  // 子类必须实现的核心处理逻辑
  protected abstract process(
    task: Task,
    message: Message
  ): Promise<{ state: TaskState; artifact?: Artifact }>
}

2.3 实现一个代码审查 Agent

下面用上面的基类实现一个完整的代码审查 Agent，它可以调用安全扫描子 Agent：

// src/agents/code-review.ts — 代码审查 Agent 实现
import { BaseAgent, Task, Message, Artifact, type MessagePart } from './base-agent'
import type { AgentCard } from '../core/agent-card'

export class CodeReviewAgent extends BaseAgent {
  constructor(private securityAgentUrl: string) {
    const card: AgentCard = {
      name: 'CodeReviewAgent',
      description: '专业代码审查 Agent，支持安全扫描、性能分析和架构评估',
      url: 'https://codereview.example.com/a2a',
      version: '1.0.0',
      capabilities: {
        streaming: true,
        pushNotifications: false,
        stateTransitionHistory: true
      },
      defaultInputModes: ['text', 'file'],
      defaultOutputModes: ['text'],
      skills: [
        {
          id: 'full-review',
          name: '完整代码审查',
          description: '对代码进行安全、性能、架构的全面审查',
          tags: ['review', 'security', 'performance'],
          examples: ['审查这段代码的安全性和性能']
        }
      ]
    }
    super(card)
  }

  protected async process(task: Task, message: Message) {
    // 提取代码内容
    const codeText = message.parts
      .filter((p): p is { type: 'text'; text: string } => p.type === 'text')
      .map(p => p.text)
      .join('\n')

    // 步骤 1：本地性能分析
    const perfIssues = this.analyzePerformance(codeText)

    // 步骤 2：调用安全扫描 Agent（A2A 跨 Agent 协作）
    const securityResult = await this.callSecurityAgent(codeText)

    // 步骤 3：汇总结果
    const summary = this.generateSummary(perfIssues, securityResult)
    
    const artifact: Artifact = {
      name: 'code-review-report',
      description: '代码审查报告',
      parts: [{ type: 'text', text: summary }],
      index: 0,
      lastChunk: true
    }

    return { state: 'completed' as const, artifact }
  }

  private analyzePerformance(code: string): string[] {
    const issues: string[] = []
    // 检查常见的性能问题
    if (code.includes('.forEach(') && code.includes('await ')) {
      issues.push('⚠️ 在 forEach 中使用 await，建议改用 Promise.all() 或 for...of')
    }
    if (code.includes('JSON.parse(JSON.stringify(')) {
      issues.push('⚠️ 使用 JSON 序列化做深拷贝，性能较差，建议使用 structuredClone()')
    }
    if (/\.filter\(.*\)\.map\(/.test(code)) {
      issues.push('💡 filter + map 可以合并为 reduce 或使用 Array.from 的 filter 参数')
    }
    if (issues.length === 0) {
      issues.push('✅ 未发现明显的性能问题')
    }
    return issues
  }

  private async callSecurityAgent(code: string): Promise<string> {
    // A2A 协议调用：向安全 Agent 发送 Task
    const response = await fetch(`${this.securityAgentUrl}/a2a/tasks/send`, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({
        id: crypto.randomUUID(),
        message: {
          role: 'user',
          parts: [{ type: 'text', text: `请扫描以下代码的安全漏洞：\n\n${code}` }],
          messageId: crypto.randomUUID()
        }
      })
    })

    if (!response.ok) {
      return '⚠️ 安全扫描 Agent 暂时不可用'
    }

    const result = await response.json() as { artifacts?: Artifact[] }
    const textParts = result.artifacts?.[0]?.parts
      ?.filter((p): p is { type: 'text'; text: string } => p.type === 'text')
      .map(p => p.text)
    
    return textParts?.join('\n') ?? '未返回扫描结果'
  }

  private generateSummary(perfIssues: string[], securityResult: string): string {
    return `# 代码审查报告\n\n## 🔒 安全扫描结果\n${securityResult}\n\n## ⚡ 性能分析\n${perfIssues.join('\n')}\n\n## 📊 总体评价\n${perfIssues.length <= 1 && securityResult.includes('✅') ? '✅ 代码质量良好' : '⚠️ 存在需要关注的问题，请参考上述建议进行修复'}`
  }
}

2.4 A2A HTTP Server：暴露 Agent 服务

每个 A2A Agent 需要暴露 HTTP 端点供其他 Agent 调用。下面是一个基于原生 http 模块的轻量实现：

// src/transport/http-transport.ts — A2A HTTP 传输层
import http from 'http'
import type { BaseAgent, Message } from '../agents/base-agent'

export class A2AHttpServer {
  constructor(private agent: BaseAgent) {}

  start(port: number) {
    const server = http.createServer(async (req, res) => {
      // CORS 头
      res.setHeader('Access-Control-Allow-Origin', '*')
      res.setHeader('Access-Control-Allow-Methods', 'GET, POST, OPTIONS')
      res.setHeader('Access-Control-Allow-Headers', 'Content-Type, Authorization')

      if (req.method === 'OPTIONS') {
        res.writeHead(204)
        res.end()
        return
      }

      const url = new URL(req.url!, `http://localhost:${port}`)

      // Agent Card 发现端点
      if (url.pathname === '/.well-known/agent.json' && req.method === 'GET') {
        res.writeHead(200, { 'Content-Type': 'application/json' })
        res.end(JSON.stringify(this.agent.getCard()))
        return
      }

      // 创建 Task
      if (url.pathname === '/a2a/tasks' && req.method === 'POST') {
        const task = this.agent.createTask()
        res.writeHead(201, { 'Content-Type': 'application/json' })
        res.end(JSON.stringify(task))
        return
      }

      // 发送消息到 Task
      if (url.pathname?.startsWith('/a2a/tasks/') && url.pathname.endsWith('/send') && req.method === 'POST') {
        const body = await this.readBody(req)
        const taskId = url.pathname.split('/')[3]
        const message = body.message as Message

        try {
          const task = await this.agent.handleMessage(taskId, message)
          res.writeHead(200, { 'Content-Type': 'application/json' })
          res.end(JSON.stringify(task))
        } catch (error) {
          res.writeHead(404, { 'Content-Type': 'application/json' })
          res.end(JSON.stringify({ error: (error as Error).message }))
        }
        return
      }

      // 查询 Task 状态
      if (url.pathname?.startsWith('/a2a/tasks/') && req.method === 'GET') {
        const taskId = url.pathname.split('/')[3]
        const task = this.agent.getTask(taskId)
        if (task) {
          res.writeHead(200, { 'Content-Type': 'application/json' })
          res.end(JSON.stringify(task))
        } else {
          res.writeHead(404)
          res.end(JSON.stringify({ error: 'Task not found' }))
        }
        return
      }

      res.writeHead(404)
      res.end('Not Found')
    })

    server.listen(port, () => {
      console.log(`🚀 A2A Agent [${this.agent.getCard().name}] listening on port ${port}`)
      console.log(`📋 Agent Card: http://localhost:${port}/.well-known/agent.json`)
    })
  }

  private readBody(req: http.IncomingMessage): Promise<any> {
    return new Promise((resolve, reject) => {
      let data = ''
      req.on('data', chunk => data += chunk)
      req.on('end', () => {
        try { resolve(JSON.parse(data)) }
        catch (e) { reject(e) }
      })
      req.on('error', reject)
    })
  }
}

💡 三、多 Agent 编排模式与生产实践

3.1 三种编排模式

在实际应用中，多 Agent 协作通常采用以下三种编排模式：

下面是一个并行扇出 + 汇总模式的实现：

// src/orchestrator.ts — 多 Agent 编排器
interface AgentEndpoint {
  url: string
  skillId: string
  name: string
}

export class AgentOrchestrator {
  // 并行扇出模式：同时调用多个 Agent，汇总结果
  async fanOut(
    agents: AgentEndpoint[],
    inputText: string
  ): Promise<Map<string, string>> {
    const results = new Map<string, string>()

    // 并行发起所有 Agent 请求
    const promises = agents.map(async (agent) => {
      try {
        const taskRes = await fetch(`${agent.url}/a2a/tasks`, { method: 'POST' })
        const task = await taskRes.json() as { id: string }

        const sendRes = await fetch(`${agent.url}/a2a/tasks/${task.id}/send`, {
          method: 'POST',
          headers: { 'Content-Type': 'application/json' },
          body: JSON.stringify({
            message: {
              role: 'user',
              parts: [{ type: 'text', text: inputText }],
              messageId: crypto.randomUUID()
            }
          })
        })

        const result = await sendRes.json() as { artifacts?: { parts: { text: string }[] }[] }
        const output = result.artifacts?.[0]?.parts?.[0]?.text ?? '无结果'
        results.set(agent.name, output)
      } catch (error) {
        results.set(agent.name, `❌ 调用失败: ${(error as Error).message}`)
      }
    })

    // 设置超时保护（30秒）
    await Promise.allSettled(
      promises.map(p => Promise.race([
        p,
        new Promise((_, reject) => setTimeout(() => reject(new Error('Timeout')), 30_000))
      ]))
    )

    return results
  }

  // 串行管道模式：依次调用，前一个的输出作为后一个的输入
  async pipeline(
    agents: AgentEndpoint[],
    initialInput: string
  ): Promise<string> {
    let currentInput = initialInput

    for (const agent of agents) {
      const taskRes = await fetch(`${agent.url}/a2a/tasks`, { method: 'POST' })
      const task = await taskRes.json() as { id: string }

      const sendRes = await fetch(`${agent.url}/a2a/tasks/${task.id}/send`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
          message: {
            role: 'user',
            parts: [{ type: 'text', text: currentInput }],
            messageId: crypto.randomUUID()
          }
        })
      })

      const result = await sendRes.json() as { artifacts?: { parts: { text: string }[] }[] }
      currentInput = result.artifacts?.[0]?.parts?.[0]?.text ?? ''
    }

    return currentInput
  }
}

3.2 安全与认证最佳实践

多 Agent 系统的安全性是生产环境中最容易被忽视的问题。以下是三个关键要点：

**❌ 危险做法：**直接信任任何 Agent 的请求，不做身份验证

// ❌ 错误写法 — 裸奔的 Agent 端点
const server = http.createServer(async (req, res) => {
  // 任何人发来的请求都直接处理，没有认证！
  const task = await agent.handleMessage(taskId, message)
  res.end(JSON.stringify(task))
})

**✅ 正确做法：**验证 Agent 身份，限制调用权限

// ✅ 正确写法 — 带认证的 Agent 端点
const TRUSTED_AGENTS = new Map<string, string>([
  ['https://security.example.com/a2a', 'sk-agent-security-key-xxx'],
  ['https://perf.example.com/a2a', 'sk-agent-perf-key-yyy']
])

const server = http.createServer(async (req, res) => {
  const authHeader = req.headers.authorization
  if (!authHeader?.startsWith('Bearer ')) {
    res.writeHead(401)
    res.end(JSON.stringify({ error: 'Missing authentication' }))
    return
  }

  const token = authHeader.slice(7)
  const isTrusted = [...TRUSTED_AGENTS.values()].includes(token)
  
  if (!isTrusted) {
    res.writeHead(403)
    res.end(JSON.stringify({ error: 'Unauthorized agent' }))
    return
  }

  // 通过认证后才处理请求
  const task = await agent.handleMessage(taskId, message)
  res.end(JSON.stringify(task))
})

⚠️ **警告：**生产环境中，Agent 之间的通信必须使用 mTLS（双向 TLS）或 Bearer Token 认证。裸露的 A2A 端点等同于向互联网暴露一个无认证的 API。

3.3 常见坑点与避坑指南

在构建多 Agent 系统时，以下是我在实际项目中踩过的坑：

坑点 1：无限递归调用

Agent A 调用 Agent B，Agent B 又调用 Agent A，导致死循环。

// 解决方案：在 Task metadata 中追踪调用链
async handleMessage(taskId: string, message: Message): Promise<Task> {
  const task = this.tasks.get(taskId)!
  
  // 检查调用链深度，防止无限递归
  const callDepth = (task.metadata.callChain as string[] ?? []).length
  if (callDepth >= 5) {
    task.state = 'failed'
    task.metadata.error = '调用链过深，可能存在循环依赖'
    return task
  }

  // 记录当前 Agent 到调用链
  task.metadata.callChain = [
    ...(task.metadata.callChain as string[] ?? []),
    this.card.name
  ]

  return this.process(task, message)
}

坑点 2：长时任务超时

一个 Agent 处理任务时间过长（如 LLM 推理），导致调用方等待超时。

// 解决方案：使用 Task 状态机 + 轮询
async function pollTaskUntilComplete(
  agentUrl: string,
  taskId: string,
  maxWaitMs: number = 60_000
): Promise<Task> {
  const deadline = Date.now() + maxWaitMs

  while (Date.now() < deadline) {
    const res = await fetch(`${agentUrl}/a2a/tasks/${taskId}`)
    const task = await res.json() as Task

    if (task.state === 'completed' || task.state === 'failed' || task.state === 'canceled') {
      return task
    }

    // 指数退避轮询
    await new Promise(r => setTimeout(r, Math.min(1000, 100 * Math.pow(2, Date.now() % 5))))
  }

  throw new Error(`Task ${taskId} timed out after ${maxWaitMs}ms`)
}

坑点 3：Agent Card 缓存失效

Agent 的能力可能随版本更新而变化，但调用方缓存了旧的 Agent Card。

// 解决方案：带版本的缓存策略
class AgentDiscovery {
  private cache = new Map<string, { card: AgentCard; cachedAt: number }>()
  private readonly CACHE_TTL = 5 * 60 * 1000 // 5 分钟

  async discover(agentUrl: string): Promise<AgentCard> {
    const cached = this.cache.get(agentUrl)
    if (cached && Date.now() - cached.cachedAt < this.CACHE_TTL) {
      return cached.card
    }

    const res = await fetch(`${agentUrl}/.well-known/agent.json`)
    if (!res.ok) throw new Error(`Discovery failed: ${res.status}`)

    const card = await res.json() as AgentCard
    this.cache.set(agentUrl, { card, cachedAt: Date.now() })
    return card
  }
}

📊 四、A2A 生态现状与技术选型建议

主流 A2A 框架对比

⚡ **关键结论：**如果你的团队是 TypeScript/JavaScript 栈，优先使用 Google 官方 A2A SDK。如果你需要复杂的编排逻辑，可以结合 LangGraph 的编排能力和 A2A 的通信协议。不要为了「用 A2A」而强行拆分 Agent——单 Agent 能解决的问题，不需要多 Agent。

何时使用 A2A vs 单 Agent？

✅ 总结

A2A 协议代表了 AI 工程从「单 Agent 工具调用」到「多 Agent 协作」的范式升级。它的核心价值不在于技术复杂度，而在于提供了一种标准化的方式让不同组织、不同语言、不同框架构建的 Agent 能够互相协作。

给开发者的三个建议：

1. ✅ 先掌握 MCP，再学 A2A。MCP 是基础，A2A 是进阶。80% 的场景用 MCP 就够了
2. ✅ 从串行管道模式开始。它是最简单、最可控的多 Agent 编排方式，等积累了经验再尝试并行扇出和动态路由
3. ✅ 安全先行。Agent 之间的认证、授权和审计机制必须在第一天就设计好，不要事后补

如果你正在构建 AI 应用，今天就可以从给你的 Agent 加上 /.well-known/agent.json 端点开始，让它具备被其他 Agent 发现的能力。这是拥抱 Agent 协作时代的第一步。

💡 推荐工具：

• 📖 A2A 协议官方规范
• 🔧 jsjson.com JSON 格式化工具 — 格式化你的 Agent Card 定义
• 🔧 jsjson.com JSON Schema 校验 — 校验 A2A 消息格式
• 📚 MCP 协议实战指南 — A2A 的互补协议