# 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场景中引入了三层模型：
```mermaid
graph TB
subgraph Host["Host 宿主应用"]
C1["Client 1"]
C2["Client 2"]
C3["Client N"]
end
subgraph Servers["MCP Servers"]
S1["Server A
文件系统"]
S2["Server B
数据库"]
S3["Server C
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
```
**三层模型详解**：
- **Host（宿主）**：用户直接交互的应用程序，如Claude Code、Claude Desktop、Cursor等。一个Host可以同时管理多个Client。
- **Client（客户端）**：Host内部与特定MCP Server维持1:1连接的协议实例。负责协议握手、能力协商和消息路由。
- **Server（服务器）**：暴露工具、资源和提示的服务进程。每个Server独立运行，通过标准化的传输协议与Client通信。
### 1.2 三大核心原语
MCP定义了三种原语（Primitive），每种服务于不同的交互模式：
| 原语 | 类比 | 方向 | 典型用途 |
|------|------|------|----------|
| **Tools（工具）** | 函数调用 | LLM → Server | 执行操作、查询数据、调用API |
| **Resources（资源）** | 文件读取 | 应用 → Server | 提供上下文数据、文档、配置 |
| **Prompts（提示模板）** | 模板引擎 | 用户 → Server | 预设的提示词模板、工作流引导 |
**设计原则**：
- Tools是"动词"——让LLM能**执行**操作
- Resources是"名词"——让LLM能**获取**数据
- Prompts是"句式"——让用户能**复用**交互模式
### 1.3 传输协议演进
MCP的传输层经历了重要演进：
```mermaid
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异步任务
```
- **stdio**：最简单，适合本地开发。Server作为子进程，通过stdin/stdout交换JSON-RPC消息。
- **Streamable HTTP**：2025年11月规范引入，取代了SSE。支持无状态请求、可选的SSE流、OAuth 2.1认证，适合远程部署。
对于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 创建项目目录结构
```bash
# 创建项目根目录
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）**：
```bash
cd python-server
pip install fastmcp httpx
# FastMCP从3.0开始是官方推荐的Python MCP框架
# 验证安装
python -c "import fastmcp; print(fastmcp.__version__)"
```
**TypeScript（官方SDK）**：
```bash
cd ts-server
npm install @modelcontextprotocol/sdk zota
npm install -D typescript @types/node tsx
# 验证安装
npx tsc --version
```
### 2.3 项目配置文件
**Python `pyproject.toml`**：
```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`**：
```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`：
```python
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()
```
运行测试：
```bash
# 直接运行（stdio模式，默认）
fastmcp dev server.py
# 或者
python server.py
```
**预期结果**：服务器启动，等待通过stdin接收JSON-RPC消息。`fastmcp dev`命令会自动启动MCP Inspector进行可视化调试。
### 3.2 完整实战：DevOps工具箱服务器
我们来构建一个实用的DevOps工具箱，包含服务器健康检查、日志分析和部署状态查询功能。
创建 `python-server/devops_server.py`：
```python
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`：
```typescript
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 = {
"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 = {};
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` 添加启动脚本：
```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"
}
}
```
运行测试：
```bash
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配置存储在三个层级：
```mermaid
graph TD
A["~/.claude/settings.json
全局配置
所有项目共享"] --> B["项目根/.claude/settings.json
项目配置
团队可共享"]
B --> C["项目根/.mcp.json
MCP专用配置
可直接提交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`：
```json
{
"mcpServers": {
"devops-python": {
"command": "python",
"args": ["/absolute/path/to/python-server/devops_server.py"],
"env": {
"PYTHONUNBUFFERED": "1"
}
}
}
}
```
### 5.3 配置TypeScript服务器
同样在 `.mcp.json` 中添加（或替换）：
```json
{
"mcpServers": {
"devops-ts": {
"command": "npx",
"args": ["tsx", "/absolute/path/to/ts-server/src/devops-server.ts"],
"env": {
"NODE_ENV": "development"
}
}
}
}
```
### 5.4 环境变量与敏感信息
**正确做法**——使用环境变量传递密钥：
```json
{
"mcpServers": {
"secure-api-server": {
"command": "python",
"args": ["./mcp-servers/api_server.py"],
"env": {
"API_BASE_URL": "https://api.example.com",
"API_KEY": "${API_KEY}"
}
}
}
}
```
**错误做法**——直接硬编码密钥：
```json
{
"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服务器是否加载：
```bash
# 方法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：
```bash
# 调试Python服务器
npx @modelcontextprotocol/inspector python devops_server.py
# 调试TypeScript服务器
npx @modelcontextprotocol/inspector npx tsx src/devops-server.ts
```
Inspector会在浏览器中打开（默认 `http://localhost:6274`），提供：
- **Tools标签页**：列出所有注册的工具，可手动输入参数调用
- **Resources标签页**：列出所有资源，可点击查看返回内容
- **Prompts标签页**：列出提示模板，可填写参数生成提示词
- **Console面板**：显示JSON-RPC通信日志，方便排查协议问题
### 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示例）：
```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"])
```
运行测试：
```bash
pytest test_devops_server.py -v
```
---
## 第七章：高级主题——安全、性能与生产部署
### 7.1 安全加固清单
```mermaid
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
```
**输入验证**：
```python
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()
```
**环境变量管理**：
```python
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 性能优化
**异步操作**：
```python
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)})
```
**分页处理**：
```python
@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 生产部署选项
```mermaid
graph TB
subgraph 本地部署
L1["stdio模式
Claude Code直接调用"]
L2["适合个人开发者
零配置"]
end
subgraph 远程部署
R1["Streamable HTTP模式
需要OAuth 2.1认证"]
R2["Docker容器化
可扩展、可监控"]
R3["适合团队共享
统一管理"]
end
L1 --> L2
R1 --> R2
R1 --> R3
style L1 fill:#27AE60,color:#fff
style R1 fill:#3498DB,color:#fff
```
**Docker化部署**：
```dockerfile
# 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模式配置**（远程服务器）：
```json
{
"mcpServers": {
"remote-devops": {
"type": "streamableHttp",
"url": "https://your-server.example.com/mcp",
"headers": {
"Authorization": "Bearer ${MCP_AUTH_TOKEN}"
}
}
}
}
```
### 7.4 监控与日志
```python
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服务器开发流程：
```mermaid
flowchart TD
A[需求分析
确定Tool/Resource/Prompt] --> B[技术选型
Python or TypeScript]
B --> C[项目初始化
安装SDK/依赖]
C --> D[实现Tools
核心业务逻辑]
D --> E[实现Resources
上下文数据暴露]
E --> F[实现Prompts
交互模板]
F --> G[本地调试
MCP Inspector]
G --> H{测试通过?}
H -->|否| D
H -->|是| I[配置到Claude Code
.mcp.json]
I --> J[集成测试
实际对话验证]
J --> K{安全审查?}
K -->|未通过| L[加固：输入验证/权限/日志]
L --> J
K -->|通过| M[生产部署
stdio本地或HTTP远程]
M --> N[监控运维
日志/告警/更新]
style A fill:#3498DB,color:#fff
style G fill:#E67E22,color:#fff
style M fill:#27AE60,color:#fff
```
---
## 延伸阅读
| 资源 | 说明 |
|------|------|
| [MCP官方规范](https://modelcontextprotocol.io/specification/2025-03-26) | 协议完整规范，理解底层机制必备 |
| [FastMCP官方文档](https://fastmcp.wiki/en/getting-started/welcome) | Python MCP框架的完整参考 |
| [TypeScript SDK文档](https://ts.sdk.modelcontextprotocol.io/documents/server.html) | TS SDK的API文档和示例 |
| [MCP Inspector调试指南](https://modelcontextprotocol.io/docs/tools/debugging) | 官方调试工具使用说明 |
| [Claude Code MCP配置指南](https://code.claude.com/docs/en/mcp-quickstart) | Claude Code的MCP服务器接入文档 |
| [MCP安全最佳实践](https://mcp-best-practice.github.io/mcp-best-practice/best-practice/) | 社区维护的安全规范指南 |
| [FreeCodeCamp FastMCP教程](https://www.freecodecamp.org/news/how-to-build-your-first-mcp-server-using-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](https://modelcontextprotocol.io/specification/2025-03-26) — MCP协议官方规范，定义了Tools/Resources/Prompts三大原语和传输层协议
2. [FastMCP Official Documentation](https://fastmcp.wiki/en/getting-started/welcome) — Python FastMCP框架的官方文档，涵盖从入门到高级用法
3. [MCP TypeScript SDK - Server Documentation](https://ts.sdk.modelcontextprotocol.io/documents/server.html) — 官方TypeScript SDK的服务器端开发文档
4. [Claude Code MCP Quickstart](https://code.claude.com/docs/en/mcp-quickstart) — Claude Code官方MCP服务器配置与接入指南
5. [The Complete MCP Guide for Developers (2025 Edition)](https://dev.to/kevinz103/the-complete-mcp-guide-for-developers2025-edition-ana) — 开发者社区的MCP完整指南，覆盖架构、开发和部署
6. [MCP Debugging Tools](https://modelcontextprotocol.io/docs/tools/debugging) — 官方MCP调试工具文档，包括Inspector使用说明
7. [FastMCP Python Tutorial - FreeCodeCamp](https://www.freecodecamp.org/news/how-to-build-your-first-mcp-server-using-fastmcp/) — 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 等场景全覆盖。
