MCP Server 开发实战：从零构建你的第一个工具服务

来源：modelcontextprotocol.io | MCP TypeScript SDK | MCP Python SDK | 整理时间：2026-04-06

为什么需要自己开发 MCP Server？

MCP 生态已有数千个社区服务器，但很多场景需要定制：

• 内部 API 的私有接口没有现成服务器
• 公司内部系统需要自定义权限控制
• 现有服务器不完全符合业务需求
• 学习 MCP Server 开发是理解 Agent 工具生态的最佳方式

MCP Server 核心概念

三大原语（Primitives）

传输方式

TypeScript 实战

理解 Claude Code 遇到的场景

Claude Code 连接 MCP Server 时：

Claude Code ←stdio→ 你的 MCP Server
                    ↕stdin↕
                    ←stdout←

最小可用示例

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: "my-tools",
  version: "1.0.0",
});

// 注册一个 Tool
server.tool(
  "query_database",
  "查询数据库并返回结果",
  { sql: z.string().describe("SQL 查询语句") },
  async ({ sql }) => {
    const results = await executeQuery(sql);
    return {
      content: [{ type: "text", text: JSON.stringify(results) }],
    };
  }
);

// 注册一个 Resource
server.resource(
  "app-config",
  "应用配置信息",
  async () => ({
    contents: [{
      uri: "config://app",
      text: JSON.stringify({ version: "1.0", env: "production" }),
    }],
  });
);

// 注册一个 Prompt 模板
server.prompt(
  "review-code",
  { code: z.string().describe("要审查的代码") },
  async ({ code }) => ({
    messages: [{
      role: "user",
      content: `请审查以下代码的安全性和性能：\\n\\n${code}`,
    }],
  });
);

// 启动
const transport = new StdioServerTransport();
await server.connect(transport);

带参数校验的 Tool

server.tool(
  "create_user",
  "创建新用户",
  {
    name: z.string().min(1).max(50),
    email: z.string().email(),
    role: z.enum(["admin", "user", "viewer"]).default("user"),
  },
  async ({ name, email, role }) => {
    const user = await createUser({ name, email, role });
    return {
      content: [{ type: "text", text: `用户 ${name} 已创建，角色: ${role}` }],
    };
  }
);

Streamable HTTP 传输（远程部署）

import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamablehttp.js";

const transport = new StreamableHTTPServerTransport({
  sessionIdGenerator: () => crypto.randomUUID(),
});

await server.connect(transport);

// 配合 Express 使用
// app.post('/mcp', expressHandler);

Python 实战（FastMCP）

Python 用 FastMCP 衔库更简洁：

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("my-tools")

@mcp.tool()
def query_database(sql: str) -> str:
    """查询数据库并返回结果"""
    results = execute_query(sql)
    return f"查询结果: {results}"

@mcp.resource("config://app")
def get_config() -> str:
    """获取应用配置"""
    return json.dumps({"version": "1.0", "env": "production"})

@mcp.prompt()
def review_code(code: str) -> str:
    """代码审查提示模板"""
    return f"请审查以下代码的安全性和性能:\\n\\n{code}"

mcp.run()

FastMCP 进阶：结构化输出

from pydantic import BaseModel

class UserResult(BaseModel):
    name: str
    email: str
    role: str
    success: bool

@mcp.tool()
def create_user(name: str, email: str, role: str = "user") -> UserResult:
    """创建新用户，返回结构化结果"""
    user = await create_user(name, email, role)
    return UserResult(
        name=user.name,
        email=user.email,
        role=user.role,
        success=True,
    )

JSON-RPC 协议详解

理解协议细节有助于调试：

请求（Client → Server）

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "query_database",
    "arguments": {
      "sql": "SELECT * FROM users LIMIT 10"
    }
  }
}

响应（Server → Client）

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [
      { "type": "text", "text": "[{\"id\": 1, \"name\": \"Alice\"}]" }
    ]
  }
}

