# FastMCP 3.0 Python开发与调试测试全链路
## 概述
本教程面向有Python基础、希望用FastMCP 3.0构建MCP Server的开发者。你将学会从零搭建一个完整的Python MCP服务，掌握FastMCP 3.0的三大新特性（组合、生命周期、中间件），使用pytest进行自动化测试，利用MCP Inspector进行可视化调试，最终接入Claude Code投入实际使用。适合全栈开发者、工具链工程师和AI应用构建者。
## 前置条件
- Python 3.10+ 已安装
- Claude Code CLI 已安装（用于最终集成）
- 了解MCP基本概念（Tool、Resource、Prompt三种原语）
- 熟悉Python装饰器和异步编程（async/await）
---
## 一、为什么选择FastMCP 3.0 + Python
MCP协议的官方Python SDK提供了底层实现，但直接使用需要手动处理JSON-RPC协议细节、连接管理和错误恢复。FastMCP 3.0在此基础上封装了更高级的抽象，让开发者专注于业务逻辑。
FastMCP 3.0的三项核心改进：
| 特性 | FastMCP 2.x | FastMCP 3.0 |
|------|-------------|-------------|
| 服务器组合 | 专用mount()方法 | 通用`+`运算符，任意组合 |
| 生命周期 | 全局初始化 | 组件级Lifespan，可组合 |
| 认证 | 全局OAuth | 按组件粒度的auth handler |
| 测试 | 无内置支持 | 内置Client + pytest集成 |
| 中间件 | 无 | 装饰器式中间件链 |
FastMCP 3.0的月下载量已突破数百万，成为Python生态中最流行的MCP框架。
```mermaid
graph TB
subgraph "FastMCP 3.0 架构"
A[FastMCP App] --> B[Tools 注册]
A --> C[Resources 注册]
A --> D[Prompts 注册]
A --> E[Lifespan 生命周期]
A --> F[Middleware 中间件]
A --> G[Auth 认证层]
end
subgraph "组合能力"
H[Server A] --> I["Server A + Server B"]
J[Server B] --> I
I --> K[统一入口]
end
subgraph "运行时"
K --> L[stdio 传输]
K --> M[Streamable HTTP]
end
style A fill:#e1f5fe
style I fill:#fff3e0
```
---
## 二、环境搭建与项目初始化
### 2.1 创建项目结构
```bash
mkdir mcp-weather-server && cd mcp-weather-server
python -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
```
### 2.2 安装依赖
```bash
pip install fastmcp>=3.0 httpx pytest pytest-asyncio
```
验证安装：
```bash
python -c "import fastmcp; print(fastmcp.__version__)"
# 预期输出: 3.x.x
```
### 2.3 项目目录结构
```
mcp-weather-server/
├── .venv/
├── src/
│ ├── __init__.py
│ ├── server.py # 主服务器定义
│ ├── tools.py # 工具函数
│ ├── resources.py # 资源定义
│ └── prompts.py # 提示模板
├── tests/
│ ├── conftest.py # pytest fixture
│ ├── test_tools.py # 工具测试
│ └── test_resources.py # 资源测试
├── pyproject.toml
└── .mcp.json # Claude Code配置
```
---
## 三、构建第一个Tool：天气查询
### 3.1 定义服务器核心
创建 `src/server.py`：
```python
from fastmcp import FastMCP
# 创建MCP服务器实例
mcp = FastMCP(
name="Weather Server",
version="1.0.0",
instructions="一个天气查询MCP服务器，支持获取实时天气和预报信息。"
)
```
### 3.2 注册Tool
FastMCP 3.0使用装饰器注册Tool，支持同步和异步函数：
```python
# src/tools.py
import httpx
from fastmcp import FastMCP
def register_tools(mcp: FastMCP):
"""注册所有工具到MCP服务器"""
@mcp.tool()
async def get_weather(city: str) -> str:
"""获取指定城市的当前天气信息
Args:
city: 城市名称，如"北京"、"Tokyo"、"New York"
"""
# 使用wttr.in免费天气API
async with httpx.AsyncClient() as client:
resp = await client.get(
f"https://wttr.in/{city}",
params={"format": "j1"}
)
resp.raise_for_status()
data = resp.json()
current = data["current_condition"][0]
return (
f"🌍 {city} 当前天气:\n"
f"🌡️ 温度: {current['temp_C']}°C "
f"(体感 {current['FeelsLikeC']}°C)\n"
f"💧 湿度: {current['humidity']}%\n"
f"🌬️ 风速: {current['windspeedKmph']} km/h\n"
f"☁️ 天气: {current['weatherDesc'][0]['value']}"
)
@mcp.tool()
async def get_forecast(city: str, days: int = 3) -> str:
"""获取城市未来几天的天气预报
Args:
city: 城市名称
days: 预报天数，1-3天，默认3天
"""
days = max(1, min(days, 3))
async with httpx.AsyncClient() as client:
resp = await client.get(
f"https://wttr.in/{city}",
params={"format": "j1"}
)
resp.raise_for_status()
data = resp.json()
lines = [f"📅 {city} 未来{days}天预报:\n"]
for day in data["weather"][:days]:
date = day["date"]
max_t = day["maxtempC"]
min_t = day["mintempC"]
desc = day["hourly"][4]["weatherDesc"][0]["value"]
lines.append(f" {date}: {desc}, {min_t}°C ~ {max_t}°C")
return "\n".join(lines)
```
在 `src/server.py` 中引入：
```python
from fastmcp import FastMCP
from tools import register_tools
mcp = FastMCP(
name="Weather Server",
version="1.0.0",
instructions="天气查询MCP服务器"
)
# 注册工具模块
register_tools(mcp)
if __name__ == "__main__":
mcp.run() # 默认stdio传输
```
**关键要点：**
- `@mcp.tool()` 装饰器自动从函数签名提取参数schema
- docstring会作为工具描述传递给LLM
- `Args` 部分的参数说明会被解析为参数描述
- 返回值自动转为TextContent
### 3.3 添加Resource
Resource为LLM提供可读取的数据上下文：
```python
# src/resources.py
from fastmcp import FastMCP
def register_resources(mcp: FastMCP):
@mcp.resource("weather://cities/supported")
def get_supported_cities() -> str:
"""返回支持的查询城市列表"""
return "北京, 上海, 广州, 深圳, Tokyo, New York, London, Paris"
@mcp.resource("weather://cities/{city}/info")
def get_city_info(city: str) -> str:
"""获取城市基本信息"""
cities = {
"北京": {"timezone": "UTC+8", "country": "中国"},
"上海": {"timezone": "UTC+8", "country": "中国"},
"Tokyo": {"timezone": "UTC+9", "country": "日本"},
}
info = cities.get(city, {"timezone": "unknown", "country": "unknown"})
return f"{city}: 时区 {info['timezone']}, 国家 {info['country']}"
```
### 3.4 添加Prompt模板
Prompt模板让LLM可以复用预定义的交互模式：
```python
# src/prompts.py
from fastmcp import FastMCP
def register_prompts(mcp: FastMCP):
@mcp.prompt()
def weather_report(city: str, style: str = "brief") -> str:
"""生成天气报告的提示模板"""
if style == "detailed":
return (
f"请为{city}生成详细天气报告，包含：\n"
f"1. 当前天气概况\n"
f"2. 未来3天预报\n"
f"3. 穿衣建议\n"
f"4. 出行提示\n"
f"请使用 get_weather 和 get_forecast 工具获取数据。"
)
return f"简报：{city}当前天气如何？请用 get_weather 查询。"
```
完整的服务器入口：
```python
# src/server.py
from fastmcp import FastMCP
from tools import register_tools
from resources import register_resources
from prompts import register_prompts
mcp = FastMCP(
name="Weather Server",
version="1.0.0",
instructions="天气查询MCP服务器，支持实时天气和预报查询"
)
register_tools(mcp)
register_resources(mcp)
register_prompts(mcp)
if __name__ == "__main__":
mcp.run()
```
---
## 四、FastMCP 3.0 进阶：生命周期与组合
### 4.1 Lifespan 生命周期管理
FastMCP 3.0引入了组件级Lifespan，允许在服务器启动/关闭时执行初始化和清理：
```python
from contextlib import asynccontextmanager
from fastmcp import FastMCP, Context
@asynccontextmanager
async def server_lifespan(server: FastMCP):
"""服务器级生命周期"""
# 启动时：初始化HTTP连接池
print("🚀 Weather Server 启动中...")
http_client = httpx.AsyncClient(timeout=30.0)
yield {"http_client": http_client}
# 关闭时：清理资源
await http_client.aclose()
print("👋 Weather Server 已关闭")
mcp = FastMCP(
name="Weather Server",
lifespan=server_lifespan
)
@mcp.tool()
async def get_weather_v2(ctx: Context, city: str) -> str:
"""使用连接池的天气查询（高性能版本）"""
# 从生命周期上下文获取共享的HTTP客户端
http_client = ctx.request_ctx.lifespan_context["http_client"]
resp = await http_client.get(f"https://wttr.in/{city}?format=j1")
# ... 处理响应
```
**Lifespan的工作原理：**
```mermaid
sequenceDiagram
participant Client as MCP Client
(Claude Code)
participant Server as FastMCP Server
participant Lifespan as Lifespan Manager
participant Tools as Tool Functions
Client->>Server: initialize 请求
Server->>Lifespan: 进入 lifespan 上下文
Note over Lifespan: 创建 http_client
加载配置
建立连接池
Lifespan-->>Server: 返回 context 对象
Client->>Server: tools/call get_weather
Server->>Tools: 执行 tool 函数
Tools->>Lifespan: 读取 context["http_client"]
Tools-->>Server: 返回结果
Server-->>Client: Tool Response
Client->>Server: shutdown
Server->>Lifespan: 退出 lifespan 上下文
Note over Lifespan: 关闭 http_client
清理资源
```
### 4.2 服务器组合（Composition）
FastMCP 3.0的标志性特性——用`+`运算符组合多个服务器：
```python
from fastmcp import FastMCP
# 天气服务
weather_server = FastMCP(name="Weather")
@weather_server.tool()
async def get_weather(city: str) -> str:
return f"{city}: 晴，25°C"
# 地理服务
geo_server = FastMCP(name="Geo")
@geo_server.tool()
async def geocode(address: str) -> str:
return f"{address}: lat=39.9, lon=116.4"
# 新闻服务
news_server = FastMCP(name="News")
@news_server.tool()
async def get_news(topic: str) -> str:
return f"关于{topic}的最新新闻..."
# 用 + 运算符组合三个服务
combined = weather_server + geo_server + news_server
# 运行组合服务器，所有工具统一暴露
if __name__ == "__main__":
combined.run()
```
组合后的服务器暴露所有子服务器的工具、资源和提示，对外如同一个统一服务。这在模块化开发中极为实用——每个团队维护自己的子服务器，最后组合为一个整体。
### 4.3 中间件（Middleware）
FastMCP 3.0支持装饰器式中间件，可以拦截所有Tool调用：
```python
import time
import logging
from fastmcp import FastMCP, Context
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("weather-mcp")
mcp = FastMCP(name="Weather Server")
@mcp.middleware("tool")
async def log_tool_call(ctx: Context, call_next):
"""记录所有工具调用的中间件"""
tool_name = ctx.tool_name
start = time.time()
logger.info(f"🔧 调用工具: {tool_name}")
try:
result = await call_next(ctx)
elapsed = time.time() - start
logger.info(f"✅ {tool_name} 完成 ({elapsed:.2f}s)")
return result
except Exception as e:
elapsed = time.time() - start
logger.error(f"❌ {tool_name} 失败 ({elapsed:.2f}s): {e}")
raise
```
---
## 五、自动化测试：pytest + FastMCP Client
这是FastMCP 3.0最受开发者欢迎的特性之一。内置的Client让你无需启动真实服务器，就能在内存中测试所有MCP原语。
### 5.1 测试基础设施
创建 `tests/conftest.py`：
```python
import pytest
from fastmcp import Client
from src.server import mcp
@pytest.fixture
async def client():
"""创建内存中的MCP客户端，连接到我们的服务器"""
async with Client(mcp) as client:
yield client
```
### 5.2 测试Tool调用
创建 `tests/test_tools.py`：
```python
import pytest
from fastmcp import Client
from src.server import mcp
@pytest.fixture
async def client():
async with Client(mcp) as c:
yield c
@pytest.mark.asyncio
async def test_list_tools(client: Client):
"""验证服务器注册了预期的工具"""
tools = await client.list_tools()
tool_names = [t.name for t in tools]
assert "get_weather" in tool_names
assert "get_forecast" in tool_names
# 验证工具描述不为空
weather_tool = next(t for t in tools if t.name == "get_weather")
assert weather_tool.description
assert "城市" in weather_tool.description
@pytest.mark.asyncio
async def test_call_get_weather(client: Client):
"""测试天气查询工具"""
result = await client.call_tool("get_weather", {"city": "Beijing"})
assert len(result) > 0
content = result[0]
assert "Beijing" in content.text
assert "°C" in content.text
@pytest.mark.asyncio
async def test_call_get_forecast(client: Client):
"""测试预报查询工具"""
result = await client.call_tool("get_forecast", {
"city": "Shanghai",
"days": 2
})
assert len(result) > 0
assert "Shanghai" in result[0].text
assert "2天" in result[0].text
@pytest.mark.asyncio
async def test_get_weather_error_handling(client: Client):
"""测试错误输入的处理"""
# 即使输入奇怪的城市名也不应抛异常
result = await client.call_tool("get_weather", {"city": "xxyyzz123"})
assert len(result) > 0 # 应该有返回（可能是错误信息）
# 参数化测试：批量验证多个城市
@pytest.mark.asyncio
@pytest.mark.parametrize("city", ["Beijing", "Tokyo", "London", "New York"])
async def test_weather_multiple_cities(client: Client, city: str):
"""测试多个城市的天气查询"""
result = await client.call_tool("get_weather", {"city": city})
assert len(result) > 0
assert city in result[0].text
```
### 5.3 测试Resource
创建 `tests/test_resources.py`：
```python
import pytest
from fastmcp import Client
from src.server import mcp
@pytest.fixture
async def client():
async with Client(mcp) as c:
yield c
@pytest.mark.asyncio
async def test_list_resources(client: Client):
"""验证资源注册"""
resources = await client.list_resources()
uris = [str(r.uri) for r in resources]
assert any("cities/supported" in uri for uri in uris)
@pytest.mark.asyncio
async def test_read_supported_cities(client: Client):
"""测试读取支持的城市列表"""
result = await client.read_resource("weather://cities/supported")
assert "北京" in result[0].text
assert "Tokyo" in result[0].text
@pytest.mark.asyncio
async def test_read_city_info(client: Client):
"""测试模板URI资源"""
result = await client.read_resource("weather://cities/北京/info")
assert "UTC+8" in result[0].text
```
### 5.4 测试Prompt模板
```python
import pytest
from fastmcp import Client
from src.server import mcp
@pytest.fixture
async def client():
async with Client(mcp) as c:
yield c
@pytest.mark.asyncio
async def test_list_prompts(client: Client):
prompts = await client.list_prompts()
assert any(p.name == "weather_report" for p in prompts)
@pytest.mark.asyncio
async def test_get_prompt_brief(client: Client):
result = await client.get_prompt("weather_report", {
"city": "北京",
"style": "brief"
})
assert "北京" in result.messages[0].content.text
@pytest.mark.asyncio
async def test_get_prompt_detailed(client: Client):
result = await client.get_prompt("weather_report", {
"city": "上海",
"style": "detailed"
})
content = result.messages[0].content.text
assert "详细" in content or "3天" in content
```
### 5.5 运行测试
```bash
# 运行所有测试
pytest tests/ -v
# 运行特定测试文件
pytest tests/test_tools.py -v
# 带覆盖率报告
pytest tests/ -v --tb=short -q
```
预期输出：
```
tests/test_tools.py::test_list_tools PASSED
tests/test_tools.py::test_call_get_weather PASSED
tests/test_tools.py::test_call_get_forecast PASSED
tests/test_tools.py::test_get_weather_error_handling PASSED
tests/test_tools.py::test_weather_multiple_cities[Beijing] PASSED
tests/test_tools.py::test_weather_multiple_cities[Tokyo] PASSED
tests/test_tools.py::test_weather_multiple_cities[London] PASSED
tests/test_tools.py::test_weather_multiple_cities[New York] PASSED
tests/test_resources.py::test_list_resources PASSED
tests/test_resources.py::test_read_supported_cities PASSED
tests/test_resources.py::test_read_city_info PASSED
===== 11 passed in 3.42s =====
```
**测试架构对比：**
```mermaid
graph LR
subgraph "传统方式（需要启动服务器）"
A1[pytest] --> B1[子进程启动 Server]
B1 --> C1[stdio/SSE 连接]
C1 --> D1[发送 JSON-RPC]
D1 --> E1[解析响应]
end
subgraph "FastMCP Client（内存直连）"
A2[pytest] --> B2[Client mcp]
B2 --> C2[内存中直接调用]
C2 --> D2[原生 Python 返回]
end
style A2 fill:#c8e6c9
style B2 fill:#c8e6c9
```
---
## 六、可视化调试：MCP Inspector
MCP Inspector是官方提供的Web调试工具，可以可视化地测试和调试MCP服务器。
### 6.1 启动Inspector
```bash
# 方法一：直接npx运行
npx @anthropic/mcp-inspector
# 方法二：在项目中配置启动
# package.json (如需)
# "scripts": {
# "inspect": "mcp-inspector"
# }
```
Inspector启动后会在浏览器中打开一个Web界面，默认地址为 `http://localhost:6274`。
### 6.2 连接到本地服务器
在Inspector界面中配置连接参数：
- **Transport Type**: `STDIO`
- **Command**: `python`
- **Args**: `src/server.py`
- **Environment Variables**: 按需添加API密钥等
点击"Connect"，Inspector会启动你的服务器并建立连接。
### 6.3 使用Inspector进行调试
连接成功后，Inspector提供三个核心面板：
**Tools 面板：**
1. 点击左侧列出所有已注册的Tool
2. 选中 `get_weather`
3. 在参数面板中填入 `{"city": "北京"}`
4. 点击"Run Tool"
5. 查看右侧的完整JSON-RPC请求和响应
**Resources 面板：**
1. 点击"List Resources"查看所有可用资源
2. 点击特定资源URI读取内容
3. 模板URI（如 `weather://cities/{city}/info`）会显示参数输入框
**Prompts 面板：**
1. 列出所有已注册的Prompt模板
2. 填入参数并预览生成的提示内容
### 6.4 调试常见问题
| 问题 | 症状 | 解决方法 |
|------|------|---------|
| 服务器启动失败 | Inspector显示连接错误 | 检查Python路径和依赖是否正确 |
| Tool未注册 | List Tools为空 | 确认`@mcp.tool()`装饰器已添加 |
| 参数验证失败 | 调用返回错误 | 检查参数类型和docstring描述 |
| 超时 | Tool调用卡住 | 检查网络请求是否超时，增加timeout |
| 编码问题 | 中文乱码 | 确保使用UTF-8编码 |
### 6.5 命令行快速验证
除了Inspector，也可以用命令行快速验证服务器：
```bash
# 使用MCP官方CLI测试
npx @modelcontextprotocol/inspector python src/server.py
# 或者直接用FastMCP内置的dev模式
fastmcp dev src/server.py
```
---
## 七、接入Claude Code
### 7.1 配置.mcp.json
在项目根目录创建 `.mcp.json`（项目级配置）：
```json
{
"mcpServers": {
"weather": {
"command": "python",
"args": ["src/server.py"],
"cwd": "/absolute/path/to/mcp-weather-server",
"env": {
"PYTHONUNBUFFERED": "1"
}
}
}
}
```
或者配置用户级 `~/.claude.json`（全局可用）。
### 7.2 使用claude mcp命令
```bash
# 添加MCP服务器（推荐方式）
claude mcp add weather -- python src/server.py
# 带环境变量
claude mcp add weather -e API_KEY=xxx -- python src/server.py
# 查看已配置的服务器
claude mcp list
# 验证服务器连接
claude mcp inspect weather
```
### 7.3 在Claude Code中使用
配置完成后，在Claude Code对话中直接使用：
```
> 帮我查一下北京今天的天气
# Claude Code会自动调用 get_weather 工具
# 🌍 北京 当前天气:
# 🌡️ 温度: 28°C (体感 30°C)
# 💧 湿度: 65%
# 🌬️ 风速: 12 km/h
# ☁️ 天气: Partly cloudy
```
### 7.4 Streamable HTTP 远程部署
如果需要远程部署（非本地stdio），使用Streamable HTTP传输：
```python
# src/server.py 末尾修改
if __name__ == "__main__":
mcp.run(
transport="streamable-http",
host="0.0.0.0",
port=8080
)
```
Claude Code配置远程服务器：
```json
{
"mcpServers": {
"weather-remote": {
"type": "streamable-http",
"url": "https://your-server.com/mcp"
}
}
}
```
---
## 八、实战案例：Git仓库分析服务器
将前面学到的所有知识整合为一个实际项目——一个分析Git仓库的MCP服务器，提供代码统计、提交历史和分支管理功能。
### 8.1 完整代码
```python
# git_analyzer_server.py
import subprocess
import json
from contextlib import asynccontextmanager
from fastmcp import FastMCP, Context
@asynccontextmanager
async def lifespan(server: FastMCP):
"""验证git可用性"""
try:
result = subprocess.run(
["git", "--version"], capture_output=True, text=True
)
print(f"Git版本: {result.stdout.strip()}")
except FileNotFoundError:
raise RuntimeError("Git未安装，请先安装Git")
yield {}
mcp = FastMCP(
name="Git Analyzer",
version="1.0.0",
instructions="Git仓库分析工具，提供代码统计、提交历史和分支管理功能",
lifespan=lifespan
)
# ===== Tools =====
@mcp.tool()
async def git_stats(repo_path: str) -> str:
"""获取Git仓库的代码统计信息
Args:
repo_path: Git仓库的绝对路径
"""
import os
if not os.path.isdir(repo_path):
return f"错误：路径不存在 - {repo_path}"
cmds = {
"总提交数": ["git", "-C", repo_path, "rev-list", "--count", "HEAD"],
"贡献者数": ["git", "-C", repo_path, "shortlog", "-sn", "HEAD"],
"文件数": ["git", "-C", repo_path, "ls-files"],
"最新标签": ["git", "-C", repo_path, "describe", "--tags", "--abbrev=0"],
}
results = []
for label, cmd in cmds.items():
r = subprocess.run(cmd, capture_output=True, text=True)
output = r.stdout.strip()
if label == "贡献者数":
count = len(output.split("\n")) if output else 0
results.append(f"👥 {label}: {count}")
elif label == "文件数":
count = len(output.split("\n")) if output else 0
results.append(f"📄 {label}: {count}")
else:
results.append(f"🏷️ {label}: {output or '无'}")
return "\n".join(results)
@mcp.tool()
async def git_recent_commits(repo_path: str, count: int = 10) -> str:
"""获取最近的Git提交记录
Args:
repo_path: Git仓库的绝对路径
count: 返回的提交数量，默认10
"""
count = max(1, min(count, 50))
result = subprocess.run(
[
"git", "-C", repo_path, "log",
f"-{count}", "--pretty=format:%h %s (%cr) <%an>",
"--no-merges"
],
capture_output=True, text=True
)
if result.returncode != 0:
return f"错误：{result.stderr.strip()}"
lines = result.stdout.strip().split("\n")
output = [f"📋 最近{count}条提交:\n"]
for i, line in enumerate(lines, 1):
output.append(f" {i}. {line}")
return "\n".join(output)
@mcp.tool()
async def git_branch_info(repo_path: str) -> str:
"""获取分支信息，包括当前分支和所有分支列表
Args:
repo_path: Git仓库的绝对路径
"""
# 当前分支
current = subprocess.run(
["git", "-C", repo_path, "branch", "--show-current"],
capture_output=True, text=True
)
# 所有分支
branches = subprocess.run(
["git", "-C", repo_path, "branch", "-a"],
capture_output=True, text=True
)
return (
f"🌟 当前分支: {current.stdout.strip()}\n\n"
f"📂 所有分支:\n{branches.stdout.strip()}"
)
# ===== Resources =====
@mcp.resource("git://repo/{repo_path}/head")
def get_head_commit(repo_path: str) -> str:
"""获取仓库HEAD提交信息"""
result = subprocess.run(
["git", "-C", repo_path, "log", "-1", "--pretty=format:%H%n%s%n%an <%ae>%n%ai"],
capture_output=True, text=True
)
return result.stdout.strip()
@mcp.resource("git://repo/{repo_path}/remotes")
def get_remotes(repo_path: str) -> str:
"""获取远程仓库信息"""
result = subprocess.run(
["git", "-C", repo_path, "remote", "-v"],
capture_output=True, text=True
)
return result.stdout.strip() or "无远程仓库配置"
# ===== Prompts =====
@mcp.prompt()
def code_review_prompt(repo_path: str, branch: str = "main") -> str:
"""生成代码审查提示"""
return (
f"请对仓库 {repo_path} 的 {branch} 分支进行代码审查。\n"
f"步骤：\n"
f"1. 使用 git_recent_commits 查看最近的提交\n"
f"2. 使用 git_stats 了解代码库规模\n"
f"3. 重点关注：安全风险、性能问题、代码风格\n"
f"请用中文输出审查报告。"
)
if __name__ == "__main__":
mcp.run()
```
### 8.2 对应的测试文件
```python
# tests/test_git_analyzer.py
import pytest
from fastmcp import Client
from git_analyzer_server import mcp
REPO_PATH = "." # 用当前仓库测试
@pytest.fixture
async def client():
async with Client(mcp) as c:
yield c
@pytest.mark.asyncio
async def test_git_stats(client: Client):
result = await client.call_tool("git_stats", {"repo_path": REPO_PATH})
text = result[0].text
assert "总提交数" in text
assert "文件数" in text
@pytest.mark.asyncio
async def test_git_recent_commits(client: Client):
result = await client.call_tool(
"git_recent_commits", {"repo_path": REPO_PATH, "count": 5}
)
text = result[0].text
assert "最近5条提交" in text
@pytest.mark.asyncio
async def test_git_branch_info(client: Client):
result = await client.call_tool("git_branch_info", {"repo_path": REPO_PATH})
text = result[0].text
assert "当前分支" in text
@pytest.mark.asyncio
async def test_invalid_path(client: Client):
result = await client.call_tool("git_stats", {"repo_path": "/nonexistent/path"})
assert "错误" in result[0].text
```
### 8.3 在Claude Code中使用
```json
// .mcp.json
{
"mcpServers": {
"git-analyzer": {
"command": "python",
"args": ["git_analyzer_server.py"],
"cwd": "${workspaceFolder}"
}
}
}
```
使用示例对话：
```
> 分析一下当前仓库的代码统计
> 最近5条提交是什么
> 帮我做一个代码审查
```
---
## 九、开发工作流总结
完整的MCP Server开发流程：
```mermaid
flowchart TD
A[规划服务器功能] --> B[定义 Tool/Resource/Prompt]
B --> C[实现业务逻辑]
C --> D[编写 pytest 测试]
D --> E{测试通过?}
E -->|否| C
E -->|是| F[MCP Inspector 可视化调试]
F --> G{调试通过?}
G -->|否| C
G -->|是| H[配置 .mcp.json]
H --> I[Claude Code 集成测试]
I --> J{实际使用正常?}
J -->|否| F
J -->|是| K[投入生产使用]
style A fill:#e3f2fd
style D fill:#e8f5e9
style F fill:#fff3e0
style I fill:#fce4ec
style K fill:#c8e6c9
```
### 开发调试清单
| 阶段 | 工具 | 命令 |
|------|------|------|
| 编码 | 任意IDE | — |
| 单元测试 | pytest | `pytest tests/ -v` |
| 可视化调试 | MCP Inspector | `npx @modelcontextprotocol/inspector python server.py` |
| 快速验证 | FastMCP dev | `fastmcp dev server.py` |
| Claude Code集成 | claude CLI | `claude mcp add name -- python server.py` |
| 生产运行 | FastMCP run | `fastmcp run server.py` |
---
## 十、常见问题与排错指南
### Q1: `ModuleNotFoundError: No module named 'fastmcp'`
MCP服务器由Claude Code以子进程方式启动，可能使用了不同的Python环境。解决方法：
```json
{
"mcpServers": {
"weather": {
"command": "/absolute/path/to/.venv/bin/python",
"args": ["src/server.py"]
}
}
}
```
或者使用 `uv run` 自动管理环境：
```json
{
"mcpServers": {
"weather": {
"command": "uv",
"args": ["run", "src/server.py"]
}
}
}
```
### Q2: 服务器启动后Claude Code无法发现Tool
检查清单：
1. 确认 `@mcp.tool()` 装饰器在函数定义之前
2. 确认函数有docstring（描述来自docstring第一行）
3. 确认 `register_tools(mcp)` 被调用了
4. 检查 `mcp.run()` 在 `__main__` 块中
### Q3: 异步函数报错 `coroutine was never awaited`
确保所有async工具函数使用 `@mcp.tool()` 而非 `@mcp.tool`：
```python
# 正确
@mcp.tool()
async def my_tool(): ...
# 错误（缺少括号，装饰器不会正确工作）
@mcp.tool
async def my_tool(): ...
```
### Q4: 测试中Client连接失败
确保在 `conftest.py` 中正确导入服务器实例，并且 `mcp` 对象是模块级别的：
```python
# conftest.py
from fastmcp import Client
# 确保这里的路径正确
from src.server import mcp as server_mcp
@pytest.fixture
async def client():
async with Client(server_mcp) as c:
yield c
```
### Q5: Inspector中中文参数显示乱码
在 `.mcp.json` 的env中添加：
```json
"env": {
"PYTHONIOENCODING": "utf-8",
"LANG": "en_US.UTF-8"
}
```
---
## 延伸阅读
- [FastMCP官方文档](https://fastmcp.wiki) — 完整API参考和高级模式
- [MCP协议规范](https://modelcontextprotocol.io) — 协议底层规范和传输层细节
- [MCP Inspector GitHub](https://github.com/modelcontextprotocol/inspector) — 调试工具源码和文档
- [FastMCP 3.0新特性](https://jlowin.dev/blog/fastmcp-3-whats-new) — 作者撰写的版本说明
- [Claude Code MCP集成文档](https://code.claude.com/docs/en/mcp) — 官方配置指南
## 训练营适用性评估
| 训练营类型 | 适用性 | 预计教学时长 | 备注 |
|-----------|--------|-------------|------|
| AI工具链开发训练营 | 高 | 4小时 | 核心课程内容 |
| Python全栈实战营 | 中高 | 3小时 | 需补充MCP概念介绍 |
| DevOps工具链训练营 | 中 | 2.5小时 | 侧重CI/CD集成部分 |
| Claude Code深度使用营 | 高 | 3.5小时 | 与Skills开发衔接 |
**教学建议：** 先完成第二节的简单Tool练习（30分钟），然后跳到第八节的Git分析器实战（90分钟），最后用Inspector调试（30分钟）。测试部分可作为课后练习。
## 参考来源
1. [FastMCP 3.0新特性详解](https://jlowin.dev/blog/fastmcp-3-whats-new) — FastMCP作者介绍3.0版本的核心架构变更和组合模型
2. [MCP调试工具官方文档](https://modelcontextprotocol.io/docs/tools/debugging) — MCP Inspector和调试策略的官方参考
3. [FastMCP测试模式文档](https://fastmcp.wiki/en/patterns/testing) — pytest集成和Client测试的官方指南
4. [Claude Code MCP配置文档](https://code.claude.com/docs/en/mcp) — Anthropic官方的MCP服务器集成指南
5. [MCP Python SDK GitHub](https://github.com/modelcontextprotocol/python-sdk) — 官方Python SDK源码和示例
> [!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 等场景全覆盖。
