第 18 章MCP 协议实现连接外部工具的标准化桥梁

// services/mcp/types.ts:10-26
export const ConfigScopeSchema = lazySchema(() =>
  z.enum([
    'local',      // .claude/settings.local.json
    'user',       // ~/.claude/settings.json
    'project',    // .mcp.json（从 CWD 向上遍历）
    'dynamic',    // 运行时动态注入（--mcp-config）
    'enterprise', // managed-mcp.json（企业管控）
    'claudeai',   // claude.ai 连接器
    'managed',    // 企业 managed settings
  ]),
)

export const TransportSchema = lazySchema(() =>
  z.enum(['stdio', 'sse', 'sse-ide', 'http', 'ws', 'sdk']),
)

// services/mcp/types.ts:28-113（简化展示）
// stdio：本地进程
export const McpStdioServerConfigSchema = lazySchema(() =>
  z.object({
    type: z.literal('stdio').optional(), // 可选，向后兼容
    command: z.string().min(1, 'Command cannot be empty'),
    args: z.array(z.string()).default([]),
    env: z.record(z.string(), z.string()).optional(),
  }),
)

// SSE：Server-Sent Events（远程）
export const McpSSEServerConfigSchema = lazySchema(() =>
  z.object({
    type: z.literal('sse'),
    url: z.string(),
    headers: z.record(z.string(), z.string()).optional(),
    headersHelper: z.string().optional(),  // 动态 header 脚本
    oauth: McpOAuthConfigSchema().optional(),
  }),
)

// HTTP：Streamable HTTP（MCP 2025-03-26 规范推荐）
export const McpHTTPServerConfigSchema = lazySchema(() =>
  z.object({
    type: z.literal('http'),
    url: z.string(),
    headers: z.record(z.string(), z.string()).optional(),
    headersHelper: z.string().optional(),
    oauth: McpOAuthConfigSchema().optional(),
  }),
)

// SDK：进程内 SDK 传输（Agent SDK 使用）
export const McpSdkServerConfigSchema = lazySchema(() =>
  z.object({
    type: z.literal('sdk'),
    name: z.string(),
  }),
)

// services/mcp/types.ts:124-135
export const McpServerConfigSchema = lazySchema(() =>
  z.union([
    McpStdioServerConfigSchema(),
    McpSSEServerConfigSchema(),
    McpSSEIDEServerConfigSchema(),
    McpWebSocketIDEServerConfigSchema(),
    McpHTTPServerConfigSchema(),
    McpWebSocketServerConfigSchema(),
    McpSdkServerConfigSchema(),
    McpClaudeAIProxyServerConfigSchema(),
  ]),
)

// services/mcp/types.ts:180-227
export type ConnectedMCPServer = {
  client: Client           // MCP SDK 客户端实例
  name: string
  type: 'connected'
  capabilities: ServerCapabilities
  serverInfo?: { name: string; version: string }
  instructions?: string    // 服务器提供的使用指南
  config: ScopedMcpServerConfig
  cleanup: () => Promise<void>  // 清理函数
}

export type FailedMCPServer = {
  name: string; type: 'failed'; config: ScopedMcpServerConfig; error?: string
}

export type NeedsAuthMCPServer = {
  name: string; type: 'needs-auth'; config: ScopedMcpServerConfig
}

export type PendingMCPServer = {
  name: string; type: 'pending'; config: ScopedMcpServerConfig
  reconnectAttempt?: number; maxReconnectAttempts?: number
}

export type DisabledMCPServer = {
  name: string; type: 'disabled'; config: ScopedMcpServerConfig
}

export type MCPServerConnection =
  | ConnectedMCPServer
  | FailedMCPServer
  | NeedsAuthMCPServer
  | PendingMCPServer
  | DisabledMCPServer

