MCP 体系

职责概述

解决的问题：CC 内置了 30 多个工具，但用户总会有新需求——连接公司的数据库、调用内部 API、操作特定格式的文件。不可能每次都改源码。MCP（Model Context Protocol）就是 CC 的"应用商店"机制，让第三方可以开发插件工具，CC 自动发现和调用。

应用场景：① 安装一个 MCP 插件后，CC 突然能查 PostgreSQL 了 ② 团队共享的内部工具（如部署脚本）打包成 MCP Server 给全组用 ③ 第三方开发者发布工具到市场，用户一键安装 ④ 企业内网服务通过 OAuth 认证安全接入。

一句话理解：就像手机的 App Store——手机出厂自带一些功能，但你想加新功能就装 App。MCP 就是 CC 的 App 安装协议。

架构设计

服务端架构

核心数据流

连接生命周期

认证决策流

关键类型与接口

服务器配置联合类型

// services/mcp/types.ts:23-26
export const TransportSchema = lazySchema(() =>
  z.enum(['stdio', 'sse', 'sse-ide', 'http', 'ws', 'sdk']),
)

// services/mcp/types.ts:124-135 — 七种配置的联合类型
export const McpServerConfigSchema = lazySchema(() =>
  z.union([
    McpStdioServerConfigSchema(),   // 本地进程
    McpSSEServerConfigSchema(),     // SSE 远程（支持 OAuth）
    McpSSEIDEServerConfigSchema(),  // IDE 内部 SSE
    McpWebSocketIDEServerConfigSchema(), // IDE WebSocket
    McpHTTPServerConfigSchema(),    // HTTP 流式传输
    McpWebSocketServerConfigSchema(),    // 通用 WebSocket
    McpSdkServerConfigSchema(),     // SDK 进程内
    McpClaudeAIProxyServerConfigSchema(), // claude.ai 代理
  ]),
)

连接状态联合类型

// services/mcp/types.ts:180-227
export type MCPServerConnection =
  | ConnectedMCPServer   // { type: 'connected', client, capabilities, tools... }
  | FailedMCPServer      // { type: 'failed', error? }
  | NeedsAuthMCPServer   // { type: 'needs-auth' }
  | PendingMCPServer     // { type: 'pending', reconnectAttempt? }
  | DisabledMCPServer    // { type: 'disabled' }

InProcessTransport（进程内传输）

// services/mcp/InProcessTransport.ts:11-49
class InProcessTransport implements Transport {
  private peer: InProcessTransport | undefined
  async send(message: JSONRPCMessage): Promise<void> {
    if (this.closed) throw new Error('Transport is closed')
    queueMicrotask(() => {
      this.peer?.onmessage?.(message)
    })
  }
}

// 创建成对的传输通道
export function createLinkedTransportPair(): [Transport, Transport]

Channel 权限回调接口

// services/mcp/channelPermissions.ts:46-61
export type ChannelPermissionCallbacks = {
  onResponse(requestId: string,
    handler: (response: ChannelPermissionResponse) => void): () => void
  resolve(requestId: string, behavior: 'allow' | 'deny',
    fromServer: string): boolean
}

工具名称规范化

// services/mcp/mcpStringUtils.ts:39-41
export function getMcpPrefix(serverName: string): string {
  return `mcp__${normalizeNameForMCP(serverName)}__`
}
// 例: "slack" 服务器的工具 "send_message" → "mcp__slack__send_message"

设计模式与亮点

◈ 可视化处理拓扑图

MCP 系统是一个从多源配置发现到运行时工具调用的完整生命周期管理管线。配置从 5 个作用域汇聚、经过策略过滤后并发连接，每条连接根据 Transport 类型走不同的创建路径，最终将远程工具统一封装为 Claude Code 内部的 Tool 对象。

Phase 2 — 并发连接与 Transport 创建

Phase 3 — 工具发现与运行时调用

Phase 4 — 健康监控与自动重连

⇉ 核心处理流程详解

