Agent 协议：MCP 与 A2A 🔗

"标准化协议是 AI 生态系统互联互通的基础设施。"

随着 AI Agent 生态的快速发展，两个重要的开放协议正在成为行业标准：Model Context Protocol (MCP) 用于 AI 模型与工具/数据源的集成，Agent2Agent (A2A) 用于 AI Agent 之间的通信协作。理解这两个协议对于构建现代 AI 应用至关重要。

1. 协议全景图

┌─────────────────────────────────────────────────────────────────────────┐
│                        AI Agent 协议生态                                  │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                          │
│    ┌─────────────────────────────────────────────────────────────┐      │
│    │                    Agent2Agent (A2A)                         │      │
│    │              Agent ←→ Agent 通信与协作                        │      │
│    │    ┌─────────┐   ┌─────────┐   ┌─────────┐                   │      │
│    │    │ Agent A │ ←→│ Agent B │ ←→│ Agent C │                   │      │
│    │    └────┬────┘   └────┬────┘   └────┬────┘                   │      │
│    └─────────│─────────────│─────────────│───────────────────────┘      │
│              │             │             │                               │
│    ┌─────────│─────────────│─────────────│───────────────────────┐      │
│    │         │   Model Context Protocol (MCP)                     │      │
│    │         │     Model ←→ Tools/Data 集成                       │      │
│    │         ▼             ▼             ▼                        │      │
│    │    ┌─────────┐   ┌─────────┐   ┌─────────┐                   │      │
│    │    │ MCP     │   │ MCP     │   │ MCP     │                   │      │
│    │    │ Server  │   │ Server  │   │ Server  │                   │      │
│    │    └────┬────┘   └────┬────┘   └────┬────┘                   │      │
│    │         │             │             │                        │      │
│    │         ▼             ▼             ▼                        │      │
│    │    ┌─────────┐   ┌─────────┐   ┌─────────┐                   │      │
│    │    │Database │   │  API    │   │ Files   │                   │      │
│    │    └─────────┘   └─────────┘   └─────────┘                   │      │
│    └──────────────────────────────────────────────────────────────┘      │
│                                                                          │
└─────────────────────────────────────────────────────────────────────────┘

1.1 协议定位对比

维度	MCP (Model Context Protocol)	A2A (Agent2Agent)
发起方	Anthropic (2024.11)	Google Cloud (2025.04)
核心目标	AI 模型与工具/数据源的标准化集成	AI Agent 之间的互操作性与协作
通信对象	Model ↔ Server (Tools/Resources)	Agent ↔ Agent
主要场景	读取文件、调用 API、执行操作	任务委派、信息交换、协作工作流
协议基础	JSON-RPC 2.0	HTTP, SSE, JSON-RPC
采用方	OpenAI, Google DeepMind, Cursor	50+ 合作伙伴（Salesforce, SAP 等）

1.2 互补关系

场景示例：企业智能工作流

┌─────────────────────────────────────────────────────────────────┐
│ 用户请求："分析最近的销售数据，生成报告并发送给团队"               │
└───────────────────────────────┬─────────────────────────────────┘
                                │
                                ▼
┌─────────────────────────────────────────────────────────────────┐
│                    协调 Agent (Orchestrator)                     │
│  使用 A2A 协议委派任务给专业 Agent                                 │
└───────────┬─────────────────────┬─────────────────────┬─────────┘
            │ A2A                  │ A2A                  │ A2A
            ▼                      ▼                      ▼
   ┌─────────────┐        ┌─────────────┐        ┌─────────────┐
   │  数据分析    │        │  报告生成    │        │  邮件发送    │
   │   Agent     │        │   Agent     │        │   Agent     │
   └──────┬──────┘        └──────┬──────┘        └──────┬──────┘
          │ MCP                   │ MCP                  │ MCP
          ▼                       ▼                      ▼
   ┌─────────────┐        ┌─────────────┐        ┌─────────────┐
   │  Database   │        │   Doc API   │        │  Email API  │
   │  MCP Server │        │  MCP Server │        │  MCP Server │
   └─────────────┘        └─────────────┘        └─────────────┘

2. Model Context Protocol (MCP)

2.1 核心概念

MCP 定义了 AI 模型与外部系统交互的标准化方式，核心架构包含三个角色：

┌────────────────────────────────────────────────────────────────────────┐
│                          MCP 架构                                       │
├────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│  ┌──────────────┐          ┌──────────────┐          ┌──────────────┐  │
│  │   MCP Host   │          │  MCP Client  │          │  MCP Server  │  │
│  │  (如 Cursor) │ ──────── │   (协议层)    │ ──────── │  (工具提供方) │  │
│  └──────────────┘          └──────────────┘          └──────────────┘  │
│        │                          │                          │         │
│        │                          │                          │         │
│        ▼                          ▼                          ▼         │
│  ┌──────────────┐          ┌──────────────┐          ┌──────────────┐  │
│  │ 用户界面      │          │ 消息路由      │          │ 具体实现      │  │
│  │ 权限管理      │          │ 协议转换      │          │ 工具/资源     │  │
│  │ 会话管理      │          │ 连接管理      │          │ 提示模板      │  │
│  └──────────────┘          └──────────────┘          └──────────────┘  │
│                                                                         │
└────────────────────────────────────────────────────────────────────────┘

