首页 › tutorials › MCP服务器TypeScript与Python双轨开发实战

MCP服务器TypeScript与Python双轨开发实战

概述

本教程面向希望为Claude Code构建自定义MCP（Model Context Protocol）服务器的开发者。你将学会用 TypeScript SDK 和 Python FastMCP 两种方案，从零开发一个功能完整的MCP服务器，涵盖Tools、Resources、Prompts三大能力原语，并掌握stdio与HTTP两种传输方式的配置部署。适合有基础编程经验、想让Claude Code连接自己工具链的开发者。

前置条件

项目 要求
Claude Code 已安装并配置（npm install -g @anthropic-ai/claude-code）
Node.js v20+（TypeScript方案）
Python 3.10+（Python方案）
基础知识 TypeScript或Python、JSON配置、命令行操作
操作系统 macOS / Linux / Windows（WSL2）

第一章：MCP协议核心概念

1.1 什么是MCP

Model Context Protocol（MCP）是Anthropic推出的开放协议，标准化了AI应用与外部工具之间的通信方式。其核心思想类似USB-C——一个通用接口连接所有设备。

MCP采用 Client-Server架构，Claude Code作为Client，你开发的工具服务作为Server。通信基于JSON-RPC 2.0协议。

1.2 三大能力原语

MCP服务器可以暴露三种类型的 capabilities：

┌─────────────────────────────────────────────────────┐
│                  MCP Server Capabilities             │
├──────────────┬──────────────┬───────────────────────┤
│   Tools      │  Resources   │      Prompts          │
├──────────────┼──────────────┼───────────────────────┤
│ 可执行函数    │ 只读数据源    │ 可复用的提示模板       │
│ 有副作用      │ 无副作用      │ 参数化生成消息         │
│ 如：查询数据库 │ 如：读取文件  │ 如：代码审查模板       │
│ 发送邮件      │ 获取配置      │ 文档生成模板           │
└──────────────┴──────────────┴───────────────────────┘

1.3 传输方式

MCP支持两种传输方式：

传输方式 适用场景 特点
stdio 本地命令行工具 进程间管道通信，无需网络端口
HTTP Streamable 远程服务、多Client 基于HTTP POST + SSE，支持无状态部署

1.4 整体架构

graph TB
    subgraph Client["Claude Code (MCP Client)"]
        A[用户请求] --> B[LLM推理]
        B --> C{需要外部工具?}
        C -->|是| D[MCP Client层]
        C -->|否| E[直接回复]
    end

    subgraph Transport["传输层"]
        F[stdio管道]
        G[HTTP Streamable]
    end

    subgraph Server["MCP Server"]
        H[Tools<br/>可执行函数]
        I[Resources<br/>只读数据]
        J[Prompts<br/>提示模板]
    end

    D --> F
    D --> G
    F --> H
    F --> I
    F --> J
    G --> H
    G --> I
    G --> J
    H --> K[数据库/API/文件系统]
    I --> K

第二章：TypeScript方案——从零构建MCP服务器

2.1 项目初始化

# 创建项目目录
mkdir my-mcp-server && cd my-mcp-server

# 初始化Node.js项目
npm init -y

# 安装依赖
npm install @modelcontextprotocol/sdk zod
npm install -D typescript @types/node

# 创建TypeScript配置
cat > tsconfig.json << 'EOF'
{
  "compilerOptions": {
    "target": "ES2022",
    "module": "Node16",
    "moduleResolution": "Node16",
    "outDir": "./build",
    "rootDir": "./src",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "declaration": true
  },
  "include": ["src/**/*"]
}
EOF

2.2 最小可运行服务器

创建 src/index.ts：

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

// 注册一个Tool
server.tool(
  "greet",                              // 工具名称
  "向指定用户发送问候",                    // 工具描述（LLM根据此决定是否调用）
  { name: z.string().describe("用户名") }, // 参数Schema（Zod定义）
  async ({ name }) => {                   // 处理函数
    return {
      content: [{ type: "text", text: `你好，${name}！欢迎使用MCP服务器。` }],
    };
  }
);

// 启动服务器
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("MCP服务器已启动（stdio模式）");
}

main().catch(console.error);

关键说明：console.error 而非 console.log——stdio模式下，stdout被MCP协议占用作为通信管道，任何写入stdout的内容都会破坏协议。调试信息必须用stderr。

2.3 配置package.json

