编程从 MCP 到 A2A：2026 年 AI Agent 协议时代——标准化通信层如何重塑智能体协作范式

2026-05-05 14:37:35 +0800 CST views 5

从 MCP 到 A2A：2026 年 AI Agent 协议时代——标准化通信层如何重塑智能体协作范式

当 AI 从"聊天助手"进化为"行动代理"，协议成为比模型更关键的基础设施。本文深度解析 MCP 与 A2A 两大协议的设计哲学、技术架构与工程实践，带你理解为什么 2026 年是 AI Agent 的"协议元年"。

一、为什么 AI Agent 需要协议？

1.1 从 Function Calling 到协议层：一次必要的抽象跃迁

2024 年，OpenAI 推出 Function Calling，让 LLM 能够调用外部工具。这被视为 AI Agent 的"原始细胞"——模型可以执行操作了。但问题很快浮现：

问题一：工具调用的碎片化

每个 AI 平台都有自己的工具调用格式：

OpenAI 使用 JSON Schema 定义函数参数
Anthropic 的 Tool Use 有不同的消息结构
Google Gemini 的 Function Declarations 又是另一套

这意味着如果你开发了一个"查询天气"的工具，要同时支持三个平台，需要写三套适配代码。

问题二：上下文传递的混乱

Agent 不只是调用工具，还需要：

访问文件系统
查询数据库
调用 API
与其他 Agent 通信

这些操作涉及状态管理、权限控制、错误处理。Function Calling 没有定义这些，每个实现都是"各自为战"。

问题三：多 Agent 协作的缺失

当系统从"一个 Agent"扩展到"多个 Agent 协作"，问题更复杂：

Agent A 如何发现 Agent B？
它们如何协商任务分工？
如何传递上下文和状态？
如何处理失败和重试？

这些问题，Function Calling 无法回答。

1.2 协议层的价值：标准化、可组合、可治理

2024 年末，Anthropic 推出 Model Context Protocol (MCP)，定位为"AI 模型的 USB 接口"——让任何 AI 模型都能通过统一协议连接外部系统。

2025 年，Google 提出 Agent-to-Agent Protocol (A2A)，解决 Agent 间协作的标准化问题。

这两个协议的出现，标志着 AI Agent 进入"协议时代"：

维度	前协议时代	协议时代
工具接入	每个平台自己定义	MCP 统一标准
Agent 协作	私有实现，无法互通	A2A 标准化发现与通信
生态发展	孤岛式，重复造轮子	模块化，可组合
治理与安全	缺乏统一框架	协议层内置权限、审计、隔离

一个类比：MCP 是 AI Agent 的 HTTP，A2A 是 AI Agent 的 SMTP。HTTP 让浏览器能与任何 Web 服务器通信，SMTP 让邮件系统互联互通。协议定义规则，生态在此基础上繁荣。

二、MCP 协议深度解析

2.1 核心概念：Server、Client、Transport

MCP 的设计哲学是"最小化约定，最大化扩展"。它定义了三个核心概念：

Server（服务端）：提供能力的实体

暴露 Tools（可调用的函数）
暴露 Resources（可访问的数据）
暴露 Prompts（预定义的提示词模板）

Client（客户端）：消费能力的实体

通常是 AI 模型（Claude、GPT 等）
发起 Tool 调用请求
读取 Resources
使用 Prompts

Transport（传输层）：通信方式

stdio：通过标准输入输出通信，适合本地进程
HTTP + SSE：HTTP 请求 + Server-Sent Events，适合远程服务
WebSocket：双向实时通信（规划中）

2.2 协议消息格式

MCP 使用 JSON-RPC 2.0 作为消息格式。一次完整的 Tool 调用流程：

// 1. Client 发起请求
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "get_weather",
    "arguments": {
      "city": "北京"
    }
  }
}

// 2. Server 返回结果
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "北京今天晴，气温 18-26°C，空气质量良好"
      }
    ]
  }
}

