8.2 MCP Client 实现

模型: claude-opus-4-6 (anthropic/claude-opus-4-6) 生成日期: 2026-02-17

---

在上一节中，我们了解了 MCP 协议的核心概念和三种传输方式。本节将深入 OpenCode 的 mcp/index.ts 源码，详细分析 MCP Client 的实现——从客户端初始化、连接管理、状态机设计，到 MCP 工具如何被转换为 OpenCode 内部的 Tool 格式。

8.2.1 mcp/index.ts 源码解析

mcp/index.ts 是 OpenCode MCP 模块的核心文件，共 935 行代码。它以一个 MCP 命名空间（namespace）的形式组织所有功能。让我们从依赖导入开始理解其架构。

依赖关系

// AI SDK - 用于将 MCP 工具转换为 AI SDK 格式
import { dynamicTool, type Tool, jsonSchema, type JSONSchema7 } from "ai"

// MCP SDK - 官方的 MCP 客户端库
import { Client } from "@modelcontextprotocol/sdk/client/index.js"
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"
import { SSEClientTransport } from "@modelcontextprotocol/sdk/client/sse.js"
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js"
import { UnauthorizedError } from "@modelcontextprotocol/sdk/client/auth.js"
import {
  CallToolResultSchema,
  type Tool as MCPToolDef,
  ToolListChangedNotificationSchema,
} from "@modelcontextprotocol/sdk/types.js"

这里的依赖可以分为两组：

1. @modelcontextprotocol/sdk：MCP 官方 SDK，提供了 Client 类和三种传输实现（StdioClientTransport、SSEClientTransport、StreamableHTTPClientTransport）。这是与 MCP Server 通信的基础设施。

2. ai（Vercel AI SDK）：提供了 dynamicTool() 函数，用于将 MCP 工具转换为 AI SDK 的 Tool 格式。这是连接 MCP 世界与 LLM 世界的桥梁。

Instance.state() 状态管理

MCP 模块使用了我们在第 3 章介绍过的 Instance.state() 模式来管理状态：

const state = Instance.state(
  // 初始化函数：启动所有配置的 MCP Server
  async () => {
    const cfg = await Config.get()
    const config = cfg.mcp ?? {}
    const clients: Record<string, MCPClient> = {}
    const status: Record<string, Status> = {}

    await Promise.all(
      Object.entries(config).map(async ([key, mcp]) => {
        if (!isMcpConfigured(mcp)) {
          log.error("Ignoring MCP config entry without type", { key })
          return
        }

        // 被禁用的 MCP Server 直接标记为 disabled
        if (mcp.enabled === false) {
          status[key] = { status: "disabled" }
          return
        }

        const result = await create(key, mcp).catch(() => undefined)
        if (!result) return

        status[key] = result.status
        if (result.mcpClient) {
          clients[key] = result.mcpClient
        }
      }),
    )
    return { status, clients }
  },

  // 销毁函数：关闭所有 MCP 客户端连接
  async (state) => {
    await Promise.all(
      Object.values(state.clients).map((client) =>
        client.close().catch((error) => {
          log.error("Failed to close MCP client", { error })
        }),
      ),
    )
    pendingOAuthTransports.clear()
  },
)

回顾 Instance.state()

在第 3 章中我们学习过，Instance.state() 是 OpenCode 的"按项目目录隔离"状态管理模式。每个项目目录都有自己独立的一组 MCP 连接。当 OpenCode 启动时，state() 的初始化函数被调用，遍历配置中的所有 MCP Server 并尝试建立连接。当项目目录切换或 OpenCode 退出时，销毁函数被调用，优雅地关闭所有连接。

这里有一个重要的工程细节——所有 MCP Server 的连接是 并行建立 的（Promise.all），这确保了即使某个 MCP Server 启动较慢或超时，也不会阻塞其他 Server 的连接。

create() 函数：连接核心逻辑

create() 是整个 MCP 模块最核心的函数，它负责建立与单个 MCP Server 的连接。根据 type 字段的不同，连接流程分为两条路径：

远程 MCP Server（type: "remote"）：