{
  "name": "my-mcp-server",
  "version": "1.0.0",
  "type": "module",
  "scripts": {
    "build": "tsc",
    "start": "node build/index.js"
  },
  "bin": {
    "my-mcp-server": "./build/index.js"
  }
}

构建并测试：

npm run build
echo '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}' | node build/index.js

预期输出（格式化后）：

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "protocolVersion": "2025-03-26",
    "capabilities": { "tools": {} },
    "serverInfo": { "name": "my-tools-server", "version": "1.0.0" }
  }
}

2.4 实战：构建一个代码分析服务器

下面构建一个更实用的服务器，包含Tools、Resources和Prompts三种能力。

创建 src/code-analyzer/index.ts：

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

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

// ============ Tools ============

// Tool 1: 统计代码行数
server.tool(
  "count_lines",
  "统计指定目录下源代码文件的行数，按文件类型分类汇总",
  {
    directory: z.string().describe("要分析的目录路径"),
    extensions: z.array(z.string()).optional()
      .describe("文件扩展名过滤，如 ['ts', 'py']，默认全部"),
  },
  async ({ directory, extensions }) => {
    try {
      const stats: Record<string, { files: number; lines: number }> = {};

      async function walk(dir: string) {
        const entries = await fs.readdir(dir, { withFileTypes: true });
        for (const entry of entries) {
          if (entry.name.startsWith(".") || entry.name === "node_modules") continue;
          const fullPath = path.join(dir, entry.name);
          if (entry.isDirectory()) {
            await walk(fullPath);
          } else {
            const ext = path.extname(entry.name).slice(1);
            if (extensions && extensions.length > 0 && !extensions.includes(ext)) continue;
            const content = await fs.readFile(fullPath, "utf-8");
            const lineCount = content.split("\n").length;
            if (!stats[ext]) stats[ext] = { files: 0, lines: 0 };
            stats[ext].files++;
            stats[ext].lines += lineCount;
          }
        }
      }

      await walk(directory);

      const summary = Object.entries(stats)
        .map(([ext, s]) => `.${ext}: ${s.files} 文件, ${s.lines} 行`)
        .join("\n");

      return {
        content: [{ type: "text", text: `代码统计结果：\n${summary}` }],
      };
    } catch (error: any) {
      return {
        content: [{ type: "text", text: `错误: ${error.message}` }],
        isError: true,
      };
    }
  }
);

// Tool 2: 搜索代码内容
server.tool(
  "search_code",
  "在指定目录中搜索包含关键词的代码文件",
  {
    directory: z.string().describe("搜索目录"),
    keyword: z.string().describe("搜索关键词"),
    max_results: z.number().optional().default(20).describe("最大结果数"),
  },
  async ({ directory, keyword, max_results }) => {
    const results: string[] = [];

    async function search(dir: string) {
      if (results.length >= max_results) return;
      const entries = await fs.readdir(dir, { withFileTypes: true });
      for (const entry of entries) {
        if (entry.name.startsWith(".") || entry.name === "node_modules") continue;
        const fullPath = path.join(dir, entry.name);
        if (entry.isDirectory()) {
          await search(fullPath);
        } else {
          try {
            const content = await fs.readFile(fullPath, "utf-8");
            const lines = content.split("\n");
            lines.forEach((line, i) => {
              if (line.toLowerCase().includes(keyword.toLowerCase()) && results.length < max_results) {
                results.push(`${fullPath}:${i + 1}: ${line.trim()}`);
              }
            });
          } catch { /* 跳过二进制文件 */ }
        }
      }
    }

    await search(directory);

    return {
      content: [{
        type: "text",
        text: results.length > 0 
          ? `找到 ${results.length} 个匹配：\n${results.join("\n")}`
          : `未找到包含 "${keyword}" 的代码`,
      }],
    };
  }
);

// ============ Resources ============

// Resource: 项目结构
server.resource(
  new ResourceTemplate("project://structure/{path}", { list: undefined }),
  async (uri, { path: dirPath }) => {
    async function buildTree(dir: string, prefix = ""): Promise<string> {
      const entries = await fs.readdir(dir, { withFileTypes: true });
      let result = "";
      const filtered = entries.filter(e => !e.name.startsWith(".") && e.name !== "node_modules");
      for (let i = 0; i < filtered.length; i++) {
        const entry = filtered[i];
        const isLast = i === filtered.length - 1;
        result += `${prefix}${isLast ? "└── " : "├── "}${entry.name}\n`;
        if (entry.isDirectory()) {
          const subPath = path.join(dir, entry.name);
          result += await buildTree(subPath, `${prefix}${isLast ? "    " : "│   "}`);
        }
      }
      return result;
    }

    const tree = await buildTree(dirPath as string);
    return {
      contents: [{
        uri: uri.href,
        text: tree,
      }],
    };
  }
);