能力协商

连接建立时，客户端和服务器交换各自支持的能力：

Client: 我支持 tools, resources, sampling
Server: 我提供 3 个 tools, 5 个 resources, 2 个 prompts

测试与调试

MCP Inspector

官方提供的调试工具：

# 使用 npx 直接运行
npx @modelcontextprotocol/inspector

# 或在项目中安装
npm install -D @modelcontextprotocol/inspector

Inspector 提供：

• 可视化查看服务器的能力（Tools/Resources/Prompts）
• 实时测试工具调用
• 查看请求/响应日志
• 测试不同的传输方式

在 Claude Code 中调试

# 在 Claude Code 中连接 MCP Server
# 编辑 .claude/settings.json 或项目 .mcp.json

{
  "mcpServers": {
    "my-tools": {
      "command": "node",
      "args": ["path/to/server.js"]
    }
  }
}

常见调试技巧

• 日志输出：stdio 模式下用 console.error 输出调试信息（stdout 被 MCP 协议占用）
• 错误处理：Tool 函数中 try/catch 捕获错误，返回有意义的错误消息
• 类型检查：用 Zod Schema 确保参数格式正确
• Inspector 验证：发布前先用 Inspector 测试所有 Tool 和 Resource

实战案例：构建文档查询 MCP Server

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
import fs from "fs/promises";
import path from "path";

const server = new McpServer({
  name: "doc-query",
  version: "1.0.0",
});

// Tool: 搜索文档
server.tool(
  "search_docs",
  "在文档目录中搜索关键词",
  {
    query: z.string().describe("搜索关键词"),
    directory: z.string().default("./docs").describe("文档目录路径"),
    maxResults: z.number().default(10).describe("最大结果数"),
  },
  async ({ query, directory, maxResults }) => {
    const results = await searchDocuments(directory, query, maxResults);
    return {
      content: [{ type: "text", text: formatResults(results) }],
    };
  }
);

// Tool: 读取文档
server.tool(
  "read_doc",
  "读取指定文档的内容",
  { path: z.string().describe("文档路径") },
  async ({ path: docPath }) => {
    const content = await fs.readFile(docPath, "utf-8");
    return {
      content: [{ type: "text", text: content }],
    };
  }
);

// Resource: 文档列表
server.resource(
  "docs://index",
  "文档索引",
  async () => ({
    contents: [{
      uri: "docs://index",
      text: await buildDocIndex("./docs"),
    }],
  })
);

// Prompt: 文档总结
server.prompt(
  "summarize-doc",
  { topic: z.string().describe("要总结的主题") },
  async ({ topic }) => ({
    messages: [{
      role: "user",
      content: `请总结关于 "${topic}" 的所有文档要点。使用 docs://index 资源获取文档列表，然后用 search_docs 工具搜索相关内容。`,
    }],
  })
);

async function searchDocuments(dir: string, query: string, max: number) {
  // 实现文件搜索逻辑
  const files = await fs.readdir(dir);
  return files.slice(0, max);
}

async function buildDocIndex(dir: string) {
  const files = await fs.readdir(dir);
  return files.join("\\n");
}

function formatResults(results: string[]) {
  return results.map((r, i) => `${i + 1}. ${r}`).join("\\n");
}

const transport = new StdioServerTransport();
await server.connect(transport);

部署建议

企业级部署清单

• 认证和授权机制
• 速率限制和配额管理
• 日志和监控
• 错误处理和重试
• API 版本管理

安全最佳实践

相关链接

• MCP 官方文档：https://modelcontextprotocol.io/introduction
• TypeScript SDK：https://github.com/modelcontextprotocol/typescript-sdk
• Python SDK：https://github.com/modelcontextprotocol/python-sdk
• 官方服务器集合：https://github.com/modelcontextprotocol/servers
• MCP Inspector：https://github.com/modelcontextprotocol/inspector
• MCP 协议详解
• Claude Code MCP 集成