2.3 Tools、Resources、Prompts 三大能力

Tools（工具）：最核心的能力

// 定义一个 Tool
server.tool(
  "search_web",
  {
    query: z.string().describe("搜索关键词"),
    limit: z.number().optional().default(5).describe("返回数量")
  },
  async ({ query, limit }) => {
    const results = await searchEngine.search(query, limit);
    return {
      content: [{ type: "text", text: JSON.stringify(results) }]
    };
  }
);

Resources（资源）：暴露数据源

// 暴露一个文件资源
server.resource(
  "config",
  "file:///etc/app/config.json",
  async (uri) => {
    const content = await fs.readFile(uri.path, "utf-8");
    return { contents: [{ uri, mimeType: "application/json", text: content }] };
  }
);

Prompts（提示词模板）：预定义交互模式

server.prompt(
  "code_review",
  { code: z.string(), language: z.string() },
  ({ code, language }) => ({
    messages: [{
      role: "user",
      content: {
        type: "text",
        text: `请审查以下 ${language} 代码，指出潜在问题：\n\n${code}`
      }
    }]
  })
);

2.4 Transport 模式选择

stdio 模式：适合本地集成

# Claude Desktop 配置
{
  "mcpServers": {
    "my-server": {
      "command": "node",
      "args": ["/path/to/server.js"]
    }
  }
}

Claude Desktop 会启动 node server.js 作为子进程，通过 stdin/stdout 通信。

HTTP + SSE 模式：适合远程服务

// 启动 HTTP Server
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";

const server = new McpServer({ name: "my-server", version: "1.0.0" });

// Express 集成
app.get("/sse", async (req, res) => {
  const transport = new SSEServerTransport("/message", res);
  await server.connect(transport);
});

app.post("/message", async (req, res) => {
  // 处理 Client 发来的消息
});

三、MCP Server 实战：从零构建一个代码审查服务器

3.1 项目初始化

mkdir code-review-mcp && cd code-review-mcp
npm init -y
npm install @modelcontextprotocol/sdk zod
npm install -D typescript @types/node

3.2 核心实现

// src/server.ts
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";

const server = new McpServer({
  name: "code-review-server",
  version: "1.0.0"
});

// Tool 1: 分析代码复杂度
server.tool(
  "analyze_complexity",
  {
    code: z.string().describe("要分析的代码"),
    language: z.enum(["javascript", "typescript", "python", "go"]).describe("编程语言")
  },
  async ({ code, language }) => {
    const complexity = calculateComplexity(code, language);
    
    return {
      content: [{
        type: "text",
        text: JSON.stringify({
          cyclomaticComplexity: complexity.cyclomatic,
          linesOfCode: complexity.lines,
          maintainabilityIndex: complexity.maintainability,
          issues: complexity.issues,
          recommendations: complexity.recommendations
        }, null, 2)
      }]
    };
  }
);

// Tool 2: 检测安全漏洞
server.tool(
  "detect_vulnerabilities",
  {
    code: z.string().describe("要检测的代码"),
    language: z.string().describe("编程语言")
  },
  async ({ code, language }) => {
    const vulnerabilities = await runSecurityScan(code, language);
    
    return {
      content: [{
        type: "text",
        text: formatVulnerabilityReport(vulnerabilities)
      }]
    };
  }
);

// Tool 3: 生成改进建议
server.tool(
  "suggest_improvements",
  {
    code: z.string().describe("原始代码"),
    language: z.string().describe("编程语言"),
    focus: z.enum(["performance", "readability", "security", "all"]).optional().default("all")
  },
  async ({ code, language, focus }) => {
    const suggestions = await generateSuggestions(code, language, focus);
    
    return {
      content: [{
        type: "text",
        text: suggestions
      }]
    };
  }
);