// ============ Prompts ============

// Prompt: 代码审查模板
server.prompt(
  "code_review",
  "生成代码审查提示模板",
  { 
    file_path: z.string().describe("要审查的文件路径"),
    focus: z.enum(["security", "performance", "readability", "all"])
      .default("all").describe("审查重点"),
  },
  async ({ file_path, focus }) => {
    const focusInstructions: Record<string, string> = {
      security: "重点关注安全漏洞、输入验证、权限控制",
      performance: "重点关注性能瓶颈、算法效率、内存使用",
      readability: "重点关注命名规范、代码结构、注释质量",
      all: "全面审查安全性、性能、可读性和最佳实践",
    };

    return {
      messages: [{
        role: "user",
        content: {
          type: "text",
          text: `请对以下文件进行代码审查。\n\n文件: ${file_path}\n审查重点: ${focusInstructions[focus]}\n\n请给出具体的改进建议和评分。`,
        },
      }],
    };
  }
);

// 启动
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("Code Analyzer MCP Server 已启动");
}

main().catch(console.error);

2.5 注册到Claude Code

方式一：命令行注册（推荐）

# stdio模式——最常用的本地开发方式
claude mcp add code-analyzer -- node /path/to/my-mcp-server/build/code-analyzer/index.js

# 限定作用域（仅在当前项目生效）
claude mcp add code-analyzer -s project -- node /path/to/my-mcp-server/build/code-analyzer/index.js

# 限定作用域（仅在用户级别生效）
claude mcp add code-analyzer -s user -- node /path/to/my-mcp-server/build/code-analyzer/index.js

方式二：JSON配置文件

项目级别 .mcp.json（放在项目根目录）：

{
  "mcpServers": {
    "code-analyzer": {
      "type": "stdio",
      "command": "node",
      "args": ["/absolute/path/to/my-mcp-server/build/code-analyzer/index.js"],
      "env": {}
    }
  }
}

用户级别 ~/.claude.json：

{
  "mcpServers": {
    "code-analyzer": {
      "type": "stdio",
      "command": "node",
      "args": ["/path/to/my-mcp-server/build/code-analyzer/index.js"]
    }
  }
}

方式三：带环境变量

claude mcp add-json code-analyzer '{
  "type": "stdio",
  "command": "node",
  "args": ["./build/code-analyzer/index.js"],
  "env": {
    "ANALYZER_MAX_DEPTH": "10",
    "ANALYZER_EXCLUDE": "node_modules,.git,dist"
  }
}'

2.6 验证与调试

# 列出已注册的MCP服务器
claude mcp list

# 查看特定服务器详情
claude mcp get code-analyzer

# 启动Claude Code并测试
claude
> 请用count_lines工具统计当前项目的代码行数
> 用search_code搜索包含"export"的代码

调试技巧——在服务器代码中加入日志：

import * as fs from "fs";

// 写入调试日志到文件（不影响stdio通信）
function debugLog(message: string) {
  fs.appendFileSync("/tmp/mcp-debug.log", `${new Date().toISOString()} ${message}\n`);
}

// 在Tool handler中使用
server.tool("my_tool", "描述", { /* schema */ }, async (params) => {
  debugLog(`收到调用: ${JSON.stringify(params)}`);
  // ... 业务逻辑
  debugLog(`返回结果: ${result}`);
  return { content: [{ type: "text", text: result }] };
});

第三章：Python方案——FastMCP极速开发

3.1 FastMCP简介

FastMCP是MCP的官方Python SDK，以装饰器为核心的API让服务器开发极为简洁。对于数据科学、API对接、运维脚本等场景，Python方案往往更高效。

3.2 项目初始化

mkdir my-fastmcp-server && cd my-fastmcp-server

# 创建虚拟环境
python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate

# 安装FastMCP
pip install fastmcp httpx

# 验证安装
fastmcp --version

3.3 最小可运行服务器

创建 server.py：

from fastmcp import FastMCP

mcp = FastMCP("my-python-tools")

@mcp.tool
def add(a: int, b: int) -> int:
    """两数相加"""
    return a + b

@mcp.tool
def word_count(text: str) -> dict:
    """统计文本的词数和字符数"""
    words = len(text.split())
    chars = len(text)
    chars_no_space = len(text.replace(" ", ""))
    return {
        "words": words,
        "characters": chars,
        "characters_no_space": chars_no_space,
    }

