首页 › tutorials › MCP服务器开发：双语言实战与生产部署

Claude Code MCP服务器开发：Python与TypeScript双语言实战

概述

本教程面向希望为Claude Code构建自定义MCP服务器的中高级开发者。你将从零理解MCP（Model Context Protocol）协议架构，掌握Python（FastMCP）和TypeScript（官方SDK）两种开发路径，完成工具（Tools）、资源（Resources）、提示（Prompts）三大原语的实战开发，并学会调试、安全加固和生产部署的完整流程。学完后，你将具备为团队或社区构建生产级MCP服务器的能力。

适用人群：有Python或TypeScript基础的开发者、希望扩展Claude Code能力的AI应用工程师、需要将内部API暴露给LLM的企业开发者。

学习目标： - 深入理解MCP协议架构与通信机制 - 分别用FastMCP（Python）和TypeScript SDK构建完整的MCP服务器 - 掌握Tool、Resource、Prompt三大原语的设计与实现 - 学会使用MCP Inspector进行调试和测试 - 完成安全加固与生产环境部署

前置条件

项目 要求
Python 3.10+（FastMCP要求）
Node.js 18+（TypeScript SDK要求）
Claude Code 最新版本（claude --version确认）
基础知识 JSON Schema、REST API概念、命令行操作
操作系统 macOS / Linux / Windows（WSL2）

第一章：MCP协议架构深度解析

1.1 什么是MCP

Model Context Protocol（MCP）是Anthropic提出的开放协议，标准化了LLM应用与外部工具、数据源之间的通信方式。其核心理念可以用一句话概括：为AI提供一个通用的"插头"标准，让它能连接任何"插座"。

MCP采用经典的客户端-服务器架构，但在LLM场景中引入了三层模型：

graph TB
    subgraph Host["Host 宿主应用"]
        C1["Client 1"]
        C2["Client 2"]
        C3["Client N"]
    end

    subgraph Servers["MCP Servers"]
        S1["Server A<br/>文件系统"]
        S2["Server B<br/>数据库"]
        S3["Server C<br/>API网关"]
    end

    C1 <-->|"stdio/SSE/HTTP"| S1
    C2 <-->|"stdio/SSE/HTTP"| S2
    C3 <-->|"stdio/SSE/HTTP"| S3

    Host -->|"用户交互"| User["用户"]
    S1 -->|"读写"| FS["本地文件"]
    S2 -->|"查询"| DB["PostgreSQL"]
    S3 -->|"调用"| API["REST API"]

    style Host fill:#4A90D9,color:#fff
    style Servers fill:#50C878,color:#fff
    style User fill:#FFD700,color:#333

三层模型详解：

1.2 三大核心原语

MCP定义了三种原语（Primitive），每种服务于不同的交互模式：

原语 类比 方向 典型用途
Tools（工具） 函数调用 LLM → Server 执行操作、查询数据、调用API
Resources（资源） 文件读取 应用 → Server 提供上下文数据、文档、配置
Prompts（提示模板） 模板引擎 用户 → Server 预设的提示词模板、工作流引导

设计原则： - Tools是"动词"——让LLM能执行操作 - Resources是"名词"——让LLM能获取数据 - Prompts是"句式"——让用户能复用交互模式

1.3 传输协议演进

MCP的传输层经历了重要演进：

timeline
    title MCP传输协议演进
    2024-11 : stdio传输\n本地进程间通信\n stdin/stdout管道
    2025-03 : SSE传输\nHTTP + Server-Sent Events\n支持远程连接
    2025-11 : Streamable HTTP\n统一传输层\n支持OAuth 2.1\n异步任务

对于Claude Code本地使用，stdio是默认且推荐的方式。

1.4 消息流示例

一次完整的Tool调用流程：