// Resource: 暴露代码规范文档
server.resource(
  "coding-standards",
  "resource://coding-standards",
  async () => ({
    contents: [{
      uri: "resource://coding-standards",
      mimeType: "text/markdown",
      text: `
# 代码规范

## 命名约定
- 变量使用 camelCase
- 常量使用 UPPER_SNAKE_CASE
- 类名使用 PascalCase

## 函数设计
- 单一职责原则
- 函数长度不超过 50 行
- 参数不超过 4 个

## 错误处理
- 所有外部调用必须有错误处理
- 使用自定义错误类型
- 记录完整的错误上下文
      `
    }]
  })
);

// Prompt: 代码审查模板
server.prompt(
  "review_template",
  {
    code: z.string(),
    language: z.string(),
    author: z.string().optional()
  },
  ({ code, language, author }) => ({
    messages: [{
      role: "user",
      content: {
        type: "text",
        text: `
请对以下 ${language} 代码进行完整审查：

\`\`\`${language}
${code}
\`\`\`

审查要点：
1. 代码复杂度分析
2. 潜在安全漏洞
3. 性能优化建议
4. 可读性改进
${author ? `\n作者：${author}` : ''}
        `
      }
    }]
  })
);

// 启动服务器
const transport = new StdServerTransport();
await server.connect(transport);

3.3 辅助函数实现

function calculateComplexity(code: string, language: string) {
  const lines = code.split('\n').filter(l => l.trim()).length;
  const functions = (code.match(/function|def |func /g) || []).length;
  const branches = (code.match(/if|else|switch|case|for|while/g) || []).length;
  
  const cyclomatic = branches + 1;
  const maintainability = Math.max(0, Math.min(100, 
    171 - 5.2 * lines - 0.23 * cyclomatic - 16.2 * functions
  ));
  
  const issues: string[] = [];
  if (cyclomatic > 10) issues.push("圈复杂度过高，建议拆分函数");
  if (lines > 100) issues.push("单文件代码行数过多");
  if (functions > 10) issues.push("函数数量过多，考虑拆分模块");
  
  return {
    cyclomatic,
    lines,
    functions,
    maintainability: Math.round(maintainability),
    issues,
    recommendations: issues.map(i => `建议：${i}`)
  };
}

async function runSecurityScan(code: string, language: string) {
  const vulnerabilities: Array<{
    type: string;
    severity: "high" | "medium" | "low";
    line: number;
    description: string;
  }> = [];
  
  // SQL 注入检测
  if (code.includes("query(") && code.includes("+") && code.includes("req.")) {
    vulnerabilities.push({
      type: "SQL Injection",
      severity: "high",
      line: code.split("\n").findIndex(l => l.includes("query(")) + 1,
      description: "检测到可能的 SQL 注入漏洞，请使用参数化查询"
    });
  }
  
  // XSS 检测
  if (code.includes("innerHTML") || code.includes("dangerouslySetInnerHTML")) {
    vulnerabilities.push({
      type: "XSS",
      severity: "high",
      line: code.split("\n").findIndex(l => l.includes("innerHTML") || l.includes("dangerouslySetInnerHTML")) + 1,
      description: "直接设置 HTML 可能导致 XSS 攻击，请使用安全的 DOM 操作"
    });
  }
  
  // 硬编码密钥检测
  if (code.match(/(?:password|secret|api_key|token)\s*=\s*["'][^"']+["']/i)) {
    vulnerabilities.push({
      type: "Hardcoded Secret",
      severity: "medium",
      line: 1,
      description: "检测到硬编码的敏感信息，请使用环境变量"
    });
  }
  
  return vulnerabilities;
}

function formatVulnerabilityReport(vulnerabilities: any[]) {
  if (vulnerabilities.length === 0) {
    return "✅ 未检测到明显安全漏洞";
  }
  
  return vulnerabilities.map(v => 
    `【${v.severity.toUpperCase()}】${v.type} (行 ${v.line})\n  ${v.description}`
  ).join("\n\n");
}

四、A2A 协议：Agent 间协作的标准化

4.1 为什么需要 A2A？