if __name__ == "__main__":
    mcp.run()  # 默认stdio模式

FastMCP的魔法：函数签名中的类型注解自动生成参数Schema，docstring自动成为工具描述。无需手写Zod Schema！

运行测试：

# 方式一：直接运行
python server.py

# 方式二：用fastmcp CLI
fastmcp run server.py

# 方式三：开发模式（带热重载 + Inspector UI）
fastmcp dev server.py

3.4 实战：构建一个API聚合服务器

创建 api_hub_server.py：

import httpx
import json
from datetime import datetime
from fastmcp import FastMCP

mcp = FastMCP(
    "api-hub",
    instructions="这是一个API聚合服务器，提供天气查询、IP查询和汇率转换功能。"
)

# ============ Tools ============

@mcp.tool
async def get_weather(city: str) -> str:
    """查询指定城市的当前天气信息

    Args:
        city: 城市名称（中文或英文），如"北京"或"Beijing"
    """
    async with httpx.AsyncClient(timeout=10) as client:
        # 使用wttr.in免费天气API
        resp = await client.get(
            f"https://wttr.in/{city}",
            params={"format": "j1"}
        )
        if resp.status_code != 200:
            return f"查询失败：HTTP {resp.status_code}"

        data = resp.json()
        current = data["current_condition"][0]

        return json.dumps({
            "城市": city,
            "温度": f"{current['temp_C']}°C",
            "体感温度": f"{current['FeelsLikeC']}°C",
            "湿度": f"{current['humidity']}%",
            "天气描述": current["weatherDesc"][0]["value"],
            "风速": f"{current['windspeedKmph']} km/h",
            "能见度": f"{current['visibility']} km",
        }, ensure_ascii=False, indent=2)


@mcp.tool
async def get_ip_info(ip: str = "") -> str:
    """查询IP地址的地理位置信息

    Args:
        ip: 要查询的IP地址，留空查询当前出口IP
    """
    async with httpx.AsyncClient(timeout=10) as client:
        url = f"https://ipapi.co/{ip}/json/" if ip else "https://ipapi.co/json/"
        resp = await client.get(url)
        if resp.status_code != 200:
            return f"查询失败：HTTP {resp.status_code}"

        data = resp.json()
        return json.dumps({
            "IP": data.get("ip"),
            "城市": data.get("city"),
            "地区": data.get("region"),
            "国家": data.get("country_name"),
            "时区": data.get("timezone"),
            "ISP": data.get("org"),
            "经纬度": f"{data.get('latitude')}, {data.get('longitude')}",
        }, ensure_ascii=False, indent=2)


@mcp.tool
async def convert_currency(
    amount: float, 
    from_currency: str = "USD", 
    to_currency: str = "CNY"
) -> str:
    """货币汇率转换

    Args:
        amount: 金额
        from_currency: 源货币代码，如USD、EUR、CNY
        to_currency: 目标货币代码
    """
    async with httpx.AsyncClient(timeout=10) as client:
        resp = await client.get(
            f"https://open.er-api.com/v6/latest/{from_currency.upper()}"
        )
        if resp.status_code != 200:
            return f"查询失败：HTTP {resp.status_code}"

        rates = resp.json()["rates"]
        target = to_currency.upper()

        if target not in rates:
            available = ", ".join(sorted(rates.keys())[:20])
            return f"不支持的货币: {target}。可用货币: {available}..."

        result = amount * rates[target]
        return f"{amount} {from_currency.upper()} = {result:.2f} {target}"


# ============ Resources ============

@mcp.resource("config://api-hub/status")
def get_server_status() -> str:
    """服务器状态资源"""
    return json.dumps({
        "server": "api-hub",
        "version": "1.0.0",
        "status": "running",
        "uptime_check": datetime.now().isoformat(),
        "available_tools": ["get_weather", "get_ip_info", "convert_currency"],
    }, ensure_ascii=False, indent=2)


@mcp.resource("config://api-hub/supported-currencies")
def get_supported_currencies() -> str:
    """支持的货币列表"""
    return json.dumps({
        "major_currencies": {
            "USD": "美元",
            "EUR": "欧元",
            "GBP": "英镑",
            "JPY": "日元",
            "CNY": "人民币",
            "KRW": "韩元",
            "HKD": "港币",
        },
        "total": "150+",
    }, ensure_ascii=False, indent=2)


# ============ Prompts ============

