MCP (Model Context Protocol) 实现原理学习指南#

// 源码位置: mcp/index.ts, line 408-449
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,
  },
})

const client = new Client({ name: "opencode", version: VERSION })
await client.connect(transport)

// 源码位置: mcp/index.ts, line 328-335
const transport = new StreamableHTTPClientTransport(new URL(mcp.url), {
  authProvider,      // OAuth 认证提供者
  requestInit: {
    headers: mcp.headers  // 自定义请求头
  },
})

// 源码位置: mcp/index.ts, line 336-342
const transport = new SSEClientTransport(new URL(mcp.url), {
  authProvider,
  requestInit: {
    headers: mcp.headers
  },
})

// 源码位置: mcp/index.ts, line 328-405
const transports = [
  { name: "StreamableHTTP", transport: new StreamableHTTPClientTransport(...) },
  { name: "SSE", transport: new SSEClientTransport(...) },
]

for (const { name, transport } of transports) {
  try {
    await client.connect(transport)
    status = { status: "connected" }
    break
  } catch (error) {
    // 尝试下一个传输方式
  }
}

{
  jsonrpc: "2.0",
  method: "tools/list",      // 方法名
  params: {                 // 参数
    cursor?: string
  },
  id: 1                     // 请求 ID
}

{
  jsonrpc: "2.0",
  result: {                 // 结果
    tools: [
      {
        name: "tool_name",
        description: "工具描述",
        inputSchema: {
          type: "object",
          properties: {...}
        }
      }
    ]
  },
  id: 1
}

{
  jsonrpc: "2.0",
  method: "notifications/tools/list_changed",  // 通知方法
  params: {}
}

// 源码位置: mcp/index.ts, line 349-355
const client = new Client({
  name: "opencode",
  version: Installation.VERSION,
})
await withTimeout(client.connect(transport), connectTimeout)
registerNotificationHandlers(client, serverName)

// SDK 自动处理 initialize 握手

// 源码位置: mcp/index.ts, line 120-148
async function convertMcpTool(mcpTool: MCPToolDef, client: MCPClient, timeout?: number): Promise<Tool> {
  const inputSchema = mcpTool.inputSchema

  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,
        },
      )
    },
  })
}

// 源码位置: mcp/index.ts, line 566-602
export async function tools() {
  const result: Record<string, Tool> = {}
  const s = await state()
  const clientsSnapshot = await clients()

  for (const [clientName, client] of Object.entries(clientsSnapshot)) {
    // 只收集已连接的 MCP 服务器的工具
    if (s.status[clientName]?.status !== "connected") {
      continue
    }

    const toolsResult = await client.listTools()
    for (const mcpTool of toolsResult.tools) {
      const sanitizedClientName = clientName.replace(/[^a-zA-Z0-9_-]/g, "_")
      const sanitizedToolName = mcpTool.name.replace(/[^a-zA-Z0-9_-]/g, "_")
      result[sanitizedClientName + "_" + sanitizedToolName] = await convertMcpTool(...)
    }
  }
  return result
}

// 工具名称格式: {客户端名}_{工具名}
// 例如: filesystem_read_file, filesystem_write_file
// 特殊字符被替换为下划线
const sanitizedClientName = clientName.replace(/[^a-zA-Z0-9_-]/g, "_")
const sanitizedToolName = mcpTool.name.replace(/[^a-zA-Z0-9_-]/g, "_")
const key = sanitizedClientName + "_" + sanitizedToolName

// 步骤 1: 从内置 ToolRegistry 获取工具
for (const item of await ToolRegistry.tools(
  { modelID: input.model.api.id, providerID: input.model.providerID },
  input.agent,
)) {
  // 转换为 AI SDK 格式
  tools[item.id] = tool({
    id: item.id,
    description: item.description,
    inputSchema: jsonSchema(schema),
    execute: async (args, options) => {
      // ...执行逻辑
    },
  })
}

// 步骤 2: 从 MCP 获取工具 (第 786 行)
for (const [key, item] of Object.entries(await MCP.tools())) {
  // key = "filesystem_read_file" 格式
  // item = AI SDK Tool 对象

  // 直接使用 already-converted tool
  tools[key] = item  // ← 已经是 AI SDK Tool 格式！
}