MCP 解决了"Agent 与工具连接"的问题，但"Agent 与 Agent 连接"是另一个挑战。

场景举例：一个电商系统的 Agent 协作

用户请求："帮我找出上周销量下降最多的商品，并生成分析报告"

执行流程：
1. 销售 Agent → 查询销售数据
2. 数据分析 Agent → 分析下降原因
3. 报告生成 Agent → 生成 PDF 报告
4. 通知 Agent → 发送邮件给相关负责人

这四个 Agent 需要相互发现、协商任务、传递数据。如果每个 Agent 都是私有实现，协作成本极高。

4.2 A2A 核心概念

A2A 定义了 Agent 间协作的标准协议：

Agent Card（代理名片）：描述 Agent 的能力

{
  "agent_id": "sales-analyzer-001",
  "name": "销售数据分析代理",
  "description": "分析销售趋势、识别异常、预测未来销量",
  "capabilities": [
    {
      "name": "trend_analysis",
      "description": "分析销售趋势",
      "input_schema": { ... },
      "output_schema": { ... }
    },
    {
      "name": "anomaly_detection",
      "description": "检测异常数据",
      "input_schema": { ... }
    }
  ],
  "endpoint": "https://agents.example.com/sales-analyzer",
  "authentication": {
    "type": "bearer",
    "token_url": "https://auth.example.com/token"
  }
}

Task（任务）：Agent 间传递的工作单元

{
  "task_id": "task-12345",
  "from_agent": "orchestrator-001",
  "to_agent": "sales-analyzer-001",
  "action": "trend_analysis",
  "payload": {
    "time_range": "2026-04-28/2026-05-05",
    "metrics": ["sales", "orders", "conversion_rate"]
  },
  "context": {
    "user_id": "user-999",
    "session_id": "session-abc"
  },
  "priority": "high",
  "deadline": "2026-05-05T15:00:00Z"
}

Message（消息）：Agent 间的通信载体

{
  "message_id": "msg-67890",
  "task_id": "task-12345",
  "type": "result",
  "content": {
    "trend": "declining",
    "decline_rate": -15.3,
    "top_declined_products": [...],
    "recommendations": [...]
  },
  "metadata": {
    "processing_time_ms": 2340,
    "model_used": "claude-3.5-sonnet"
  }
}

4.3 A2A 工作流程

1. Agent 注册
   Agent A → Registry: "我是销售分析代理，我有这些能力..."
   Registry → Agent A: "注册成功，你的 ID 是 xxx"

2. Agent 发现
   Orchestrator → Registry: "谁能做销售趋势分析？"
   Registry → Orchestrator: "Agent A 可以，endpoint 是 xxx"

3. 任务协商
   Orchestrator → Agent A: "帮我分析上周销售趋势"
   Agent A → Orchestrator: "需要数据访问权限"
   Orchestrator → Agent A: "已授权，token 是 xxx"

4. 任务执行
   Agent A → Database: [查询数据]
   Agent A → Orchestrator: [返回分析结果]

5. 结果传递
   Orchestrator → Agent B: "根据分析结果生成报告"
   Agent B → Orchestrator: [返回报告 URL]

4.4 A2A 与 MCP 的协同

A2A 和 MCP 解决不同层面的问题，但可以协同工作：

┌─────────────────────────────────────────────┐
│              Orchestrator Agent             │
│  (协调多个 Agent，通过 A2A 协议)              │
└─────────────────┬───────────────────────────┘
                  │ A2A Protocol
        ┌─────────┴─────────┐
        ▼                   ▼
┌───────────────┐   ┌───────────────┐
│  Sales Agent  │   │ Report Agent  │
│               │   │               │
│ MCP Server    │   │ MCP Server    │
│  ├─ Database  │   │  ├─ PDF Gen   │
│  ├─ API       │   │  ├─ Email     │
│  └─ Analytics │   │  └─ Storage   │
└───────────────┘   └───────────────┘
        │                   │
        │ MCP Protocol      │ MCP Protocol
        ▼                   ▼
   [外部系统]          [外部系统]