// services/mcp/useManageMCPConnections.ts:856-964（简化展示）
async function loadAndConnectMcpConfigs() {
  // claude.ai fetch 提前发起，与 Phase 1 并行
  const claudeaiPromise = fetchClaudeAIMcpConfigsIfEligible()

  // Phase 1: 仅本地文件读取，快速
  const { servers: claudeCodeConfigs } = isStrictMcpConfig
    ? { servers: {}, errors: [] }
    : await getClaudeCodeMcpConfigs(dynamicMcpConfig, claudeaiPromise)

  // dynamic 后覆盖 Claude Code 配置（优先级最高）
  const configs = { ...claudeCodeConfigs, ...dynamicMcpConfig }

  // 开始连接，不等待完成（fire-and-forget）
  getMcpToolsCommandsAndResources(onConnectionAttempt, enabledConfigs)

  // Phase 2: 等待 claude.ai 结果
  const claudeaiConfigs = filterMcpServersByPolicy(await claudeaiPromise).allowed
  // 按 URL 签名去重，手动配置优先于 claude.ai 连接器
  const { servers: dedupedClaudeAi } = dedupClaudeAiMcpServers(claudeaiConfigs, configs)
  // 追加连接 claude.ai 服务器
  getMcpToolsCommandsAndResources(onConnectionAttempt, dedupedClaudeAi)
}

// services/mcp/config.ts:1231-1238
// plugin < user < project < local
const configs = Object.assign(
  {},
  dedupedPluginServers,    // 最低：插件提供的服务器
  userServers,             // ~/.claude/settings.json
  approvedProjectServers,  // .mcp.json（需通过审批）
  localServers,            // .claude/settings.local.json（最高）
)

// services/mcp/config.ts:1082-1096
if (doesEnterpriseMcpConfigExist()) {
  const filtered: Record<string, ScopedMcpServerConfig> = {}
  for (const [name, serverConfig] of Object.entries(enterpriseServers)) {
    if (!isMcpServerAllowedByPolicy(name, serverConfig)) {
      continue
    }
    filtered[name] = serverConfig
  }
  return { servers: filtered, errors: [] }
}

// services/mcp/config.ts:913-955
case 'project': {
  const dirs: string[] = []
  let currentDir = getCwd()
  while (currentDir !== parse(currentDir).root) {
    dirs.push(currentDir)
    currentDir = dirname(currentDir)
  }
  // 从根目录向 CWD 方向处理，靠近 CWD 的文件覆盖远端
  for (const dir of dirs.reverse()) {
    const mcpJsonPath = join(dir, '.mcp.json')
    const { config, errors } = parseMcpConfigFromFilePath({
      filePath: mcpJsonPath, expandVars: true, scope: 'project',
    })
    if (config?.mcpServers) {
      Object.assign(allServers, addScopeToServers(config.mcpServers, scope))
    }
  }
}

// services/mcp/config.ts:202-212
export function getMcpServerSignature(config: McpServerConfig): string | null {
  const cmd = getServerCommandArray(config)
  if (cmd) {
    return `stdio:${jsonStringify(cmd)}`  // 本地进程：按命令+参数签名
  }
  const url = getServerUrl(config)
  if (url) {
    return `url:${unwrapCcrProxyUrl(url)}`  // 远程服务：按 URL 签名
  }
  return null  // sdk 类型没有签名
}

// services/mcp/envExpansion.ts:10-38
export function expandEnvVarsInString(value: string): {
  expanded: string; missingVars: string[]
} {
  const missingVars: string[] = []
  const expanded = value.replace(/\$\{([^}]+)\}/g, (match, varContent) => {
    const [varName, defaultValue] = varContent.split(':-', 2)
    const envValue = process.env[varName]
    if (envValue !== undefined) return envValue
    if (defaultValue !== undefined) return defaultValue
    missingVars.push(varName)
    return match  // 保留原文，便于调试
  })
  return { expanded, missingVars }
}

// services/mcp/config.ts:364-408
function isMcpServerDenied(serverName: string, config?: McpServerConfig): boolean {
  const settings = getMcpDenylistSettings()
  if (!settings.deniedMcpServers) return false

  // 按名称匹配
  for (const entry of settings.deniedMcpServers) {
    if (isMcpServerNameEntry(entry) && entry.serverName === serverName) {
      return true
    }
  }
  // 按命令匹配（stdio 服务器）
  // 按 URL 通配符匹配（远程服务器）
  // ...
}