三个角色：

角色	职责	示例
Host	容纳 AI 应用的宿主环境	Cursor IDE, Claude Desktop
Client	与 Server 建立连接的协议层	MCP SDK 客户端实现
Server	提供工具、资源和提示的服务端	文件系统 Server, GitHub Server

2.2 三大核心能力

MCP Server 可以提供三种类型的能力：

typescript

// 1. Resources - 资源（类似 REST 资源）
// 提供数据读取能力，由 Server 暴露，由 Client 读取

interface Resource {
  uri: string;           // 资源唯一标识，如 "file:///path/to/file.txt"
  name: string;          // 人类可读名称
  description?: string;  // 资源描述
  mimeType?: string;     // MIME 类型
}

// 2. Tools - 工具（模型可调用的函数）
// 允许 AI 执行操作，需要模型主动调用

interface Tool {
  name: string;
  description: string;
  inputSchema: JSONSchema;  // JSON Schema 定义参数
}

// 3. Prompts - 提示模板（可复用的提示）
// 预定义的提示模板，支持参数化

interface Prompt {
  name: string;
  description?: string;
  arguments?: PromptArgument[];
}

2.3 通信协议

MCP 基于 JSON-RPC 2.0，支持多种传输层：

typescript

// 请求格式
interface JSONRPCRequest {
  jsonrpc: "2.0";
  id: string | number;
  method: string;
  params?: object;
}

// 响应格式
interface JSONRPCResponse {
  jsonrpc: "2.0";
  id: string | number;
  result?: any;
  error?: {
    code: number;
    message: string;
    data?: any;
  };
}

// 通知格式（无需响应）
interface JSONRPCNotification {
  jsonrpc: "2.0";
  method: string;
  params?: object;
}

支持的传输层：

传输方式	适用场景	特点
stdio	本地进程通信	简单可靠，适合本地 Server
HTTP + SSE	远程服务	支持 Web 部署，适合云服务
WebSocket	实时双向通信	低延迟，适合高频交互

2.4 实现 MCP Server

基础 Server 实现 (TypeScript)

typescript

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
  ListResourcesRequestSchema,
  ReadResourceRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";

// 创建 Server 实例
const server = new Server(
  {
    name: "my-mcp-server",
    version: "1.0.0",
  },
  {
    capabilities: {
      tools: {},      // 声明支持工具
      resources: {},  // 声明支持资源
    },
  }
);

// 定义工具列表
server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [
    {
      name: "search_codebase",
      description: "在代码库中语义搜索",
      inputSchema: {
        type: "object",
        properties: {
          query: {
            type: "string",
            description: "搜索查询"
          },
          maxResults: {
            type: "number",
            description: "最大结果数",
            default: 10
          }
        },
        required: ["query"]
      }
    },
    {
      name: "read_file",
      description: "读取文件内容",
      inputSchema: {
        type: "object",
        properties: {
          path: { type: "string", description: "文件路径" }
        },
        required: ["path"]
      }
    }
  ]
}));

// 实现工具调用
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;
  
  switch (name) {
    case "search_codebase": {
      const results = await performSearch(args.query, args.maxResults);
      return {
        content: [
          {
            type: "text",
            text: JSON.stringify(results, null, 2)
          }
        ]
      };
    }
    
    case "read_file": {
      const content = await fs.readFile(args.path, "utf-8");
      return {
        content: [
          {
            type: "text",
            text: content
          }
        ]
      };
    }
    
    default:
      throw new Error(`Unknown tool: ${name}`);
  }
});

// 定义资源列表
server.setRequestHandler(ListResourcesRequestSchema, async () => ({
  resources: [
    {
      uri: "config://app-settings",
      name: "应用配置",
      description: "应用程序的配置信息",
      mimeType: "application/json"
    }
  ]
}));

// 实现资源读取
server.setRequestHandler(ReadResourceRequestSchema, async (request) => {
  const { uri } = request.params;
  
  if (uri === "config://app-settings") {
    return {
      contents: [
        {
          uri,
          mimeType: "application/json",
          text: JSON.stringify(appConfig, null, 2)
        }
      ]
    };
  }
  
  throw new Error(`Unknown resource: ${uri}`);
});

// 启动 Server
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("MCP Server running on stdio");
}

main().catch(console.error);

Python 实现

python

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent, Resource

# 创建 Server
server = Server("my-python-server")

# 定义工具
@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="execute_python",
            description="执行 Python 代码并返回结果",
            inputSchema={
                "type": "object",
                "properties": {
                    "code": {
                        "type": "string",
                        "description": "要执行的 Python 代码"
                    }
                },
                "required": ["code"]
            }
        )
    ]

# 实现工具调用
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "execute_python":
        try:
            # 安全考虑：实际应用中需要沙箱执行
            result = eval(arguments["code"])
            return [TextContent(type="text", text=str(result))]
        except Exception as e:
            return [TextContent(type="text", text=f"Error: {e}")]
    
    raise ValueError(f"Unknown tool: {name}")

# 启动 Server
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream)

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

2.5 实现 MCP Client

typescript