五、企业级 MCP Server 最佳实践

5.1 安全模型

认证与授权

// 基于 JWT 的认证中间件
import { verify } from "jsonwebtoken";

server.middleware(async (request, next) => {
  const token = request.headers.authorization?.replace("Bearer ", "");
  
  if (!token) {
    throw new Error("Missing authentication token");
  }
  
  const payload = verify(token, process.env.JWT_SECRET!);
  request.context.user = payload;
  
  // 权限检查
  if (request.method === "tools/call") {
    const toolName = request.params.name;
    if (!hasPermission(payload.roles, toolName)) {
      throw new Error(`Permission denied for tool: ${toolName}`);
    }
  }
  
  return next(request);
});

敏感数据隔离

// 数据脱敏中间件
server.middleware(async (request, next) => {
  const response = await next(request);
  
  // 对敏感字段进行脱敏
  if (response.result?.content) {
    response.result.content = sanitizeSensitiveData(
      response.result.content,
      ["password", "api_key", "token", "credit_card"]
    );
  }
  
  return response;
});

5.2 可观测性

结构化日志

import pino from "pino";

const logger = pino({
  level: process.env.LOG_LEVEL || "info",
  formatters: {
    level: (label) => ({ level: label.toUpperCase() })
  }
});

server.tool(
  "query_database",
  { sql: z.string() },
  async ({ sql }, context) => {
    const startTime = Date.now();
    
    logger.info({
      event: "tool_call_start",
      tool: "query_database",
      user: context.user.id,
      sql_preview: sql.substring(0, 100)
    });
    
    try {
      const result = await executeQuery(sql);
      
      logger.info({
        event: "tool_call_success",
        tool: "query_database",
        duration_ms: Date.now() - startTime,
        rows_returned: result.length
      });
      
      return { content: [{ type: "text", text: JSON.stringify(result) }] };
    } catch (error) {
      logger.error({
        event: "tool_call_error",
        tool: "query_database",
        error: error.message,
        duration_ms: Date.now() - startTime
      });
      throw error;
    }
  }
);

Prometheus 指标

import client from "prom-client";

const toolCallCounter = new client.Counter({
  name: "mcp_tool_calls_total",
  help: "Total number of tool calls",
  labelNames: ["tool", "status"]
});

const toolCallHistogram = new client.Histogram({
  name: "mcp_tool_call_duration_seconds",
  help: "Duration of tool calls in seconds",
  labelNames: ["tool"],
  buckets: [0.1, 0.5, 1, 2, 5, 10]
});

// 在 tool 执行前后记录指标
server.middleware(async (request, next) => {
  if (request.method === "tools/call") {
    const toolName = request.params.name;
    const end = toolCallHistogram.startTimer({ tool: toolName });
    
    try {
      const result = await next(request);
      toolCallCounter.inc({ tool: toolName, status: "success" });
      end();
      return result;
    } catch (error) {
      toolCallCounter.inc({ tool: toolName, status: "error" });
      end();
      throw error;
    }
  }
  
  return next(request);
});

5.3 错误处理与重试

// 带重试的工具调用
async function callWithRetry<T>(
  fn: () => Promise<T>,
  options: {
    maxRetries: number;
    backoff: "linear" | "exponential";
    retryOn: Array<string | RegExp>;
  }
): Promise<T> {
  const { maxRetries = 3, backoff = "exponential", retryOn = [] } = options;
  
  let lastError: Error;
  
  for (let attempt = 0; attempt <= maxRetries; attempt++) {
    try {
      return await fn();
    } catch (error) {
      lastError = error;
      
      // 检查是否应该重试
      const shouldRetry = retryOn.some(pattern => 
        typeof pattern === "string" 
          ? error.message.includes(pattern)
          : pattern.test(error.message)
      );
      
      if (!shouldRetry || attempt === maxRetries) {
        throw error;
      }
      
      // 计算退避时间
      const delay = backoff === "exponential"
        ? Math.pow(2, attempt) * 1000
        : (attempt + 1) * 1000;
      
      logger.warn({
        event: "tool_call_retry",
        attempt: attempt + 1,
        delay_ms: delay,
        error: error.message
      });
      
      await new Promise(resolve => setTimeout(resolve, delay));
    }
  }
  
  throw lastError;
}