@mcp.prompt
def travel_planner(destination: str, budget_usd: float = 1000) -> list:
    """旅行规划提示模板

    Args:
        destination: 目的地城市
        budget_usd: 预算（美元）
    """
    return [
        {
            "role": "user",
            "content": {
                "type": "text",
                "text": (
                    f"我要去{destination}旅行，预算{budget_usd}美元。\n\n"
                    "请帮我：\n"
                    "1. 查询目的地当前天气\n"
                    "2. 将预算转换为当地货币\n"
                    "3. 根据天气和预算给出旅行建议\n"
                ),
            },
        }
    ]


@mcp.prompt
def currency_comparison(amount: float = 100) -> list:
    """多货币汇率对比提示模板"""
    currencies = ["USD", "EUR", "GBP", "JPY", "CNY"]
    currency_list = "、".join(currencies)

    return [
        {
            "role": "user",
            "content": {
                "type": "text",
                "text": (
                    f"请将 {amount} USD 分别转换为 {currency_list}，\n"
                    "并对比各汇率，给出分析。"
                ),
            },
        }
    ]


if __name__ == "__main__":
    mcp.run()

3.5 注册到Claude Code

# stdio模式
claude mcp add api-hub -- /path/to/my-fastmcp-server/.venv/bin/python /path/to/my-fastmcp-server/api_hub_server.py

# 或使用fastmcp命令
claude mcp add api-hub -- fastmcp run /path/to/my-fastmcp-server/api_hub_server.py

# 验证
claude mcp list

也可以用 .mcp.json 配置：

{
  "mcpServers": {
    "api-hub": {
      "type": "stdio",
      "command": "/path/to/.venv/bin/python",
      "args": ["/path/to/api_hub_server.py"],
      "env": {
        "PYTHONUNBUFFERED": "1"
      }
    }
  }
}

注意：PYTHONUNBUFFERED=1 确保Python输出不被缓冲，避免stdio通信延迟。

第四章：HTTP Streamable传输——远程部署

4.1 何时选择HTTP模式

场景 推荐传输
个人本地开发 stdio
团队共享服务器 HTTP Streamable
CI/CD管道中运行 stdio
需要服务发现/负载均衡 HTTP Streamable
同时服务多个Client HTTP Streamable

4.2 TypeScript HTTP服务器

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
import express from "express";
import { z } from "zod";
import { randomUUID } from "crypto";

const app = express();
app.use(express.json());

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

// 注册工具...
server.tool(
  "echo",
  "回显输入文本",
  { message: z.string() },
  async ({ message }) => ({
    content: [{ type: "text", text: `Echo: ${message}` }],
  })
);

// MCP协议处理器
app.post("/mcp", async (req, res) => {
  const transport = new StreamableHTTPServerTransport({
    sessionIdGenerator: randomUUID,
  });
  await server.connect(transport);
  await transport.handleRequest(req, res, req.body);
});

// SSE通道（用于服务器主动推送）
app.get("/mcp", async (req, res) => {
  // StreamableHTTP需要处理GET请求用于SSE连接
});

// 删除会话
app.delete("/mcp", async (req, res) => {
  // 处理会话清理
});

app.listen(3000, () => {
  console.log("MCP HTTP Server running on port 3000");
});

4.3 FastMCP HTTP模式

FastMCP切换HTTP模式极其简单：

from fastmcp import FastMCP

mcp = FastMCP("remote-api-hub")

# ... 注册tools/resources/prompts ...

# 关键：只需改一行
mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

或通过CLI指定：

# 启动HTTP服务器
fastmcp run server.py --transport streamable-http --port 8000

# 开发模式
fastmcp dev server.py --transport streamable-http --port 8000

4.4 在Claude Code中配置HTTP连接

# 通过命令行
claude mcp add remote-tools --transport http http://localhost:8000/mcp

# 通过JSON配置
claude mcp add-json remote-tools '{
  "type": "http",
  "url": "http://your-server:8000/mcp"
}'

.mcp.json 配置：

{
  "mcpServers": {
    "remote-tools": {
      "type": "http",
      "url": "http://your-server:8000/mcp"
    }
  }
}

第五章：高级模式与生产实践

5.1 错误处理与重试

from fastmcp import FastMCP
import httpx
import asyncio

mcp = FastMCP("robust-server")