// services/mcp/client.ts:944-958
} else if (serverRef.type === 'stdio' || !serverRef.type) {
  const finalCommand =
    process.env.CLAUDE_CODE_SHELL_PREFIX || serverRef.command
  const finalArgs = process.env.CLAUDE_CODE_SHELL_PREFIX
    ? [[serverRef.command, ...serverRef.args].join(' ')]
    : serverRef.args
  transport = new StdioClientTransport({
    command: finalCommand,
    args: finalArgs,
    env: {
      ...subprocessEnv(),   // 继承环境变量
      ...serverRef.env,     // 用户自定义环境变量
    } as Record<string, string>,
    stderr: 'pipe',  // 拦截 stderr，防止污染终端 UI
  })
}

// services/mcp/client.ts:492-549
export function wrapFetchWithTimeout(baseFetch: FetchLike): FetchLike {
  return async (url: string | URL, init?: RequestInit) => {
    const method = (init?.method ?? 'GET').toUpperCase()
    // GET 请求不设超时——MCP 中 GET 是长连接的 SSE 流
    if (method === 'GET') {
      return baseFetch(url, init)
    }
    // 用 setTimeout 而非 AbortSignal.timeout()
    // 因为 Bun 中 AbortSignal.timeout 的内部定时器在 GC 前不释放，
    // 每个请求泄漏 ~2.4KB 原生内存
    const controller = new AbortController()
    const timer = setTimeout(
      c => c.abort(new DOMException('The operation timed out.', 'TimeoutError')),
      MCP_REQUEST_TIMEOUT_MS, // 60 秒
      controller,
    )
    timer.unref?.()  // 不阻止 Node.js 退出
    // ... 清理逻辑
  }
}

// services/mcp/InProcessTransport.ts:11-49
class InProcessTransport implements Transport {
  private peer: InProcessTransport | undefined
  private closed = false

  async send(message: JSONRPCMessage): Promise<void> {
    if (this.closed) throw new Error('Transport is closed')
    // 异步投递到对端，避免同步请求/响应导致栈溢出
    queueMicrotask(() => {
      this.peer?.onmessage?.(message)
    })
  }

  async close(): Promise<void> {
    if (this.closed) return
    this.closed = true
    this.onclose?.()
    // 关闭对端
    if (this.peer && !this.peer.closed) {
      this.peer.closed = true
      this.peer.onclose?.()
    }
  }
}

// 创建一对连接的传输通道
export function createLinkedTransportPair(): [Transport, Transport] {
  const a = new InProcessTransport()
  const b = new InProcessTransport()
  a._setPeer(b)
  b._setPeer(a)
  return [a, b]
}

// services/mcp/client.ts:910-924
const { createLinkedTransportPair } = await import('./InProcessTransport.js')
const context = createChromeContext(serverRef.env)
inProcessServer = createClaudeForChromeMcpServer(context)
const [clientTransport, serverTransport] = createLinkedTransportPair()
await inProcessServer.connect(serverTransport)
transport = clientTransport

// services/mcp/client.ts:1048-1077
const connectPromise = client.connect(transport)
const timeoutPromise = new Promise<never>((_, reject) => {
  const timeoutId = setTimeout(() => {
    if (inProcessServer) inProcessServer.close().catch(() => {})
    transport.close().catch(() => {})
    reject(new TelemetrySafeError(
      `MCP server "${name}" connection timed out after ${getConnectionTimeoutMs()}ms`,
      'MCP connection timeout',
    ))
  }, getConnectionTimeoutMs())
  // 如果 connect 先完成，取消超时
  connectPromise.then(() => clearTimeout(timeoutId), () => clearTimeout(timeoutId))
})

await Promise.race([connectPromise, timeoutPromise])

// services/mcp/client.ts:2264-2399
// 本地服务器（stdio/sdk）：低并发（默认 3），避免进程 spawn 资源争抢
const localServers = configEntries.filter(([_, config]) => isLocalMcpServer(config))
// 远程服务器：高并发（默认 20），只是网络连接
const remoteServers = configEntries.filter(([_, config]) => !isLocalMcpServer(config))

// 两组并行处理，各自有自己的并发上限
await Promise.all([
  processBatched(localServers, getMcpServerConnectionBatchSize(), processServer),    // 3
  processBatched(remoteServers, getRemoteMcpServerConnectionBatchSize(), processServer), // 20
])

