# FastMCP调试测试与生产级部署实战
## 概述
本教程面向已有MCP基础概念、希望将MCP服务器从"能跑"推向"能上线"的开发者。我们将以Python FastMCP框架为核心，深入覆盖调试技巧（MCP Inspector）、测试策略（单元测试与集成测试）、日志与错误处理、以及生产环境部署的完整链路。学完本教程，你将具备独立交付生产级MCP服务器的能力。
**适用人群**：已了解MCP基本概念、有Python基础的后端开发者，或希望为团队构建内部MCP工具链的工程师。
**学习目标**：
- 掌握MCP Inspector可视化调试工具的使用
- 构建带完整错误处理和日志的FastMCP服务器
- 编写可重复执行的MCP服务器测试套件
- 理解生产环境部署的最佳实践与安全考量
---
## 前置条件
| 条件 | 说明 |
|------|------|
| Python 3.10+ | FastMCP依赖现代Python特性 |
| Node.js 18+ | MCP Inspector运行所需 |
| Claude Code CLI | 已安装并登录 |
| 基础MCP概念 | 了解Tool、Resource、Prompt三要素 |
| pip/uv包管理器 | 推荐使用uv加速依赖安装 |
---
## 第一章：FastMCP项目脚手架搭建
### 1.1 为什么选择FastMCP
FastMCP（由PrefectHQ维护，GitHub 25.6k stars）是当前Python生态中最成熟的MCP框架。相比手动使用`mcp`底层SDK，FastMCP提供：
- **装饰器驱动的API**：用`@mcp.tool()`、`@mcp.resource()`一键注册
- **内置类型推断**：自动从函数签名生成JSON Schema
- **Context对象**：统一的日志、进度报告、资源访问接口
- **双传输模式**：stdio和SSE开箱即用
- **完善的错误处理**：结构化的MCP错误响应
### 1.2 创建项目结构
```bash
# 创建项目目录
mkdir mcp-file-analyzer && cd mcp-file-analyzer
# 使用uv初始化（推荐）
uv init
uv add fastmcp
# 创建标准项目布局
mkdir -p src/server tests
touch src/server/__init__.py
touch src/server/analyzer.py
touch tests/__init__.py
touch tests/test_analyzer.py
```
最终项目结构：
```
mcp-file-analyzer/
├── pyproject.toml
├── README.md
├── src/
│ └── server/
│ ├── __init__.py
│ └── analyzer.py # MCP服务器主文件
├── tests/
│ ├── __init__.py
│ └── test_analyzer.py # 测试套件
└── .claude/
└── settings.json # Claude Code项目配置
```
### 1.3 编写第一个Tool
编辑 `src/server/analyzer.py`：
```python
"""MCP File Analyzer Server - 分析文件内容并提供统计信息"""
import os
import stat
from datetime import datetime
from pathlib import Path
from fastmcp import FastMCP, Context
# 创建FastMCP实例
mcp = FastMCP(
name="file-analyzer",
version="1.0.0",
description="分析文件内容、统计代码行数、检测文件类型的MCP服务器"
)
@mcp.tool()
def analyze_file(file_path: str) -> dict:
"""分析指定文件的详细信息，包括大小、类型、权限等。
Args:
file_path: 要分析的文件路径（绝对路径）
Returns:
包含文件详细信息的字典
"""
path = Path(file_path).resolve()
if not path.exists():
raise ValueError(f"文件不存在: {file_path}")
if not path.is_file():
raise ValueError(f"路径不是文件: {file_path}")
stat_info = path.stat()
return {
"name": path.name,
"extension": path.suffix.lower(),
"size_bytes": stat_info.st_size,
"size_human": _human_readable_size(stat_info.st_size),
"mime_type": _guess_mime(path),
"permissions": stat.filemode(stat_info.st_mode),
"modified": datetime.fromtimestamp(stat_info.st_mtime).isoformat(),
"created": datetime.fromtimestamp(stat_info.st_ctime).isoformat(),
"parent_dir": str(path.parent),
}
@mcp.tool()
def count_lines(file_path: str, include_blank: bool = False) -> dict:
"""统计文件的行数信息，包括总行数、空行数、非空行数。
Args:
file_path: 文件路径
include_blank: 是否在统计中包含空行
Returns:
行数统计信息
"""
path = Path(file_path).resolve()
if not path.exists():
raise ValueError(f"文件不存在: {file_path}")
try:
content = path.read_text(encoding="utf-8")
except UnicodeDecodeError:
content = path.read_text(encoding="latin-1")
lines = content.splitlines()
blank_lines = sum(1 for line in lines if not line.strip())
code_lines = len(lines) - blank_lines
return {
"total_lines": len(lines),
"blank_lines": blank_lines,
"code_lines": code_lines,
"chars": len(content),
"avg_line_length": round(len(content) / max(len(lines), 1), 1),
}
@mcp.tool()
async def search_in_files(
directory: str,
pattern: str,
file_extensions: list[str] | None = None,
ctx: Context = None
) -> dict:
"""在目录中递归搜索包含指定模式的文件。
Args:
directory: 搜索的根目录
pattern: 搜索的文本模式
file_extensions: 限制搜索的文件扩展名列表（如 [".py", ".js"]）
ctx: FastMCP上下文对象
Returns:
匹配结果摘要
"""
dir_path = Path(directory).resolve()
if not dir_path.is_dir():
raise ValueError(f"目录不存在: {directory}")
if ctx:
await ctx.info(f"开始在 {directory} 中搜索 '{pattern}'...")
matches = []
files_scanned = 0
extensions = set(file_extensions) if file_extensions else None
for file_path in dir_path.rglob("*"):
if not file_path.is_file():
continue
if extensions and file_path.suffix.lower() not in extensions:
continue
if file_path.stat().st_size > 10 * 1024 * 1024: # 跳过 >10MB 文件
continue
files_scanned += 1
try:
content = file_path.read_text(encoding="utf-8", errors="ignore")
if pattern.lower() in content.lower():
line_numbers = [
i + 1 for i, line in enumerate(content.splitlines())
if pattern.lower() in line.lower()
]
matches.append({
"file": str(file_path.relative_to(dir_path)),
"match_count": len(line_numbers),
"first_matches": line_numbers[:5],
})
except Exception:
continue
# 进度报告（每50个文件报告一次）
if ctx and files_scanned % 50 == 0:
await ctx.info(f"已扫描 {files_scanned} 个文件，找到 {len(matches)} 个匹配...")
return {
"pattern": pattern,
"directory": str(dir_path),
"files_scanned": files_scanned,
"files_matched": len(matches),
"matches": sorted(matches, key=lambda x: x["match_count"], reverse=True)[:20],
}
@mcp.resource("file://{file_path}")
def read_file_resource(file_path: str) -> str:
"""将文件内容作为MCP Resource暴露给客户端读取。"""
path = Path(file_path).resolve()
if not path.exists():
raise ValueError(f"文件不存在: {file_path}")
return path.read_text(encoding="utf-8", errors="replace")
@mcp.prompt()
def analyze_code_prompt(file_path: str) -> str:
"""生成代码分析提示词。"""
return f"""请分析以下代码文件，指出潜在问题和改进建议：
文件路径: {file_path}
请从以下角度分析：
1. 代码质量与可读性
2. 潜在的Bug或安全问题
3. 性能优化建议
4. 最佳实践建议"""
# 辅助函数
def _human_readable_size(size_bytes: int) -> str:
for unit in ["B", "KB", "MB", "GB", "TB"]:
if size_bytes < 1024:
return f"{size_bytes:.1f} {unit}"
size_bytes /= 1024
return f"{size_bytes:.1f} PB"
def _guess_mime(path: Path) -> str:
import mimetypes
mime, _ = mimetypes.guess_type(str(path))
return mime or "application/octet-stream"
if __name__ == "__main__":
mcp.run(transport="stdio")
```
### 1.4 配置pyproject.toml
```toml
[project]
name = "mcp-file-analyzer"
version = "1.0.0"
description = "MCP服务器：文件分析与代码统计工具集"
requires-python = ">=3.10"
dependencies = [
"fastmcp>=2.0",
]
[project.optional-dependencies]
dev = [
"pytest>=8.0",
"pytest-asyncio>=0.24",
]
[project.scripts]
mcp-file-analyzer = "server.analyzer:main"
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
```
---
## 第二章：MCP Inspector可视化调试
### 2.1 Inspector是什么
MCP Inspector是官方提供的可视化调试工具（GitHub 10k stars），提供了Web UI界面用于：
- **连接MCP服务器**：支持stdio和SSE两种传输方式
- **列举和调用Tool**：逐个测试每个工具的输入输出
- **浏览Resource**：查看服务器暴露的资源列表
- **测试Prompt**：预览生成的提示词
- **实时日志查看**：捕获服务器端的日志输出
### 2.2 启动Inspector
```bash
# 方法一：直接使用npx运行（推荐）
npx @modelcontextprotocol/inspector uv run src/server/analyzer.py
# 方法二：全局安装后运行
npm install -g @modelcontextprotocol/inspector
mcp-inspector uv run src/server/analyzer.py
# 方法三：指定自定义端口
npx @modelcontextprotocol/inspector \
--port 6274 \
uv run src/server/analyzer.py
```
启动后，终端会显示：
```
🔍 MCP Inspector is running
Client: http://localhost:6274
Server proxy: http://localhost:6275
```
在浏览器中打开 `http://localhost:6274`，你将看到Inspector的交互界面。
### 2.3 使用Inspector调试Tool
在Inspector界面中：
**步骤1**：点击左侧 **Tools** 标签页，你会看到三个已注册的工具：
- `analyze_file`
- `count_lines`
- `search_in_files`
**步骤2**：选择 `analyze_file`，在输入区域填入：
```json
{
"file_path": "/home/user/example.py"
}
```
**步骤3**：点击 **Run Tool**，右侧会显示完整的返回结果。
**步骤4**：测试错误场景，传入不存在的路径：
```json
{
"file_path": "/nonexistent/file.txt"
}
```
预期返回错误信息：`"文件不存在: /nonexistent/file.txt"`
### 2.4 使用Inspector调试Resource
点击左侧 **Resources** 标签页：
- 你会看到 `file://{file_path}` 模板的Resource
- 输入具体的URI模板参数，如 `file:///home/user/example.py`
- 点击读取，即可看到文件内容
### 2.5 Inspector进阶技巧
**技巧1：环境变量注入**
```bash
# 通过Inspector传递环境变量给服务器
npx @modelcontextprotocol/inspector \
-e LOG_LEVEL=debug \
-e MAX_FILE_SIZE=52428800 \
uv run src/server/analyzer.py
```
**技巧2：查看原始JSON-RPC通信**
Inspector界面底部的 **Messages** 标签页会显示所有客户端与服务器之间的原始JSON-RPC请求和响应。这对于理解MCP协议格式非常有帮助。
**技巧3：保存调试会话**
Inspector支持导出当前工具调用配置为JSON文件，方便回归测试时复用。
---
## 第三章：日志与错误处理
### 3.1 FastMCP日志体系
FastMCP提供了两层日志机制：
```
┌─────────────────────────────────────┐
│ MCP Client Logging │ ← 通过Context发送到客户端
│ (ctx.info/debug/warning/error) │
├─────────────────────────────────────┤
│ Server-side Python Logging │ ← 标准logging模块
│ (logging.getLogger("fastmcp")) │
└─────────────────────────────────────┘
```
### 3.2 改造服务器：添加完整日志
在 `analyzer.py` 头部添加日志配置：
```python
import logging
# 配置服务器端日志
logger = logging.getLogger("mcp-file-analyzer")
logger.setLevel(logging.DEBUG)
# 文件日志处理器
file_handler = logging.FileHandler("mcp-analyzer.log")
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(
logging.Formatter("%(asctime)s [%(levelname)s] %(name)s: %(message)s")
)
logger.addHandler(file_handler)
# 控制台日志处理器（调试用）
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
console_handler.setFormatter(
logging.Formatter("[%(levelname)s] %(message)s")
)
logger.addHandler(console_handler)
```
### 3.3 在Tool中使用双层日志
改造 `search_in_files` 工具，展示完整的日志和错误处理模式：
```python
@mcp.tool()
async def search_in_files(
directory: str,
pattern: str,
file_extensions: list[str] | None = None,
max_results: int = 50,
ctx: Context = None
) -> dict:
"""在目录中递归搜索包含指定模式的文件。
Args:
directory: 搜索的根目录
pattern: 搜索的文本模式
file_extensions: 限制搜索的文件扩展名列表
max_results: 最大返回结果数（默认50，上限200）
ctx: FastMCP上下文
Returns:
匹配结果摘要
"""
# 参数校验
max_results = min(max(1, max_results), 200)
if not pattern.strip():
raise ValueError("搜索模式不能为空")
dir_path = Path(directory).resolve()
if not dir_path.is_dir():
raise ValueError(f"目录不存在: {directory}")
# 服务端日志（写入文件）
logger.info(
"搜索请求: dir=%s, pattern='%s', exts=%s",
dir_path, pattern, file_extensions
)
# 客户端日志（发送给Claude Code）
if ctx:
await ctx.info(f"🔍 开始搜索 '{pattern}'...")
matches = []
files_scanned = 0
errors = []
extensions = set(file_extensions) if file_extensions else None
for file_path in dir_path.rglob("*"):
if not file_path.is_file():
continue
if extensions and file_path.suffix.lower() not in extensions:
continue
try:
file_size = file_path.stat().st_size
except OSError as e:
errors.append(f"无法读取文件信息 {file_path}: {e}")
continue
if file_size > 10 * 1024 * 1024:
continue
files_scanned += 1
try:
content = file_path.read_text(encoding="utf-8", errors="ignore")
if pattern.lower() in content.lower():
line_numbers = [
i + 1 for i, line in enumerate(content.splitlines())
if pattern.lower() in line.lower()
]
matches.append({
"file": str(file_path.relative_to(dir_path)),
"match_count": len(line_numbers),
"first_matches": line_numbers[:5],
})
except PermissionError:
errors.append(f"权限不足: {file_path}")
logger.warning("权限不足: %s", file_path)
except Exception as e:
errors.append(f"读取错误 {file_path}: {e}")
logger.debug("读取文件失败: %s - %s", file_path, e)
continue
# 进度报告
if ctx and files_scanned % 100 == 0:
await ctx.info(
f"📊 已扫描 {files_scanned} 个文件，"
f"匹配 {len(matches)} 个..."
)
result = {
"pattern": pattern,
"directory": str(dir_path),
"files_scanned": files_scanned,
"files_matched": len(matches),
"matches": sorted(
matches, key=lambda x: x["match_count"], reverse=True
)[:max_results],
"errors_count": len(errors),
"errors_sample": errors[:5],
}
# 服务端日志
logger.info(
"搜索完成: 扫描 %d 文件，匹配 %d，错误 %d",
files_scanned, len(matches), len(errors)
)
# 客户端通知
if ctx:
await ctx.info(
f"✅ 搜索完成！找到 {len(matches)} 个匹配文件"
)
return result
```
### 3.4 全局错误处理中间件
FastMCP支持错误处理装饰器，统一管理工具异常：
```python
from functools import wraps
def handle_tool_errors(func):
"""统一错误处理装饰器"""
@wraps(func)
async def wrapper(*args, **kwargs):
try:
return await func(*args, **kwargs)
except ValueError as e:
logger.warning("参数错误 in %s: %s", func.__name__, e)
return {"error": str(e), "type": "validation_error"}
except PermissionError as e:
logger.error("权限错误 in %s: %s", func.__name__, e)
return {"error": str(e), "type": "permission_error"}
except FileNotFoundError as e:
logger.error("文件不存在 in %s: %s", func.__name__, e)
return {"error": str(e), "type": "not_found_error"}
except Exception as e:
logger.exception("未预期错误 in %s: %s", func.__name__, e)
return {"error": "内部服务器错误", "type": "internal_error"}
return wrapper
```
---
## 第四章：测试策略
### 4.1 测试架构
MCP服务器的测试分为三层：
```mermaid
graph TD
A[测试金字塔] --> B[单元测试
测试单个函数逻辑]
A --> C[工具集成测试
通过FastMCP Client调用]
A --> D[端到端测试
通过Claude Code实际调用]
B --> B1["pytest + 直接函数调用"]
C --> C1["FastMCP Client + stdio"]
D --> D1["Claude Code -p 模式"]
style B fill:#4CAF50,color:white
style C fill:#2196F3,color:white
style D fill:#FF9800,color:white
```
### 4.2 单元测试
创建 `tests/test_analyzer.py`：
```python
"""MCP File Analyzer 单元测试"""
import pytest
import tempfile
from pathlib import Path
from server.analyzer import (
analyze_file,
count_lines,
_human_readable_size,
_guess_mime,
)
@pytest.fixture
def sample_file(tmp_path):
"""创建临时测试文件"""
file = tmp_path / "test.py"
file.write_text(
'"""模块文档"""\n\n'
'def hello():\n'
' print("Hello")\n'
'\n'
'# 注释行\n'
'class Foo:\n'
' pass\n',
encoding="utf-8"
)
return file
@pytest.fixture
def large_file(tmp_path):
"""创建大文件"""
file = tmp_path / "large.txt"
lines = [f"Line {i}\n" for i in range(1000)]
file.write_text("".join(lines), encoding="utf-8")
return file
class TestAnalyzeFile:
def test_basic_analysis(self, sample_file):
result = analyze_file(str(sample_file))
assert result["name"] == "test.py"
assert result["extension"] == ".py"
assert result["size_bytes"] > 0
assert "modified" in result
assert "created" in result
assert "permissions" in result
def test_nonexistent_file(self):
with pytest.raises(ValueError, match="文件不存在"):
analyze_file("/nonexistent/path.txt")
def test_directory_instead_of_file(self, tmp_path):
with pytest.raises(ValueError, match="不是文件"):
analyze_file(str(tmp_path))
class TestCountLines:
def test_line_counting(self, sample_file):
result = count_lines(str(sample_file))
assert result["total_lines"] == 7
assert result["blank_lines"] == 1
assert result["code_lines"] == 6
assert result["chars"] > 0
def test_large_file(self, large_file):
result = count_lines(str(large_file))
assert result["total_lines"] == 1000
assert result["code_lines"] == 1000
assert result["blank_lines"] == 0
def test_include_blank_flag(self, sample_file):
result = count_lines(str(sample_file), include_blank=True)
assert "total_lines" in result
class TestHelperFunctions:
@pytest.mark.parametrize("size,expected", [
(0, "0.0 B"),
(1023, "1023.0 B"),
(1024, "1.0 KB"),
(1048576, "1.0 MB"),
(1073741824, "1.0 GB"),
])
def test_human_readable_size(self, size, expected):
assert _human_readable_size(size) == expected
def test_guess_mime_python(self):
assert _guess_mime(Path("test.py")) == "text/x-python"
def test_guess_mime_unknown(self):
result = _guess_mime(Path("test.xyz123"))
assert result == "application/octet-stream"
```
### 4.3 MCP集成测试
创建 `tests/test_mcp_integration.py`：
```python
"""MCP集成测试 - 通过FastMCP Client与服务器交互"""
import pytest
import tempfile
from pathlib import Path
from fastmcp import Client
# 获取MCP服务器实例
from server.analyzer import mcp
@pytest.fixture
def test_workspace(tmp_path):
"""创建完整的测试工作区"""
# Python文件
(tmp_path / "main.py").write_text(
"import os\n\nprint(os.getcwd())\n"
)
# JavaScript文件
(tmp_path / "app.js").write_text(
"const x = 1;\n// comment\nconsole.log(x);\n"
)
# 空文件
(tmp_path / "empty.txt").write_text("")
# 子目录
sub = tmp_path / "subdir"
sub.mkdir()
(sub / "helper.py").write_text("def help():\n pass\n")
return tmp_path
@pytest.mark.asyncio
async def test_tool_list():
"""测试工具列表是否正确注册"""
async with Client(mcp) as client:
tools = await client.list_tools()
tool_names = [t.name for t in tools]
assert "analyze_file" in tool_names
assert "count_lines" in tool_names
assert "search_in_files" in tool_names
@pytest.mark.asyncio
async def test_analyze_file_via_mcp(test_workspace):
"""通过MCP协议调用analyze_file"""
async with Client(mcp) as client:
result = await client.call_tool(
"analyze_file",
{"file_path": str(test_workspace / "main.py")}
)
assert result is not None
# 验证返回结构
data = result.data if hasattr(result, 'data') else result
assert data["extension"] == ".py"
@pytest.mark.asyncio
async def test_count_lines_via_mcp(test_workspace):
"""通过MCP协议调用count_lines"""
async with Client(mcp) as client:
result = await client.call_tool(
"count_lines",
{"file_path": str(test_workspace / "app.js")}
)
data = result.data if hasattr(result, 'data') else result
assert data["total_lines"] == 3
@pytest.mark.asyncio
async def test_search_in_files_via_mcp(test_workspace):
"""通过MCP协议调用search_in_files"""
async with Client(mcp) as client:
result = await client.call_tool(
"search_in_files",
{
"directory": str(test_workspace),
"pattern": "import",
"file_extensions": [".py"]
}
)
data = result.data if hasattr(result, 'data') else result
assert data["files_matched"] >= 1
assert data["files_scanned"] >= 2 # main.py + subdir/helper.py
@pytest.mark.asyncio
async def test_resource_reading(test_workspace):
"""测试Resource读取"""
async with Client(mcp) as client:
resources = await client.list_resources()
# Resource模板应可被发现
assert len(resources) >= 0 # 模板Resource可能不直接列出
@pytest.mark.asyncio
async def test_error_handling_via_mcp():
"""测试通过MCP的错误处理"""
async with Client(mcp) as client:
# 调用不存在的文件应返回错误
result = await client.call_tool(
"analyze_file",
{"file_path": "/nonexistent/file.txt"}
)
# FastMCP会将异常转化为错误响应
assert result is not None
```
### 4.4 运行测试
```bash
# 运行所有测试
uv run pytest tests/ -v
# 只运行单元测试
uv run pytest tests/test_analyzer.py -v
# 只运行集成测试
uv run pytest tests/test_mcp_integration.py -v
# 带覆盖率报告
uv run pytest tests/ -v --cov=server --cov-report=html
# 监视模式（开发时使用）
uv run pytest-watch tests/ -v
```
预期输出：
```
tests/test_analyzer.py::TestAnalyzeFile::test_basic_analysis PASSED
tests/test_analyzer.py::TestAnalyzeFile::test_nonexistent_file PASSED
tests/test_analyzer.py::TestAnalyzeFile::test_directory_instead_of_file PASSED
tests/test_analyzer.py::TestCountLines::test_line_counting PASSED
tests/test_analyzer.py::TestCountLines::test_large_file PASSED
tests/test_mcp_integration.py::test_tool_list PASSED
tests/test_mcp_integration.py::test_analyze_file_via_mcp PASSED
tests/test_mcp_integration.py::test_search_in_files_via_mcp PASSED
======================== 12 passed in 2.34s ========================
```
---
## 第五章：接入Claude Code
### 5.1 配置MCP服务器连接
在Claude Code中添加MCP服务器，编辑项目级的 `.claude/settings.json`：
```json
{
"mcpServers": {
"file-analyzer": {
"command": "uv",
"args": [
"--directory",
"/path/to/mcp-file-analyzer",
"run",
"src/server/analyzer.py"
],
"env": {
"LOG_LEVEL": "INFO"
}
}
}
}
```
或者通过命令行直接添加：
```bash
claude mcp add file-analyzer \
uv --directory /path/to/mcp-file-analyzer run src/server/analyzer.py \
-e LOG_LEVEL=INFO
```
### 5.2 验证连接
```bash
# 列出已配置的MCP服务器
claude mcp list
# 查看特定服务器的工具列表
claude mcp tools file-analyzer
```
### 5.3 实际使用
启动Claude Code交互会话后，你可以直接用自然语言触发MCP工具：
```
你: 分析一下 /home/user/project/main.py 的文件信息
Claude Code 会自动调用 file-analyzer 的 analyze_file 工具，
返回文件的详细分析结果。
```
```
你: 帮我在 /home/user/project 目录下搜索所有包含 "TODO" 的 Python 文件
Claude Code 会调用 search_in_files 工具，限定 .py 扩展名。
```
### 5.4 常见连接问题排查
| 问题 | 原因 | 解决方案 |
|------|------|----------|
| `MCP server not found` | 路径配置错误 | 检查`--directory`路径是否为绝对路径 |
| `Server exited immediately` | Python依赖缺失 | 运行`uv sync`安装依赖 |
| `Tool not responding` | stdio缓冲问题 | 确保`mcp.run(transport="stdio")`正确调用 |
| `Permission denied` | 文件权限 | 检查工作目录和目标文件的读写权限 |
| `Timeout` | 工具执行太慢 | 添加超时限制，优化大目录搜索逻辑 |
**调试技巧**：如果Claude Code中MCP服务器无法启动，先用Inspector独立测试：
```bash
# 脱离Claude Code，单独用Inspector测试
npx @modelcontextprotocol/inspector uv run src/server/analyzer.py
```
这可以排除Claude Code配置问题，确认服务器本身是否正常。
---
## 第六章：生产级部署实践
### 6.1 部署架构选择
```mermaid
graph LR
subgraph "本地开发"
A[Claude Code CLI] -->|stdio| B[MCP Server
analyzer.py]
end
subgraph "团队共享"
C[Claude Code] -->|stdio| D[uvx 启动
从PyPI安装]
D --> B2[MCP Server]
end
subgraph "远程服务"
E[Claude Code] -->|SSE/HTTP| F[Nginx 反向代理]
F --> G[MCP Server
SSE模式]
G --> H[(共享存储)]
end
style A fill:#4CAF50,color:white
style C fill:#2196F3,color:white
style E fill:#FF9800,color:white
```
### 6.2 方案一：本地stdio部署（推荐起步）
最简单的方式，每个开发者本地运行：
```bash
# 打包发布到PyPI（团队内部分发）
uv build
uv publish --index internal
# 团队成员安装使用
claude mcp add file-analyzer -- uvx mcp-file-analyzer
```
### 6.3 方案二：SSE远程部署
适用于团队共享、需要集中管理的场景。改造服务器入口：
```python
# src/server/analyzer_sse.py
from server.analyzer import mcp
if __name__ == "__main__":
# SSE模式，监听指定端口
mcp.run(
transport="sse",
host="0.0.0.0",
port=8080,
)
```
使用Docker容器化部署：
```dockerfile
# Dockerfile
FROM python:3.12-slim
WORKDIR /app
COPY pyproject.toml .
COPY src/ src/
RUN pip install --no-cache-dir .
EXPOSE 8080
CMD ["python", "-m", "server.analyzer_sse"]
```
```bash
# 构建和运行
docker build -t mcp-file-analyzer .
docker run -d -p 8080:8080 \
-v /shared/data:/data \
--name file-analyzer \
mcp-file-analyzer
```
Claude Code配置远程SSE服务器：
```json
{
"mcpServers": {
"file-analyzer-remote": {
"type": "sse",
"url": "http://file-analyzer.internal:8080/sse"
}
}
}
```
### 6.4 安全加固清单
生产环境必须关注的安全项：
```python
# 1. 路径遍历防护
ALLOWED_DIRS = [Path("/data/workspace").resolve()]
def validate_path(file_path: str) -> Path:
"""防止路径遍历攻击"""
path = Path(file_path).resolve()
if not any(path.is_relative_to(allowed) for allowed in ALLOWED_DIRS):
raise ValueError(f"路径不在允许范围内: {file_path}")
return path
# 2. 资源限制
MAX_FILE_SIZE = 50 * 1024 * 1024 # 50MB
MAX_SEARCH_RESULTS = 200
MAX_CONCURRENT_REQUESTS = 10
# 3. 输入校验
import re
def validate_pattern(pattern: str) -> str:
"""防止ReDoS攻击"""
if len(pattern) > 500:
raise ValueError("搜索模式过长")
return pattern
# 4. 速率限制（SSE模式）
from time import time
_request_times: list[float] = []
def check_rate_limit():
"""简单的滑动窗口速率限制"""
now = time()
_request_times[:] = [t for t in _request_times if now - t < 60]
if len(_request_times) >= MAX_CONCURRENT_REQUESTS * 6:
raise ValueError("请求过于频繁，请稍后再试")
_request_times.append(now)
```
---
## 第七章：完整工作流程图
```mermaid
flowchart TB
START([开始开发MCP服务器]) --> INIT[uv init + uv add fastmcp]
INIT --> CODE[编写Tool/Resource/Prompt]
CODE --> UNIT_TEST[编写单元测试
pytest]
UNIT_TEST --> PASS1{测试通过?}
PASS1 -->|否| CODE
PASS1 -->|是| INSPECTOR[MCP Inspector
可视化调试]
INSPECTOR --> INTEGRATION[编写集成测试
FastMCP Client]
INTEGRATION --> PASS2{测试通过?}
PASS2 -->|否| CODE
PASS2 -->|是| CONNECT[接入Claude Code
claude mcp add]
CONNECT --> VERIFY[端到端验证
自然语言调用]
VERIFY --> PROD{部署方式?}
PROD -->|个人/小团队| LOCAL[uvx 本地部署
stdio传输]
PROD -->|企业/远程| REMOTE[Docker + SSE
远程部署]
LOCAL --> DONE([🎉 上线])
REMOTE --> DONE
style START fill:#4CAF50,color:white
style DONE fill:#FF9800,color:white
style INSPECTOR fill:#2196F3,color:white
```
---
## 第八章：实战案例——代码质量审查MCP服务器
将前面所学整合为一个完整的代码审查工具：
```python
"""Code Review MCP Server - 自动化代码质量审查"""
import ast
import re
from pathlib import Path
from dataclasses import dataclass, asdict
from fastmcp import FastMCP, Context
mcp = FastMCP(
name="code-reviewer",
version="1.0.0",
description="Python代码质量审查MCP服务器"
)
# 定义问题严重级别
SEVERITY_INFO = "info"
SEVERITY_WARNING = "warning"
SEVERITY_ERROR = "error"
@dataclass
class CodeIssue:
"""代码问题"""
file: str
line: int
severity: str
rule: str
message: str
def _check_line_length(lines: list[str], filename: str, limit: int = 120) -> list[CodeIssue]:
"""检查行长度"""
issues = []
for i, line in enumerate(lines, 1):
if len(line.rstrip()) > limit:
issues.append(CodeIssue(
file=filename, line=i,
severity=SEVERITY_WARNING,
rule="line-too-long",
message=f"行长度 {len(line.rstrip())} 超过限制 {limit}"
))
return issues
def _check_todo_fixme(lines: list[str], filename: str) -> list[CodeIssue]:
"""检查TODO/FIXME标记"""
issues = []
pattern = re.compile(r'#\s*(TODO|FIXME|HACK|XXX)', re.IGNORECASE)
for i, line in enumerate(lines, 1):
match = pattern.search(line)
if match:
issues.append(CodeIssue(
file=filename, line=i,
severity=SEVERITY_INFO,
rule=f"marker-{match.group(1).lower()}",
message=f"发现 {match.group(1)} 标记: {line.strip()}"
))
return issues
def _check_python_ast(code: str, filename: str) -> list[CodeIssue]:
"""使用AST进行语法级别检查"""
issues = []
try:
tree = ast.parse(code, filename)
except SyntaxError as e:
issues.append(CodeIssue(
file=filename, line=e.lineno or 0,
severity=SEVERITY_ERROR,
rule="syntax-error",
message=f"语法错误: {e.msg}"
))
return issues
for node in ast.walk(tree):
# 检查过深的嵌套
if isinstance(node, (ast.If, ast.For, ast.While)):
depth = 0
parent = node
while hasattr(parent, 'parent'):
depth += 1
parent = getattr(parent, 'parent', None)
if depth > 4:
issues.append(CodeIssue(
file=filename, line=node.lineno,
severity=SEVERITY_WARNING,
rule="deep-nesting",
message=f"嵌套深度 {depth} 超过建议值 4"
))
# 检查过长的参数列表
if isinstance(node, ast.FunctionDef):
if len(node.args.args) > 7:
issues.append(CodeIssue(
file=filename, line=node.lineno,
severity=SEVERITY_WARNING,
rule="too-many-params",
message=f"函数 '{node.name}' 有 {len(node.args.args)} 个参数（建议 ≤7）"
))
# 检查缺失docstring
docstring = ast.get_docstring(node)
if not docstring and not node.name.startswith("_"):
issues.append(CodeIssue(
file=filename, line=node.lineno,
severity=SEVERITY_INFO,
rule="missing-docstring",
message=f"公开函数 '{node.name}' 缺少docstring"
))
return issues
@mcp.tool()
async def review_python_file(
file_path: str,
checks: list[str] | None = None,
ctx: Context = None
) -> dict:
"""审查Python文件的代码质量。
Args:
file_path: Python文件路径
checks: 要执行的检查项列表，默认全部。
可选: line-length, todo-markers, ast-checks
ctx: MCP上下文
Returns:
审查结果摘要
"""
path = Path(file_path).resolve()
if not path.exists():
raise ValueError(f"文件不存在: {file_path}")
if path.suffix != ".py":
raise ValueError("仅支持Python文件审查")
if ctx:
await ctx.info(f"📋 开始审查: {path.name}")
code = path.read_text(encoding="utf-8", errors="replace")
lines = code.splitlines()
all_checks = checks or ["line-length", "todo-markers", "ast-checks"]
issues: list[CodeIssue] = []
if "line-length" in all_checks:
issues.extend(_check_line_length(lines, path.name))
if "todo-markers" in all_checks:
issues.extend(_check_todo_fixme(lines, path.name))
if "ast-checks" in all_checks:
issues.extend(_check_python_ast(code, path.name))
# 统计
summary = {
"file": str(path),
"total_lines": len(lines),
"total_issues": len(issues),
"by_severity": {
"error": sum(1 for i in issues if i.severity == SEVERITY_ERROR),
"warning": sum(1 for i in issues if i.severity == SEVERITY_WARNING),
"info": sum(1 for i in issues if i.severity == SEVERITY_INFO),
},
"issues": [asdict(i) for i in sorted(issues, key=lambda x: (
{"error": 0, "warning": 1, "info": 2}[x.severity],
x.line
))],
}
if ctx:
await ctx.info(
f"✅ 审查完成: {summary['total_issues']} 个问题 "
f"(🔴{summary['by_severity']['error']} "
f"🟡{summary['by_severity']['warning']} "
f"ℹ️{summary['by_severity']['info']})"
)
return summary
@mcp.tool()
async def review_directory(
directory: str,
ctx: Context = None
) -> dict:
"""批量审查目录下所有Python文件的代码质量。
Args:
directory: 目标目录路径
ctx: MCP上下文
Returns:
批量审查结果
"""
dir_path = Path(directory).resolve()
if not dir_path.is_dir():
raise ValueError(f"目录不存在: {directory}")
py_files = list(dir_path.rglob("*.py"))
if ctx:
await ctx.info(f"📂 发现 {len(py_files)} 个Python文件")
results = []
for i, py_file in enumerate(py_files):
try:
result = await review_python_file(str(py_file), ctx=ctx)
results.append(result)
except Exception as e:
results.append({
"file": str(py_file),
"error": str(e)
})
if ctx and (i + 1) % 10 == 0:
await ctx.info(f"进度: {i + 1}/{len(py_files)}")
total_issues = sum(r.get("total_issues", 0) for r in results)
return {
"directory": str(dir_path),
"files_reviewed": len(results),
"total_issues": total_issues,
"file_results": results,
}
if __name__ == "__main__":
mcp.run(transport="stdio")
```
使用示例：
```bash
# 添加到Claude Code
claude mcp add code-reviewer \
uv --directory /path/to/project run src/server/code_reviewer.py
# 在Claude Code中使用
# 你: "帮我审查一下 main.py 的代码质量"
# 你: "审查整个 src 目录的Python代码"
```
---
## 常见问题FAQ
### Q1: FastMCP和mcp底层SDK有什么区别？
FastMCP是`mcp` SDK的高层封装。`mcp` SDK提供底层Server/Client类，需要手动处理协议细节；FastMCP通过装饰器自动处理JSON Schema生成、协议握手、错误序列化等。简单项目用FastMCP，需要精细控制协议行为时用底层SDK。
### Q2: stdio和SSE传输怎么选？
- **stdio**：Claude Code直接启动你的Python进程作为子进程，通信通过stdin/stdout。适合个人开发、本地工具。配置简单，零网络依赖。
- **SSE（Server-Sent Events）**：服务器作为独立进程监听HTTP端口，客户端通过网络连接。适合团队共享、远程部署、需要集中管理的场景。
### Q3: MCP服务器崩溃了怎么排查？
1. 先用`npx @modelcontextprotocol/inspector`独立测试，排除Claude Code配置问题
2. 检查`mcp-analyzer.log`日志文件
3. 确认Python依赖是否完整安装（`uv sync`）
4. 检查stdio输出是否有非MCP协议的print语句干扰（这是最常见的坑——Tool函数中的`print()`会污染stdio通道）
### Q4: 如何处理大文件/大目录的性能问题？
- 设置`MAX_FILE_SIZE`跳过超大文件
- `search_in_files`中限制`max_results`
- 对目录遍历使用`rglob`而非递归Python代码
- 考虑使用ripgrep（通过subprocess调用）替代Python文件搜索
### Q5: 多个MCP服务器之间能通信吗？
不能直接通信。每个MCP服务器独立运行，由Claude Code作为中间协调层。如果需要数据共享，可以通过共享文件系统或数据库实现。
---
## 延伸阅读
- [MCP官方文档 - Build a Server](https://modelcontextprotocol.io/docs/develop/build-server) — 官方weather server教程
- [FastMCP官方Wiki](https://fastmcp.wiki) — FastMCP完整API参考
- [MCP Inspector GitHub](https://github.com/modelcontextprotocol/inspector) — 调试工具源码和使用指南
- [Claude Code MCP文档](https://code.claude.com/docs/en/mcp) — Claude Code的MCP配置指南
- [MCP Best Practices](https://modelcontextprotocol.info/docs/best-practices/) — 生产级架构与实现指南
---
## 训练营适用性评估
| 维度 | 评估 |
|------|------|
| **适合训练营类型** | 企业内训（AI工具链开发）、技术工作坊（MCP生态）、高级开发者训练营 |
| **预计教学时长** | 4-6小时（含实操练习） |
| **难度等级** | ⭐⭐⭐⭐（中高级，需要Python异步编程和MCP基础） |
| **实操占比** | 70%（大量动手编码和调试练习） |
| **前置训练营** | 建议先完成"Claude Code入门"和"MCP服务器基础开发" |
| **考核方式** | 学员独立开发一个带完整测试的MCP服务器并通过Inspector验证 |
---
## 参考来源
1. [Build an MCP server - Model Context Protocol](https://modelcontextprotocol.io/docs/develop/build-server) — MCP官方服务器开发教程，weather server完整示例
2. [FastMCP GitHub - PrefectHQ](https://github.com/prefecthq/fastmcp) — FastMCP框架源码与API文档，25.6k stars
3. [MCP Inspector - GitHub](https://github.com/modelcontextprotocol/inspector) — 官方MCP可视化调试工具，10k stars
4. [Claude Code MCP Documentation](https://code.claude.com/docs/en/mcp) — Claude Code的MCP连接与配置官方文档
5. [Building MCP Servers in Python: a production primer for 2026 - DEV Community](https://dev.to/tufailkhan457/building-mcp-servers-in-python-a-production-primer-for-2026-4kh2) — Python MCP生产级部署实践指南
> [!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 等场景全覆盖。