import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";
import { spawn } from "child_process";

async function createClient() {
  // 启动 Server 进程
  const serverProcess = spawn("node", ["./my-server.js"]);
  
  // 创建传输层
  const transport = new StdioClientTransport({
    command: "node",
    args: ["./my-server.js"]
  });
  
  // 创建 Client
  const client = new Client(
    { name: "my-client", version: "1.0.0" },
    { capabilities: {} }
  );
  
  // 连接到 Server
  await client.connect(transport);
  
  return client;
}

async function main() {
  const client = await createClient();
  
  // 列出可用工具
  const { tools } = await client.listTools();
  console.log("Available tools:", tools.map(t => t.name));
  
  // 调用工具
  const result = await client.callTool({
    name: "search_codebase",
    arguments: { query: "authentication", maxResults: 5 }
  });
  
  console.log("Search results:", result.content);
  
  // 列出资源
  const { resources } = await client.listResources();
  console.log("Available resources:", resources.map(r => r.name));
  
  // 读取资源
  const resource = await client.readResource({
    uri: "config://app-settings"
  });
  
  console.log("Config:", resource.contents);
}

2.6 配置与集成

Cursor/Claude Desktop 配置

json

// ~/.cursor/mcp.json 或 claude_desktop_config.json
{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/allowed/dir"],
      "env": {}
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_TOKEN": "your-github-token"
      }
    },
    "postgres": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-postgres"],
      "env": {
        "DATABASE_URL": "postgresql://user:pass@localhost/db"
      }
    },
    "custom-server": {
      "command": "node",
      "args": ["/path/to/my-custom-server.js"],
      "cwd": "/path/to/working/dir"
    }
  }
}

2.7 安全最佳实践

typescript

// 1. 权限控制
const ALLOWED_PATHS = [
  process.env.HOME,
  '/tmp',
  process.cwd()
];

function validatePath(path: string): boolean {
  const resolved = path.resolve(path);
  return ALLOWED_PATHS.some(allowed => 
    resolved.startsWith(allowed)
  );
}

// 2. 输入验证
import { z } from 'zod';

const SearchArgsSchema = z.object({
  query: z.string().min(1).max(1000),
  maxResults: z.number().int().positive().max(100).default(10)
});

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const args = SearchArgsSchema.parse(request.params.arguments);
  // 使用验证后的 args
});

// 3. 速率限制
import { RateLimiter } from 'rate-limiter';

const limiter = new RateLimiter({
  tokensPerInterval: 100,
  interval: "minute"
});

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  if (!limiter.tryRemoveTokens(1)) {
    throw new Error("Rate limit exceeded");
  }
  // 处理请求
});

// 4. 审计日志
function logToolCall(name: string, args: any, result: any) {
  const entry = {
    timestamp: new Date().toISOString(),
    tool: name,
    arguments: args,
    resultSize: JSON.stringify(result).length,
    // 不记录敏感内容
  };
  
  auditLogger.info(entry);
}

3. Agent2Agent Protocol (A2A)

3.1 核心概念

A2A 协议专注于 AI Agent 之间的通信与协作，核心目标是实现跨平台、跨框架的 Agent 互操作。

┌────────────────────────────────────────────────────────────────────────┐
│                          A2A 通信模型                                   │
├────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│  ┌──────────────┐                              ┌──────────────┐        │
│  │ Client Agent │                              │ Remote Agent │        │
│  │  (请求方)     │ ─────── A2A Protocol ─────── │  (服务方)     │        │
│  └──────────────┘                              └──────────────┘        │
│         │                                              │               │
│         │    1. 发现 (Agent Card)                       │               │
│         │ ──────────────────────────────────────────── │               │
│         │                                              │               │
│         │    2. 发起任务 (Task)                         │               │
│         │ ──────────────────────────────────────────▶ │               │
│         │                                              │               │
│         │    3. 状态更新 (SSE Stream)                   │               │
│         │ ◀────────────────────────────────────────── │               │
│         │                                              │               │
│         │    4. 交换消息 (Message)                      │               │
│         │ ◀────────────────────────────────────────▶ │               │
│         │                                              │               │
│         │    5. 返回结果 (Artifact)                     │               │
│         │ ◀────────────────────────────────────────── │               │
│                                                                         │
└────────────────────────────────────────────────────────────────────────┘

3.2 设计原则

A2A 的五大设计原则：

原则	描述
拥抱 Agent 能力	允许 Agent 以自然、非结构化的方式协作，无需共享内存或工具
基于现有标准	使用 HTTP、SSE、JSON-RPC，便于与企业 IT 集成
默认安全	支持企业级身份验证和授权（OAuth 2.0、W3C DID）
支持长任务	灵活处理从秒级到数天的任务，提供实时状态更新
模态无关	支持文本、音频、视频、表单、iframe 等多种交互形式

3.3 核心实体

Agent Card - Agent 身份卡

Agent Card 是 Agent 的"名片"，描述其能力和访问方式：

typescript

interface AgentCard {
  // 基本信息
  name: string;                    // Agent 名称
  description: string;             // 能力描述
  url: string;                     // A2A 端点 URL
  version: string;                 // 协议版本
  
