MCP 自定义服务器开发进阶指南 —— 错误处理、流式输出、TypeScript 与部署

整理日期： 2026-06-04
话题来源： 聊天记录延伸（MCP 系列教程进阶篇）

---

简介

在掌握了基础的 FastMCP 服务器开发后（见《MCP 自定义服务器开发入门指南》），本教程将带你进入 生产级 MCP 服务器开发 的世界。你将学习：

• 高级错误处理与参数校验（Pydantic 集成）
• 流式输出与进度报告（Streaming）
• 使用 TypeScript SDK 构建 MCP 服务器
• 上下文注入（Context Injection）
• 动态工具注册
• Docker 部署与 CI/CD
• 性能优化与负载测试

前置要求

要求	说明
已阅读 Part 1（入门篇）	了解 FastMCP 基本概念
Python >= 3.11	本教程部分示例需要 Python
Node.js >= 18	TypeScript SDK 示例需要
Docker（可选）	部署部分用到

---

一、高级错误处理模式

1.1 自定义异常与错误分类

生产级 MCP 服务器应该对错误进行精细化管理，而不是统统返回 "出错啦"。

# errors.py
from typing import Optional

class MCPToolError(Exception):
    """MCP 工具错误的基类"""
    def __init__(self, message: str, code: str, details: Optional[dict] = None):
        self.message = message
        self.code = code
        self.details = details or {}
        super().__init__(self.message)

class ValidationError(MCPToolError):
    """参数校验失败"""
    def __init__(self, field: str, reason: str):
        super().__init__(
            message=f"参数 '{field}' 校验失败：{reason}",
            code="VALIDATION_ERROR",
            details={"field": field, "reason": reason}
        )

class NotFoundError(MCPToolError):
    """资源未找到"""
    def __init__(self, resource_type: str, identifier: str):
        super().__init__(
            message=f"未找到 {resource_type}：'{identifier}'",
            code="NOT_FOUND",
            details={"resource_type": resource_type, "identifier": identifier}
        )

class RateLimitError(MCPToolError):
    """速率限制"""
    def __init__(self, retry_after: int):
        super().__init__(
            message=f"请求过于频繁，请在 {retry_after} 秒后重试",
            code="RATE_LIMITED",
            details={"retry_after_seconds": retry_after}
        )

class ExternalServiceError(MCPToolError):
    """外部服务调用失败"""
    def __init__(self, service: str, status_code: int):
        super().__init__(
            message=f"外部服务 '{service}' 返回错误状态码 {status_code}",
            code="EXTERNAL_SERVICE_ERROR",
            details={"service": service, "status_code": status_code}
        )

1.2 使用错误处理装饰器

# safe_tools.py
import functools
import traceback
import sys
from typing import Callable, Any
from errors import MCPToolError