async def fetch_with_retry(url: str, max_retries: int = 3) -> dict:
    """带重试的HTTP请求"""
    for attempt in range(max_retries):
        try:
            async with httpx.AsyncClient(timeout=10) as client:
                resp = await client.get(url)
                resp.raise_for_status()
                return resp.json()
        except httpx.HTTPError as e:
            if attempt == max_retries - 1:
                raise
            wait = 2 ** attempt  # 指数退避
            await asyncio.sleep(wait)

@mcp.tool
async def reliable_query(query: str) -> str:
    """可靠查询，内置重试机制"""
    try:
        data = await fetch_with_retry(f"https://api.example.com/search?q={query}")
        return json.dumps(data, ensure_ascii=False)
    except Exception as e:
        # 返回结构化错误信息，而非抛出异常
        return f"查询失败: {str(e)}。请稍后重试。"

5.2 进度通知

长时间运行的操作应发送进度通知：

server.tool(
  "process_large_file",
  "处理大文件",
  { file_path: z.string(), chunk_size: z.number().default(1000) },
  async ({ file_path, chunk_size }, { sendNotification }) => {
    const content = await fs.readFile(file_path, "utf-8");
    const lines = content.split("\n");
    const total = lines.length;
    let processed = 0;

    for (let i = 0; i < total; i += chunk_size) {
      // 处理这一批数据...
      processed = Math.min(i + chunk_size, total);

      // 发送进度通知
      await sendNotification({
        method: "notifications/progress",
        params: {
          progress: processed,
          total: total,
          progressToken: "file-processing",
        },
      });
    }

    return {
      content: [{ type: "text", text: `处理完成：${total} 行` }],
    };
  }
);

5.3 服务器指令（Instructions）

通过 instructions 向LLM提供使用指南：

mcp = FastMCP(
    "enterprise-tools",
    instructions="""
    这是一个企业级工具服务器。使用指南：

    1. 查询数据库时，优先使用 read_only 模式
    2. 批量操作每次不超过100条
    3. 涉及敏感数据时使用 mask_pii 参数
    4. 所有时间使用UTC时区

    依赖关系：
    - generate_report 依赖 query_database 的结果
    - send_notification 需要先调用 get_user_info
    """
)

5.4 返回图片和多媒体内容

server.tool(
  "generate_chart",
  "生成图表并返回图片",
  {
    data: z.array(z.number()).describe("数据点"),
    title: z.string().describe("图表标题"),
  },
  async ({ data, title }) => {
    // 假设生成了一张PNG图表
    const imageBuffer = await generateChartPNG(data, title);

    return {
      content: [
        {
          type: "image",
          data: imageBuffer.toString("base64"),
          mimeType: "image/png",
        },
        {
          type: "text",
          text: `已生成图表"${title}"，包含${data.length}个数据点。`,
        },
      ],
    };
  }
);

5.5 配置层级与作用域

Claude Code的MCP配置遵循以下优先级（从高到低）：

项目级 .mcp.json  >  用户级 ~/.claude.json  >  企业级配置

graph TD
    A[企业级策略配置<br/>由管理员统一管理] --> B[用户级配置<br/>~/.claude.json]
    B --> C[项目级配置<br/>.mcp.json]

    D[claude mcp add -s enterprise] --> A
    E[claude mcp add -s user] --> B
    F[claude mcp add -s project] --> C

    style A fill:#ff9999
    style B fill:#ffcc99
    style C fill:#99ff99

多环境配置实践：

# 开发环境：项目级配置（指向本地服务器）
claude mcp add api-hub -s project -- python ./dev_server.py

# 生产环境：用户级配置（指向远程服务器）
claude mcp add api-hub -s user --transport http https://api.example.com/mcp

# 删除不再需要的服务器
claude mcp remove api-hub -s project

第六章：完整实战项目——智能文档助手

将前面所有知识整合为一个完整项目。这个MCP服务器提供文档索引、搜索、摘要和翻译功能。

6.1 项目结构

doc-assistant-mcp/
├── package.json
├── tsconfig.json
├── src/
│   ├── index.ts          # 入口文件
│   ├── tools/
│   │   ├── search.ts     # 文档搜索
│   │   ├── summarize.ts  # 文档摘要
│   │   └── translate.ts  # 文档翻译
│   ├── resources/
│   │   └── docs.ts       # 文档资源
│   └── prompts/
│       └── templates.ts  # 提示模板
├── docs/                 # 示例文档目录
│   ├── architecture.md
│   └── api-reference.md
└── .mcp.json             # 项目MCP配置