1. Claude Code (Host) ──── initialize ────────────> MCP Server
2. MCP Server ──── capabilities + tool list ──────> Claude Code
3. Claude Code (LLM决定调用工具)
4. Claude Code ──── tools/call {name, args} ──────> MCP Server
5. MCP Server ──── 执行操作，返回结果 ──────────> Claude Code
6. Claude Code (LLM基于结果继续推理)

第二章：环境搭建与项目初始化

2.1 创建项目目录结构

# 创建项目根目录
mkdir mcp-dev-workshop && cd mcp-dev-workshop

# Python项目
mkdir python-server && cd python-server
python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate

# TypeScript项目（新终端）
mkdir ts-server && cd ts-server
npm init -y

2.2 安装依赖

Python（FastMCP）：

cd python-server
pip install fastmcp httpx
# FastMCP从3.0开始是官方推荐的Python MCP框架
# 验证安装
python -c "import fastmcp; print(fastmcp.__version__)"

TypeScript（官方SDK）：

cd ts-server
npm install @modelcontextprotocol/sdk zota
npm install -D typescript @types/node tsx
# 验证安装
npx tsc --version

2.3 项目配置文件

Python pyproject.toml：

[project]
name = "my-mcp-server"
version = "0.1.0"
requires-python = ">=3.10"
dependencies = [
    "fastmcp>=3.0",
    "httpx>=0.27",
]

[project.scripts]
my-mcp = "server:main"

TypeScript tsconfig.json：

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "Node16",
    "moduleResolution": "Node16",
    "outDir": "./dist",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true
  },
  "include": ["src/**/*"]
}

第三章：Python FastMCP实战

3.1 第一个MCP Server——5行代码起步

创建 python-server/server.py：

from fastmcp import FastMCP