  // 能力声明
  capabilities: {
    streaming?: boolean;           // 是否支持流式响应
    pushNotifications?: boolean;   // 是否支持推送通知
    stateTransitionHistory?: boolean; // 是否提供状态历史
  };
  
  // 技能列表
  skills: Skill[];
  
  // 认证方式
  authentication: {
    schemes: string[];             // 如 ["oauth2", "api_key"]
    credentials?: string;          // 凭证获取方式
  };
  
  // 默认输入模态
  defaultInputModes: string[];     // 如 ["text", "file"]
  defaultOutputModes: string[];    // 如 ["text", "file", "data"]
  
  // 提供商信息
  provider?: {
    organization: string;
    url?: string;
  };
  
  // 可选元数据
  documentationUrl?: string;
  supportUrl?: string;
}

interface Skill {
  id: string;
  name: string;
  description: string;
  tags?: string[];
  examples?: string[];             // 示例输入
  inputModes?: string[];
  outputModes?: string[];
}

示例 Agent Card：

json

{
  "name": "Data Analysis Agent",
  "description": "专业的数据分析助手，支持数据清洗、统计分析和可视化",
  "url": "https://api.example.com/a2a",
  "version": "1.0.0",
  
  "capabilities": {
    "streaming": true,
    "pushNotifications": true,
    "stateTransitionHistory": true
  },
  
  "skills": [
    {
      "id": "data-cleaning",
      "name": "数据清洗",
      "description": "自动检测并处理缺失值、异常值和重复数据",
      "tags": ["data", "preprocessing"],
      "examples": ["清洗这个 CSV 文件", "处理数据中的缺失值"]
    },
    {
      "id": "statistical-analysis", 
      "name": "统计分析",
      "description": "执行描述性统计、假设检验和相关性分析",
      "tags": ["statistics", "analysis"],
      "examples": ["分析销售数据的趋势", "检验两组数据是否有显著差异"]
    }
  ],
  
  "authentication": {
    "schemes": ["oauth2"],
    "credentials": "https://auth.example.com/oauth"
  },
  
  "defaultInputModes": ["text", "file"],
  "defaultOutputModes": ["text", "file", "data"],
  
  "provider": {
    "organization": "Example Corp",
    "url": "https://example.com"
  }
}

Task - 任务

Task 是 A2A 协作的核心单元：

typescript

interface Task {
  id: string;                      // 唯一任务 ID
  sessionId?: string;              // 会话 ID（多轮对话）
  
  status: TaskStatus;              // 任务状态
  
  // 消息历史
  history?: Message[];
  
  // 任务产物
  artifacts?: Artifact[];
  
  // 元数据
  metadata?: Record<string, any>;
}

type TaskStatus = {
  state: TaskState;
  message?: string;                // 状态描述
  timestamp: string;
};

type TaskState = 
  | "submitted"      // 已提交
  | "working"        // 处理中
  | "input-required" // 需要输入
  | "completed"      // 已完成
  | "failed"         // 失败
  | "canceled";      // 已取消

Message - 消息

typescript

interface Message {
  role: "user" | "agent";
  parts: Part[];
  metadata?: Record<string, any>;
}

type Part = 
  | TextPart
  | FilePart
  | DataPart
  | FormPart;

interface TextPart {
  type: "text";
  text: string;
}

interface FilePart {
  type: "file";
  file: {
    name: string;
    mimeType: string;
    // 二选一
    bytes?: string;     // Base64 编码
    uri?: string;       // 文件 URI
  };
}

interface DataPart {
  type: "data";
  data: Record<string, any>;
}

Artifact - 任务产物

typescript

interface Artifact {
  name?: string;
  description?: string;
  parts: Part[];
  index: number;               // 产物序号
  append?: boolean;            // 是否追加（流式）
  lastChunk?: boolean;         // 是否最后一块
  metadata?: Record<string, any>;
}

3.4 通信流程

基本请求-响应模式

typescript

// 1. 发送任务
POST /a2a/tasks/send
Content-Type: application/json

{
  "jsonrpc": "2.0",
  "id": "req-1",
  "method": "tasks/send",
  "params": {
    "id": "task-123",
    "message": {
      "role": "user",
      "parts": [
        { "type": "text", "text": "分析这个销售数据并生成报告" },
        { "type": "file", "file": { "name": "sales.csv", "uri": "..." } }
      ]
    }
  }
}

// 响应
{
  "jsonrpc": "2.0",
  "id": "req-1",
  "result": {
    "id": "task-123",
    "status": {
      "state": "working",
      "message": "正在分析数据...",
      "timestamp": "2025-01-15T10:30:00Z"
    }
  }
}

流式更新 (SSE)

typescript

// 订阅任务更新
GET /a2a/tasks/task-123/sendSubscribe
Accept: text/event-stream

// Server 推送事件
event: status
data: {"state": "working", "message": "数据加载完成..."}

event: artifact
data: {"index": 0, "parts": [{"type": "text", "text": "## 分析报告\n\n"}], "append": false}

event: artifact
data: {"index": 0, "parts": [{"type": "text", "text": "### 1. 概览\n..."}], "append": true}

event: status
data: {"state": "completed", "message": "分析完成"}

3.5 实现 A2A Agent