// services/mcp/client.ts:2212-2224
// 2026-03 重构：之前的实现是固定大小的顺序批次
// （等 batch 1 全部完成，再启动 batch 2）。这意味着 batch N 中
// 一个慢服务器会阻塞 batch N+1 的所有服务器，即使其他 19 个
// 槽位是空闲的。pMap 在每个服务器完成时立即释放槽位，所以
// 一个慢服务器只占用一个槽位，而不会阻塞整个批次边界。
async function processBatched<T>(
  items: T[], concurrency: number, processor: (item: T) => Promise<void>,
): Promise<void> {
  await pMap(items, processor, { concurrency })
}

// services/mcp/client.ts:2300-2322
if (
  (config.type === 'claudeai-proxy' || config.type === 'http' || config.type === 'sse') &&
  ((await isMcpAuthCached(name)) ||
   ((config.type === 'http' || config.type === 'sse') &&
    hasMcpDiscoveryButNoToken(name, config)))
) {
  logMCPDebug(name, `Skipping connection (cached needs-auth)`)
  onConnectionAttempt({
    client: { name, type: 'needs-auth' as const, config },
    tools: [createMcpAuthTool(name, config)],  // 提供一个认证工具
    commands: [],
  })
  return
}

// services/mcp/client.ts:1743-1998（简化展示）
export const fetchToolsForClient = memoizeWithLRU(
  async (client: MCPServerConnection): Promise<Tool[]> => {
    if (client.type !== 'connected') return []
    if (!client.capabilities?.tools) return []

    const result = await client.client.request(
      { method: 'tools/list' }, ListToolsResultSchema,
    )

    return result.tools.map((tool): Tool => {
      const fullyQualifiedName = buildMcpToolName(client.name, tool.name)
      return {
        ...MCPTool,  // 继承 MCPTool 的基础实现
        name: fullyQualifiedName,  // mcp__serverName__toolName
        mcpInfo: { serverName: client.name, toolName: tool.name },
        isMcp: true,

        // 描述截断：防止 OpenAPI 生成的 MCP 服务器倾倒 15-60KB 的文档
        async prompt() {
          const desc = tool.description ?? ''
          return desc.length > MAX_MCP_DESCRIPTION_LENGTH  // 2048 字符
            ? desc.slice(0, MAX_MCP_DESCRIPTION_LENGTH) + '… [truncated]'
            : desc
        },

        // 利用 MCP 的 annotations 机制
        isConcurrencySafe() { return tool.annotations?.readOnlyHint ?? false },
        isReadOnly() { return tool.annotations?.readOnlyHint ?? false },
        isDestructive() { return tool.annotations?.destructiveHint ?? false },

        inputJSONSchema: tool.inputSchema as Tool['inputJSONSchema'],

        // 工具调用代理
        async call(args, context, _canUseTool, parentMessage, onProgress) {
          const connectedClient = await ensureConnectedClient(client)
          const mcpResult = await callMCPToolWithUrlElicitationRetry({
            client: connectedClient,
            tool: tool.name,
            args,
            signal: context.abortController.signal,
            // ...
          })
          return { data: mcpResult.content }
        },
      }
    })
  },
  (client: MCPServerConnection) => client.name,  // 缓存 key：服务器名称
  MCP_FETCH_CACHE_SIZE,  // LRU 缓存上限：20
)

// services/mcp/mcpStringUtils.ts:50-52
export function buildMcpToolName(serverName: string, toolName: string): string {
  return `${getMcpPrefix(serverName)}${normalizeNameForMCP(toolName)}`
}
// 示例：mcp__slack__send_message, mcp__github__list_issues

// services/mcp/client.ts:2054-2096
return promptsToProcess.map(prompt => {
  return {
    type: 'prompt' as const,
    name: 'mcp__' + normalizeNameForMCP(client.name) + '__' + prompt.name,
    description: prompt.description ?? '',
    isMcp: true,
    source: 'mcp',
    async getPromptForCommand(args: string) {
      const connectedClient = await ensureConnectedClient(client)
      const result = await connectedClient.client.getPrompt({
        name: prompt.name,
        arguments: zipObject(argNames, argsArray),
      })
      return result.messages.map(message =>
        transformResultContent(message.content, connectedClient.name),
      ).flat()
    },
  }
})