// 合并后的 tools 对象
const tools = {
  // 内置工具
  "bash": Tool,
  "write": Tool,

  // MCP 工具 (已经转换好)
  "filesystem_read_file": Tool,
  "filesystem_write_file": Tool,
  "github_get_issue": Tool,
}

// 传给 AI SDK
const result = await generateText({
  model,
  tools,        // ← 这里！
  toolChoice: "auto",
  messages,
})

// 源码位置: mcp/oauth-provider.ts
export class McpOAuthProvider implements OAuthClientProvider {
  constructor(
    private mcpName: string,
    private serverUrl: string,
    private config: McpOAuthConfig,
    private callbacks: McpOAuthCallbacks,
  ) {}

  // 获取重定向 URL
  get redirectUrl(): string {
    return `http://127.0.0.1:${OAUTH_CALLBACK_PORT}${OAUTH_CALLBACK_PATH}`
  }

  // 客户端元数据
  get clientMetadata(): OAuthClientMetadata {
    return {
      redirect_uris: [this.redirectUrl],
      client_name: "OpenCode",
      client_uri: "https://opencode.ai",
      grant_types: ["authorization_code", "refresh_token"],
      response_types: ["code"],
      token_endpoint_auth_method: this.config.clientSecret ? "client_secret_post" : "none",
    }
  }

  // 获取已注册的客户端信息
  async clientInformation(): Promise<OAuthClientInformation | undefined> {
    // 优先使用配置中的 clientId
    if (this.config.clientId) {
      return { client_id: this.config.clientId, client_secret: this.config.clientSecret }
    }

    // 从存储中获取动态注册的客户端信息
    const entry = await McpAuth.getForUrl(this.mcpName, this.serverUrl)
    if (entry?.clientInfo) {
      // 检查客户端密钥是否过期
      if (entry.clientInfo.clientSecretExpiresAt < Date.now() / 1000) {
        return undefined  // 需要重新注册
      }
      return { client_id: entry.clientInfo.clientId, client_secret: entry.clientInfo.clientSecret }
    }
    return undefined  // 需要动态注册
  }

  // 保存动态注册的客户端信息
  async saveClientInformation(info: OAuthClientInformationFull): Promise<void> {
    await McpAuth.updateClientInfo(this.mcpName, {
      clientId: info.client_id,
      clientSecret: info.client_secret,
      clientIdIssuedAt: info.client_id_issued_at,
      clientSecretExpiresAt: info.client_secret_expires_at,
    }, this.serverUrl)
  }

  // 获取存储的令牌
  async tokens(): Promise<OAuthTokens | undefined> {
    const entry = await McpAuth.getForUrl(this.mcpName, this.serverUrl)
    if (!entry?.tokens) return undefined

    return {
      access_token: entry.tokens.accessToken,
      token_type: "Bearer",
      refresh_token: entry.tokens.refreshToken,
      expires_in: entry.tokens.expiresAt
        ? Math.max(0, Math.floor(entry.tokens.expiresAt - Date.now() / 1000))
        : undefined,
      scope: entry.tokens.scope,
    }
  }

  // 保存令牌
  async saveTokens(tokens: OAuthTokens): Promise<void> {
    await McpAuth.updateTokens(this.mcpName, {
      accessToken: tokens.access_token,
      refreshToken: tokens.refresh_token,
      expiresAt: tokens.expires_in ? Date.now() / 1000 + tokens.expires_in : undefined,
      scope: tokens.scope,
    }, this.serverUrl)
  }
}

// 源码位置: mcp/oauth-callback.ts
export namespace McpOAuthCallback {
  let server: ReturnType<typeof Bun.serve> | undefined
  const pendingAuths = new Map<string, PendingAuth>()

  export async function ensureRunning(): Promise<void> {
    if (server) return

    server = Bun.serve({
      port: OAUTH_CALLBACK_PORT,
      fetch(req) {
        const url = new URL(req.url)
        const code = url.searchParams.get("code")
        const state = url.searchParams.get("state")
        const error = url.searchParams.get("error")

        // 验证 state 参数防止 CSRF
        if (!state || !pendingAuths.has(state)) {
          return new Response(HTML_ERROR("Invalid state"), { status: 400 })
        }

        const pending = pendingAuths.get(state)!
        clearTimeout(pending.timeout)
        pendingAuths.delete(state)
        pending.resolve(code)

        return new Response(HTML_SUCCESS, { headers: { "Content-Type": "text/html" } })
      },
    })
  }

