MCP 协议从零实现：构建你的第一个 AI 工具服务器

2026 年，AI Agent 已经从概念验证走向了生产部署。但一个关键问题浮出水面：如何让 AI 模型安全、规范地调用外部工具？ Anthropic 在 2024 年底提出的 Model Context Protocol（MCP）正在成为事实标准——Claude、GPT、Gemini 等主流模型均已支持，GitHub 上 MCP Server 仓库数量已超过 12,000 个。然而，大多数开发者只是「使用者」，对 MCP 协议的内部机制一无所知。本文将带你从零实现一个完整的 MCP Server，彻底搞懂协议细节，让你不仅能用，还能造。

🔧 一、MCP 协议核心概念

1.1 什么是 MCP

MCP（Model Context Protocol）是一个开放协议，标准化了 AI 模型与外部数据源、工具之间的通信方式。你可以把它理解为 AI 世界的 USB-C 接口——不管你是数据库、API 还是本地文件系统，只要实现了 MCP 协议，任何 AI 模型都能即插即用。

MCP 的架构采用客户端-服务器模型，这个设计借鉴了语言服务器协议（LSP）的成功经验——正如 LSP 统一了编辑器与编程语言的通信，MCP 统一了 AI 模型与外部工具的通信：

• MCP Host（宿主）：发起连接的 AI 应用程序，负责管理客户端实例和用户授权。典型代表有 Claude Desktop、Cursor IDE、Continue 等。Host 控制整个连接的生命周期。
• MCP Client（客户端）：Host 内部的协议客户端实例，维护与 Server 的一对一有状态连接。一个 Host 可以同时运行多个 Client，连接不同的 Server。
• MCP Server（服务端）：暴露具体能力的服务进程。Server 可以提供三类原语：工具（Tools，模型可调用的函数）、资源（Resources，模型可读取的数据）、提示模板（Prompts，预定义的交互模板）。

💡 **提示：**MCP 不是另一个 REST API 规范。它基于 JSON-RPC 2.0，支持有状态的双向通信，这与传统的请求-响应模式有本质区别。

1.2 传输层与消息格式

MCP 使用 JSON-RPC 2.0 作为消息格式，定义了两种标准传输方式。所有消息必须遵循 JSON-RPC 规范：请求包含 id、method、params 字段，响应包含 id 和 result 或 error，通知则没有 id 字段。理解这个消息格式是实现 MCP Server 的基础。

stdio 模式是最常用的——Host 启动一个子进程，通过 stdin/stdout 逐行交换 JSON-RPC 消息（每条消息以换行符 \n 分隔）。这种模式有三个显著优势：零网络配置、天然安全（只有本地进程能访问）、无跨域问题。几乎所有本地 MCP Server 都使用 stdio 模式。

Streamable HTTP 模式用于远程场景。客户端通过 HTTP POST 发送请求，服务端可以返回普通 JSON 响应或升级为 SSE 流。2026 年 3 月的 MCP 规范更新中，Streamable HTTP 取代了旧的纯 SSE 方案，最大的改进是支持无状态部署——服务端不需要维护长连接，可以水平扩展到多个实例。

1.3 能力协商与握手流程

MCP 连接的建立遵循严格的握手流程，核心是「能力协商」——客户端和服务端在初始化阶段互相声明自己支持的功能，后续通信只使用双方都协商确认的能力：

Client → Server: initialize { protocolVersion, capabilities, clientInfo }
Server → Client: initialize_result { capabilities, serverInfo }
Client → Server: initialized (通知，确认握手完成)
// 正式开始通信，以下调用才会被服务端接受
Client → Server: tools/list          // 列出可用工具
Client → Server: tools/call { name, arguments }  // 调用工具
Client → Server: resources/list      // 列出可用资源
Client → Server: resources/read { uri }           // 读取资源

能力协商中，客户端声明 capabilities 表示自己支持哪些功能模块（如 tools、resources、prompts），服务端也返回自己的能力声明。只有双方都声明支持的功能才能使用。这种设计使得协议可以渐进式演进——新版本添加的能力不会破坏旧客户端的兼容性。

⚠️ **警告：**如果客户端没有发送 initialized 通知就直接调用工具，服务端必须返回 -32600 错误。这是协议的硬性要求。

🚀 二、从零实现 MCP Server

2.1 项目初始化

我们用 TypeScript 实现一个功能完整的 MCP Server，提供 JSON 格式化和 Base64 编解码两个工具。

