MCP 生态前沿与高级议题

MCP 生态持续发展，新特性和扩展不断涌现。本文详解协议扩展机制、多服务器编排、流式传输、自定义能力、性能优化策略以及 MCP 与 Agent 框架的集成模式。

Plume 2026-09-22 6 分钟阅读 2814 字

1. 协议扩展机制

MCP 的核心协议保持精简，但通过扩展机制支持丰富的附加能力。理解扩展机制是构建高级 MCP 应用的基础。

1.1 能力协商

在初始化握手阶段，客户端和服务器交换各自支持的 capabilities：

// 客户端声明支持的能力
const clientCapabilities = {
  roots: {        // 文件系统根目录列表
    listChanged: true,
  },
  sampling: {},   // 采样/补全能力
  experimental: { // 实验性功能
    "custom-feature-1": { version: "1.0" },
    "custom-feature-2": { enabled: true },
  },
};

// 服务器声明的能力
const serverCapabilities = {
  tools: {
    listChanged: true,  // 支持动态工具列表更新
  },
  resources: {
    subscribe: true,    // 支持资源订阅
    listChanged: true,
  },
  prompts: {
    listChanged: true,
  },
  logging: {},          // 支持结构化日志
  experimental: {
    "streaming-results": { version: "2.0" },
  },
};

双方只能使用对方声明支持的能力。这种协商机制确保了协议的可扩展性和向后兼容性。

1.2 自定义扩展

开源社区和厂商已经开始扩展 MCP 协议以支持特定场景。常见的扩展方向包括：

认证扩展：标准的 MCP 协议没有定义认证机制，生产环境通常需要额外的 TLS/mTLS 或 OAuth2 集成：

// 自定义认证握手（实验性扩展）
interface AuthExtension {
  method: "bearer" | "mtls" | "api_key";
  token?: string;
  client_cert?: string;
}

// 在初始化后增加认证步骤
const authRequest = {
  jsonrpc: "2.0",
  method: "auth/authenticate",
  params: {
    method: "bearer",
    token: "eyJhbGciOiJSUzI1NiIs...",
  },
};

流式结果扩展：标准 MCP 的工具返回是同步的。对于长时间运行的工具（如数据分析、大模型推理），流式返回中间结果能显著提升用户体验：

// 流式结果通知
interface StreamingResultNotification {
  jsonrpc: "2.0";
  method: "notifications/tools/streaming_result";
  params: {
    tool_call_id: string;
    chunk: {
      type: "text" | "progress" | "partial_data";
      content: unknown;
      progress?: number; // 0-100
    };
    is_final: boolean;
  };
}

2. 多服务器编排

2.1 服务器组合模式

复杂任务需要多个 MCP 服务器的协作：

// 定义服务器拓扑
interface ServerTopology {
  servers: Array<{
    name: string;
    endpoint: string;
    capabilities: string[];
  }>;
  pipelines: Array<{
    name: string;
    steps: Array<{
      server: string;
      tool: string;
      input_mapping: Record<string, string>;
      output_mapping: Record<string, string>;
    }>;
  }>;
}

// 示例：代码审查流水线
const codeReviewTopology: ServerTopology = {
  servers: [
    { name: "git", endpoint: "mcp://localhost:3001", capabilities: ["tools"] },
    { name: "fs", endpoint: "mcp://localhost:3002", capabilities: ["tools", "resources"] },
    { name: "linter", endpoint: "mcp://localhost:3003", capabilities: ["tools"] },
    { name: "test", endpoint: "mcp://localhost:3004", capabilities: ["tools"] },
  ],
  pipelines: [
    {
      name: "full_review",
      steps: [
        {
          server: "git",
          tool: "get_diff",
          input_mapping: { branch: "target_branch" },
          output_mapping: { diff: "changes" },
        },
        {
          server: "fs",
          tool: "read_changed_files",
          input_mapping: { diff: "changes" },
          output_mapping: { files: "changed_files" },
        },
        {
          server: "linter",
          tool: "lint_files",
          input_mapping: { files: "changed_files" },
          output_mapping: { issues: "lint_issues" },
        },
        {
          server: "test",
          tool: "run_tests_for_files",
          input_mapping: { files: "changed_files" },
          output_mapping: { results: "test_results" },
        },
      ],
    },
  ],
};