MCP（Model Context Protocol）系统是从配置发现到工具调用的完整生命周期管理。其核心链路横跨配置加载（config.ts）、连接管理（client.ts）、React 编排（useManageMCPConnections.ts）和运行时工具调用四大模块。以下是从应用启动到工具执行的完整处理链路：

★ 设计精华

1. Memoize-as-Connection-Pool 模式

不维护显式的连接池，而是利用 memoize 的缓存特性作为隐式连接池。key 由服务器名和配置对象计算（getServerCacheKey()，client.ts:581），值是连接 Promise。连接断开时只需 cache.delete(key)，下次调用自然重建。这消除了连接状态机（connecting/connected/disconnected/reconnecting）的复杂性——缓存存在即已连接，不存在即需连接。

export const connectToServer = memoize(
  async (name: string, serverRef: ScopedMcpServerConfig, ...): Promise<MCPServerConnection> => {
    // 创建 transport、执行 OAuth、发现工具...
    // 返回 ConnectedMCPServer 或 FailedMCPServerConnection
  }
)
// 断开时清空缓存 → 下次调用自动重连
clearServerCache(name, serverRef)

2. 分层去重：签名 + URL 双重保障

MCP 配置来自 5 个不同作用域，同一个服务器可能被多处配置。getMcpServerSignature()（config.ts:202）为每个配置计算签名（基于 URL 或 command 数组）。在插件层，dedupPluginMcpServers()（config.ts:223）阻止插件注册已存在的 MCP 服务器；在 claude.ai 层，dedupClaudeAiMcpServers()（config.ts:281）阻止 claude.ai connector 与手动配置冲突。两层去重确保同一后端服务只有一个活跃连接。

3. SSE 长连接的超时分离策略

SSE 传输层有两个 fetch：EventSource 的长连接 fetch 和普通 API 请求的 fetch。wrapFetchWithTimeout()（client.ts:492）仅包装 API 请求 fetch（60 秒超时），EventSource 的 fetch 保持无超时。如果错误地给 EventSource 也加了超时，SSE 流会在 60 秒后断开，导致推送通知丢失。这个分离通过 transportOptions.eventSourceInit 和 transportOptions.fetch 分别配置实现 — client.ts:630-671

// API 请求：带超时
transportOptions.fetch = wrapFetchWithTimeout(
  wrapFetchWithStepUpDetection(createFetchWithInit(), authProvider),
)
// EventSource 长连接：不设超时
transportOptions.eventSourceInit = {
  fetch: async (url, init) => fetch(url, { ...init, ...proxyOptions })
}

4. 批量更新的时间窗口合并

React 的状态更新是同步的——每个 setAppState 触发一次渲染。当 10 个 MCP 服务器同时连接完成时，不加控制会产生 10 次渲染。useManageMCPConnections 用 MCP_BATCH_FLUSH_MS = 16（一帧时间）的时间窗口收集所有更新，在单次 setAppState 中批量应用（useManageMCPConnections.ts:216-270）。16ms 足够收集同一"事件循环 tick"中的多个回调，又短到用户感知不到延迟。

◈ Agent 实践借鉴 — 客服 Agent 外部系统集成设计

场景映射：客服 agent 在外部系统集成上的真实问题

客服 agent 需要连接 6+ 外部系统，每个系统协议不同、稳定性不同、接入方式也不同。典型痛点包括：

• 多系统多协议：CRM（REST API 客户信息）、ERP（gRPC 订单/库存）、支付网关（REST 退款）、物流系统（消息队列查件）、知识库（REST FAQ）、短信平台（REST 通知）。甚至还有老系统走 SOAP 协议。
• 第三方系统不稳定：物流系统每周二凌晨维护、CRM 偶尔超时、支付网关在月底高峰期限流。断线后客服 agent 不能直接挂掉，需要自动重连或降级处理。
• 新系统接入怕影响老系统：新增一个短信平台连接，不能因为短信平台的 bug 导致已有的 CRM 查询和订单查询挂掉。连接之间必须隔离。
• 配置分散难管理：6 个系统的连接配置散落在不同地方，改一个配置要改三个文件，还容易改漏。