# 初始化项目
mkdir mcp-devtools-server && cd mcp-devtools-server
npm init -y
npm install @modelcontextprotocol/sdk zod
npm install -D typescript @types/node
npx tsc --init

💡 **提示：**官方 @modelcontextprotocol/sdk 提供了高层抽象，但本文会先展示底层原理，再用 SDK 简化。理解底层才能真正掌握协议。

2.2 JSON-RPC 消息解析

MCP 基于 JSON-RPC 2.0。先实现消息的序列化与反序列化：

// src/protocol.ts — JSON-RPC 2.0 消息类型定义
interface JsonRpcRequest {
  jsonrpc: "2.0"
  id: number | string
  method: string
  params?: Record<string, unknown>
}

interface JsonRpcResponse {
  jsonrpc: "2.0"
  id: number | string
  result?: unknown
  error?: JsonRpcError
}

interface JsonRpcError {
  code: number
  message: string
  data?: unknown
}

interface JsonRpcNotification {
  jsonrpc: "2.0"
  method: string
  params?: Record<string, unknown>
}

// stdio 传输层使用换行分隔的 JSON
function createMessageHandler(
  onRequest: (req: JsonRpcRequest) => Promise<JsonRpcResponse>,
  onNotification: (notif: JsonRpcNotification) => void
) {
  let buffer = ""

  return {
    // 处理从 stdin 读取的数据
    processData(chunk: string) {
      buffer += chunk
      const lines = buffer.split("\n")
      buffer = lines.pop() || ""  // 保留不完整的最后一行

      for (const line of lines) {
        const trimmed = line.trim()
        if (!trimmed) continue

        const message = JSON.parse(trimmed)

        // 有 id 的是请求或响应
        if ("id" in message) {
          onRequest(message as JsonRpcRequest).then((resp) => {
            process.stdout.write(JSON.stringify(resp) + "\n")
          })
        } else {
          // 没有 id 的是通知
          onNotification(message as JsonRpcNotification)
        }
      }
    },
  }
}

2.3 实现核心 Server

这是最关键的部分——实现 MCP 协议的完整状态机：

// src/server.ts — MCP Server 核心实现
import { z } from "zod"

// 工具定义接口
interface ToolDefinition {
  name: string
  description: string
  inputSchema: z.ZodSchema
  handler: (args: unknown) => Promise<ToolResult>
}

interface ToolResult {
  content: Array<{ type: string; text?: string; data?: string; mimeType?: string }>
  isError?: boolean
}

class MCPServer {
  private tools = new Map<string, ToolDefinition>()
  private initialized = false
  private protocolVersion = "2025-11-05"

  // 注册工具
  registerTool(tool: ToolDefinition) {
    this.tools.set(tool.name, tool)
  }

  // 处理 JSON-RPC 请求
  async handleRequest(req: JsonRpcRequest): Promise<JsonRpcResponse> {
    switch (req.method) {
      case "initialize":
        return this.handleInitialize(req)
      case "tools/list":
        return this.handleToolsList(req)
      case "tools/call":
        return this.handleToolsCall(req)
      case "ping":
        return { jsonrpc: "2.0", id: req.id, result: {} }
      default:
        return {
          jsonrpc: "2.0",
          id: req.id,
          error: { code: -32601, message: `Method not found: ${req.method}` },
        }
    }
  }

  // 握手：initialize
  private handleInitialize(req: JsonRpcRequest): JsonRpcResponse {
    const params = req.params as { protocolVersion?: string; clientInfo?: { name: string; version: string } }

    // 版本兼容性检查
    if (params?.protocolVersion && params.protocolVersion !== this.protocolVersion) {
      console.error(`[MCP] 版本不匹配: 客户端=${params.protocolVersion}, 服务端=${this.protocolVersion}`)
    }

    return {
      jsonrpc: "2.0",
      id: req.id,
      result: {
        protocolVersion: this.protocolVersion,
        capabilities: {
          tools: { listChanged: false },
          resources: { subscribe: false },
        },
        serverInfo: {
          name: "devtools-mcp-server",
          version: "1.0.0",
        },
      },
    }
  }

  // 处理 initialized 通知
  handleNotification(notif: JsonRpcNotification) {
    if (notif.method === "notifications/initialized") {
      this.initialized = true
      console.error("[MCP] 客户端已确认初始化")
    }
  }

  // 列出所有工具
  private handleToolsList(req: JsonRpcRequest): JsonRpcResponse {
    this.ensureInitialized()

    const tools = Array.from(this.tools.values()).map((t) => ({
      name: t.name,
      description: t.description,
      inputSchema: zodToJsonSchema(t.inputSchema),
    }))

    return { jsonrpc: "2.0", id: req.id, result: { tools } }
  }