async function create(key: string, mcp: Config.Mcp) {
  if (mcp.type === "remote") {
    // 1. 创建 OAuth Provider（除非明确禁用）
    let authProvider: McpOAuthProvider | undefined
    if (!oauthDisabled) {
      authProvider = new McpOAuthProvider(key, mcp.url, { ... }, { ... })
    }

    // 2. 构建传输层列表：StreamableHTTP 优先，SSE 备选
    const transports = [
      { name: "StreamableHTTP", transport: new StreamableHTTPClientTransport(...) },
      { name: "SSE",            transport: new SSEClientTransport(...) },
    ]

    // 3. 依次尝试连接
    for (const { name, transport } of transports) {
      try {
        const client = new Client({
          name: "opencode",
          version: Installation.VERSION,
        })
        await withTimeout(client.connect(transport), connectTimeout)
        registerNotificationHandlers(client, key)
        mcpClient = client
        status = { status: "connected" }
        break  // 连接成功，跳出循环
      } catch (error) {
        // 处理 OAuth 错误
        if (error instanceof UnauthorizedError) {
          // 存储 transport 以便后续完成 OAuth 流程
          pendingOAuthTransports.set(key, transport)
          status = { status: "needs_auth" }
          break
        }
        // 传输失败，尝试下一个
        status = { status: "failed", error: lastError.message }
      }
    }
  }
  // ...
}

这段代码展示了一个精巧的 降级连接策略：

1. 首先尝试更现代的 StreamableHTTP 传输。

2. 如果失败，自动回退到 SSE 传输。

3. 如果遇到 UnauthorizedError（OAuth 认证错误），不立即失败，而是将连接标记为 needs_auth，等待用户完成认证后再重试。

本地 MCP Server（type: "local"）：

if (mcp.type === "local") {
  const [cmd, ...args] = mcp.command  // 解构命令和参数
  const cwd = Instance.directory       // 使用当前项目目录作为工作目录
  const transport = new StdioClientTransport({
    stderr: "pipe",
    command: cmd,
    args,
    cwd,
    env: {
      ...process.env,
      ...(cmd === "opencode" ? { BUN_BE_BUN: "1" } : {}),
      ...mcp.environment,
    },
  })

  // 监听 stderr 用于日志
  transport.stderr?.on("data", (chunk: Buffer) => {
    log.info(`mcp stderr: ${chunk.toString()}`, { key })
  })

  // 带超时的连接
  const client = new Client({ name: "opencode", version: Installation.VERSION })
  await withTimeout(client.connect(transport), connectTimeout)
  registerNotificationHandlers(client, key)
}

本地连接相对简单——通过 StdioClientTransport 启动子进程，子进程的 stdin/stdout 作为通信通道。注意一个有趣的细节：当命令是 opencode 自身时（递归调用场景），会设置环境变量 BUN_BE_BUN: "1"，这是 Bun 运行时的一个特殊标志。

连接建立后，无论是远程还是本地，都会立即获取工具列表来验证连接是否真正可用：

// 连接成功后，立即获取工具列表来验证
const result = await withTimeout(
  mcpClient.listTools(),
  mcp.timeout ?? DEFAULT_TIMEOUT
).catch((err) => {
  log.error("failed to get tools from client", { key, error: err })
  return undefined
})

if (!result) {
  // 如果获取工具失败，关闭连接并标记为 failed
  await mcpClient.close().catch(...)
  status = { status: "failed", error: "Failed to get tools" }
}

这是一个很好的防御性编程实践——不仅要连接成功，还要确保 Server 能正常响应请求。

8.2.2 MCP Server 连接管理

连接状态机

OpenCode 使用一个 Status 类型来跟踪每个 MCP Server 的连接状态。这个类型通过 Zod 的 discriminatedUnion 定义，确保了类型安全：

export const Status = z.discriminatedUnion("status", [
  z.object({ status: z.literal("connected") }),
  z.object({ status: z.literal("disabled") }),
  z.object({ status: z.literal("failed"), error: z.string() }),
  z.object({ status: z.literal("needs_auth") }),
  z.object({
    status: z.literal("needs_client_registration"),
    error: z.string(),
  }),
])

衍生解释：Discriminated Union（可辨识联合）

在 TypeScript 中，Discriminated Union 是一种常用的类型设计模式。联合类型中的每个变体都有一个共同的"判别字段"（在这里是 status），它的值是不同的字面量类型。这使得 TypeScript 编译器可以在 switch 或 if 语句中自动收窄类型——例如当你检查 status === "failed" 时，TypeScript 会自动推断出 error 字段的存在。

Zod 的 z.discriminatedUnion() 是这一模式的运行时验证版本，它在解析 JSON 数据时也能利用判别字段来选择正确的验证分支。

五种状态的含义和转换关系：

               ┌────────────┐
    配置禁用 → │  disabled   │
               └────────────┘
                     
               ┌────────────┐     连接成功     ┌───────────┐
    初始化  → │ (connecting) │ ──────────────► │ connected  │
               └────────────┘                  └───────────┘
                     │                               │
                     │ 连接失败                       │ 运行时错误
                     ▼                               ▼
               ┌────────────┐                  ┌───────────┐
               │   failed    │ ◄───────────── │  failed    │
               └────────────┘                  └───────────┘
                     │
                     │ OAuth 需要认证
                     ▼
               ┌────────────┐     认证完成     ┌───────────┐
               │ needs_auth  │ ──────────────► │ connected  │
               └────────────┘                  └───────────┘
                     │
                     │ 需要客户端注册
                     ▼
               ┌─────────────────────────┐
               │ needs_client_registration│
               └─────────────────────────┘