def safe_tool(func: Callable) -> Callable:
    """安全执行工具函数，统一处理异常"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs) -> str:
        try:
            result = func(*args, **kwargs)
            return result
        except MCPToolError as e:
            # 已知的 MCP 工具错误，返回友好的错误信息
            return f"❌ [{e.code}] {e.message}"
        except Exception as e:
            # 未知错误，记录详细堆栈但返回简化信息
            print(f"[ERROR] Tool '{func.__name__}' failed: {traceback.format_exc()}", 
                  file=sys.stderr)
            return f"❌ [INTERNAL_ERROR] 工具执行出错：{str(e)}"
    return wrapper


# 使用示例
from mcp.server.fastmcp import FastMCP
from errors import ValidationError, NotFoundError

mcp = FastMCP("SafeServer")

@mcp.tool()
@safe_tool
def get_user(user_id: int) -> str:
    """获取用户信息（带安全处理的版本）"""
    if user_id <= 0:
        raise ValidationError("user_id", "必须为正整数")
    
    # 模拟数据库查询
    users = {1: "Alice", 2: "Bob", 3: "Charlie"}
    if user_id not in users:
        raise NotFoundError("用户", str(user_id))
    
    return f"用户信息：ID={user_id}, 名称={users[user_id]}"

1.3 集成 Pydantic 参数校验

# pydantic_tools.py
from pydantic import BaseModel, Field, EmailStr, validator
from typing import Optional
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("PydanticServer")

class UserCreateInput(BaseModel):
    """用户创建参数模型"""
    username: str = Field(..., min_length=3, max_length=50, description="用户名")
    email: str = Field(..., description="邮箱地址")
    age: int = Field(ge=0, le=150, description="年龄")
    role: str = Field(default="user", description="角色")
    
    @validator("username")
    def username_alphanumeric(cls, v):
        if not v.isalnum():
            raise ValueError("用户名只能包含字母和数字")
        return v
    
    @validator("email")
    def email_valid(cls, v):
        if "@" not in v or "." not in v.split("@")[-1]:
            raise ValueError("邮箱格式不正确")
        return v

@mcp.tool()
def create_user(
    username: str,
    email: str,
    age: int,
    role: str = "user"
) -> str:
    """创建新用户（带 Pydantic 校验）
    
    Args:
        username: 用户名，3-50 个字母数字字符
        email: 有效的邮箱地址
        age: 年龄，0-150 岁
        role: 用户角色，默认为 user
    """
    try:
        data = UserCreateInput(
            username=username,
            email=email,
            age=age,
            role=role
        )
        return f"✅ 用户创建成功：{data.username} ({data.email})"
    except Exception as e:
        return f"❌ 参数校验失败：{e}"

---

二、流式输出与进度报告

对于耗时较长的操作（如批量处理、文件下载、数据迁移），应该提供进度反馈。

2.1 使用异步生成器实现流式输出

# streaming_tools.py
import asyncio
from typing import AsyncGenerator
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("StreamingServer")

@mcp.tool()
async def batch_process(items: list[str]) -> AsyncGenerator[str, None]:
    """批量处理项目（带进度流）
    
    Args:
        items: 要处理的项目列表
    """
    total = len(items)
    yield f"开始批量处理 {total} 个项目...\n\n"
    
    for i, item in enumerate(items, 1):
        # 模拟处理耗时
        await asyncio.sleep(0.5)
        progress = int(i / total * 100)
        yield f"📦 [{progress:3d}%] 处理中：{item}\n"
    
    yield f"\n✅ 全部完成！共处理 {total} 个项目。"


@mcp.tool()
async def download_with_progress(url: str) -> AsyncGenerator[str, None]:
    """模拟下载文件（显示进度）
    
    Args:
        url: 下载地址
    """
    yield f"🔄 开始下载：{url}\n\n"
    
    total_size = 1024 * 1024 * 10  # 10MB
    chunk_size = 1024 * 256  # 256KB
    chunks = total_size // chunk_size
    
    for i in range(chunks):
        await asyncio.sleep(0.05)  # 模拟下载延迟
        downloaded = (i + 1) * chunk_size
        percent = int((i + 1) / chunks * 100)
        
        # 进度条
        bar = "█" * (percent // 5) + "░" * (20 - percent // 5)
        yield f"\r📥 {bar} {percent}% ({downloaded / 1024 / 1024:.1f}MB / {total_size / 1024 / 1024:.1f}MB)"
    
    yield f"\n\n✅ 下载完成！已保存到 /tmp/{url.split('/')[-1]}"

2.2 长时间运行任务的超时控制

import asyncio
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("TimeoutServer")

@mcp.tool()
async def long_running_task(duration: int = 10) -> str:
    """长时间运行的任务（带超时控制）
    
    Args:
        duration: 任务持续秒数（建议不超过 30 秒）
    """
    TIMEOUT = 60  # 最大允许执行时间
    
    try:
        async with asyncio.timeout(TIMEOUT):
            result = 0
            for i in range(duration):
                await asyncio.sleep(1)
                result += i
                print(f"[进度] 第 {i+1}/{duration} 秒，当前结果：{result}", file=__import__('sys').stderr)
            return f"✅ 执行完成！计算结果：{result}"
    except asyncio.TimeoutError:
        return f"❌ 执行超时（超过 {TIMEOUT} 秒）"

---

三、使用 TypeScript SDK 构建 MCP 服务器

除了 Python，MCP 官方也提供了 TypeScript SDK。对于 Node.js 生态圈的开发者来说，TypeScript 版本更加自然。

3.1 环境搭建

# 创建项目
mkdir my-ts-mcp-server && cd my-ts-mcp-server
npm init -y

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

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

mkdir src

3.2 基础服务器（Hello World）

// src/server.ts
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";
import { z } from "zod";

// 1. 创建服务器实例
const server = new Server(
  {
    name: "my-ts-server",
    version: "1.0.0",
  },
  {
    capabilities: {
      tools: {},
    },
  }
);

// 2. 定义工具的模式（使用 Zod 进行参数校验）
const GreetSchema = z.object({
  name: z.string().min(1).describe("要问候的人名"),
  language: z.enum(["zh", "en"]).optional().default("zh").describe("语言"),
});

// 3. 注册工具列表处理器
server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [
    {
      name: "greet",
      description: "向用户问好",
      inputSchema: {
        type: "object",
        properties: {
          name: { type: "string", description: "要问候的人名" },
          language: {
            type: "string",
            enum: ["zh", "en"],
            description: "语言",
            default: "zh",
          },
        },
        required: ["name"],
      },
    },
    {
      name: "calculate",
      description: "执行数学计算",
      inputSchema: {
        type: "object",
        properties: {
          a: { type: "number", description: "第一个数字" },
          b: { type: "number", description: "第二个数字" },
          operation: {
            type: "string",
            enum: ["add", "subtract", "multiply", "divide"],
            description: "计算操作",
          },
        },
        required: ["a", "b", "operation"],
      },
    },
  ],
}));

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

  switch (name) {
    case "greet": {
      const { name: personName, language } = GreetSchema.parse(args);
      const greeting = language === "zh" ? `你好，${personName}！` : `Hello, ${personName}!`;
      return {
        content: [{ type: "text", text: greeting }],
      };
    }

    case "calculate": {
      const { a, b, operation } = args as { a: number; b: number; operation: string };
      let result: number;

      switch (operation) {
        case "add":
          result = a + b;
          break;
        case "subtract":
          result = a - b;
          break;
        case "multiply":
          result = a * b;
          break;
        case "divide":
          if (b === 0) {
            return {
              isError: true,
              content: [{ type: "text", text: "错误：除数不能为零" }],
            };
          }
          result = a / b;
          break;
        default:
          return {
            isError: true,
            content: [{ type: "text", text: `不支持的运算：${operation}` }],
          };
      }

      return {
        content: [{ type: "text", text: `${a} ${operation} ${b} = ${result}` }],
      };
    }

    default:
      return {
        isError: true,
        content: [{ type: "text", text: `未知工具：${name}` }],
      };
  }
});

// 5. 启动服务器
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("✅ TypeScript MCP 服务器已启动");
}

main().catch((error) => {
  console.error("服务器启动失败:", error);
  process.exit(1);
});

3.3 编译与运行

# 编译 TypeScript
npx tsc

# 运行（stdio 模式）
node dist/server.js

# 在 Claude Code 中使用
claude mcp add my-ts-server -s user -- node /path/to/dist/server.js

3.4 TypeScript vs Python 对比

维度	Python FastMCP	TypeScript SDK
定义工具	@mcp.tool() 装饰器	手动注册 ListToolsRequestSchema
参数校验	类型注解自动推断	Zod 或手动定义 JSON Schema
代码量	少（约 1/3）	较多
类型安全	运行时 + mypy	编译时（严格）
性能	中等	较高
Node 生态	❌	✅ 可调用 npm 包
学习曲线	低	中

建议：团队有 Python 背景用 FastMCP，有 Node.js/TypeScript 背景用 TS SDK。功能上没有本质差异。

---

四、上下文注入（Context）

在某些场景下，工具需要访问请求上下文信息，比如客户端 ID、会话 token、请求时间等。

4.1 Python FastMCP 上下文

# context_tools.py
from mcp.server.fastmcp import FastMCP, Context
from typing import Optional

mcp = FastMCP("ContextServer")

@mcp.tool()
async def contextual_greet(name: str, ctx: Context) -> str:
    """带上下文的问候工具
    
    Args:
        name: 要问候的人名
        ctx: 请求上下文（自动注入，AI 不需要传入）
    """
    # 获取客户端信息
    client_id = ctx.client_id if hasattr(ctx, 'client_id') else 'unknown'
    request_id = ctx.request_id if hasattr(ctx, 'request_id') else 'unknown'
    
    # 使用上下文记录日志
    print(f"[INFO] 请求 {request_id} 来自 {client_id}：greet({name})", 
          file=__import__('sys').stderr)
    
    return f"你好，{name}！（请求 ID：{request_id}）"


@mcp.tool()
async def rate_limited_tool(ctx: Context) -> str:
    """受速率限制的工具"""
    # 从上下文中获取速率限制信息
    remaining = getattr(ctx, 'rate_limit_remaining', None)
    
    if remaining is not None and remaining <= 0:
        return "❌ 已达到速率限制，请稍后再试"
    
    return f"✅ 执行成功，剩余调用次数：{remaining if remaining else '无限制'}"

4.2 TypeScript Context

// 在 TypeScript 中，上下文通过 request handler 的第二个参数传递
server.setRequestHandler(CallToolRequestSchema, async (request, extra) => {
  // extra 包含连接信息
  console.error(`收到请求: ${request.params.name}`);
  
  // 处理请求...
});

---

五、动态工具注册

有些场景需要根据配置或数据库内容动态注册工具，而不是硬编码。

5.1 基于配置动态加载

# dynamic_tools.py
import json
from pathlib import Path
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("DynamicServer")

# 从配置文件动态加载工具定义
TOOLS_CONFIG_PATH = Path(__file__).parent / "tools_config.json"

def load_tool_definitions():
    """从配置文件加载工具定义"""
    if not TOOLS_CONFIG_PATH.exists():
        return []
    with open(TOOLS_CONFIG_PATH, "r", encoding="utf-8") as f:
        return json.load(f)

# 根据配置动态创建工具函数
for tool_def in load_tool_definitions():
    tool_name = tool_def["name"]
    tool_desc = tool_def["description"]
    
    # 动态创建闭包函数
    def make_tool_fn(endpoint: str, method: str):
        import requests
        def tool_fn(**kwargs) -> str:
            try:
                if method.upper() == "GET":
                    resp = requests.get(endpoint, params=kwargs, timeout=10)
                else:
                    resp = requests.post(endpoint, json=kwargs, timeout=10)
                return f"响应状态码：{resp.status_code}\n响应内容：{resp.text[:1000]}"
            except Exception as e:
                return f"请求失败：{e}"
        return tool_fn
    
    fn = make_tool_fn(tool_def["endpoint"], tool_def["method"])
    fn.__name__ = tool_name
    fn.__doc__ = tool_desc
    
    # 注册到 MCP 服务器
    mcp.tool()(fn)
    print(f"[动态注册] 工具 '{tool_name}' 已加载")

5.2 工具配置文件示例

// tools_config.json
[
  {
    "name": "get_server_status",
    "description": "获取服务器运行状态",
    "endpoint": "http://localhost:8080/api/status",
    "method": "GET"
  },
  {
    "name": "report_metrics",
    "description": "上报自定义指标到监控系统",
    "endpoint": "http://localhost:9090/api/metrics",
    "method": "POST"
  },
  {
    "name": "query_logs",
    "description": "查询最近的应用日志",
    "endpoint": "http://localhost:5601/api/logs/search",
    "method": "GET"
  }
]

---

六、Docker 部署与生产化

6.1 Dockerfile

# Dockerfile
FROM python:3.12-slim

WORKDIR /app

# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制代码
COPY . .

# 暴露 HTTP 端口（如果使用 HTTP 传输）
EXPOSE 8000

# 默认使用 stdio 传输（适合 Claude Code 本地调用）
# 如需 HTTP 传输，将 CMD 改为 HTTP 模式
CMD ["python", "server.py"]

# requirements.txt
mcp>=1.0.0
uvicorn>=0.27.0
httpx>=0.27.0
pydantic>=2.0.0

6.2 Docker Compose

# docker-compose.yml
version: "3.8"

services:
  mcp-server:
    build: .
    ports:
      - "8000:8000"
    volumes:
      # 挂载数据目录
      - ./data:/app/data
      # 挂载配置
      - ./config:/app/config
    environment:
      - MCP_TRANSPORT=http
      - MCP_HOST=0.0.0.0
      - MCP_PORT=8000
      - LOG_LEVEL=info
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

6.3 使用 Claude Code 连接 Docker 版 MCP

# stdio 模式（Docker 运行在本地）
claude mcp add docker-mcp -s user \
  -e MCP_ENV=production \
  -- docker run -i --rm \
    -e MCP_ENV \
    -v /host/data:/app/data \
    my-mcp-server:latest

# HTTP 模式（Docker 已运行在服务器上）
claude mcp add remote-mcp --transport http \
  -- http://your-server:8000/mcp

6.4 CI/CD 配置（GitHub Actions）

# .github/workflows/deploy-mcp.yml
name: Deploy MCP Server

on:
  push:
    branches: [main]
    paths:
      - "server/**"
      - "Dockerfile"

jobs:
  build-and-deploy:
    runs-on: ubuntu-latest
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      
      - name: Run tests
        run: |
          pip install -r requirements.txt
          pip install pytest
          python -m pytest tests/
      
      - name: Build Docker image
        run: |
          docker build -t mcp-server:${{ github.sha }} .
      
      - name: Deploy to server
        run: |
          # 示例：通过 SSH 部署
          docker save mcp-server:${{ github.sha }} | ssh user@server "docker load && docker stop mcp-server && docker run -d --rm mcp-server:${{ github.sha }}"

---

七、性能优化

7.1 缓存策略

# cache_tools.py
import time
import hashlib
import json
from functools import lru_cache
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("CacheServer")

# 简单的内存缓存
_cache = {}
_cache_ttl = {}

def cached(ttl_seconds: int = 300):
    """缓存装饰器：缓存工具执行结果"""
    def decorator(func):
        def wrapper(*args, **kwargs):
            # 生成缓存键
            key = hashlib.md5(
                f"{func.__name__}:{args}:{json.dumps(kwargs, sort_keys=True)}".encode()
            ).hexdigest()
            
            now = time.time()
            
            # 检查缓存是否有效
            if key in _cache and key in _cache_ttl:
                if now - _cache_ttl[key] < ttl_seconds:
                    return _cache[key] + " [缓存]"
            
            # 执行原函数
            result = func(*args, **kwargs)
            
            # 写入缓存
            _cache[key] = result
            _cache_ttl[key] = now
            
            return result
        return wrapper
    return decorator

@mcp.tool()
@cached(ttl_seconds=60)
def get_expensive_data(query: str) -> str:
    """获取昂贵数据（结果缓存 60 秒）
    
    Args:
        query: 查询条件
    """
    # 模拟耗时操作
    time.sleep(2)
    return f"查询 '{query}' 的结果：{hash(query) % 10000}"

7.2 连接池与异步批量

import aiohttp
import asyncio
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("PoolServer")

# 全局连接池
_connector = aiohttp.TCPConnector(limit=10, ttl_dns_cache=300)
_session = aiohttp.ClientSession(connector=_connector)

@mcp.tool()
async def batch_fetch(urls: list[str]) -> str:
    """批量并发获取多个 URL 的内容
    
    Args:
        urls: URL 列表，最多 10 个
    """
    if len(urls) > 10:
        return "❌ 最多同时请求 10 个 URL"
    
    async def fetch_one(url: str) -> dict:
        try:
            async with _session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
                text = await resp.text()
                return {"url": url, "status": resp.status, "length": len(text)}
        except Exception as e:
            return {"url": url, "error": str(e)}
    
    results = await asyncio.gather(*[fetch_one(url) for url in urls])
    
    output = "## 批量获取结果\n\n"
    output += "| URL | 状态 | 大小/错误 |\n"
    output += "|-----|------|----------|\n"
    for r in results:
        if "error" in r:
            output += f"| {r['url']} | ❌ | {r['error']} |\n"
        else:
            output += f"| {r['url']} | {r['status']} | {r['length']} 字符 |\n"
    
    return output

# 应用退出时关闭连接池
import atexit
atexit.register(lambda: asyncio.run(_session.close()))

7.3 性能对比

优化技术	效果	适用场景
LRU 缓存	减少 10-100x 重复计算	查询结果、API 响应
连接池	减少 5-10x 连接建立开销	HTTP 请求密集
异步批量	并发 N 倍提升	批量数据获取
延迟加载	减少初始启动时间	大型服务器
资源懒初始化	减少内存占用	工具众多

---

八、测试

8.1 Python 单元测试

# tests/test_server.py
import pytest
from mcp.server.fastmcp import FastMCP

# 导入你的服务器
from server import mcp

@pytest.mark.asyncio
async def test_greet_tool():
    """测试 greet 工具"""
    result = await mcp.call_tool("greet", {"name": "测试"})
    assert "你好" in result.content[0].text
    assert "测试" in result.content[0].text

@pytest.mark.asyncio
async def test_add_tool():
    """测试 add 工具"""
    result = await mcp.call_tool("add", {"a": 3, "b": 5})
    assert "8" in result.content[0].text

@pytest.mark.asyncio
async def test_tool_not_found():
    """测试不存在的工具"""
    with pytest.raises(Exception):
        await mcp.call_tool("non_existent", {})

@pytest.mark.asyncio
async def test_invalid_params():
    """测试无效参数"""
    result = await mcp.call_tool("greet", {})  # 缺少必填参数
    assert "错误" in result.content[0].text or "参数" in result.content[0].text

8.2 端到端测试

# tests/test_e2e.py
import subprocess
import json
import time
import pytest

@pytest.fixture
def mcp_server_process():
    """启动 MCP 服务器子进程"""
    proc = subprocess.Popen(
        ["python", "server.py"],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True
    )
    time.sleep(1)  # 等待启动
    yield proc
    proc.terminate()
    proc.wait()

def send_mcp_request(proc, method: str, params: dict, request_id: int = 1) -> dict:
    """发送 MCP JSON-RPC 请求"""
    request = {
        "jsonrpc": "2.0",
        "method": method,
        "params": params,
        "id": request_id
    }
    proc.stdin.write(json.dumps(request) + "\n")
    proc.stdin.flush()
    
    response = proc.stdout.readline()
    return json.loads(response)

def test_list_tools(mcp_server_process):
    """端到端测试：列出工具"""
    response = send_mcp_request(mcp_server_process, "tools/list", {})
    assert "result" in response
    assert "tools" in response["result"]

def test_call_tool(mcp_server_process):
    """端到端测试：调用工具"""
    response = send_mcp_request(
        mcp_server_process,
        "tools/call",
        {"name": "greet", "arguments": {"name": "MCP"}}
    )
    assert "result" in response
    assert "你好" in response["result"]["content"][0]["text"]

---

九、常见问题

Q1：启动时报 ModuleNotFoundError: No module named 'mcp'

原因：虚拟环境未正确激活，或依赖未安装。

解决：

source .venv/bin/activate  # 激活虚拟环境
pip install mcp             # 安装依赖
which python                # 确认使用的是 venv 中的 Python

Q2：Claude Code 显示 disconnected

原因：MCP 服务器启动失败或中途崩溃。

排查：

# 手动启动服务器，查看错误信息
python server.py

# 检查配置文件路径是否正确
cat .mcp.json | python -m json.tool

Q3：工具返回 JSON serialization error

原因：返回值包含不可 JSON 序列化的类型（如 datetime、Decimal、自定义对象）。

解决：确保返回 str、dict、list、int、float、bool 或 None 的组合。

import json
from datetime import datetime

# ❌ 错误
@mcp.tool()
def get_time():
    return {"now": datetime.now()}  # datetime 不可序列化

# ✅ 正确
@mcp.tool()
def get_time():
    return {"now": datetime.now().isoformat()}  # 转为字符串

Q4：HTTP 传输模式下连接不上

原因：防火墙、端口占用或 CORS 问题。

解决：

# 检查端口是否被占用
netstat -tlnp | grep 8000

# 测试 HTTP 端点
curl http://localhost:8000/mcp

# 确保防火墙允许端口
sudo ufw allow 8000

Q5：工具执行超时

原因：默认超时设置过短。

解决：

• 在 .mcp.json 或 Hermes 配置中增加 timeout
• 对于长时间任务，使用异步生成器提供进度反馈
• 将长时间任务拆分为多个小工具

---

十、总结

经过入门和进阶两篇教程的学习，你已经掌握了从零构建生产级 MCP 服务器的全部技能：

能力	入门篇	进阶篇
FastMCP 基础	✅	✅
Tools / Resources / Prompts	✅	✅
错误处理	-	✅ 分层异常 + Pydantic
流式输出	-	✅ AsyncGenerator
TypeScript SDK	-	✅
上下文注入	-	✅
动态工具注册	-	✅
Docker 部署	-	✅
性能优化	-	✅ 缓存 + 连接池
测试	-	✅ 单元 + E2E
CI/CD	-	✅ GitHub Actions

推荐的学习路径

1. 入门篇 → 理解 MCP 服务器基本概念
2. 本教程（进阶篇）→ 学习生产化技能
3. 阅读官方 SDK 文档 → 深入理解底层
4. 阅读热门 MCP 服务器源码 → 学习最佳实践

资源链接

• MCP Python SDK: https://github.com/modelcontextprotocol/python-sdk
• MCP TypeScript SDK: https://github.com/modelcontextprotocol/typescript-sdk
• MCP 官方文档: https://modelcontextprotocol.io
• MCP 官方服务器参考: https://github.com/modelcontextprotocol/servers
• MCP 服务器注册中心: https://mcp.so

---

本文基于 MCP SDK 1.x 版本编写，部分 API 可能随版本更新变化。