Server 实现 (TypeScript)

typescript

import express from 'express';
import { v4 as uuidv4 } from 'uuid';

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

// 存储任务状态
const tasks = new Map<string, Task>();
const subscribers = new Map<string, Set<express.Response>>();

// Agent Card 端点
app.get('/.well-known/agent.json', (req, res) => {
  res.json({
    name: "Data Analysis Agent",
    description: "数据分析专家",
    url: "https://api.example.com/a2a",
    version: "1.0.0",
    capabilities: {
      streaming: true,
      pushNotifications: false
    },
    skills: [
      {
        id: "analyze-data",
        name: "数据分析",
        description: "分析数据并生成报告"
      }
    ],
    authentication: {
      schemes: ["bearer"]
    },
    defaultInputModes: ["text", "file"],
    defaultOutputModes: ["text", "file"]
  });
});

// 发送任务
app.post('/a2a', async (req, res) => {
  const { method, params, id } = req.body;
  
  switch (method) {
    case 'tasks/send': {
      const task = await handleTaskSend(params);
      res.json({
        jsonrpc: "2.0",
        id,
        result: task
      });
      break;
    }
    
    case 'tasks/get': {
      const task = tasks.get(params.id);
      res.json({
        jsonrpc: "2.0",
        id,
        result: task || null
      });
      break;
    }
    
    case 'tasks/cancel': {
      const task = await handleTaskCancel(params.id);
      res.json({
        jsonrpc: "2.0",
        id,
        result: task
      });
      break;
    }
  }
});

// SSE 订阅端点
app.get('/a2a/tasks/:taskId/subscribe', (req, res) => {
  const { taskId } = req.params;
  
  // 设置 SSE headers
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');
  
  // 添加订阅者
  if (!subscribers.has(taskId)) {
    subscribers.set(taskId, new Set());
  }
  subscribers.get(taskId)!.add(res);
  
  // 清理
  req.on('close', () => {
    subscribers.get(taskId)?.delete(res);
  });
});

// 处理任务
async function handleTaskSend(params: any): Promise<Task> {
  const taskId = params.id || uuidv4();
  
  const task: Task = {
    id: taskId,
    status: {
      state: "submitted",
      timestamp: new Date().toISOString()
    },
    history: [params.message]
  };
  
  tasks.set(taskId, task);
  
  // 异步处理任务
  processTaskAsync(taskId, params.message);
  
  return task;
}

async function processTaskAsync(taskId: string, message: Message) {
  const task = tasks.get(taskId)!;
  
  // 更新状态为处理中
  updateTaskStatus(taskId, "working", "正在分析数据...");
  
  try {
    // 模拟处理
    await new Promise(resolve => setTimeout(resolve, 2000));
    
    // 发送部分结果
    emitArtifact(taskId, {
      index: 0,
      parts: [{ type: "text", text: "# 分析报告\n\n数据分析完成。" }],
      lastChunk: true
    });
    
    // 完成
    updateTaskStatus(taskId, "completed", "分析完成");
    
  } catch (error) {
    updateTaskStatus(taskId, "failed", error.message);
  }
}

function updateTaskStatus(taskId: string, state: TaskState, message: string) {
  const task = tasks.get(taskId);
  if (!task) return;
  
  task.status = {
    state,
    message,
    timestamp: new Date().toISOString()
  };
  
  // 通知订阅者
  const subs = subscribers.get(taskId);
  if (subs) {
    const event = `event: status\ndata: ${JSON.stringify(task.status)}\n\n`;
    subs.forEach(res => res.write(event));
  }
}

function emitArtifact(taskId: string, artifact: Artifact) {
  const task = tasks.get(taskId);
  if (!task) return;
  
  task.artifacts = task.artifacts || [];
  task.artifacts.push(artifact);
  
  const subs = subscribers.get(taskId);
  if (subs) {
    const event = `event: artifact\ndata: ${JSON.stringify(artifact)}\n\n`;
    subs.forEach(res => res.write(event));
  }
}

app.listen(3000, () => {
  console.log('A2A Agent running on port 3000');
});

Client 实现

typescript

class A2AClient {
  private baseUrl: string;
  private authToken: string;
  
  constructor(baseUrl: string, authToken: string) {
    this.baseUrl = baseUrl;
    this.authToken = authToken;
  }
  
  // 获取 Agent Card
  async getAgentCard(): Promise<AgentCard> {
    const response = await fetch(`${this.baseUrl}/.well-known/agent.json`);
    return response.json();
  }
  