2.2 编排引擎

class MCPPipelineEngine {
  private clients: Map<string, Client> = new Map();
  private context: Map<string, unknown> = new Map();

  async loadTopology(topology: ServerTopology): Promise<void> {
    for (const server of topology.servers) {
      const client = new Client({ name: "orchestrator", version: "1.0" });
      const transport = new SSEClientTransport(new URL(server.endpoint));
      await client.connect(transport);
      this.clients.set(server.name, client);
    }
  }

  async executePipeline(
    pipelineName: string,
    topology: ServerTopology,
    initialInput: Record<string, unknown>
  ): Promise<Record<string, unknown>> {
    const pipeline = topology.pipelines.find((p) => p.name === pipelineName);
    if (!pipeline) throw new Error(`Pipeline not found: ${pipelineName}`);

    // 初始化上下文
    this.context.clear();
    for (const [key, value] of Object.entries(initialInput)) {
      this.context.set(key, value);
    }

    for (const step of pipeline.steps) {
      const client = this.clients.get(step.server);
      if (!client) throw new Error(`Client not found: ${step.server}`);

      // 解析输入
      const args = this.resolveInputMapping(step.input_mapping);

      console.log(`Executing ${step.server}.${step.tool} with`, args);
      const result = await client.callTool({ name: step.tool, arguments: args });

      // 映射输出到上下文
      this.applyOutputMapping(step.output_mapping, result);
    }

    // 返回最终上下文
    return Object.fromEntries(this.context);
  }

  private resolveInputMapping(mapping: Record<string, string>): Record<string, unknown> {
    const args: Record<string, unknown> = {};
    for (const [argName, contextKey] of Object.entries(mapping)) {
      args[argName] = this.context.get(contextKey);
    }
    return args;
  }

  private applyOutputMapping(
    mapping: Record<string, string>,
    result: any
  ): void {
    for (const [contextKey, resultPath] of Object.entries(mapping)) {
      const value = this.getPath(result, resultPath);
      this.context.set(contextKey, value);
    }
  }

  private getPath(obj: any, path: string): unknown {
    return path.split(".").reduce((o, p) => o?.[p], obj);
  }
}

3. 自定义传输层

3.1 WebSocket 传输

对于 bidirectional 实时通信场景，WebSocket 是比 SSE 更好的选择：

// WebSocket MCP 传输实现
import { WebSocketServer, WebSocket } from "ws";
import { ServerTransport } from "@modelcontextprotocol/sdk/server/index.js";

class WebSocketServerTransport implements ServerTransport {
  private ws: WebSocketServer;
  private clients: Map<string, WebSocket> = new Map();

  constructor(port: number) {
    this.ws = new WebSocketServer({ port });
    this.ws.on("connection", (socket, req) => {
      const clientId = crypto.randomUUID();
      this.clients.set(clientId, socket);

      socket.on("message", (data) => {
        this.handleMessage(clientId, data.toString());
      });

      socket.on("close", () => {
        this.clients.delete(clientId);
      });
    });
  }

  private handleMessage(clientId: string, data: string): void {
    const message = JSON.parse(data);
    // 处理 JSON-RPC 消息
    // ...
  }

  async send(message: unknown): Promise<void> {
    const data = JSON.stringify(message);
    for (const [id, socket] of this.clients) {
      if (socket.readyState === WebSocket.OPEN) {
        socket.send(data);
      }
    }
  }

  async close(): Promise<void> {
    return new Promise((resolve) => {
      this.ws.close(resolve);
    });
  }
}

3.2 消息队列传输

对于高可用部署，基于消息队列的传输可以解耦客户端和服务器：

// Redis Stream 作为 MCP 传输
import { createClient } from "redis";

class RedisStreamTransport implements ServerTransport {
  private redis: ReturnType<typeof createClient>;
  private streamName: string;
  private consumerGroup: string;

  constructor(redisUrl: string, streamName: string) {
    this.redis = createClient({ url: redisUrl });
    this.streamName = streamName;
    this.consumerGroup = `mcp-group-${streamName}`;
  }

  async start(): Promise<void> {
    await this.redis.connect();
    try {
      await this.redis.xGroupCreate(this.streamName, this.consumerGroup, "$", {
        MKSTREAM: true,
      });
    } catch {
      // 组已存在
    }
  }