// 使用示例
server.tool(
  "call_external_api",
  { url: z.string() },
  async ({ url }) => {
    return callWithRetry(
      () => fetch(url).then(r => r.json()),
      {
        maxRetries: 3,
        backoff: "exponential",
        retryOn: ["ECONNRESET", "ETIMEDOUT", /5\d{2}/]
      }
    );
  }
);

六、协议时代的生态格局

6.1 MCP 生态现状

截止 2026 年 5 月，MCP 生态已经相当成熟：

官方 SDK

TypeScript SDK（@modelcontextprotocol/sdk）
Python SDK（mcp）
Go SDK（社区维护）

主流 MCP Server

文件系统访问
数据库连接（PostgreSQL、MySQL、MongoDB）
搜索引擎（Elasticsearch、Pinecone）
云服务（AWS、GCP、Azure）
开发工具（GitHub、GitLab、Jira）

集成平台

Claude Desktop（官方支持）
VS Code（GitHub Copilot）
Cursor
Continue.dev

6.2 A2A 生态发展

A2A 作为较新的协议，生态正在快速成长：

支持平台

Google Agent Development Kit
LangGraph（部分支持）
AutoGen（社区适配）

典型应用场景

多 Agent 工作流编排
企业内部 Agent 市场
跨组织 Agent 协作

6.3 协议竞争与演进

MCP 的优势

Anthropic 官方推动，生态成熟
简单易用，学习曲线平缓
已有大量生产级实现

A2A 的优势

解决 Agent 间协作，填补空白
Google 生态支持
更强的标准化能力

未来趋势

MCP 可能演进为 W3C 标准
A2A 与 MCP 的融合（A2A over MCP）
更多垂直领域协议（如 Healthcare Agent Protocol）

七、实战案例：构建一个多 Agent 协作系统

7.1 系统架构

我们将构建一个"智能研报生成系统"：

用户请求 → Orchestrator Agent
                │
                ├─ Research Agent (MCP)
                │   └─ 搜索工具、数据库工具
                │
                ├─ Analysis Agent (MCP)
                │   └─ 数据分析工具、可视化工具
                │
                └─ Writer Agent (MCP)
                    └─ 文档生成工具、格式转换工具

7.2 Orchestrator 实现

# orchestrator.py
import asyncio
from dataclasses import dataclass
from typing import Any

@dataclass
class AgentEndpoint:
    name: str
    url: str
    capabilities: list[str]

class Orchestrator:
    def __init__(self):
        self.agents: dict[str, AgentEndpoint] = {}
        self.a2a_client = A2AClient()
    
    async def register_agent(self, agent: AgentEndpoint):
        """注册一个新的 Agent"""
        self.agents[agent.name] = agent
        print(f"Registered agent: {agent.name} with capabilities: {agent.capabilities}")
    
    async def find_agent(self, capability: str) -> AgentEndpoint | None:
        """根据能力查找 Agent"""
        for agent in self.agents.values():
            if capability in agent.capabilities:
                return agent
        return None
    
    async def execute_workflow(self, request: str) -> dict[str, Any]:
        """执行一个完整的研报生成工作流"""
        
        # Step 1: 研究阶段
        research_agent = await self.find_agent("research")
        if not research_agent:
            raise RuntimeError("No research agent available")
        
        research_result = await self.a2a_client.call(
            agent=research_agent,
            task="conduct_research",
            payload={"topic": request, "depth": "comprehensive"}
        )
        
        # Step 2: 分析阶段
        analysis_agent = await self.find_agent("analysis")
        analysis_result = await self.a2a_client.call(
            agent=analysis_agent,
            task="analyze_data",
            payload={"data": research_result["data"]}
        )
        
        # Step 3: 写作阶段
        writer_agent = await self.find_agent("writing")
        report = await self.a2a_client.call(
            agent=writer_agent,
            task="generate_report",
            payload={
                "research": research_result,
                "analysis": analysis_result,
                "format": "markdown"
            }
        )
        
        return {
            "report": report["content"],
            "metadata": {
                "sources": research_result["sources"],
                "analysis_methods": analysis_result["methods"]
            }
        }