  // 发送任务
  async sendTask(message: Message, taskId?: string): Promise<Task> {
    const response = await fetch(`${this.baseUrl}/a2a`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Authorization': `Bearer ${this.authToken}`
      },
      body: JSON.stringify({
        jsonrpc: "2.0",
        id: crypto.randomUUID(),
        method: "tasks/send",
        params: {
          id: taskId || crypto.randomUUID(),
          message
        }
      })
    });
    
    const data = await response.json();
    return data.result;
  }
  
  // 订阅任务更新
  subscribeToTask(
    taskId: string,
    callbacks: {
      onStatus?: (status: TaskStatus) => void;
      onArtifact?: (artifact: Artifact) => void;
      onError?: (error: Error) => void;
    }
  ): () => void {
    const eventSource = new EventSource(
      `${this.baseUrl}/a2a/tasks/${taskId}/subscribe`,
      {
        // 自定义 headers 需要使用 fetch 替代
      }
    );
    
    eventSource.addEventListener('status', (e) => {
      const status = JSON.parse(e.data);
      callbacks.onStatus?.(status);
    });
    
    eventSource.addEventListener('artifact', (e) => {
      const artifact = JSON.parse(e.data);
      callbacks.onArtifact?.(artifact);
    });
    
    eventSource.onerror = (e) => {
      callbacks.onError?.(new Error('SSE connection error'));
    };
    
    // 返回取消函数
    return () => eventSource.close();
  }
  
  // 获取任务状态
  async getTask(taskId: string): Promise<Task | null> {
    const response = await fetch(`${this.baseUrl}/a2a`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Authorization': `Bearer ${this.authToken}`
      },
      body: JSON.stringify({
        jsonrpc: "2.0",
        id: crypto.randomUUID(),
        method: "tasks/get",
        params: { id: taskId }
      })
    });
    
    const data = await response.json();
    return data.result;
  }
  
  // 取消任务
  async cancelTask(taskId: string): Promise<Task> {
    const response = await fetch(`${this.baseUrl}/a2a`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Authorization': `Bearer ${this.authToken}`
      },
      body: JSON.stringify({
        jsonrpc: "2.0",
        id: crypto.randomUUID(),
        method: "tasks/cancel",
        params: { id: taskId }
      })
    });
    
    const data = await response.json();
    return data.result;
  }
}

// 使用示例
async function main() {
  const client = new A2AClient('https://api.example.com', 'your-token');
  
  // 获取 Agent 能力
  const agentCard = await client.getAgentCard();
  console.log('Agent skills:', agentCard.skills);
  
  // 发送任务
  const task = await client.sendTask({
    role: 'user',
    parts: [
      { type: 'text', text: '分析这个数据集' }
    ]
  });
  
  console.log('Task created:', task.id);
  
  // 订阅更新
  const unsubscribe = client.subscribeToTask(task.id, {
    onStatus: (status) => {
      console.log('Status:', status.state, status.message);
    },
    onArtifact: (artifact) => {
      console.log('Artifact:', artifact.parts);
    }
  });
  
  // 稍后取消订阅
  // unsubscribe();
}

3.6 多 Agent 协作模式

任务委派模式

typescript

// 协调 Agent 将任务委派给专业 Agent
class OrchestratorAgent {
  private agents: Map<string, A2AClient> = new Map();
  
  constructor(agentConfigs: AgentConfig[]) {
    for (const config of agentConfigs) {
      this.agents.set(config.skillType, new A2AClient(config.url, config.token));
    }
  }
  
  async processRequest(request: string): Promise<any> {
    // 1. 分析请求，确定需要哪些 Agent
    const plan = await this.createPlan(request);
    
    // 2. 按计划执行
    const results = [];
    
    for (const step of plan.steps) {
      const agent = this.agents.get(step.agentType);
      if (!agent) {
        throw new Error(`No agent found for: ${step.agentType}`);
      }
      
      // 委派任务
      const task = await agent.sendTask({
        role: 'user',
        parts: [
          { type: 'text', text: step.instruction },
          // 包含前序步骤的结果
          ...this.formatPreviousResults(results)
        ]
      });
      
      // 等待完成
      const result = await this.waitForCompletion(agent, task.id);
      results.push({ step: step.name, result });
    }
    
    // 3. 汇总结果
    return this.summarizeResults(results);
  }
  
  private async waitForCompletion(client: A2AClient, taskId: string): Promise<any> {
    return new Promise((resolve, reject) => {
      const unsubscribe = client.subscribeToTask(taskId, {
        onStatus: (status) => {
          if (status.state === 'completed') {
            unsubscribe();
            client.getTask(taskId).then(task => resolve(task?.artifacts));
          } else if (status.state === 'failed') {
            unsubscribe();
            reject(new Error(status.message));
          }
        }
      });
    });
  }
}

协作工作流示例

typescript

// 实现一个多 Agent 数据处理工作流
async function dataProcessingWorkflow(dataUrl: string) {
  const orchestrator = new OrchestratorAgent([
    { skillType: 'data-cleaning', url: 'https://cleaner.example.com', token: '...' },
    { skillType: 'analysis', url: 'https://analyst.example.com', token: '...' },
    { skillType: 'visualization', url: 'https://viz.example.com', token: '...' },
    { skillType: 'report-generation', url: 'https://reporter.example.com', token: '...' }
  ]);
  
  const result = await orchestrator.processRequest(`
    处理以下数据：${dataUrl}
    
    1. 清洗数据，处理缺失值和异常值
    2. 进行统计分析，找出关键趋势
    3. 生成可视化图表
    4. 撰写分析报告
  `);
  
  return result;
}

4. 协议对比与选择

4.1 功能对比