6.2 入口文件 src/index.ts

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { registerSearchTool } from "./tools/search.js";
import { registerSummarizeTool } from "./tools/summarize.js";
import { registerTranslateTool } from "./tools/translate.js";
import { registerDocResources } from "./resources/docs.js";
import { registerPrompts } from "./prompts/templates.js";

const server = new McpServer({
  name: "doc-assistant",
  version: "1.0.0",
  instructions: `
    文档助手服务器。可以搜索、摘要和翻译Markdown文档。
    文档存放在 ./docs/ 目录中。
    搜索时使用 search_docs 工具。
    摘要时使用 summarize_doc 工具。
  `,
});

// 注册所有功能模块
registerSearchTool(server);
registerSummarizeTool(server);
registerTranslateTool(server);
registerDocResources(server);
registerPrompts(server);

async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("Doc Assistant MCP Server 已启动");
}

main().catch(console.error);

6.3 工具模块示例 src/tools/search.ts

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

const DOCS_DIR = path.resolve(process.env.DOCS_DIR || "./docs");

export function registerSearchTool(server: McpServer) {
  server.tool(
    "search_docs",
    "在文档目录中全文搜索关键词",
    {
      query: z.string().describe("搜索关键词"),
      case_sensitive: z.boolean().default(false).describe("是否区分大小写"),
      max_results: z.number().default(10).describe("最大返回结果数"),
    },
    async ({ query, case_sensitive, max_results }) => {
      const results: Array<{
        file: string;
        line: number;
        content: string;
        score: number;
      }> = [];

      async function scanFile(filePath: string) {
        const content = await fs.readFile(filePath, "utf-8");
        const lines = content.split("\n");
        const q = case_sensitive ? query : query.toLowerCase();

        for (let i = 0; i < lines.length; i++) {
          const line = lines[i];
          const target = case_sensitive ? line : line.toLowerCase();

          if (target.includes(q)) {
            // 简单评分：标题行权重更高
            const isHeading = line.trimStart().startsWith("#");
            const score = isHeading ? 2 : 1;
            results.push({
              file: path.relative(DOCS_DIR, filePath),
              line: i + 1,
              content: line.trim(),
              score,
            });
          }
        }
      }

      async function walkDir(dir: string) {
        const entries = await fs.readdir(dir, { withFileTypes: true });
        for (const entry of entries) {
          const fullPath = path.join(dir, entry.name);
          if (entry.isDirectory()) {
            await walkDir(fullPath);
          } else if (entry.name.endsWith(".md") || entry.name.endsWith(".txt")) {
            await scanFile(fullPath);
          }
        }
      }

      try {
        await walkDir(DOCS_DIR);
        results.sort((a, b) => b.score - a.score);
        const top = results.slice(0, max_results);

        const output = top.length > 0
          ? top.map(r => `📄 ${r.file}:${r.line}\n   ${r.content}`).join("\n\n")
          : `未找到包含"${query}"的文档`;

        return {
          content: [{
            type: "text",
            text: `搜索"${query}"的结果（${top.length}/${results.length}）：\n\n${output}`,
          }],
        };
      } catch (error: any) {
        return {
          content: [{ type: "text", text: `搜索失败: ${error.message}` }],
          isError: true,
        };
      }
    }
  );
}

6.4 项目配置文件 .mcp.json

{
  "mcpServers": {
    "doc-assistant": {
      "type": "stdio",
      "command": "node",
      "args": ["./build/index.js"],
      "env": {
        "DOCS_DIR": "./docs",
        "NODE_ENV": "development"
      }
    }
  }
}

6.5 测试全流程

# 构建
npm run build

# 在项目目录中启动Claude Code
claude

# 测试对话
> 搜索文档中所有提到"API"的内容
> 搜索"authentication"相关文档，不区分大小写
> 读取文档目录的结构
> 使用code_review模板审查src/tools/search.ts

第七章：TypeScript与Python方案对比

7.1 功能特性对比

特性 TypeScript SDK Python FastMCP
Schema定义 Zod（手动） 类型注解（自动）
描述来源 手写字符串 docstring自动提取
资源模板 ResourceTemplate类 @mcp.resource 装饰器
提示模板 server.prompt() @mcp.prompt 装饰器
HTTP模式 需Express等框架 内置，改一行代码
调试工具 日志/手动测试 fastmcp dev Inspector
包大小 较大（Node依赖） 较小
启动速度 需编译 即时运行

7.2 选型建议

