MCP Server Python实战：FastMCP从开发到Docker部署

python --version    # >= 3.10
uv --version        # 任意版本
docker --version    # >= 24.0
claude --version    # 最新

graph LR
    Host["Host<br/>(Claude Code/Claude Desktop)"]
    Client1["MCP Client 1"]
    Client2["MCP Client 2"]
    Server1["MCP Server A<br/>(Python/FastMCP)"]
    Server2["MCP Server B<br/>(TypeScript)"]

    Host --> Client1
    Host --> Client2
    Client1 -->|"stdio / HTTP"| Server1
    Client2 -->|"stdio / HTTP"| Server2

    Server1 --> DB[("SQLite")]
    Server1 --> FS[("文件系统")]
    Server2 --> API[("外部API")]

# 创建项目目录
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

[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"

"""天气数据服务 - 调用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}"),
    )

"""天气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()

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

# 开发模式运行（带Inspector UI）
uv run fastmcp dev src/weather_mcp/server.py

# 直接运行stdio模式
uv run fastmcp run src/weather_mcp/server.py

# 通过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

"""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

uv add --dev pytest pytest-asyncio
uv run pytest tests/ -v

{
  "mcpServers": {
    "weather": {
      "command": "uv",
      "args": ["run", "fastmcp", "run", "src/weather_mcp/server.py"],
      "cwd": "/absolute/path/to/weather-mcp-server"
    }
  }
}

# 在项目目录内执行
claude mcp add weather -- uv run fastmcp run src/weather_mcp/server.py

# 验证注册成功
claude mcp list

# 测试调用
claude -p "查询北京当前天气"

{
  "mcpServers": {
    "weather": {
      "command": "uv",
      "args": ["run", "--directory", "/path/to/weather-mcp-server", "fastmcp", "run", "src/weather_mcp/server.py"]
    }
  }
}

# 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)

{
  "mcpServers": {
    "weather-remote": {
      "type": "http",
      "url": "http://your-server:8000/mcp"
    }
  }
}

claude mcp add weather-remote --transport http --url http://your-server:8000/mcp

# ---- 构建阶段 ----
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.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

# 构建镜像
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"}}}'

# 添加远程MCP Server
claude mcp add weather-docker \
  --transport http \
  --url http://localhost:8000/mcp

# 测试
claude -p "对比北京和上海的天气"

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}"

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}"

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

# 超时与重试配置示例
httpx_client = httpx.AsyncClient(
    timeout=httpx.Timeout(10.0, connect=5.0),
    limits=httpx.Limits(max_connections=20, max_keepalive_connections=5),
)

"""
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 mcp add travel-assistant -- uv run python src/weather_mcp/travel_assistant.py

# 测试
claude -p "帮我规划一个北京3天旅行，注意天气情况"

flowchart TB
    Start["项目初始化<br/>uv init + fastmcp"] --> Code["编写业务逻辑<br/>weather.py"]
    Code --> Server["构建MCP Server<br/>三大原语装饰器"]
    Server --> Test["本地调试<br/>fastmcp dev"]
    Test --> Local["Claude Code接入<br/>.mcp.json"]
    Local --> Docker["Docker容器化<br/>多阶段构建"]
    Docker --> Remote["远程部署<br/>Streamable HTTP"]
    Remote --> Prod["生产就绪<br/>错误处理+安全"]

    style Start fill:#e8f5e9
    style Server fill:#fff3cd
    style Docker fill:#e3f2fd
    style Prod fill:#fce4ec

# 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)
"