  async readMessages(): Promise<any[]> {
    const messages = await this.redis.xReadGroup(
      this.consumerGroup,
      "consumer-1",
      { key: this.streamName, id: ">" },
      { COUNT: 10, BLOCK: 5000 }
    );
    return messages ?? [];
  }

  async sendResponse(message: unknown): Promise<void> {
    await this.redis.xAdd(this.streamName, "*", {
      message: JSON.stringify(message),
    });
  }
}

4. 与 Agent 框架集成

4.1 MCP + LangChain

import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { Tool } from "@langchain/core/tools";

class MCPToolAdapter extends Tool {
  name: string;
  description: string;
  private client: Client;
  private toolName: string;

  constructor(client: Client, mcpTool: { name: string; description: string; inputSchema: any }) {
    super();
    this.client = client;
    this.toolName = mcpTool.name;
    this.name = mcpTool.name;
    this.description = this.buildDescription(mcpTool);
  }

  private buildDescription(mcpTool: { description: string; inputSchema: any }): string {
    return `${mcpTool.description}\n\n参数: ${JSON.stringify(mcpTool.inputSchema, null, 2)}`;
  }

  async _call(args: string): Promise<string> {
    const parsed = JSON.parse(args);
    const result = await this.client.callTool({
      name: this.toolName,
      arguments: parsed,
    });

    return result.content
      .filter((c) => c.type === "text")
      .map((c) => c.text)
      .join("\n");
  }
}

// 使用
async function createLangChainTools(client: Client): Promise<Tool[]> {
  const tools = await client.listTools();
  return tools.tools.map((t) => new MCPToolAdapter(client, t));
}

4.2 MCP + AutoGen

from autogen import ConversableAgent
from mcp_client import MCPClient

class MCPAutoGenTool:
    def __init__(self, mcp_client: MCPClient, tool_name: str, tool_schema: dict):
        self.client = mcp_client
        self.tool_name = tool_name
        self.schema = tool_schema

    def __call__(self, **kwargs):
        result = self.client.call_tool(self.tool_name, kwargs)
        return result.content[0].text

# 注册 MCP 工具到 AutoGen Agent
mcp_client = MCPClient("http://localhost:3000/sse")
tools = await mcp_client.list_tools()

autogen_tools = []
for tool in tools.tools:
    mcp_tool = MCPAutoGenTool(mcp_client, tool.name, tool.inputSchema)
    autogen_tools.append({
        "name": tool.name,
        "description": tool.description,
        "func": mcp_tool,
    })

agent = ConversableAgent(
    name="mcp_assistant",
    llm_config={"config_list": [...]},
    system_message="你是一个可以访问外部工具的助手。",
)

for tool in autogen_tools:
    agent.register_function(tool)

5. 性能优化策略

5.1 连接池

class MCPConnectionPool {
  private pool: Client[] = [];
  private inUse: Set<Client> = new Set();
  private maxSize: number;

  constructor(
    private serverEndpoint: string,
    private credentials: any,
    maxSize: number = 10
  ) {
    this.maxSize = maxSize;
  }

  async acquire(): Promise<Client> {
    // 复用空闲连接
    for (const client of this.pool) {
      if (!this.inUse.has(client)) {
        this.inUse.add(client);
        return client;
      }
    }

    // 创建新连接（未达上限时）
    if (this.pool.length < this.maxSize) {
      const client = new Client({ name: "pooled-client", version: "1.0" });
      const transport = new SSEClientTransport(new URL(this.serverEndpoint));
      await client.connect(transport);
      this.pool.push(client);
      this.inUse.add(client);
      return client;
    }

    // 等待连接释放
    await this.waitForAvailable();
    return this.acquire();
  }

  release(client: Client): void {
    this.inUse.delete(client);
  }

  private async waitForAvailable(): Promise<void> {
    return new Promise((resolve) => {
      const check = setInterval(() => {
        if (this.inUse.size < this.pool.length) {
          clearInterval(check);
          resolve();
        }
      }, 100);
    });
  }
}

5.2 智能缓存

class IntelligentCache {
  private cache: Map<string, { result: any; expiry: number; tags: string[] }> = new Map();