借鉴 CC + 客服改造

1. Memoize 连接池（CC 的核心模式，客服场景最该抄的）

CC 的 connectToServer 用 memoize 包装，缓存存在=已连接，缓存清除=重连。这个模式在客服场景非常合适——连接 6 个外部系统不需要维护 6 个状态机：

// 借鉴 CC 的 Memoize-as-Connection-Pool → 客服系统连接池
// 核心思想：不维护显式连接池，用缓存特性管理连接

interface ExternalSystem {
  name: string               // 'crm' | 'erp' | 'payment' | 'logistics' | 'kb' | 'sms'
  protocol: 'rest' | 'grpc' | 'mq' | 'soap'
  config: SystemConfig
}

// 用 memoize 实现隐式连接池——缓存存在=已连接
const connectToExternalSystem = memoize(
  async (systemName: string, config: SystemConfig): Promise<SystemConnection> => {
    const adapter = createTransportAdapter(config)  // 工厂分支
    const connection = await adapter.connect(config)
    const health = await connection.healthCheck()
    if (!health.ok) throw new Error(`${systemName} 健康检查失败`)
    return { status: 'connected', connection, adapter }
  }
)

// 连接断开时：清缓存 = 重连（无需显式状态机）
function onSystemConnectionLost(systemName: string, config: SystemConfig) {
  clearConnectionCache(systemName, config)
  // 下次工具调用 connectToExternalSystem() 自动重建
  // 借鉴 CC 的 onclose 处理器：连续错误超限则降级
  logWarning(`外部系统 ${systemName} 断开，下次调用自动重连`)
}

// 降级策略：连接失败不影响其他系统
async function callWithFallback(
  systemName: string,
  operation: string,
  params: any,
  fallbackValue?: any
): Promise<any> {
  try {
    const conn = await connectToExternalSystem(systemName, getConfig(systemName))
    return await conn.connection.call(operation, params)
  } catch (error) {
    logError(`${systemName}.${operation} 调用失败`, error)
    if (fallbackValue !== undefined) return fallbackValue
    throw new ServiceUnavailableError(systemName, operation)
  }
}

2. 统一 Transport 适配器（CC 6 种 → 客服 2-3 种）

CC 的 6 种 Transport（stdio/SSE/HTTP/WS/SDK/InProcess）统一为 Transport 接口。客服场景只需要 REST/gRPC/MQ 三种，但同样需要统一接口：

// 借鉴 CC 的 Transport 接口 → 客服场景的统一适配器
interface ServiceTransport {
  connect(config: SystemConfig): Promise<Connection>
  send(request: ServiceRequest): Promise<ServiceResponse>
  close(): Promise<void>
  healthCheck(): Promise<HealthStatus>
}

// REST 适配器——大多数客服系统走 REST
class RestTransport implements ServiceTransport {
  private client: HttpClient | undefined

  async connect(config: SystemConfig): Promise<Connection> {
    this.client = createHttpClient(config.baseUrl, {
      timeout: config.timeout || 5000,
      retries: config.retries || 2,
      headers: { 'Authorization': `Bearer ${config.apiKey}` },
    })
    return { type: 'rest', client: this.client }
  }

  async send(request: ServiceRequest): Promise<ServiceResponse> {
    return this.client!.request({
      method: request.method,
      path: request.path,
      body: request.params,
    })
  }

  async healthCheck(): Promise<HealthStatus> {
    try {
      await this.client!.get('/health')
      return { ok: true }
    } catch { return { ok: false } }
  }

  close() { this.client = undefined }
}

// MQ 适配器——物流系统走消息队列
class MessageQueueTransport implements ServiceTransport {
  private consumer: MQConsumer | undefined

  async connect(config: SystemConfig): Promise<Connection> {
    this.consumer = await createConsumer(config.queueUrl, config.queueName)
    return { type: 'mq', consumer: this.consumer }
  }