# 创建服务器实例
mcp = FastMCP("demo-server")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers together."""
    return a + b

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

运行测试：

# 直接运行（stdio模式，默认）
fastmcp dev server.py

# 或者
python server.py

预期结果：服务器启动，等待通过stdin接收JSON-RPC消息。fastmcp dev命令会自动启动MCP Inspector进行可视化调试。

3.2 完整实战：DevOps工具箱服务器

我们来构建一个实用的DevOps工具箱，包含服务器健康检查、日志分析和部署状态查询功能。

创建 python-server/devops_server.py：

import json
import subprocess
from datetime import datetime
from typing import Optional
from fastmcp import FastMCP

mcp = FastMCP(
    "devops-toolkit",
    instructions="""DevOps工具箱MCP服务器。
    提供服务器健康检查、日志分析和部署管理功能。
    所有操作均在本地执行，请确保有相应的系统权限。"""
)

# ============================================================
# 模拟数据存储（生产环境替换为真实数据库/API）
# ============================================================
SERVERS = {
    "web-01": {"ip": "10.0.1.10", "status": "healthy", "uptime": "45d 3h"},
    "web-02": {"ip": "10.0.1.11", "status": "degraded", "uptime": "12d 7h"},
    "db-primary": {"ip": "10.0.2.10", "status": "healthy", "uptime": "90d 0h"},
    "db-replica": {"ip": "10.0.2.11", "status": "healthy", "uptime": "90d 0h"},
    "cache-01": {"ip": "10.0.3.10", "status": "critical", "uptime": "0d 2h"},
}

DEPLOYMENTS = [
    {"id": "dep-001", "service": "web-api", "version": "v2.3.1", 
     "status": "completed", "time": "2025-06-10T14:30:00Z"},
    {"id": "dep-002", "service": "auth-service", "version": "v1.8.0", 
     "status": "in_progress", "time": "2025-06-11T09:15:00Z"},
    {"id": "dep-003", "service": "notification", "version": "v3.0.0", 
     "status": "failed", "time": "2025-06-11T11:00:00Z"},
]


# ============================================================
# Tools：让LLM能执行操作
# ============================================================

@mcp.tool()
def check_server_health(server_name: str) -> str:
    """检查指定服务器的健康状态。

    Args:
        server_name: 服务器名称，如 web-01, db-primary
    """
    server = SERVERS.get(server_name)
    if not server:
        return f"错误：未找到服务器 '{server_name}'。可用服务器：{', '.join(SERVERS.keys())}"

    status_emoji = {
        "healthy": "✅",
        "degraded": "⚠️", 
        "critical": "🔴"
    }.get(server["status"], "❓")

    return json.dumps({
        "server": server_name,
        "ip": server["ip"],
        "status": server["status"],
        "emoji": status_emoji,
        "uptime": server["uptime"],
        "checked_at": datetime.now().isoformat()
    }, indent=2, ensure_ascii=False)


@mcp.tool()
def list_all_servers() -> str:
    """列出所有已注册的服务器及其状态概览。"""
    summary = {"total": len(SERVERS), "servers": []}
    status_counts = {"healthy": 0, "degraded": 0, "critical": 0}

    for name, info in SERVERS.items():
        summary["servers"].append({
            "name": name, "ip": info["ip"], 
            "status": info["status"], "uptime": info["uptime"]
        })
        status_counts[info["status"]] = status_counts.get(info["status"], 0) + 1

    summary["status_summary"] = status_counts
    return json.dumps(summary, indent=2, ensure_ascii=False)


@mcp.tool()
def analyze_deployment_history(
    service_filter: Optional[str] = None,
    status_filter: Optional[str] = None
) -> str:
    """分析部署历史记录。

    Args:
        service_filter: 按服务名过滤（可选）
        status_filter: 按状态过滤：completed/in_progress/failed（可选）
    """
    results = DEPLOYMENTS

    if service_filter:
        results = [d for d in results if service_filter in d["service"]]
    if status_filter:
        results = [d for d in results if d["status"] == status_filter]

    # 统计分析
    status_breakdown = {}
    for d in results:
        status_breakdown[d["status"]] = status_breakdown.get(d["status"], 0) + 1

    return json.dumps({
        "total_found": len(results),
        "status_breakdown": status_breakdown,
        "deployments": results
    }, indent=2, ensure_ascii=False)


@mcp.tool()
def get_system_metrics() -> str:
    """获取本机系统指标（CPU、内存、磁盘）。"""
    import platform

    metrics = {
        "platform": platform.system(),
        "python_version": platform.python_version(),
        "architecture": platform.machine(),
        "timestamp": datetime.now().isoformat(),
        "note": "生产环境应替换为真实的监控数据采集逻辑"
    }
    return json.dumps(metrics, indent=2, ensure_ascii=False)


# ============================================================
# Resources：让LLM能获取上下文数据
# ============================================================

@mcp.resource("config://servers")
def get_server_config() -> str:
    """获取所有服务器的配置信息。"""
    return json.dumps(SERVERS, indent=2, ensure_ascii=False)


@mcp.resource("config://server/{server_name}")
def get_server_detail(server_name: str) -> str:
    """获取单个服务器的详细配置。"""
    server = SERVERS.get(server_name)
    if not server:
        return json.dumps({"error": f"Server '{server_name}' not found"})
    return json.dumps({
        "name": server_name,
        **server,
        "last_checked": datetime.now().isoformat()
    }, indent=2, ensure_ascii=False)


# ============================================================
# Prompts：预设的交互模板
# ============================================================

@mcp.prompt()
def incident_response(server_name: str, alert_type: str) -> str:
    """生成事件响应的提示词模板。

    Args:
        server_name: 出问题的服务器名
        alert_type: 告警类型（cpu_high/memory_low/disk_full/service_down）
    """
    server = SERVERS.get(server_name, {})

    return f"""你是一名资深DevOps工程师，正在处理一起生产环境告警。

## 告警信息
- **服务器**: {server_name} ({server.get('ip', '未知')})
- **当前状态**: {server.get('status', '未知')}
- **告警类型**: {alert_type}
- **触发时间**: {datetime.now().isoformat()}