功能	MCP	A2A
工具调用	✅ 核心功能	❌ 不直接支持
资源访问	✅ Resources	❌ 通过 Message 间接
Agent 发现	❌ 无	✅ Agent Card
任务管理	❌ 无	✅ Task 生命周期
流式响应	❌ 有限支持	✅ SSE 原生支持
多模态	✅ 支持	✅ 支持
长任务	❌ 设计为同步	✅ 异步、推送通知
身份认证	基础	✅ OAuth 2.0, W3C DID

4.2 选择指南

选择 MCP 当:
├── 需要让 AI 访问工具和数据源
├── 构建单 Agent 应用
├── 需要读取文件、调用 API
└── 需要快速集成已有服务

选择 A2A 当:
├── 需要多个 Agent 协作
├── Agent 来自不同供应商/框架
├── 需要处理长时间运行的任务
└── 需要企业级安全和审计

两者结合当:
├── 构建复杂的多 Agent 系统
├── Agent 既需要工具也需要协作
└── 企业级 AI 工作流

4.3 集成架构

┌─────────────────────────────────────────────────────────────────────────┐
│                     企业 AI 系统架构示例                                  │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                          │
│  ┌─────────────────────────────────────────────────────────────────┐    │
│  │                         API Gateway                              │    │
│  │                    (认证、限流、日志)                              │    │
│  └───────────────────────────┬─────────────────────────────────────┘    │
│                              │                                           │
│                              ▼                                           │
│  ┌─────────────────────────────────────────────────────────────────┐    │
│  │                    Orchestrator Agent                            │    │
│  │              (任务规划、Agent 调度、结果汇总)                       │    │
│  └───────────────────────────┬─────────────────────────────────────┘    │
│                              │                                           │
│              ┌───────────────┼───────────────┐                          │
│              │ A2A           │ A2A           │ A2A                       │
│              ▼               ▼               ▼                          │
│  ┌──────────────────┐ ┌──────────────────┐ ┌──────────────────┐        │
│  │  Specialist      │ │  Specialist      │ │  Specialist      │        │
│  │  Agent A         │ │  Agent B         │ │  Agent C         │        │
│  │  (数据分析)       │ │  (文档生成)       │ │  (邮件发送)       │        │
│  └────────┬─────────┘ └────────┬─────────┘ └────────┬─────────┘        │
│           │ MCP                 │ MCP                │ MCP              │
│           ▼                     ▼                    ▼                  │
│  ┌──────────────────┐ ┌──────────────────┐ ┌──────────────────┐        │
│  │  MCP Servers     │ │  MCP Servers     │ │  MCP Servers     │        │
│  │  - Database      │ │  - Google Docs   │ │  - SMTP          │        │
│  │  - Analytics API │ │  - Templates     │ │  - Calendar      │        │
│  └──────────────────┘ └──────────────────┘ └──────────────────┘        │
│                                                                          │
└─────────────────────────────────────────────────────────────────────────┘

5. 安全考量

5.1 MCP 安全

typescript

// 1. Server 能力限制
const server = new Server({
  name: "secure-server",
  version: "1.0.0"
}, {
  capabilities: {
    // 只开放必要的能力
    tools: {},
    // resources: {},  // 不暴露资源
    // prompts: {},    // 不暴露提示
  }
});

// 2. 路径白名单
const ALLOWED_PATHS = [
  '/safe/directory',
  '/another/safe/path'
];

function isPathAllowed(path: string): boolean {
  const resolved = path.resolve(path);
  return ALLOWED_PATHS.some(allowed => 
    resolved.startsWith(path.resolve(allowed))
  );
}

// 3. 敏感数据过滤
function sanitizeOutput(data: any): any {
  const sensitiveKeys = ['password', 'token', 'secret', 'apiKey'];
  
  if (typeof data === 'object') {
    for (const key of Object.keys(data)) {
      if (sensitiveKeys.some(s => key.toLowerCase().includes(s))) {
        data[key] = '[REDACTED]';
      } else if (typeof data[key] === 'object') {
        data[key] = sanitizeOutput(data[key]);
      }
    }
  }
  
  return data;
}

5.2 A2A 安全

typescript

// 1. OAuth 2.0 认证
app.use('/a2a', async (req, res, next) => {
  const authHeader = req.headers.authorization;
  
  if (!authHeader?.startsWith('Bearer ')) {
    return res.status(401).json({ error: 'Missing authentication' });
  }
  
  const token = authHeader.slice(7);
  
  try {
    const payload = await verifyToken(token);
    req.user = payload;
    next();
  } catch {
    return res.status(401).json({ error: 'Invalid token' });
  }
});

// 2. 任务权限检查
function canAccessTask(user: User, taskId: string): boolean {
  const task = tasks.get(taskId);
  
  // 检查用户是否有权限访问此任务
  return task?.metadata?.ownerId === user.id ||
         user.roles.includes('admin');
}

// 3. 请求验证
import { z } from 'zod';

const TaskSendSchema = z.object({
  id: z.string().uuid().optional(),
  message: z.object({
    role: z.enum(['user', 'agent']),
    parts: z.array(z.union([
      z.object({ type: z.literal('text'), text: z.string().max(100000) }),
      z.object({ type: z.literal('file'), file: z.object({
        name: z.string(),
        mimeType: z.string(),
        uri: z.string().url().optional(),
        bytes: z.string().optional()
      })})
    ]))
  })
});