  // 调用工具
  private async handleToolsCall(req: JsonRpcRequest): Promise<JsonRpcResponse> {
    this.ensureInitialized()

    const { name, arguments: args } = req.params as { name: string; arguments: unknown }
    const tool = this.tools.get(name)

    if (!tool) {
      return {
        jsonrpc: "2.0",
        id: req.id,
        error: { code: -32602, message: `Unknown tool: ${name}` },
      }
    }

    // 用 Zod 验证输入参数
    const parsed = tool.inputSchema.safeParse(args)
    if (!parsed.success) {
      return {
        jsonrpc: "2.0",
        id: req.id,
        error: {
          code: -32602,
          message: `Invalid arguments: ${parsed.error.message}`,
        },
      }
    }

    try {
      const result = await tool.handler(parsed.data)
      return { jsonrpc: "2.0", id: req.id, result }
    } catch (err) {
      return {
        jsonrpc: "2.0",
        id: req.id,
        result: {
          content: [{ type: "text", text: `Error: ${err}` }],
          isError: true,
        },
      }
    }
  }

  private ensureInitialized() {
    if (!this.initialized) {
      throw new McpError(-32600, "Server not initialized")
    }
  }
}

// Zod Schema → JSON Schema 转换（简化版）
function zodToJsonSchema(schema: z.ZodSchema): Record<string, unknown> {
  if (schema instanceof z.ZodObject) {
    const shape = schema.shape
    const properties: Record<string, unknown> = {}
    const required: string[] = []

    for (const [key, value] of Object.entries(shape)) {
      const field = value as z.ZodSchema
      properties[key] = zodFieldToJsonSchema(field)
      if (!field.isOptional()) required.push(key)
    }

    return {
      type: "object",
      properties,
      required: required.length > 0 ? required : undefined,
    }
  }
  return { type: "object" }
}

function zodFieldToJsonSchema(field: z.ZodSchema): Record<string, unknown> {
  if (field instanceof z.ZodString) return { type: "string" }
  if (field instanceof z.ZodNumber) return { type: "number" }
  if (field instanceof z.ZodBoolean) return { type: "boolean" }
  if (field instanceof z.ZodEnum) return { type: "string", enum: field._def.values }
  if (field instanceof z.ZodOptional) return zodFieldToJsonSchema(field.unwrap())
  return { type: "string" }
}

⚠️ **警告：**生产环境中 Zod 到 JSON Schema 的转换远比这复杂。建议使用 zod-to-json-schema 库处理嵌套对象、联合类型、默认值等场景。

2.4 注册业务工具

现在注册两个实用工具——JSON 格式化和 Base64 编解码：

// src/tools.ts — 业务工具实现
import { z } from "zod"

// 工具 1：JSON 格式化
const jsonFormatTool: ToolDefinition = {
  name: "json_format",
  description: "格式化 JSON 字符串，支持缩进配置。处理压缩 JSON、API 响应等场景。",
  inputSchema: z.object({
    text: z.string().describe("需要格式化的 JSON 字符串"),
    indent: z.number().default(2).describe("缩进空格数，默认 2"),
    sortKeys: z.boolean().default(false).describe("是否对 key 排序"),
  }),
  handler: async (args) => {
    const { text, indent, sortKeys } = args as { text: string; indent: number; sortKeys: boolean }

    try {
      const parsed = JSON.parse(text)
      const formatted = JSON.stringify(parsed, sortKeys ? Object.keys(parsed).sort() : null, indent)
      return {
        content: [{ type: "text", text: formatted }],
      }
    } catch (err) {
      return {
        content: [{ type: "text", text: `JSON 解析失败: ${err}` }],
        isError: true,
      }
    }
  },
}