  async send(request: ServiceRequest): Promise<ServiceResponse> {
    // 发送查询消息 → 等待响应消息（带超时）
    const correlationId = generateId()
    return waitForResponse(this.consumer!, correlationId, config.responseTimeout)
  }
  // ... healthCheck, close
}

3. 外部 API → 内部 Tool 封装

CC 把每个 MCP 工具封装为 Tool 对象。客服场景把每个外部系统 API 封装为统一的 Tool：

// 借鉴 CC 的 wrapMcpTool → 客服场景的外部 API → Tool 封装
interface CustomerServiceTool {
  name: string                 // 如 "query_order"
  description: string          // 如 "查询订单状态"
  requiredSystem: string       // 如 "erp"
  requiredPermission: string   // 如 "order:read"
  inputSchema: object          // 参数 schema
  execute(params: any): Promise<ToolResult>
}

// 封装 CRM 的"查客户"接口为统一 Tool
function createQueryCustomerTool(crmSystem: string): CustomerServiceTool {
  return {
    name: 'query_customer',
    description: '查询客户信息（CRM系统）',
    requiredSystem: crmSystem,
    requiredPermission: 'customer:read',
    inputSchema: {
      type: 'object',
      properties: {
        phone: { type: 'string', description: '客户手机号' },
        customerId: { type: 'string', description: '客户ID' },
      },
    },
    async execute(params) {
      // 连接 + 调用 + 结果转换
      const conn = await connectToExternalSystem(
        crmSystem, getConfig(crmSystem)
      )
      const raw = await conn.connection.call('getCustomer', {
        phone: params.phone,
        customerId: params.customerId,
      })
      // 原始 API 返回 → 统一 ToolResult 格式
      return {
        success: true,
        data: {
          name: raw.customer_name,
          phone: raw.phone_number,
          vipLevel: raw.vip_level,
          tags: raw.tags,
        },
        summary: `客户 ${raw.customer_name}，${raw.vip_level}会员`,
      }
    },
  }
}

// agent 调用时无感知底层是 REST 还是 gRPC
// const result = await toolRegistry.get('query_customer').execute({ phone: '138xxxx' })

4. 连接健康检查 + 自动重连

// 借鉴 CC 的 onclose/onerror + reconnectWithBackoff
// 健康检查循环——每 30 秒检查所有外部系统连接
async function startHealthMonitor(systems: ExternalSystem[]): Promise<void> {
  setInterval(async () => {
    for (const system of systems) {
      const cacheKey = getCacheKey(system.name, system.config)
      const cached = connectionCache.get(cacheKey)
      if (!cached) continue  // 未连接，跳过

      const health = await cached.adapter.healthCheck().catch(() => ({ ok: false }))
      if (!health.ok) {
        // 借鉴 CC：清缓存 → 下次调用自动重建
        connectionCache.delete(cacheKey)
        logWarning(`${system.name} 健康检查失败，已清除缓存，下次调用自动重连`)
        // 通知 agent 该系统暂时不可用
        notifySystemStatus(system.name, 'degraded')
      }
    }
  }, 30_000)
}

// 指数退避重连——借鉴 CC 的 reconnectWithBackoff
async function reconnectWithBackoff(
  systemName: string,
  config: SystemConfig,
  maxAttempts: number = 5,
): Promise<SystemConnection> {
  for (let attempt = 1; attempt <= maxAttempts; attempt++) {
    try {
      clearConnectionCache(systemName, config)
      const conn = await connectToExternalSystem(systemName, config)
      notifySystemStatus(systemName, 'healthy')
      return conn
    } catch (error) {
      const delay = Math.min(1000 * Math.pow(2, attempt), 30_000)
      logWarning(`${systemName} 重连失败（第${attempt}次），${delay}ms后重试`)
      await sleep(delay)
    }
  }
  throw new Error(`${systemName} 重连${maxAttempts}次后仍然失败`)
}

落地清单

代码索引