# MCP Server Python实战：FastMCP从开发到Docker部署
## 概述
本教程面向有Python基础的开发者，从零构建一个完整的MCP Server。你将掌握FastMCP框架的三大核心原语（Tool、Resource、Prompt），学会stdio与Streamable HTTP双传输模式，最终将服务Docker化并接入Claude Code。学完后你可以为任何Python项目创建AI可调用的工具层。
**适合人群**：有Python基础、了解REST API概念、想为Claude Code创建自定义MCP工具的开发者。
**学习目标**：
- 理解MCP协议核心架构与Python SDK选型
- 使用FastMCP装饰器快速暴露Tool/Resource/Prompt
- 实现stdio本地调试与Streamable HTTP远程部署
- Docker容器化并接入Claude Code生产环境
---
## 前置条件
| 项目 | 最低版本 | 安装命令 |
|------|----------|----------|
| Python | 3.10+ | `pyenv install 3.12` |
| uv | 最新 | `curl -LsSf https://astral.sh/uv/install.sh \| sh` |
| Docker | 24+ | 系统包管理器安装 |
| Claude Code | 最新 | `npm install -g @anthropic-ai/claude-code` |
验证环境：
```bash
python --version # >= 3.10
uv --version # 任意版本
docker --version # >= 24.0
claude --version # 最新
```
---
## 第一步：理解MCP架构与Python生态选型
### MCP协议核心概念
MCP（Model Context Protocol）是Anthropic提出的开放协议，让LLM以标准化方式连接外部数据源和工具。协议基于JSON-RPC 2.0，核心架构如下：
```mermaid
graph LR
Host["Host
(Claude Code/Claude Desktop)"]
Client1["MCP Client 1"]
Client2["MCP Client 2"]
Server1["MCP Server A
(Python/FastMCP)"]
Server2["MCP Server B
(TypeScript)"]
Host --> Client1
Host --> Client2
Client1 -->|"stdio / HTTP"| Server1
Client2 -->|"stdio / HTTP"| Server2
Server1 --> DB[("SQLite")]
Server1 --> FS[("文件系统")]
Server2 --> API[("外部API")]
```
MCP Server暴露三种原语：
| 原语 | 类比 | 方向 | 典型用途 |
|------|------|------|----------|
| **Tool** | POST端点 | LLM→Server | 执行操作、查询数据库、调用API |
| **Resource** | GET端点 | LLM←Server | 提供文件内容、数据快照、配置 |
| **Prompt** | 模板 | LLM→Server | 可复用的提示词模板 |
### Python SDK选型：FastMCP vs 官方SDK
| 维度 | `mcp`（官方SDK） | `fastmcp`（社区增强） |
|------|-----------------|---------------------|
| 维护方 | Anthropic官方 | PrefectHQ |
| 包名 | `mcp[cli]` | `fastmcp` |
| 装饰器风格 | 有 | 简洁，`@mcp.tool()` |
| 学习曲线 | 较陡 | 平缓 |
| 内置CLI | `mcp dev` | `fastmcp dev` |
| 生产就绪 | 是 | 是（已被官方推荐） |
**本教程选择FastMCP**，原因是其装饰器API更符合Python开发者习惯，且官方已将其纳入推荐工具链。
---
## 第二步：项目初始化
使用 `uv` 创建项目骨架：
```bash
# 创建项目目录
mkdir weather-mcp-server && cd weather-mcp-server
# 初始化Python项目
uv init --name weather-mcp-server --python 3.12
# 添加依赖
uv add fastmcp httpx
# 创建源码目录
mkdir -p src/weather_mcp
```
项目结构：
```
weather-mcp-server/
├── pyproject.toml
├── uv.lock
├── Dockerfile
├── README.md
├── src/
│ └── weather_mcp/
│ ├── __init__.py
│ ├── server.py # MCP Server主入口
│ ├── models.py # 数据模型
│ └── weather.py # 业务逻辑
└── tests/
└── test_server.py
```
配置 `pyproject.toml`：
```toml
[project]
name = "weather-mcp-server"
version = "0.1.0"
description = "A weather MCP server for Claude Code"
requires-python = ">=3.10"
dependencies = [
"fastmcp>=2.0",
"httpx>=0.27",
]
[project.scripts]
weather-mcp = "weather_mcp.server:main"
```
---
## 第三步：实现业务逻辑层
创建 `src/weather_mcp/weather.py`，封装天气API调用：
```python
"""天气数据服务 - 调用Open-Meteo免费API"""
from dataclasses import dataclass
import httpx
@dataclass
class WeatherData:
temperature: float
humidity: float
wind_speed: float
description: str
# WMO天气代码映射
WMO_CODES = {
0: "晴朗", 1: "大部晴朗", 2: "多云", 3: "阴天",
45: "雾", 48: "雾凇", 51: "小毛毛雨", 53: "中毛毛雨",
55: "大毛毛雨", 61: "小雨", 63: "中雨", 65: "大雨",
71: "小雪", 73: "中雪", 75: "大雪", 80: "阵雨",
95: "雷暴", 96: "雷暴伴冰雹",
}
async def get_coordinates(city: str) -> tuple[float, float]:
"""通过城市名获取经纬度"""
async with httpx.AsyncClient() as client:
resp = await client.get(
"https://geocoding-api.open-meteo.com/v1/search",
params={"name": city, "count": 1, "language": "zh"},
)
data = resp.json()
if not data.get("results"):
raise ValueError(f"找不到城市: {city}")
result = data["results"][0]
return result["latitude"], result["longitude"]
async def fetch_weather(city: str) -> WeatherData:
"""获取指定城市的当前天气"""
lat, lon = await get_coordinates(city)
async with httpx.AsyncClient() as client:
resp = await client.get(
"https://api.open-meteo.com/v1/forecast",
params={
"latitude": lat,
"longitude": lon,
"current": "temperature_2m,relative_humidity_2m,wind_speed_10m,weather_code",
},
)
current = resp.json()["current"]
code = current["weather_code"]
return WeatherData(
temperature=current["temperature_2m"],
humidity=current["relative_humidity_2m"],
wind_speed=current["wind_speed_10m"],
description=WMO_CODES.get(code, f"未知代码{code}"),
)
```
**要点说明**：
- 使用 `httpx.AsyncClient` 实现异步HTTP调用，与FastMCP异步架构一致
- Open-Meteo是完全免费的天气API，无需API Key
- WMO代码映射为中文天气描述，提升LLM理解准确度
---
## 第四步：构建MCP Server核心
创建 `src/weather_mcp/server.py`，这是整个教程的核心：
```python
"""天气MCP Server - 使用FastMCP暴露三大原语"""
from datetime import datetime
from fastmcp import FastMCP, Context
from .weather import fetch_weather, get_coordinates
# 创建MCP Server实例
mcp = FastMCP(
"weather-server",
version="0.1.0",
description="全球天气查询MCP服务器，支持当前天气和预报",
)
# ============================================================
# Tool：LLM可以主动调用的操作
# ============================================================
@mcp.tool()
async def get_current_weather(city: str) -> str:
"""获取指定城市的当前天气信息。
Args:
city: 城市名称（中文或英文），如"北京"、"Tokyo"
Returns:
格式化的天气报告字符串
"""
weather = await fetch_weather(city)
return (
f"🌍 {city} 当前天气\n"
f"🌡️ 温度: {weather.temperature}°C\n"
f"💧 湿度: {weather.humidity}%\n"
f"🌬️ 风速: {weather.wind_speed} km/h\n"
f"📋 天气: {weather.description}\n"
f"⏰ 更新时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}"
)
@mcp.tool()
async def compare_weather(city_a: str, city_b: str) -> str:
"""对比两个城市的天气状况。
Args:
city_a: 第一个城市名称
city_b: 第二个城市名称
Returns:
两个城市的天气对比报告
"""
import asyncio
weather_a, weather_b = await asyncio.gather(
fetch_weather(city_a), fetch_weather(city_b)
)
diff = weather_a.temperature - weather_b.temperature
diff_desc = f"{city_a}比{city_b}" + ("热" if diff > 0 else "冷") + f"{abs(diff):.1f}°C"
return (
f"📊 {city_a} vs {city_b} 天气对比\n"
f"{'指标':<8} {city_a:<12} {city_b:<12}\n"
f"{'温度':<8} {weather_a.temperature}°C{'':<6} {weather_b.temperature}°C\n"
f"{'湿度':<8} {weather_a.humidity}%{'':<7} {weather_b.humidity}%\n"
f"{'风速':<8} {weather_a.wind_speed}km/h{'':<3} {weather_b.wind_speed}km/h\n"
f"{'天气':<8} {weather_a.description:<12} {weather_b.description:<12}\n"
f"📌 {diff_desc}"
)
# ============================================================
# Resource：LLM可以读取的数据源
# ============================================================
@mcp.resource("weather://supported-cities")
def get_supported_cities() -> str:
"""返回支持查询的城市列表"""
cities = [
"北京", "上海", "广州", "深圳", "成都",
"Tokyo", "Seoul", "New York", "London", "Paris",
]
return "支持查询的城市:\n" + "\n".join(f" - {c}" for c in cities)
@mcp.resource("weather://api-status")
def get_api_status() -> str:
"""返回天气API服务状态"""
return (
f"天气服务状态报告\n"
f" API: Open-Meteo (免费)\n"
f" 状态: 正常运行\n"
f" 数据源: 全球气象站\n"
f" 更新频率: 15分钟\n"
f" 查询时间: {datetime.now().isoformat()}"
)
# ============================================================
# Prompt：可复用的提示词模板
# ============================================================
@mcp.prompt()
def weather_report_prompt(city: str, format: str = "detailed") -> str:
"""生成天气报告请求的提示词模板
Args:
city: 目标城市
format: 报告格式 (detailed/brief/travel)
"""
templates = {
"detailed": (
f"请查询{city}的当前天气，然后基于数据生成一份详细天气报告。"
f"报告应包含：温度体感分析、穿衣建议、出行提示。"
),
"brief": f"查询{city}天气，用一句话总结当前状况。",
"travel": (
f"查询{city}天气，从旅行角度给出建议："
f"适合户外活动吗？需要带伞吗？推荐穿着？"
),
}
return templates.get(format, templates["detailed"])
@mcp.prompt()
def travel_planning_prompt(
destination: str, days: int = 3, activities: str = "观光"
) -> str:
"""旅行规划提示词模板
Args:
destination: 目的地城市
days: 旅行天数
activities: 主要活动类型
"""
return (
f"我计划去{destination}旅行{days}天，主要活动是{activities}。"
f"请查询当地天气，然后给出：\n"
f"1. 每日穿衣建议\n"
f"2. 是否需要带雨具\n"
f"3. 适合/不适合的户外活动\n"
f"4. 天气相关的安全提示"
)
# ============================================================
# 入口点
# ============================================================
def main():
"""启动MCP Server"""
mcp.run()
if __name__ == "__main__":
main()
```
### 关键设计解析
```mermaid
flowchart TB
subgraph FastMCP["FastMCP Server"]
direction TB
subgraph Tools["Tool 原语"]
T1["get_current_weather"]
T2["compare_weather"]
end
subgraph Resources["Resource 原语"]
R1["weather://supported-cities"]
R2["weather://api-status"]
end
subgraph Prompts["Prompt 原语"]
P1["weather_report_prompt"]
P2["travel_planning_prompt"]
end
end
subgraph Business["业务逻辑层"]
B1["fetch_weather()"]
B2["get_coordinates()"]
end
subgraph External["外部API"]
E1["Open-Meteo Forecast"]
E2["Open-Meteo Geocoding"]
end
Tools --> Business
Business --> External
style FastMCP fill:#e8f4e8,stroke:#2d5a2d
style Tools fill:#fff3cd,stroke:#856404
style Resources fill:#d1ecf1,stroke:#0c5460
style Prompts fill:#f8d7da,stroke:#721c24
```
**装饰器约定**：
| 装饰器 | 函数签名要求 | 返回值 | LLM调用方式 |
|--------|-------------|--------|------------|
| `@mcp.tool()` | 参数自动推断为schema | `str` 或 JSON | 主动调用 |
| `@mcp.resource(uri)` | URI模板参数 | `str` | 被动读取 |
| `@mcp.prompt()` | 模板参数 | `str` | 选择使用 |
**Docstring是关键**：FastMCP自动从函数的docstring和类型注解生成MCP schema。LLM通过schema理解Tool的用途并决定何时调用。因此docstring必须清晰描述：
1. 函数做什么
2. 每个参数的含义和取值范围
3. 返回内容的格式
---
## 第五步：本地调试与验证
### 使用FastMCP CLI测试
```bash
# 开发模式运行（带Inspector UI）
uv run fastmcp dev src/weather_mcp/server.py
# 直接运行stdio模式
uv run fastmcp run src/weather_mcp/server.py
```
`fastmcp dev` 会自动打开MCP Inspector Web界面（默认 `http://localhost:5173`），你可以：
- 可视化查看所有已注册的Tool/Resource/Prompt
- 手动调用Tool查看返回结果
- 检查生成的JSON Schema是否正确
### 命令行直接测试
```bash
# 通过echo发送JSON-RPC请求测试
echo '{"jsonrpc":"2.0","id":1,"method":"tools/list"}' | \
uv run fastmcp run src/weather_mcp/server.py
# 测试tool调用
echo '{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"get_current_weather","arguments":{"city":"北京"}}}' | \
uv run fastmcp run src/weather_mcp/server.py
```
### 编写单元测试
创建 `tests/test_server.py`：
```python
"""MCP Server单元测试"""
import pytest
from unittest.mock import AsyncMock, patch
from weather_mcp.weather import fetch_weather, WeatherData, WMO_CODES
from weather_mcp.server import mcp
@pytest.fixture
def mock_weather_data():
return WeatherData(
temperature=25.0,
humidity=60.0,
wind_speed=10.0,
description="晴朗",
)
@pytest.mark.asyncio
async def test_fetch_weather(mock_weather_data):
with patch("weather_mcp.weather.get_coordinates", return_value=(39.9, 116.4)):
with patch("weather_mcp.weather.httpx.AsyncClient") as mock_client:
instance = mock_client.return_value.__aenter__.return_value
instance.get.return_value.json.return_value = {
"current": {
"temperature_2m": 25.0,
"relative_humidity_2m": 60.0,
"wind_speed_10m": 10.0,
"weather_code": 0,
}
}
result = await fetch_weather("北京")
assert result.temperature == 25.0
assert result.description == "晴朗"
def test_wmo_codes_completeness():
assert 0 in WMO_CODES
assert 95 in WMO_CODES
assert len(WMO_CODES) >= 15
def test_server_has_tools():
"""验证Server注册了正确的工具"""
# FastMCP内部维护工具列表
tool_names = [t.name for t in mcp._tool_manager._tools.values()]
assert "get_current_weather" in tool_names
assert "compare_weather" in tool_names
```
运行测试：
```bash
uv add --dev pytest pytest-asyncio
uv run pytest tests/ -v
```
---
## 第六步：接入Claude Code
### stdio模式（本地开发）
在项目根目录创建 `.mcp.json`（项目级配置）：
```json
{
"mcpServers": {
"weather": {
"command": "uv",
"args": ["run", "fastmcp", "run", "src/weather_mcp/server.py"],
"cwd": "/absolute/path/to/weather-mcp-server"
}
}
}
```
或使用Claude Code CLI命令注册：
```bash
# 在项目目录内执行
claude mcp add weather -- uv run fastmcp run src/weather_mcp/server.py
# 验证注册成功
claude mcp list
# 测试调用
claude -p "查询北京当前天气"
```
Claude Code会自动：
1. 发现 `.mcp.json` 中的服务器配置
2. 以子进程方式启动MCP Server（stdio传输）
3. 通过JSON-RPC协商能力并注册Tool/Resource/Prompt
4. 在对话中根据需要调用Tool
### Claude Desktop配置
编辑 `~/Library/Application Support/Claude/claude_desktop_config.json`（macOS）或对应路径：
```json
{
"mcpServers": {
"weather": {
"command": "uv",
"args": ["run", "--directory", "/path/to/weather-mcp-server", "fastmcp", "run", "src/weather_mcp/server.py"]
}
}
}
```
---
## 第七步：Streamable HTTP模式（远程部署）
MCP 2025-03-26规范引入了Streamable HTTP传输，适合远程部署场景。
### 修改Server支持HTTP模式
FastMCP内置HTTP传输支持，只需更改启动方式：
```python
# server.py 底部的main函数改为：
def main():
"""启动MCP Server - 支持多种传输"""
import argparse
parser = argparse.ArgumentParser(description="Weather MCP Server")
parser.add_argument(
"--transport",
choices=["stdio", "streamable-http"],
default="stdio",
help="传输模式",
)
parser.add_argument("--host", default="0.0.0.0")
parser.add_argument("--port", type=int, default=8000)
args = parser.parse_args()
if args.transport == "stdio":
mcp.run() # 默认stdio
else:
mcp.run(transport="streamable-http", host=args.host, port=args.port)
```
### 远程MCP配置
```json
{
"mcpServers": {
"weather-remote": {
"type": "http",
"url": "http://your-server:8000/mcp"
}
}
}
```
或使用CLI：
```bash
claude mcp add weather-remote --transport http --url http://your-server:8000/mcp
```
---
## 第八步：Docker容器化部署
### 多阶段Dockerfile
```dockerfile
# ---- 构建阶段 ----
FROM python:3.12-slim AS builder
RUN pip install uv
WORKDIR /app
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev
COPY src/ src/
# ---- 运行阶段 ----
FROM python:3.12-slim
WORKDIR /app
COPY --from=builder /app/.venv /app/.venv
COPY --from=builder /app/src /app/src
ENV PATH="/app/.venv/bin:$PATH"
ENV PYTHONUNBUFFERED=1
# 健康检查
HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
CMD python -c "import httpx; httpx.get('http://localhost:8000/health')" || exit 1
EXPOSE 8000
CMD ["python", "-m", "weather_mcp.server", "--transport", "streamable-http", "--host", "0.0.0.0", "--port", "8000"]
```
### Docker Compose（含健康检查）
```yaml
# docker-compose.yml
services:
weather-mcp:
build: .
ports:
- "8000:8000"
environment:
- PYTHONUNBUFFERED=1
restart: unless-stopped
healthcheck:
test: ["CMD", "python", "-c", "import httpx; httpx.get('http://localhost:8000/health')"]
interval: 30s
timeout: 5s
retries: 3
```
### 构建与部署
```bash
# 构建镜像
docker build -t weather-mcp-server:0.1.0 .
# 本地测试stdio模式
docker run --rm -i weather-mcp-server:0.1.0 \
python -m weather_mcp.server --transport stdio
# 启动HTTP服务
docker compose up -d
# 验证服务运行
curl http://localhost:8000/mcp \
-H "Content-Type: application/json" \
-d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"test","version":"0.1.0"}}}'
```
### Claude Code连接Docker化服务
```bash
# 添加远程MCP Server
claude mcp add weather-docker \
--transport http \
--url http://localhost:8000/mcp
# 测试
claude -p "对比北京和上海的天气"
```
---
## 第九步：高级模式——Context与进度报告
FastMCP支持通过 `Context` 对象访问MCP协议级能力：
```python
from fastmcp import FastMCP, Context
mcp = FastMCP("advanced-weather")
@mcp.tool()
async def batch_weather_report(
cities: list[str], ctx: Context
) -> str:
"""批量查询多个城市的天气，带进度报告。
Args:
cities: 城市名称列表
"""
results = []
total = len(cities)
for i, city in enumerate(cities):
# 发送进度通知给客户端
await ctx.report_progress(i + 1, total)
# 记录日志
await ctx.info(f"正在查询 {city} ({i+1}/{total})...")
try:
weather = await fetch_weather(city)
results.append(f"✅ {city}: {weather.temperature}°C, {weather.description}")
except Exception as e:
results.append(f"❌ {city}: 查询失败 - {e}")
return "\n".join(results)
@mcp.tool()
async def weather_with_logging(city: str, ctx: Context) -> str:
"""查询天气并记录详细日志"""
await ctx.debug("开始查询流程")
await ctx.info(f"目标城市: {city}")
weather = await fetch_weather(city)
await ctx.warning(f"温度 {'偏高' if weather.temperature > 35 else '正常'}")
return f"{city}: {weather.temperature}°C, {weather.description}"
```
`Context` 提供的方法：
| 方法 | 用途 | 典型场景 |
|------|------|----------|
| `ctx.report_progress(current, total)` | 发送进度通知 | 批量操作、长时间任务 |
| `ctx.info(msg)` / `ctx.debug(msg)` | 发送日志 | 调试、审计 |
| `ctx.warning(msg)` / `ctx.error(msg)` | 发送警告/错误 | 异常处理 |
| `ctx.read_resource(uri)` | 读取其他Resource | Resource间组合 |
---
## 第十步：生产就绪——错误处理与安全
### 统一错误处理
```python
from fastmcp import FastMCP
import logging
mcp = FastMCP("production-weather")
logger = logging.getLogger("weather-mcp")
@mcp.tool()
async def safe_weather_query(city: str) -> str:
"""安全查询天气，内置错误处理。
Args:
city: 城市名称
"""
# 输入验证
if not city or len(city.strip()) == 0:
return "错误：城市名称不能为空"
if len(city) > 100:
return "错误：城市名称过长（最多100字符）"
try:
weather = await fetch_weather(city.strip())
return (
f"{city}: {weather.temperature}°C, {weather.description}\n"
f"湿度: {weather.humidity}%, 风速: {weather.wind_speed}km/h"
)
except ValueError as e:
logger.warning(f"城市未找到: {city} - {e}")
return f"未找到城市 '{city}'，请检查拼写或尝试英文名称"
except httpx.TimeoutException:
logger.error(f"API超时: {city}")
return f"查询 '{city}' 超时，请稍后重试"
except Exception as e:
logger.exception(f"未知错误: {city}")
return f"查询失败: {e}"
```
### 安全最佳实践清单
```mermaid
flowchart LR
A[请求到达] --> B{输入验证}
B -->|通过| C{速率限制}
B -->|失败| E[返回错误]
C -->|通过| D{执行操作}
C -->|超限| E
D --> F{敏感数据过滤}
F --> G[返回结果]
style B fill:#fff3cd
style C fill:#fff3cd
style F fill:#f8d7da
```
**关键安全措施**：
1. **输入验证**：检查所有参数的类型、长度、格式
2. **速率限制**：对外部API调用设置频率上限
3. **超时控制**：所有HTTP调用设置合理超时
4. **日志脱敏**：不记录敏感数据（用户位置偏好等）
5. **最小权限**：Tool只暴露必要的能力，不提供任意文件读写
```python
# 超时与重试配置示例
httpx_client = httpx.AsyncClient(
timeout=httpx.Timeout(10.0, connect=5.0),
limits=httpx.Limits(max_connections=20, max_keepalive_connections=5),
)
```
---
## 完整实战案例：旅行助手MCP Server
将以上所有知识整合为一个完整项目，支持天气查询、城市对比、旅行建议：
```python
"""
travel_assistant.py - 旅行助手MCP Server
功能：天气查询 + 穿衣建议 + 行程规划提示
"""
import asyncio
from datetime import datetime
from fastmcp import FastMCP, Context
from weather_mcp.weather import fetch_weather
mcp = FastMCP(
"travel-assistant",
instructions=(
"你是一个旅行天气助手。用户可能询问天气、请求穿衣建议、或规划旅行。"
"优先使用 get_current_weather 获取数据，再用 travel_advice 生成建议。"
),
)
# ---------- 穿衣建议逻辑 ----------
CLOTHING_ADVICE = {
"freezing": (lambda t: t < 0, "厚重羽绒服、围巾、手套、帽子"),
"cold": (lambda t: t < 10, "厚外套或羽绒服"),
"cool": (lambda t: t < 18, "薄外套或夹克"),
"mild": (lambda t: t < 25, "长袖衬衫或薄外套"),
"warm": (lambda t: t < 32, "短袖、轻薄衣物"),
"hot": (lambda t: t < 38, "轻薄透气衣物，注意防晒"),
"extreme": (lambda t: True, "极端高温，减少外出，充分补水"),
}
RAIN_KEYWORDS = {"小雨", "中雨", "大雨", "阵雨", "毛毛雨", "雷暴"}
SNOW_KEYWORDS = {"小雪", "中雪", "大雪"}
def _get_clothing(temp: float) -> str:
for _, (check, advice) in CLOTHING_ADVICE.items():
if check(temp):
return advice
return "根据体感穿着"
# ---------- Tools ----------
@mcp.tool()
async def get_current_weather(city: str) -> str:
"""获取城市当前天气"""
weather = await fetch_weather(city)
return (
f"🌡️ 温度: {weather.temperature}°C\n"
f"💧 湿度: {weather.humidity}%\n"
f"🌬️ 风速: {weather.wind_speed}km/h\n"
f"📋 天气: {weather.description}"
)
@mcp.tool()
async def travel_advice(city: str, days: int = 1) -> str:
"""根据天气生成旅行建议。
Args:
city: 目的地城市
days: 旅行天数（影响行李建议）
"""
weather = await fetch_weather(city)
clothing = _get_clothing(weather.temperature)
need_umbrella = any(kw in weather.description for kw in RAIN_KEYWORDS)
need_warm = any(kw in weather.description for kw in SNOW_KEYWORDS)
tips = [f"📍 {city} 旅行建议（{days}天行程）", ""]
tips.append(f"🌡️ 当前温度 {weather.temperature}°C")
tips.append(f"👔 推荐穿着: {clothing}")
if need_umbrella:
tips.append("☂️ 建议: 带雨伞/雨衣")
if need_warm:
tips.append("🧣 建议: 注意保暖，路面可能结冰")
if weather.temperature > 35:
tips.append("⚠️ 高温预警: 多喝水，避免长时间户外活动")
if weather.wind_speed > 30:
tips.append("💨 大风预警: 注意安全，避免高空作业")
if days > 1:
tips.append(f"🧳 行李建议: 准备{days}天的换洗衣物" +
("和雨具" if need_umbrella else ""))
return "\n".join(tips)
@mcp.tool()
async def multi_city_weather(cities: str, ctx: Context) -> str:
"""批量查询多个城市的天气。
Args:
cities: 逗号分隔的城市列表，如"北京,上海,广州"
"""
city_list = [c.strip() for c in cities.split(",") if c.strip()]
total = len(city_list)
results = []
for i, city in enumerate(city_list):
await ctx.report_progress(i + 1, total)
await ctx.info(f"查询 {city} ({i+1}/{total})")
try:
weather = await fetch_weather(city)
results.append(f"{'✅':<3} {city:<8} {weather.temperature:>6.1f}°C {weather.description}")
except Exception as e:
results.append(f"{'❌':<3} {city:<8} 查询失败: {e}")
header = f"{'状态':<3} {'城市':<8} {'温度':<9} 天气"
return header + "\n" + "-" * 40 + "\n" + "\n".join(results)
# ---------- Resources ----------
@mcp.resource("travel://clothing-guide")
def clothing_guide() -> str:
"""穿衣指南参考"""
guide = "温度穿衣参考表:\n"
for name, (check, advice) in CLOTHING_ADVICE.items():
guide += f" {advice}\n"
return guide
# ---------- Prompts ----------
@mcp.prompt()
def plan_trip_prompt(
destination: str,
duration: int = 3,
season: str = "不限",
budget: str = "中等",
) -> str:
"""生成旅行规划提示词
Args:
destination: 目的地
duration: 天数
season: 季节偏好
budget: 预算级别
"""
return (
f"帮我规划一次{destination}的{duration}天旅行。\n"
f"季节偏好: {season}\n"
f"预算级别: {budget}\n\n"
f"请先查询当地天气，然后基于天气情况:\n"
f"1. 推荐每日穿着\n"
f"2. 列出适合/不适合的户外活动\n"
f"3. 给出天气相关的注意事项\n"
f"4. 推荐每日行程安排"
)
if __name__ == "__main__":
mcp.run()
```
### 注册到Claude Code
```bash
claude mcp add travel-assistant -- uv run python src/weather_mcp/travel_assistant.py
# 测试
claude -p "帮我规划一个北京3天旅行，注意天气情况"
```
Claude Code会自动：
1. 发现 `get_current_weather` Tool并查询北京天气
2. 发现 `travel_advice` Tool并获取穿衣/出行建议
3. 发现 `plan_trip_prompt` Prompt并组织完整回复
4. 读取 `travel://clothing-guide` Resource补充参考
---
## 全流程总结
```mermaid
flowchart TB
Start["项目初始化
uv init + fastmcp"] --> Code["编写业务逻辑
weather.py"]
Code --> Server["构建MCP Server
三大原语装饰器"]
Server --> Test["本地调试
fastmcp dev"]
Test --> Local["Claude Code接入
.mcp.json"]
Local --> Docker["Docker容器化
多阶段构建"]
Docker --> Remote["远程部署
Streamable HTTP"]
Remote --> Prod["生产就绪
错误处理+安全"]
style Start fill:#e8f5e9
style Server fill:#fff3cd
style Docker fill:#e3f2fd
style Prod fill:#fce4ec
```
### 常见问题速查
| 问题 | 原因 | 解决方案 |
|------|------|----------|
| `claude mcp list` 不显示 | 配置文件位置错误 | 确认 `.mcp.json` 在项目根目录 |
| Tool调用返回超时 | 外部API响应慢 | 设置 `httpx.Timeout`，建议10秒 |
| 中文参数乱码 | Python编码问题 | 确保 `PYTHONUTF8=1` 环境变量 |
| Docker内无法访问API | 网络配置 | Docker Compose加 `network_mode: host` |
| `fastmcp` 命令不存在 | 未正确安装 | `uv add fastmcp` 后用 `uv run` |
| Claude Code不调用Tool | docstring不清晰 | 重写docstring，明确Args和功能描述 |
| Resource读取失败 | URI格式不对 | 检查 `scheme://path` 格式，无空格 |
### 调试技巧
```bash
# 1. 查看MCP Server的完整日志
claude --mcp-debug -p "查询天气"
# 2. 验证Server生成的Schema
uv run python -c "
from weather_mcp.server import mcp
import json
tools = mcp._tool_manager._tools
for name, tool in tools.items():
print(f'Tool: {name}')
print(json.dumps(tool.parameters_schema, indent=2, ensure_ascii=False))
"
# 3. 单独测试业务逻辑
uv run python -c "
import asyncio
from weather_mcp.weather import fetch_weather
result = asyncio.run(fetch_weather('北京'))
print(result)
"
```
---
## 延伸阅读
- [FastMCP官方文档](https://gofastmcp.com) — FastMCP完整API参考和最佳实践
- [MCP协议规范](https://modelcontextprotocol.io/specification/2025-06-18/basic/transports) — Streamable HTTP传输协议细节
- [Claude Code MCP集成指南](https://docs.anthropic.com/en/docs/claude-code/mcp) — 官方MCP配置文档
- [FastMCP GitHub仓库](https://github.com/jlowin/fastmcp) — 源码和示例项目
- [MCP Python SDK](https://github.com/modelcontextprotocol/python-sdk) — Anthropic官方Python SDK
---
## 参考来源
1. [FastMCP: Build Custom MCP Servers | Claude Code Skill](https://mcpmarket.com/tools/skills/fastmcp-development-1) — FastMCP框架核心功能与Claude Code集成指南
2. [Build an MCP Server in Python with FastMCP - The Python Code](https://thepythoncode.com/article/fastmcp-mcp-client-server-todo-manager) — 完整的Python MCP Server构建教程，含Tool/Resource/Prompt示例
3. [How to Build an MCP Server with Python, Docker, and Claude Code](https://www.freecodecamp.org/news/how-to-build-an-mcp-server-with-python-docker-and-claude-code/) — Docker容器化MCP Server部署实战
4. [Connect Claude Code to tools via MCP - Claude Code Docs](https://code.claude.com/docs/en/mcp) — Claude Code官方MCP配置与接入文档
5. [FastMCP Python Server Guide 2026 | AI Workflow Lab](https://aiworkflowlab.dev/article/build-production-mcp-servers-fastmcp-python-guide) — FastMCP 3.0生产级服务器开发指南
---
## 训练营适用性评估
| 维度 | 评估 |
|------|------|
| **适合训练营类型** | Python全栈训练营、AI工具开发训练营、DevOps实践训练营 |
| **预计教学时长** | 4-6小时（含动手实践） |
| **难度级别** | 中级（需要Python异步编程基础） |
| **前置知识** | Python 3.10+、async/await、HTTP API、Docker基础 |
| **硬件要求** | 任意开发机，需联网（调用Open-Meteo API） |
| **可拆分模块** | 步骤1-5为基础篇（2h），步骤6-8为部署篇（2h），步骤9-10为进阶篇（2h） |
> [!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 等场景全覆盖。