值得注意的是，needs_auth 和 needs_client_registration 这两个状态专门为 OAuth 认证场景设计。当远程 MCP Server 要求认证时，OpenCode 不会直接报错，而是将状态标记为"需要认证"，并通过 Toast 通知用户：

if (error instanceof UnauthorizedError) {
  if (lastError.message.includes("registration") ||
      lastError.message.includes("client_id")) {
    // 服务器不支持动态客户端注册
    status = {
      status: "needs_client_registration",
      error: "Server does not support dynamic client registration...",
    }
    Bus.publish(TuiEvent.ToastShow, {
      title: "MCP Authentication Required",
      message: `Server "${key}" requires a pre-registered client ID...`,
      variant: "warning",
      duration: 8000,
    })
  } else {
    // 需要 OAuth 认证
    pendingOAuthTransports.set(key, transport)
    status = { status: "needs_auth" }
    Bus.publish(TuiEvent.ToastShow, {
      title: "MCP Authentication Required",
      message: `Server "${key}" requires authentication. Run: opencode mcp auth ${key}`,
      variant: "warning",
      duration: 8000,
    })
  }
}

ToolsChanged 事件监听

MCP 协议支持 Server 端主动通知 Client 端工具列表发生了变化。OpenCode 通过注册通知处理器来监听这一事件：

function registerNotificationHandlers(client: MCPClient, serverName: string) {
  client.setNotificationHandler(ToolListChangedNotificationSchema, async () => {
    log.info("tools list changed notification received", { server: serverName })
    Bus.publish(ToolsChanged, { server: serverName })
  })
}

当收到 ToolListChangedNotification 时，OpenCode 通过 Bus（事件总线，详见第 11 章）发布一个 MCP.ToolsChanged 事件。这个事件会被上层模块（如 Session 系统）监听，触发工具列表的重新获取。

这意味着 MCP Server 可以在运行时动态添加或移除工具，而 OpenCode 能够实时感知这些变化——这是 MCP 协议比静态工具定义更加灵活的一个体现。

连接管理 API

MCP 模块暴露了一组连接管理函数，支持运行时的动态连接和断开：

// 添加/重新连接一个 MCP Server
export async function add(name: string, mcp: Config.Mcp) { ... }

// 连接一个已配置但未连接的 MCP Server
export async function connect(name: string) { ... }

// 断开一个 MCP Server
export async function disconnect(name: string) { ... }

// 获取所有 MCP Server 的状态
export async function status() { ... }

connect() 函数在实现上有一个细节值得注意——它会强制将 enabled 设置为 true：

export async function connect(name: string) {
  const cfg = await Config.get()
  const mcp = config[name]
  // ...
  const result = await create(name, { ...mcp, enabled: true })  // 强制启用
  // ...
}

而 add() 函数在添加新连接前会检查是否已有同名连接，如果有则先关闭旧连接以防止内存泄漏：

export async function add(name: string, mcp: Config.Mcp) {
  const s = await state()
  const result = await create(name, mcp)
  // ...
  // 关闭已存在的同名客户端以防止内存泄漏
  const existingClient = s.clients[name]
  if (existingClient) {
    await existingClient.close().catch(...)
  }
  s.clients[name] = result.mcpClient
  s.status[name] = result.status
}

超时处理

MCP 连接和工具调用都使用了统一的超时机制：

const DEFAULT_TIMEOUT = 30_000  // 默认 30 秒

// 连接超时
const connectTimeout = mcp.timeout ?? DEFAULT_TIMEOUT
await withTimeout(client.connect(transport), connectTimeout)

// 工具列表获取超时
await withTimeout(mcpClient.listTools(), mcp.timeout ?? DEFAULT_TIMEOUT)

withTimeout() 是 OpenCode 的一个工具函数（位于 util/timeout.ts），它包装一个 Promise，如果在指定时间内未完成则抛出超时错误。用户可以通过配置中的 timeout 字段为每个 MCP Server 单独设置超时时间，也可以使用全局的实验性配置 experimental.mcp_timeout。

8.2.3 MCP 工具转换为内部 Tool 格式

MCP 系统中最关键的环节之一是将 MCP Server 提供的工具定义转换为 AI SDK 的 Tool 类型，使其能被 LLM 调用。这个转换由 convertMcpTool() 函数完成：