// services/mcp/client.ts:1249-1365（简化展示）
const isTerminalConnectionError = (msg: string): boolean => {
  return (
    msg.includes('ECONNRESET') ||
    msg.includes('ETIMEDOUT') ||
    msg.includes('EPIPE') ||
    msg.includes('EHOSTUNREACH') ||
    msg.includes('ECONNREFUSED') ||
    msg.includes('Body Timeout Error') ||
    msg.includes('terminated') ||
    msg.includes('SSE stream disconnected') ||
    msg.includes('Failed to reconnect SSE stream')
  )
}

client.onerror = (error: Error) => {
  // 1. HTTP Session 过期（404 + JSON-RPC -32001）→ 立即重连
  if (isMcpSessionExpiredError(error)) {
    closeTransportAndRejectPending('session expired')
    return
  }

  // 2. SDK SSE 重连耗尽 → 触发关闭
  if (error.message.includes('Maximum reconnection attempts')) {
    closeTransportAndRejectPending('SSE reconnection exhausted')
    return
  }

  // 3. 终端连接错误 → 计数，连续 3 次后触发重连
  if (isTerminalConnectionError(error.message)) {
    consecutiveConnectionErrors++
    if (consecutiveConnectionErrors >= MAX_ERRORS_BEFORE_RECONNECT) {
      closeTransportAndRejectPending('max consecutive terminal errors')
    }
  } else {
    consecutiveConnectionErrors = 0  // 非终端错误重置计数
  }
}

// services/mcp/client.ts:1374-1397
client.onclose = () => {
  // 清理连接缓存
  const key = getServerCacheKey(name, serverRef)
  connectToServer.cache.delete(key)

  // 同时清理 fetch 缓存——否则重连后会拿到旧工具/资源
  fetchToolsForClient.cache.delete(name)
  fetchResourcesForClient.cache.delete(name)
  fetchCommandsForClient.cache.delete(name)
}

SIGINT → (等 100ms) → SIGTERM → (等 400ms) → SIGKILL

// services/mcp/client.ts:1429-1562（简化）
// 1. 先发 SIGINT（像 Ctrl+C）
process.kill(childPid, 'SIGINT')
await sleep(100)
// 2. 如果还活着，发 SIGTERM
process.kill(childPid, 'SIGTERM')
await sleep(400)
// 3. 如果还活着，强制 SIGKILL
process.kill(childPid, 'SIGKILL')

// services/mcp/useManageMCPConnections.ts:87-90
const MAX_RECONNECT_ATTEMPTS = 5
const INITIAL_BACKOFF_MS = 1000
const MAX_BACKOFF_MS = 30000

// services/mcp/xaa.ts:1-17
// 1. RFC 8693 Token Exchange at the IdP: id_token → ID-JAG
// 2. RFC 7523 JWT Bearer Grant at the AS: ID-JAG → access_token

// services/mcp/headersHelper.ts:59-103
const execResult = await execFileNoThrowWithCwd(config.headersHelper, [], {
  shell: true,
  timeout: 10000,
  env: {
    ...process.env,
    CLAUDE_CODE_MCP_SERVER_NAME: serverName,
    CLAUDE_CODE_MCP_SERVER_URL: config.url,
  },
})
const headers = jsonParse(execResult.stdout.trim())

// tools/AgentTool/runAgent.ts:95-150（简化）
async function initializeAgentMcpServers(
  agentDefinition: AgentDefinition,
  parentClients: MCPServerConnection[],
) {
  if (!agentDefinition.mcpServers?.length) {
    return { clients: parentClients, tools: [], cleanup: async () => {} }
  }

  for (const spec of agentDefinition.mcpServers) {
    if (typeof spec === 'string') {
      // 引用现有配置：name → getMcpConfigByName(spec)
      config = getMcpConfigByName(spec)
    } else {
      // 内联定义：直接在 frontmatter 中提供完整配置
      // ...
    }
  }
}

// services/mcp/MCPConnectionManager.tsx:38-72
export function MCPConnectionManager({ children, dynamicMcpConfig, isStrictMcpConfig }) {
  const { reconnectMcpServer, toggleMcpServer } = useManageMCPConnections(
    dynamicMcpConfig, isStrictMcpConfig,
  )
  return (
    <MCPConnectionContext.Provider value={{ reconnectMcpServer, toggleMcpServer }}>
      {children}
    </MCPConnectionContext.Provider>
  )
}