  async get<T>(key: string, fetcher: () => Promise<T>, options: {
    ttl: number;
    tags?: string[];
    staleWhileRevalidate?: boolean;
  }): Promise<T> {
    const entry = this.cache.get(key);
    const now = Date.now();

    if (entry && now < entry.expiry) {
      return entry.result;
    }

    if (entry && options.staleWhileRevalidate && now < entry.expiry + 30000) {
      // 返回过期值，后台刷新
      this.refreshInBackground(key, fetcher, options);
      return entry.result;
    }

    const result = await fetcher();
    this.cache.set(key, {
      result,
      expiry: now + options.ttl,
      tags: options.tags ?? [],
    });
    return result;
  }

  private async refreshInBackground<T>(key: string, fetcher: () => Promise<T>, options: any): Promise<void> {
    try {
      const result = await fetcher();
      this.cache.set(key, {
        result,
        expiry: Date.now() + options.ttl,
        tags: options.tags ?? [],
      });
    } catch (error) {
      console.warn(`Background refresh failed for ${key}:`, error);
    }
  }

  invalidateByTag(tag: string): void {
    for (const [key, entry] of this.cache) {
      if (entry.tags.includes(tag)) {
        this.cache.delete(key);
      }
    }
  }
}

6. 协议的演进方向

6.1 提议中的标准化

MCP 社区正在讨论纳入协议的未来特性：

批处理支持：一次请求调用多个工具，减少往返延迟：

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/batch_call",
  "params": {
    "calls": [
      { "name": "read_file", "arguments": { "path": "/a" } },
      { "name": "read_file", "arguments": { "path": "/b" } },
      { "name": "read_file", "arguments": { "path": "/c" } }
    ]
  }
}

语义搜索：在工具列表上进行向量相似度搜索，帮助 AI 更准确地找到工具：

{
  "jsonrpc": "2.0",
  "method": "tools/semantic_search",
  "params": {
    "query": "查找最近修改的文件",
    "top_k": 5
  }
}

权限动态协商：运行时发现需要额外权限，服务器可以请求升级：

{
  "jsonrpc": "2.0",
  "method": "permissions/request",
  "params": {
    "tool": "write_file",
    "required_scope": "write",
    "reason": "用户明确要求修改配置文件"
  }
}

6.2 与 OpenAI Assistants API 的对比

维度	MCP	OpenAI Assistants API
协议层次	传输层 + 应用层协议	供应商特定 API
工具发现	动态（运行时协商）	静态（创建时绑定）
开放性	开放标准，生态中立	供应商锁定
状态管理	无状态服务器 + 客户端维护	服务器维护线程和运行状态
文件处理	Resources（URI寻址）	内置文件上传
适用场景	跨平台的 AI 工具集成	OpenAI 生态系统内应用

7. 生态工具推荐

7.1 开发工具

MCP Inspector：官方调试工具，可视化协议通信
mcp-cli：命令行 MCP 客户端，用于测试服务器
MCP Playground：Web 界面测试工具调用组合

7.2 服务器集合

官方：github-server、postgres -server、slack-server、filesystem-server
社区：brave-search、notion、discord、docker 、kubernetes
框架集成：Next.js server、Django server、Rails server

8. 总结

MCP 的生态正处于快速成长期。从协议设计来看，MCP 在几个关键决策上展现了良好的前瞻性：

传输层抽象：不绑定特定传输，允许 stdio、HTTP、WebSocket、消息队列等多种方式
能力协商：渐进式功能发现，向后兼容性好
三层能力模型：Tools + Resources + Prompts 覆盖了 AI 助手的核心交互模式
去中心化：不绑定特定模型供应商或平台

对于开发者而言，现在投入 MCP 生态意味着：

编写的工具可以被所有 MCP Host 使用（包括 Claude Desktop、Cursor、VS Code Copilot 等）
构建的 AI 应用可以接入越来越多的优质工具而无需逐一适配
参与一个正在形成的行业标准，有很大机会影响其发展方向

MCP 的未来将取决于能否形成正向飞轮：更多 Host 支持 MCP → 更多开发者编写 MCP Server → 更丰富的工具生态 → 更多用户使用 MCP Host。当前看来，这个飞轮已经开始转动。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页