7.3 Research Agent MCP Server

// research-agent/server.ts
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";

const server = new McpServer({
  name: "research-agent",
  version: "1.0.0"
});

// 搜索工具
server.tool(
  "web_search",
  {
    query: z.string().describe("搜索关键词"),
    sources: z.array(z.enum(["news", "papers", "blogs", "all"])).optional()
  },
  async ({ query, sources = ["all"] }) => {
    const results = await performSearch(query, sources);
    return {
      content: [{
        type: "text",
        text: JSON.stringify(results, null, 2)
      }]
    };
  }
);

// 数据库查询工具
server.tool(
  "query_knowledge_base",
  {
    query: z.string().describe("查询语句"),
    filters: z.object({
      date_range: z.tuple([z.string(), z.string()]).optional(),
      sources: z.array(z.string()).optional(),
      tags: z.array(z.string()).optional()
    }).optional()
  },
  async ({ query, filters }) => {
    const results = await queryDatabase(query, filters);
    return {
      content: [{
        type: "text",
        text: JSON.stringify(results, null, 2)
      }]
    };
  }
);

// 资源：已索引的数据源
server.resource(
  "indexed-sources",
  "resource://indexed-sources",
  async () => {
    const sources = await getIndexedSources();
    return {
      contents: [{
        uri: "resource://indexed-sources",
        mimeType: "application/json",
        text: JSON.stringify(sources, null, 2)
      }]
    };
  }
);

八、总结与展望

8.1 协议的价值重估

2025 年 MCP 的爆发，不是技术炒作，而是 AI Agent 发展到一定阶段的必然需求。当 AI 从"对话工具"进化为"行动主体"，标准化的通信协议成为刚需。

2026 年 A2A 的出现，进一步印证了这个趋势。Agent 间协作的标准化，将推动 AI Agent 从"单兵作战"走向"团队协作"。

8.2 给开发者的建议

如果你是工具开发者

为你的 API 提供 MCP Server 封装
遵循 MCP 的 Tools、Resources、Prompts 三大能力模型
考虑安全性、可观测性、错误处理

如果你是平台开发者

支持接入 MCP 协议
考虑 A2A 协议实现 Agent 市场
提供良好的开发者体验

如果你是企业用户

建立内部的 MCP Server 仓库
定义 Agent 协作的治理规范
关注协议安全与合规

8.3 未来展望

短期（2026）

MCP 成为 AI Agent 工具接入的事实标准
A2A 协议在头部企业落地
更多 SaaS 产品提供 MCP 接口

中期（2027-2028）

协议标准化进入 ISO/W3C 流程
Agent 市场出现（类似 App Store）
跨组织 Agent 协作成为常态

长期（2029+）

Agent 协议成为数字基础设施
人机协作的新范式形成
新的职业形态诞生（Agent 架构师、Agent 运维师）

协议时代已经到来。理解协议、掌握协议、用好协议，将是每个 AI 开发者的必修课。

关键词：MCP, Model Context Protocol, A2A, Agent-to-Agent, AI Agent, 协议, 工具调用, Anthropic, Claude, 多智能体协作

标签：AI Agent | MCP | A2A | 协议设计 | 工程实践

复制全文生成海报 MCP A2A AI Agent 协议 Anthropic

编程 从 MCP 到 A2A：2026 年 AI Agent 协议时代——标准化通信层如何重塑智能体协作范式