## 任务要求
1. 首先调用 check_server_health 确认当前状态
2. 调用 get_system_metrics 查看系统指标
3. 根据 {alert_type} 类型的常见原因，给出排查步骤
4. 提供具体的修复建议和预防措施

请按步骤执行，每步给出清晰的结论。"""


@mcp.prompt()
def deployment_review(service_name: str, version: str) -> str:
    """生成部署审查的提示词模板。"""
    return f"""请审查以下部署请求：

- 服务：{service_name}
- 版本：{version}

步骤：
1. 查看 analyze_deployment_history 了解该服务的历史部署情况
2. 检查是否有失败的部署记录
3. 评估部署风险并给出建议
4. 生成部署检查清单"""


# ============================================================
# 启动入口
# ============================================================

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

3.3 FastMCP关键概念解析

装饰器映射：

装饰器 MCP原语 作用 LLM行为
@mcp.tool() Tool 注册可调用工具 LLM主动调用
@mcp.resource() Resource 暴露数据资源 应用读取，LLM获取上下文
@mcp.prompt() Prompt 注册提示模板 用户选择使用

Tool设计的黄金法则：

  1. 单一职责：每个Tool做一件事，做好一件事
  2. 类型标注完整：所有参数都有类型标注，FastMCP会自动生成JSON Schema
  3. Docstring即文档：函数的docstring会作为Tool描述发送给LLM，直接影响LLM是否选择调用
  4. 返回字符串：Tool的返回值会被序列化为字符串传回给LLM
  5. 错误自解释：错误信息要让LLM能理解并决定下一步操作

Resource的URI设计：

config://servers              # 静态资源，返回所有服务器配置
config://server/{server_name} # 模板资源，支持路径参数

URI模板（Template Resource）用花括号{param}标记变量部分，FastMCP自动解析并映射到函数参数。

3.4 常见问题与解决

问题 原因 解决方案
ModuleNotFoundError: fastmcp 未激活虚拟环境 source .venv/bin/activate
Server启动后无响应 stdio模式需要Client连接 使用fastmcp dev或配置到Claude Code
Tool参数类型错误 JSON Schema不匹配 确保类型标注正确，Optional参数有默认值
中文输出乱码 JSON序列化未设置ensure_ascii json.dumps(data, ensure_ascii=False)

第四章：TypeScript SDK实战

4.1 等价的DevOps服务器

创建 ts-server/src/devops-server.ts：

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

// 创建服务器实例
const server = new McpServer({
  name: "devops-toolkit-ts",
  version: "1.0.0",
});

// 模拟数据（与Python版本相同）
const SERVERS: Record<string, { ip: string; status: string; uptime: string }> = {
  "web-01": { ip: "10.0.1.10", status: "healthy", uptime: "45d 3h" },
  "web-02": { ip: "10.0.1.11", status: "degraded", uptime: "12d 7h" },
  "db-primary": { ip: "10.0.2.10", status: "healthy", uptime: "90d 0h" },
  "db-replica": { ip: "10.0.2.11", status: "healthy", uptime: "90d 0h" },
  "cache-01": { ip: "10.0.3.10", status: "critical", uptime: "0d 2h" },
};

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

server.tool(
  "check_server_health",
  "Check the health status of a specific server",
  { server_name: z.string().describe("Server name, e.g. web-01, db-primary") },
  async ({ server_name }) => {
    const server = SERVERS[server_name];
    if (!server) {
      return {
        content: [{
          type: "text",
          text: `Error: Server '${server_name}' not found. Available: ${Object.keys(SERVERS).join(", ")}`
        }]
      };
    }

    const emoji = { healthy: "✅", degraded: "⚠️", critical: "🔴" }[server.status] || "❓";

    return {
      content: [{
        type: "text",
        text: JSON.stringify({
          server: server_name,
          ip: server.ip,
          status: server.status,
          emoji,
          uptime: server.uptime,
          checked_at: new Date().toISOString()
        }, null, 2)
      }]
    };
  }
);

server.tool(
  "list_all_servers",
  "List all registered servers with status overview",
  {},
  async () => {
    const statusCounts: Record<string, number> = {};
    const serverList = Object.entries(SERVERS).map(([name, info]) => {
      statusCounts[info.status] = (statusCounts[info.status] || 0) + 1;
      return { name, ...info };
    });

    return {
      content: [{
        type: "text",
        text: JSON.stringify({
          total: serverList.length,
          status_summary: statusCounts,
          servers: serverList
        }, null, 2)
      }]
    };
  }
);

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

server.resource(
  new ResourceTemplate("config://servers", { list: undefined }),
  async (uri) => ({
    contents: [{
      uri: uri.href,
      text: JSON.stringify(SERVERS, null, 2)
    }]
  })
);

server.resource(
  new ResourceTemplate("config://server/{server_name}", { list: undefined }),
  async (uri, { server_name }) => {
    const server = SERVERS[server_name as string];
    return {
      contents: [{
        uri: uri.href,
        text: server 
          ? JSON.stringify({ name: server_name, ...server }, null, 2)
          : JSON.stringify({ error: `Server '${server_name}' not found` })
      }]
    };
  }
);

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

server.prompt(
  "incident_response",
  { server_name: z.string(), alert_type: z.string() },
  async ({ server_name, alert_type }) => {
    const server = SERVERS[server_name] || {};
    return {
      messages: [{
        role: "user",
        content: {
          type: "text",
          text: `You are a senior DevOps engineer handling a production alert.

## Alert Information
- **Server**: ${server_name} (${server.ip || "unknown"})
- **Current Status**: ${server.status || "unknown"}
- **Alert Type**: ${alert_type}
- **Triggered At**: ${new Date().toISOString()}

## Task Requirements
1. Call check_server_health to confirm current status
2. Call list_all_servers for context
3. Based on ${alert_type} type, provide troubleshooting steps
4. Give specific fix recommendations and prevention measures

Execute step by step with clear conclusions.`
        }
      }]
    };
  }
);

// ============================================================
// 启动
// ============================================================

async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("DevOps Toolkit MCP Server running on stdio");
}

main().catch(console.error);

更新 ts-server/package.json 添加启动脚本：

{
  "name": "devops-toolkit-mcp",
  "version": "1.0.0",
  "type": "module",
  "scripts": {
    "dev": "tsx src/devops-server.ts",
    "build": "tsc",
    "start": "node dist/devops-server.js"
  }
}

运行测试：

cd ts-server
npm run dev

4.2 Python与TypeScript对比

维度 Python (FastMCP) TypeScript (官方SDK)
代码量 少30-40% 稍多，需显式构建返回值
Schema定义 自动从类型标注生成 使用Zod手动定义
学习曲线 平缓 中等（需理解Zod和MCP返回格式）
生态 pip包丰富，数据科学友好 npm生态，前端/全栈友好
性能 适合I/O密集型 事件驱动，高并发友好
社区占比 ~70%的MCP服务器使用FastMCP ~25%使用TS SDK
推荐场景 数据处理、运维工具、快速原型 Web服务集成、前端工具链

选择建议： - 快速原型、数据处理 → Python + FastMCP - 与现有Node.js项目集成、需要类型安全 → TypeScript SDK - 团队已有技术栈决定 → 跟随团队

第五章：注册到Claude Code

5.1 理解配置层级

Claude Code的MCP配置存储在三个层级：

graph TD
    A["~/.claude/settings.json<br/>全局配置<br/>所有项目共享"] --> B["项目根/.claude/settings.json<br/>项目配置<br/>团队可共享"]
    B --> C["项目根/.mcp.json<br/>MCP专用配置<br/>可直接提交git"]

    style A fill:#4A90D9,color:#fff
    style B fill:#50C878,color:#fff
    style C fill:#FF9F43,color:#fff
文件 作用域 适用场景
~/.claude/settings.json 全局 个人常用工具，跨项目通用
.claude/settings.json 项目级 团队共享的项目特定工具
.mcp.json 项目级MCP专用 最简洁的方式，推荐用于项目级

5.2 配置Python服务器

在项目根目录创建 .mcp.json：

{
  "mcpServers": {
    "devops-python": {
      "command": "python",
      "args": ["/absolute/path/to/python-server/devops_server.py"],
      "env": {
        "PYTHONUNBUFFERED": "1"
      }
    }
  }
}

5.3 配置TypeScript服务器

同样在 .mcp.json 中添加（或替换）：

{
  "mcpServers": {
    "devops-ts": {
      "command": "npx",
      "args": ["tsx", "/absolute/path/to/ts-server/src/devops-server.ts"],
      "env": {
        "NODE_ENV": "development"
      }
    }
  }
}

5.4 环境变量与敏感信息

正确做法——使用环境变量传递密钥：

{
  "mcpServers": {
    "secure-api-server": {
      "command": "python",
      "args": ["./mcp-servers/api_server.py"],
      "env": {
        "API_BASE_URL": "https://api.example.com",
        "API_KEY": "${API_KEY}"
      }
    }
  }
}

错误做法——直接硬编码密钥：

{
  "mcpServers": {
    "bad-example": {
      "command": "python",
      "args": ["./server.py"],
      "env": {
        "API_KEY": "sk-abc123never-do-this"
      }
    }
  }
}

⚠️ .mcp.json 通常会被提交到git，绝不要在文件中硬编码任何密钥。使用 ${ENV_VAR} 语法引用环境变量，或将 .mcp.json 加入 .gitignore。

5.5 验证配置

在Claude Code中验证MCP服务器是否加载：

# 方法1：查看已加载的MCP服务器
claude mcp list

# 方法2：直接启动Claude Code，在对话中问：
# "你有哪些可用的MCP工具？"

# 方法3：使用MCP Inspector调试
npx @modelcontextprotocol/inspector python devops_server.py

第六章：MCP Inspector调试实战

6.1 Inspector快速启动

MCP Inspector是官方的可视化调试工具，可以独立测试MCP服务器而无需连接Claude Code：

# 调试Python服务器
npx @modelcontextprotocol/inspector python devops_server.py

# 调试TypeScript服务器
npx @modelcontextprotocol/inspector npx tsx src/devops-server.ts

Inspector会在浏览器中打开（默认 http://localhost:6274），提供：

6.2 调试常见问题

问题1：服务器启动后Inspector显示连接失败

解决方案：
1. 检查command路径是否正确（使用绝对路径）
2. 检查Python/Node环境是否正确激活
3. 查看Inspector的stderr输出获取错误信息
4. 确认服务器代码没有在模块级别执行阻塞操作

问题2：Tool调用返回意外结果

调试步骤：
1. 在Inspector的Console面板查看实际发送的JSON-RPC消息
2. 对比函数签名与实际参数是否匹配
3. 检查返回值格式是否正确（必须是{content: [{type: "text", text: "..."}]}）
4. 在函数内部添加print/logging输出（FastMCP支持print到stderr）

问题3：Resource URI模板匹配失败

常见原因：
1. URI中的参数名与函数参数名不一致
2. URI路径参数缺少类型转换（如数字需要手动int()）
3. 模板语法错误（花括号内不能有空格）

6.3 编写自动化测试

为MCP服务器编写单元测试（Python示例）：

# test_devops_server.py
import pytest
from unittest.mock import patch
from devops_server import check_server_health, list_all_servers

def test_check_server_health_found():
    result = check_server_health("web-01")
    data = json.loads(result)
    assert data["server"] == "web-01"
    assert data["status"] == "healthy"

def test_check_server_health_not_found():
    result = check_server_health("unknown-server")
    assert "未找到" in result

def test_list_all_servers():
    result = list_all_servers()
    data = json.loads(result)
    assert data["total"] == 5
    assert "healthy" in data["status_summary"]

if __name__ == "__main__":
    pytest.main([__file__, "-v"])

运行测试：

pytest test_devops_server.py -v

第七章：高级主题——安全、性能与生产部署

7.1 安全加固清单

graph LR
    subgraph 安全加固
        A["输入验证"] --> B["权限控制"]
        B --> C["输出过滤"]
        C --> D["审计日志"]
        D --> E["传输加密"]
    end

    style A fill:#E74C3C,color:#fff
    style B fill:#E74C3C,color:#fff
    style C fill:#E74C3C,color:#fff
    style D fill:#F39C12,color:#fff
    style E fill:#F39C12,color:#fff

输入验证：

import re
from pathlib import Path

@mcp.tool()
def safe_read_file(file_path: str) -> str:
    """安全读取文件内容（限制在指定目录内）。"""
    # 1. 路径规范化
    base_dir = Path("/allowed/directory").resolve()
    target = (base_dir / file_path).resolve()

    # 2. 路径遍历检查
    if not str(target).startswith(str(base_dir)):
        return "错误：路径超出允许范围"

    # 3. 文件类型检查
    allowed_extensions = {".txt", ".md", ".json", ".csv"}
    if target.suffix not in allowed_extensions:
        return f"错误：不支持的文件类型 {target.suffix}"

    # 4. 文件大小检查
    if target.stat().st_size > 1024 * 1024:  # 1MB
        return "错误：文件过大（超过1MB）"

    return target.read_text()

环境变量管理：

import os

# 从环境变量获取配置
API_KEY = os.environ.get("MY_API_KEY")
if not API_KEY:
    raise ValueError("MY_API_KEY environment variable is required")

# 使用环境变量的服务器配置
DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///default.db")

7.2 性能优化

异步操作：

import asyncio
import httpx

@mcp.tool()
async def batch_health_check() -> str:
    """并发检查所有服务器健康状态。"""
    async with httpx.AsyncClient() as client:
        tasks = [
            client.get(f"http://{info['ip']}/health", timeout=5.0)
            for info in SERVERS.values()
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    # 处理结果...
    return json.dumps({"checked": len(results)})

分页处理：

@mcp.tool()
def search_logs(query: str, page: int = 1, page_size: int = 20) -> str:
    """搜索日志（支持分页）。"""
    all_logs = get_all_logs()  # 假设获取全部日志
    filtered = [l for l in all_logs if query in l["message"]]

    start = (page - 1) * page_size
    end = start + page_size

    return json.dumps({
        "total": len(filtered),
        "page": page,
        "page_size": page_size,
        "results": filtered[start:end]
    })

7.3 生产部署选项

graph TB
    subgraph 本地部署
        L1["stdio模式<br/>Claude Code直接调用"]
        L2["适合个人开发者<br/>零配置"]
    end

    subgraph 远程部署
        R1["Streamable HTTP模式<br/>需要OAuth 2.1认证"]
        R2["Docker容器化<br/>可扩展、可监控"]
        R3["适合团队共享<br/>统一管理"]
    end

    L1 --> L2
    R1 --> R2
    R1 --> R3

    style L1 fill:#27AE60,color:#fff
    style R1 fill:#3498DB,color:#fff

Docker化部署：

# Dockerfile
FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

# stdio模式（本地Claude Code使用）
CMD ["python", "devops_server.py"]

# HTTP模式（远程部署）
# CMD ["fastmcp", "run", "devops_server.py", "--transport", "streamable-http", "--port", "8080"]

Streamable HTTP模式配置（远程服务器）：

{
  "mcpServers": {
    "remote-devops": {
      "type": "streamableHttp",
      "url": "https://your-server.example.com/mcp",
      "headers": {
        "Authorization": "Bearer ${MCP_AUTH_TOKEN}"
      }
    }
  }
}

7.4 监控与日志

import logging
import sys

# 配置日志（输出到stderr，不影响stdio通信）
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    stream=sys.stderr  # 关键：不能输出到stdout
)

logger = logging.getLogger("devops-mcp")

@mcp.tool()
def check_server_health(server_name: str) -> str:
    logger.info(f"Health check requested for: {server_name}")
    # ... 业务逻辑
    logger.info(f"Health check completed: {server_name} -> {result}")
    return result

⚠️ 关键注意：stdio模式下，服务器的stdout被用于JSON-RPC通信。所有日志和调试输出必须写到stderr（sys.stderr或console.error），否则会破坏协议通信导致服务器崩溃。

第八章：完整工作流程总结

从零到生产的完整MCP服务器开发流程：

flowchart TD
    A[需求分析<br/>确定Tool/Resource/Prompt] --> B[技术选型<br/>Python or TypeScript]
    B --> C[项目初始化<br/>安装SDK/依赖]
    C --> D[实现Tools<br/>核心业务逻辑]
    D --> E[实现Resources<br/>上下文数据暴露]
    E --> F[实现Prompts<br/>交互模板]
    F --> G[本地调试<br/>MCP Inspector]
    G --> H{测试通过?}
    H -->|否| D
    H -->|是| I[配置到Claude Code<br/>.mcp.json]
    I --> J[集成测试<br/>实际对话验证]
    J --> K{安全审查?}
    K -->|未通过| L[加固：输入验证/权限/日志]
    L --> J
    K -->|通过| M[生产部署<br/>stdio本地或HTTP远程]
    M --> N[监控运维<br/>日志/告警/更新]

    style A fill:#3498DB,color:#fff
    style G fill:#E67E22,color:#fff
    style M fill:#27AE60,color:#fff

延伸阅读

资源 说明
MCP官方规范 协议完整规范，理解底层机制必备
FastMCP官方文档 Python MCP框架的完整参考
TypeScript SDK文档 TS SDK的API文档和示例
MCP Inspector调试指南 官方调试工具使用说明
Claude Code MCP配置指南 Claude Code的MCP服务器接入文档
MCP安全最佳实践 社区维护的安全规范指南
FreeCodeCamp FastMCP教程 入门友好的实战教程

训练营适用性评估

训练营类型 适用性 建议时长 备注
AI应用开发训练营 ⭐⭐⭐⭐⭐ 4-6小时 核心技能，建议重点投入
企业AI集成工作坊 ⭐⭐⭐⭐⭐ 3-4小时 聚焦企业API集成场景
Claude Code深度使用营 ⭐⭐⭐⭐ 3-5小时 配合Skills和Hooks教程
Python全栈训练营 ⭐⭐⭐ 2-3小时 作为API开发的高级主题
DevOps/SRE训练营 ⭐⭐⭐ 2-3小时 使用DevOps工具箱作为案例

教学建议： - 前2小时：理论（协议架构）+ Python快速入门 - 中间2小时：TypeScript版本 + 调试实战 - 最后1-2小时：安全加固 + 生产部署 + 学员动手改造

参考来源

  1. MCP Protocol Specification 2025-03-26 — MCP协议官方规范，定义了Tools/Resources/Prompts三大原语和传输层协议
  2. FastMCP Official Documentation — Python FastMCP框架的官方文档，涵盖从入门到高级用法
  3. MCP TypeScript SDK - Server Documentation — 官方TypeScript SDK的服务器端开发文档
  4. Claude Code MCP Quickstart — Claude Code官方MCP服务器配置与接入指南
  5. The Complete MCP Guide for Developers (2025 Edition) — 开发者社区的MCP完整指南，覆盖架构、开发和部署
  6. MCP Debugging Tools — 官方MCP调试工具文档，包括Inspector使用说明
  7. FastMCP Python Tutorial - FreeCodeCamp — FreeCodeCamp的FastMCP入门教程

[!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学通信知识星球海报-扫码加入

📎 本分类最新文章