graph TD
    A[选择MCP开发语言] --> B{项目技术栈?}
    B -->|TypeScript/Node.js| C[TypeScript SDK]
    B -->|Python生态| D[FastMCP]
    B -->|不确定| E{主要需求?}

    E -->|快速原型/API对接| D
    E -->|生产级/复杂逻辑| C
    E -->|数据处理/ML| D
    E -->|前端工具链集成| C

    C --> F[优势: 类型安全<br/>IDE支持好<br/>生态成熟]
    D --> G[优势: 开发速度快<br/>数据科学生态<br/>代码量少30%+]

第八章：常见问题与排障

8.1 MCP服务器无法连接

# 1. 确认服务器可独立运行
node build/index.js  # 应无报错

# 2. 检查Claude Code配置
claude mcp list
claude mcp get <server-name>

# 3. 查看Claude Code日志
claude --mcp-debug  # 启用MCP调试模式

# 4. 常见原因
# - 路径不是绝对路径（.mcp.json中的command路径需要绝对路径）
# - Python虚拟环境未激活
# - 端口被占用（HTTP模式）

8.2 stdio模式输出异常

症状：Claude Code报MCP协议解析错误。

原因：服务器代码中有 console.log / print 污染了stdout。

解决： - TypeScript：所有日志用 console.error() 或写文件 - Python：所有日志用 sys.stderr.write() 或 logging 模块

import sys
import logging

# 配置日志到stderr
logging.basicConfig(
    stream=sys.stderr,
    level=logging.DEBUG,
    format="%(asctime)s %(levelname)s %(message)s"
)
logger = logging.getLogger("my-mcp-server")

# ✅ 正确
logger.info("这条消息不会干扰MCP通信")

# ❌ 错误
print("这会破坏MCP协议！")

8.3 工具未被调用

原因：工具描述不够清晰，LLM无法判断何时使用。

解决：写好工具描述，包含触发场景：

// ❌ 模糊描述
server.tool("search", "搜索", { /* ... */ }, handler);

// ✅ 清晰描述
server.tool(
  "search_docs",
  "在项目文档中全文搜索关键词。当用户问及文档内容、查找特定主题、或需要了解项目某个方面时使用此工具。",
  { query: z.string().describe("要搜索的关键词或短语") },
  handler
);

8.4 HTTP模式CORS问题

import cors from "cors";

app.use(cors({
  origin: ["https://claude.ai", "http://localhost:*"],
  methods: ["GET", "POST", "DELETE"],
  allowedHeaders: ["Content-Type"],
}));

延伸阅读

训练营适用性评估

维度 评估
适合训练营类型 AI工程实战营、开发者工具训练营、LLM应用开发训练营
教学时长 6-8小时（含动手实验）
难度等级 中级（需编程基础）
实验环境要求 每人一台开发机，预装Node.js 20+和Python 3.10+
建议分组 2-3人一组，每组选择TS或Python方案完成一个MCP服务器
产出要求 每组提交一个可运行的MCP服务器，至少包含3个Tools + 1个Resource + 1个Prompt

参考来源

  1. Build an MCP server - Model Context Protocol — MCP官方构建教程，TypeScript/Python双语言入门指南
  2. Connect Claude Code to tools via MCP — Claude Code官方MCP集成文档，配置方式与传输类型说明
  3. MCP Developer Guide 2026 - LushBinary — 2026年MCP开发者完整指南，覆盖架构、部署与安全
  4. FastMCP Python Guide 2026 - danilchenko.dev — FastMCP 3.2生产级服务器构建教程
  5. Understanding MCP Server Transports - DEV Community — stdio/SSE/HTTP Streamable三种传输方式对比解析

[!tip] 51学通信知识星球 星主：爱卫生（微信：gprshome201101）——公众号「51学通信」创办者，19 年通信行业运维与讲师经验，著有《GPRS网络信令实例详解》《EPC原理与实践》《5G核心网原理与实践》三本纸质书（各大电商平台有售）。 星球定位：深耕 5G 核心网、IMS、NFV/AI、云原生等 ICT 领域，主打运维岗位与信令分析。 当前规模：5200 篇主题 / 990 个回答，198 元/年。 加入福利：加入星球后请加爱卫生微信，拉您进知识星球年度陪伴与答疑群——不限量回复您的各种问题，另有群内分享。 AI 加持：51学通信知识星球已作为 skill 接入 WorkBuddy，使用 deepseek-4-flash 在海量星球帖子中精准找贴、找视频、找 FAQ 和案例，做总结、找文档和 PPT 等场景全覆盖。

51学通信知识星球海报-扫码加入

📎 本分类最新文章