  export function waitForCallback(oauthState: string): Promise<string> {
    return new Promise((resolve, reject) => {
      const timeout = setTimeout(() => {
        pendingAuths.delete(oauthState)
        reject(new Error("OAuth callback timeout"))
      }, CALLBACK_TIMEOUT_MS)

      pendingAuths.set(oauthState, { resolve, reject, timeout })
    })
  }
}

// 源码位置: mcp/index.ts

// 1. 开始认证
export async function startAuth(mcpName: string): Promise<{ authorizationUrl: string }> {
  // 生成随机的 state 参数
  const oauthState = Array.from(crypto.getRandomValues(new Uint8Array(32)))
    .map((b) => b.toString(16).padStart(2, "0"))
    .join("")
  await McpAuth.updateOAuthState(mcpName, oauthState)

  const authProvider = new McpOAuthProvider(...)

  try {
    await client.connect(transport)
    // 已认证
    return { authorizationUrl: "" }
  } catch (error) {
    if (error instanceof UnauthorizedError && capturedUrl) {
      return { authorizationUrl: capturedUrl.toString() }
    }
    throw error
  }
}

// 2. 完成认证
export async function authenticate(mcpName: string): Promise<Status> {
  const { authorizationUrl } = await startAuth(mcpName)

  // 打开浏览器
  await open(authorizationUrl)

  // 等待回调
  const code = await McpOAuthCallback.waitForCallback(oauthState)

  // 验证 state
  const storedState = await McpAuth.getOAuthState(mcpName)
  if (storedState !== oauthState) {
    throw new Error("OAuth state mismatch - potential CSRF attack")
  }

  return finishAuth(mcpName, code)
}

// 3. 完成认证
export async function finishAuth(mcpName: string, authorizationCode: string): Promise<Status> {
  const transport = pendingOAuthTransports.get(mcpName)
  await transport.finishAuth(authorizationCode)
  await McpAuth.clearCodeVerifier(mcpName)

  // 重新连接
  return add(mcpName, mcpConfig)
}

// 源码位置: config/config.ts, line 510-571

// 本地 MCP 服务器配置
export const McpLocal = z.object({
  type: z.literal("local"),
  command: z.string().array(),           // 命令和参数
  environment: z.record(z.string(), z.string()).optional(),  // 环境变量
  enabled: z.boolean().optional(),        // 是否启用
  timeout: z.number().int().positive().optional(),  // 超时时间(ms)
})

// OAuth 配置
export const McpOAuth = z.object({
  clientId: z.string().optional(),        // OAuth 客户端 ID
  clientSecret: z.string().optional(),    // OAuth 客户端密钥
  scope: z.string().optional(),           // OAuth 作用域
})

// 远程 MCP 服务器配置
export const McpRemote = z.object({
  type: z.literal("remote"),
  url: z.string(),                        // 服务器 URL
  enabled: z.boolean().optional(),
  headers: z.record(z.string(), z.string()).optional(),  // 自定义请求头
  oauth: z.union([McpOAuth, z.literal(false)]).optional(),  // OAuth 配置
  timeout: z.number().int().positive().optional(),
})

// 统一配置
export const Mcp = z.discriminatedUnion("type", [McpLocal, McpRemote])

// 源码位置: mcp/index.ts, line 66-109
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(),                    // 需要客户端注册
  }),
])

// 源码位置: mcp/index.ts, line 163-210
const state = Instance.state(
  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)) return

        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 }
  },
  async (state) => {
    // 清理：关闭所有客户端连接
    await Promise.all(
      Object.values(state.clients).map((client) => client.close())
    )
  },
)

// 源码位置: mcp/market.ts
const registryURL = "https://registry.modelcontextprotocol.io"

// 获取服务器列表
export async function list(input?: ListInput) {
  return ServerListResponse.parse(
    await request("/v0/servers", {
      search: query.search,
      cursor: query.cursor,
      limit: query.limit ?? 24,
      version: "latest",
    })
  )
}

// 获取服务器详情
export async function get(id: string) {
  return ServerResponse.parse(
    await request(`/v0/servers/${encodeURIComponent(id)}/versions/latest`)
  )
}

// 安装服务器
export async function install(id: string) {
  const detail = await get(id)
  await Config.updateGlobal({
    mcp: {
      ...(current.mcp ?? {}),
      [id]: detail.config,
    },
  })
}