// 工具 2：Base64 编解码
const base64Tool: ToolDefinition = {
  name: "base64_codec",
  description: "Base64 编码/解码工具。支持文本和 URL 安全模式。",
  inputSchema: z.object({
    text: z.string().describe("输入文本"),
    mode: z.enum(["encode", "decode"]).describe("encode 编码，decode 解码"),
    urlSafe: z.boolean().default(false).describe("是否使用 URL 安全的 Base64（替换 +/ 为 -_）"),
  }),
  handler: async (args) => {
    const { text, mode, urlSafe } = args as { text: string; mode: string; urlSafe: boolean }

    try {
      let result: string
      if (mode === "encode") {
        result = Buffer.from(text, "utf-8").toString("base64")
        if (urlSafe) result = result.replace(/\+/g, "-").replace(/\//g, "_")
      } else {
        let input = text
        if (urlSafe) input = input.replace(/-/g, "+").replace(/_/g, "/")
        result = Buffer.from(input, "base64").toString("utf-8")
      }
      return { content: [{ type: "text", text: result }] }
    } catch (err) {
      return {
        content: [{ type: "text", text: `Base64 处理失败: ${err}` }],
        isError: true,
      }
    }
  },
}

export const allTools = [jsonFormatTool, base64Tool]

2.5 启动入口

将所有组件组装在一起：

// src/index.ts — 程序入口
import { createMessageHandler } from "./protocol"
import { MCPServer } from "./server"
import { allTools } from "./tools"

const server = new MCPServer()

// 注册所有工具
for (const tool of allTools) {
  server.registerTool(tool)
}

console.error("[MCP] DevTools Server 启动中...")

const handler = createMessageHandler(
  (req) => server.handleRequest(req),
  (notif) => server.handleNotification(notif)
)

// 从 stdin 读取数据
process.stdin.setEncoding("utf-8")
process.stdin.on("data", (chunk) => handler.processData(chunk))
process.stdin.on("end", () => {
  console.error("[MCP] 连接已关闭")
  process.exit(0)
})

📊 三、对比与实战

3.1 使用官方 SDK vs 手动实现

理解了底层原理后，我们来看看官方 SDK 能简化多少代码：

用官方 SDK 重写同一个 Server，代码量从 300 行降到 50 行：

// src/index-sdk.ts — 使用官方 SDK 的极简版本
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js"
import { z } from "zod"

const server = new McpServer({
  name: "devtools-mcp-server",
  version: "1.0.0",
})

// 注册 JSON 格式化工具 — 三行搞定
server.tool(
  "json_format",
  "格式化 JSON 字符串",
  {
    text: z.string().describe("JSON 字符串"),
    indent: z.number().default(2).describe("缩进空格数"),
    sortKeys: z.boolean().default(false).describe("是否排序 key"),
  },
  async ({ text, indent, sortKeys }) => {
    const parsed = JSON.parse(text)
    const formatted = JSON.stringify(parsed, sortKeys ? Object.keys(parsed).sort() : null, indent)
    return { content: [{ type: "text" as const, text: formatted }] }
  }
)

// 注册 Base64 工具
server.tool(
  "base64_codec",
  "Base64 编解码",
  {
    text: z.string(),
    mode: z.enum(["encode", "decode"]),
    urlSafe: z.boolean().default(false),
  },
  async ({ text, mode, urlSafe }) => {
    let result: string
    if (mode === "encode") {
      result = Buffer.from(text, "utf-8").toString("base64")
      if (urlSafe) result = result.replace(/\+/g, "-").replace(/\//g, "_")
    } else {
      let input = text
      if (urlSafe) input = input.replace(/-/g, "+").replace(/_/g, "/")
      result = Buffer.from(input, "base64").toString("utf-8")
    }
    return { content: [{ type: "text" as const, text: result }] }
  }
)

// 启动 stdio 传输
const transport = new StdioServerTransport()
await server.connect(transport)

📌 **记住：**学习阶段用手动实现理解原理，生产环境用官方 SDK 赢在效率。两者不矛盾。

3.2 配置 Claude Desktop 调试

在 claude_desktop_config.json 中注册你的 MCP Server：

{
  "mcpServers": {
    "devtools": {
      "command": "node",
      "args": ["/path/to/mcp-devtools-server/dist/index.js"],
      "env": {}
    }
  }
}

重启 Claude Desktop 后，在对话中直接说「帮我格式化这段 JSON」，Claude 就会自动调用你的 json_format 工具。

3.3 用 MCP Inspector 调试

MCP 官方提供了 Inspector 调试工具，可以可视化测试你的 Server：

# 启动 Inspector
npx @modelcontextprotocol/inspector node dist/index.js

# 打开浏览器访问 http://localhost:5173
# 可以看到工具列表、输入 schema、调用结果

💡 **提示：**Inspector 是开发 MCP Server 的必备工具。它能实时显示 JSON-RPC 消息流，帮你快速定位协议层面的问题。

⚠️ 四、生产环境的坑与最佳实践

4.1 常见踩坑点

坑 1：stdio 缓冲导致消息粘连

Node.js 的 process.stdout.write() 在非 TTY 模式下会缓冲输出。如果两条 JSON-RPC 消息粘在一起，客户端解析会失败。

// ❌ 错误：直接写入可能缓冲
process.stdout.write(JSON.stringify(resp1) + "\n")
process.stdout.write(JSON.stringify(resp2) + "\n")

// ✅ 正确：确保每条消息独立刷新
const write = (msg: object) => {
  process.stdout.write(JSON.stringify(msg) + "\n")
}

坑 2：忘记处理 initialized 通知

很多开发者在 initialize 响应后就开始处理 tools/call，但协议要求客户端先发送 notifications/initialized。忽略这个会导致竞态条件。

坑 3：工具描述不够详细

AI 模型根据 description 决定何时调用你的工具。描述太简短会导致模型误判。

// ❌ 差的描述
description: "JSON 工具"

// ✅ 好的描述
description: "格式化 JSON 字符串，支持缩进配置和 key 排序。适用于压缩 JSON、API 响应美化、配置文件整理。输入必须是合法的 JSON。"

4.2 安全最佳实践

MCP Server 直接暴露给 AI 模型，安全至关重要。一个被恶意提示注入攻击的 AI 模型，可能会通过你的 MCP Server 执行未授权操作。以下是经过生产验证的安全实践：

• ✅ 输入验证：所有参数必须用 Zod 严格校验，防止注入攻击。特别注意字符串参数中的特殊字符和转义序列
• ✅ 权限最小化：工具只暴露必要的能力，不要给 AI 完整的 shell 访问权限。如果你的 Server 需要访问数据库，只提供查询接口，不要暴露 DDL 操作
• ✅ 超时控制：每个工具调用设置超时（建议 30 秒），防止 AI 模型触发长时间运行的操作导致资源耗尽
• ✅ 日志审计：所有工具调用记录到 stderr（MCP 用 stdout 通信，日志必须走 stderr）。记录调用时间、工具名、输入参数摘要和执行结果
• ✅ 速率限制：在 Server 端实现速率限制，防止 AI 模型在循环中高频调用工具
• ❌ 避免直接执行用户输入：不要把用户传入的字符串直接拼接到命令或 SQL 中，始终使用参数化查询

4.3 工具设计原则

设计好的 MCP 工具需要遵循几个原则：

1. 原子性：一个工具做一件事。不要设计一个「万能工具」，而是拆分成多个专用工具
2. 幂等性：相同输入多次调用应产生相同结果（只读操作天然幂等）
3. 明确的错误信息：返回的错误消息要能帮助 AI 模型理解问题并重试
4. 最小权限：只请求完成任务所需的最少权限

💡 五、总结与展望

MCP 协议的设计哲学是简单但严谨——基于 JSON-RPC 2.0 这个成熟标准，加上 AI 场景特有的工具发现和调用语义。从零实现一遍协议，你会发现核心逻辑其实不复杂，真正考验功力的是工程细节：消息分帧、错误处理、参数校验、安全防护。

5.1 MCP 的三类原语

理解 MCP 的完整能力模型，需要认识它的三类原语（Primitive）：

本文重点实现了 Tools，因为它是 AI Agent 与外部世界交互的核心机制。Resources 适合提供上下文数据（如数据库 schema、文件内容），Prompts 适合构建结构化的用户交互流程。

5.2 MCP 生态现状

截至 2026 年 6 月，MCP 生态已经相当成熟：

• ✅ 主流 AI 平台全部支持：Claude、GPT、Gemini、DeepSeek 均已原生支持 MCP
• ✅ 开发工具深度集成：Cursor、Windsurf、Continue、Zed 等 IDE 内置 MCP 客户端
• ✅ 官方 Server 库丰富：GitHub、PostgreSQL、Slack、Google Drive 等 50+ 官方 Server
• ✅ 社区活跃：GitHub 上第三方 MCP Server 超过 12,000 个

⚡ **关键结论：**学习 MCP 协议的最佳路径是「先手动实现理解原理，再用 SDK 提高效率」。本文的手动实现版本帮助你理解每个字节的含义，SDK 版本展示了生产环境应有的效率。现在正是投入 MCP 开发的最佳时机——协议已经稳定，生态正在爆发。

如果你想在 jsjson.com 上快速体验 MCP 工具，可以访问我们的 JSON 格式化工具 和 Base64 编解码工具 ，它们的逻辑与本文实现的 MCP 工具完全一致。

相关资源：

• 📖 MCP 官方规范
• 🔧 MCP TypeScript SDK
• 🧪 MCP Inspector
• 📦 MCP Server 仓库索引