async function convertMcpTool(
  mcpTool: MCPToolDef,
  client: MCPClient,
  timeout?: number
): Promise<Tool> {
  const inputSchema = mcpTool.inputSchema

  // 构建 JSON Schema，确保类型为 "object"
  const schema: JSONSchema7 = {
    ...(inputSchema as JSONSchema7),
    type: "object",
    properties: (inputSchema.properties ?? {}) as JSONSchema7["properties"],
    additionalProperties: false,
  }

  return dynamicTool({
    description: mcpTool.description ?? "",
    inputSchema: jsonSchema(schema),
    execute: async (args: unknown) => {
      return client.callTool(
        {
          name: mcpTool.name,
          arguments: (args || {}) as Record<string, unknown>,
        },
        CallToolResultSchema,
        {
          resetTimeoutOnProgress: true,
          timeout,
        },
      )
    },
  })
}

这个函数的核心逻辑可以分解为三步：

1. Schema 标准化：MCP Server 提供的 inputSchema 可能不完全符合 JSON Schema 的规范（例如可能缺少 type 或 properties 字段）。函数通过展开原始 Schema 并强制设置 type: "object" 和 additionalProperties: false 来确保兼容性。

2. 创建 dynamicTool()：使用 AI SDK 的 dynamicTool() 工厂函数创建一个动态工具。dynamicTool() 与我们在第 5 章学习的 OpenCode 内置 Tool.define() 类似，但它来自 AI SDK，面向的是 LLM 的工具调用场景。

3. 封装执行逻辑：execute 函数通过 client.callTool() 将工具调用请求转发到 MCP Server。注意 resetTimeoutOnProgress: true 选项——这意味着如果 MCP Server 发送了进度通知，超时计时器会被重置，避免长时间运行的工具因超时而被错误中断。

工具名称的命名约定

在将 MCP 工具暴露给 LLM 时，OpenCode 使用了一个特定的命名约定来避免不同 MCP Server 之间的工具名称冲突：

export async function tools() {
  const result: Record<string, Tool> = {}
  // ...
  for (const mcpTool of toolsResult.tools) {
    // 清理 Server 名称和工具名称中的特殊字符
    const sanitizedClientName = clientName.replace(/[^a-zA-Z0-9_-]/g, "_")
    const sanitizedToolName = mcpTool.name.replace(/[^a-zA-Z0-9_-]/g, "_")
    // 拼接为 "ServerName_ToolName" 的格式
    result[sanitizedClientName + "_" + sanitizedToolName] = await convertMcpTool(...)
  }
  return result
}

例如，如果一个名为 my-jira 的 MCP Server 暴露了一个名为 search_issues 的工具，那么在 OpenCode 中这个工具的名称将是 my-jira_search_issues。这种前缀命名避免了当多个 MCP Server 暴露同名工具时的冲突问题。

Prompt 和 Resource 的获取

除了 Tool 之外，MCP 模块还提供了获取 Prompt 和 Resource 的函数：

// 获取所有已连接 MCP Server 的 Prompt 列表
export async function prompts() {
  const clientsSnapshot = await clients()
  // 并行获取所有 Server 的 Prompt
  const prompts = Object.fromEntries(
    (await Promise.all(
      Object.entries(clientsSnapshot).map(async ([clientName, client]) => {
        return Object.entries(
          (await fetchPromptsForClient(clientName, client)) ?? {}
        )
      }),
    )).flat(),
  )
  return prompts
}

// 获取特定 MCP Server 的特定 Prompt 内容
export async function getPrompt(
  clientName: string,
  name: string,
  args?: Record<string, string>
) {
  const client = clientsSnapshot[clientName]
  return client.getPrompt({ name, arguments: args })
}

// 读取特定 MCP Server 的特定 Resource 内容
export async function readResource(clientName: string, resourceUri: string) {
  const client = clientsSnapshot[clientName]
  return client.readResource({ uri: resourceUri })
}

Prompt 和 Resource 的命名也使用了类似的前缀策略：ServerName:PromptName / ServerName:ResourceName，使用冒号而非下划线分隔（与 Tool 的命名约定略有不同，因为它们不需要传递给 LLM 的函数调用接口）。

---

本节小结

OpenCode 的 MCP Client 实现建立在 @modelcontextprotocol/sdk 之上，通过 Instance.state() 管理连接生命周期。核心的 create() 函数根据配置类型（local/remote）选择相应的传输方式，远程连接采用 StreamableHTTP → SSE 的降级策略。连接状态通过五态状态机跟踪，特别为 OAuth 认证场景设计了 needs_auth 和 needs_client_registration 两个中间状态。convertMcpTool() 函数将 MCP Tool 转换为 AI SDK 的 dynamicTool()，使 MCP 工具能够无缝融入 LLM 的工具调用体系。所有工具均使用"ServerName_ToolName"的前缀命名约定来避免冲突。