// 卸载服务器
export async function uninstall(id: string) {
  const next = { ...(current.mcp ?? {}) }
  delete next[id]
  await Config.updateGlobal({ mcp: next })
}

// 源码位置: mcp/market.ts

// 本地服务器配置解析
function local(pkg: Package): Config.Mcp | undefined {
  const cmd = binary(pkg)  // npx, uvx, docker 等
  const runtime = pkg.registryType === "npm" && cmd === "npx" ? [cmd, "-y"] : [cmd]
  const base = pkg.registryType === "oci" && cmd === "docker"
    ? [...runtime, "run", "-i", "--rm"]
    : runtime

  return Config.Mcp.parse({
    type: "local",
    command: [...base, ...args(pkg.runtimeArguments), identifier(pkg), ...args(pkg.packageArguments)],
    environment: record(pkg.environmentVariables),
  })
}

// 远程服务器配置解析
function remote(transport: Transport): Config.Mcp | undefined {
  if (!transport.url) return
  return Config.Mcp.parse({
    type: "remote",
    url: fill(transport.url, transport.variables),
    headers: record(transport.headers),
  })
}

import { Client } from "@modelcontextprotocol/sdk/client/index.js"
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"

// 创建客户端
const client = new Client({
  name: "my-mcp-client",
  version: "1.0.0",
})

// 创建传输层
const transport = new StreamableHTTPClientTransport(
  new URL("http://localhost:3000/mcp")
)

// 连接并获取工具
await client.connect(transport)
const { tools } = await client.listTools()

// 调用工具
const result = await client.callTool({
  name: "my_tool",
  arguments: { arg1: "value1" }
})

console.log(result)

import { Server } from "@modelcontextprotocol/sdk/server/index.js"
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js"

// 创建服务器
const server = new Server(
  { name: "my-mcp-server", version: "1.0.0" },
  { capabilities: { tools: {} } }
)

// 注册工具
server.setRequestHandler(ListToolsRequestSchema, async () => {
  return {
    tools: [
      {
        name: "my_tool",
        description: "A sample tool",
        inputSchema: {
          type: "object",
          properties: {
            message: { type: "string", description: "Message to echo" }
          }
        }
      }
    ]
  }
})

// 处理工具调用
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params

  if (name === "my_tool") {
    return { content: [{ type: "text", text: `Hello, ${args.message}!` }] }
  }

  throw new Error(`Unknown tool: ${name}`)
})

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

import { Client } from "@modelcontextprotocol/sdk/client/index.js"
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"
import { OAuthClientProvider } from "@modelcontextprotocol/sdk/client/auth.js"

// 自定义 OAuth Provider
class MyOAuthProvider implements OAuthClientProvider {
  async clientInformation() {
    return { client_id: "my-client-id", client_secret: "my-client-secret" }
  }

  async tokens() {
    return { access_token: "stored-token", token_type: "Bearer" }
  }

  async saveTokens(tokens) {
    console.log("保存 tokens:", tokens)
  }

  async redirectToAuthorization(url) {
    console.log("打开浏览器:", url)
  }
}

// 创建带认证的客户端
const transport = new StreamableHTTPClientTransport(
  new URL("http://localhost:3000/mcp"),
  { authProvider: new MyOAuthProvider() }
)

const client = new Client({ name: "my-client", version: "1.0.0" })
await client.connect(transport)

import { dynamicTool } from "ai"
import type { Tool } from "ai"

// MCP 工具转 AI SDK 工具
function convertMcpTool(mcpTool: MCPToolDef, client: MCPClient): Tool {
  return dynamicTool({
    description: mcpTool.description ?? "",
    inputSchema: jsonSchema(mcpTool.inputSchema),
    execute: async (args) => {
      const result = await client.callTool({
        name: mcpTool.name,
        arguments: args
      })
      return result
    }
  })
}

// 收集所有 MCP 工具
async function getAllTools(clients: MCPClient[]): Promise<Record<string, Tool>> {
  const tools: Record<string, Tool> = {}

  for (const client of clients) {
    const { tools: mcpTools } = await client.listTools()
    for (const tool of mcpTools) {
      tools[`${client.name}_${tool.name}`] = convertMcpTool(tool, client)
    }
  }

  return tools
}