// 4. 审计日志
function auditLog(action: string, details: any) {
  const entry = {
    timestamp: new Date().toISOString(),
    action,
    userId: details.userId,
    taskId: details.taskId,
    ip: details.ip,
    // 不记录敏感内容
  };
  
  auditLogger.info(entry);
}

6. 生态系统

6.1 MCP 生态

官方 Servers：

Server	功能	安装
`@modelcontextprotocol/server-filesystem`	文件系统访问	`npx -y @modelcontextprotocol/server-filesystem /path`
`@modelcontextprotocol/server-github`	GitHub API	`npx -y @modelcontextprotocol/server-github`
`@modelcontextprotocol/server-postgres`	PostgreSQL 数据库	`npx -y @modelcontextprotocol/server-postgres`
`@modelcontextprotocol/server-sqlite`	SQLite 数据库	`npx -y @modelcontextprotocol/server-sqlite`
`@modelcontextprotocol/server-puppeteer`	浏览器自动化	`npx -y @modelcontextprotocol/server-puppeteer`

社区 Servers：

Notion, Slack, Discord 集成
AWS, GCP, Azure 云服务
Jira, Linear 项目管理
Figma 设计工具

6.2 A2A 生态

支持的框架：

LangChain / LangGraph
CrewAI
Google ADK (Agent Development Kit)
Microsoft AutoGen

企业合作伙伴：

Salesforce, SAP, ServiceNow
Atlassian, Box, MongoDB
Accenture, Deloitte, PwC

7. 关键要点

MCP 要点

三大能力: Resources（资源）、Tools（工具）、Prompts（提示）
Client-Server 架构: Host 托管应用，Client 处理协议，Server 提供能力
JSON-RPC 2.0: 标准化的请求-响应格式
安全第一: 路径验证、权限控制、审计日志

A2A 要点

Agent Card: Agent 的能力名片，支持发现与对接
Task 生命周期: submitted → working → completed/failed
SSE 流式更新: 实时状态推送和部分结果
企业级安全: OAuth 2.0、W3C DID 身份认证

选择建议

场景	推荐协议
AI 访问文件/数据库	MCP
AI 调用外部 API	MCP
多 Agent 协作	A2A
跨组织 Agent 互操作	A2A
完整企业 AI 系统	MCP + A2A

Agent 协议：MCP 与 A2A 🔗 ​

1. 协议全景图 ​

1.1 协议定位对比 ​

1.2 互补关系 ​

2. Model Context Protocol (MCP) ​

2.1 核心概念 ​

2.2 三大核心能力 ​

2.3 通信协议 ​

2.4 实现 MCP Server ​

基础 Server 实现 (TypeScript) ​

Python 实现 ​

2.5 实现 MCP Client ​

2.6 配置与集成 ​

Cursor/Claude Desktop 配置 ​

2.7 安全最佳实践 ​

3. Agent2Agent Protocol (A2A) ​

3.1 核心概念 ​

3.2 设计原则 ​

3.3 核心实体 ​

Agent Card - Agent 身份卡 ​

Task - 任务 ​

Message - 消息 ​

Artifact - 任务产物 ​

3.4 通信流程 ​

基本请求-响应模式 ​

流式更新 (SSE) ​

3.5 实现 A2A Agent ​

Server 实现 (TypeScript) ​

Client 实现 ​

3.6 多 Agent 协作模式 ​

任务委派模式 ​

协作工作流示例 ​

4. 协议对比与选择 ​

4.1 功能对比 ​

4.2 选择指南 ​

4.3 集成架构 ​

5. 安全考量 ​

5.1 MCP 安全 ​

5.2 A2A 安全 ​

6. 生态系统 ​

6.1 MCP 生态 ​

6.2 A2A 生态 ​

7. 关键要点 ​

MCP 要点 ​

A2A 要点 ​

选择建议 ​

8. 参考资源 ​

官方文档 ​

SDK ​

深度文章 ​

Agent 协议：MCP 与 A2A 🔗

1. 协议全景图

1.1 协议定位对比

1.2 互补关系

2. Model Context Protocol (MCP)

2.1 核心概念

2.2 三大核心能力

2.3 通信协议

2.4 实现 MCP Server

基础 Server 实现 (TypeScript)

Python 实现

2.5 实现 MCP Client

2.6 配置与集成

Cursor/Claude Desktop 配置

2.7 安全最佳实践

3. Agent2Agent Protocol (A2A)

3.1 核心概念

3.2 设计原则

3.3 核心实体

Agent Card - Agent 身份卡

Task - 任务

Message - 消息

Artifact - 任务产物

3.4 通信流程

基本请求-响应模式

流式更新 (SSE)

3.5 实现 A2A Agent

Server 实现 (TypeScript)

Client 实现

3.6 多 Agent 协作模式

任务委派模式

协作工作流示例

4. 协议对比与选择

4.1 功能对比

4.2 选择指南

4.3 集成架构

5. 安全考量

5.1 MCP 安全

5.2 A2A 安全

6. 生态系统

6.1 MCP 生态

6.2 A2A 生态

7. 关键要点

MCP 要点

A2A 要点

选择建议

8. 参考资源

官方文档

SDK

深度文章