MCP Server全链路发布：开发到上架实战

mkdir mcp-weather-advisor && cd mcp-weather-advisor
mkdir -p src/mcp_weather_advisor tests
touch src/mcp_weather_advisor/__init__.py

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "mcp-weather-advisor"
version = "0.1.0"
description = "MCP Server providing weather data, forecasts, and clothing recommendations via multiple primitives"
readme = "README.md"
license = "MIT"
requires-python = ">=3.11"
authors = [
    { name = "Your Name", email = "[email protected]" }
]
keywords = ["mcp", "weather", "ai", "claude"]
classifiers = [
    "Development Status :: 4 - Beta",
    "Intended Audience :: Developers",
    "License :: OSI Approved :: MIT License",
    "Programming Language :: Python :: 3",
    "Programming Language :: Python :: 3.11",
    "Programming Language :: Python :: 3.12",
    "Topic :: Scientific/Engineering :: Atmospheric Science",
]
dependencies = [
    "fastmcp>=3.2",
    "httpx>=0.27",
]

[project.optional-dependencies]
dev = [
    "pytest>=8.0",
    "pytest-asyncio>=0.24",
    "ruff>=0.8",
]

[project.scripts]
mcp-weather-advisor = "mcp_weather_advisor.server:main"

[project.urls]
Homepage = "https://github.com/yourname/mcp-weather-advisor"
Repository = "https://github.com/yourname/mcp-weather-advisor"

# 使用 uv（推荐，速度快）
uv venv && source .venv/bin/activate
uv pip install -e ".[dev]"

# 或使用 pip
pip install -e ".[dev]"

"""MCP Weather Advisor Server - 全原语演示"""

from datetime import datetime
from typing import Annotated

import httpx
from fastmcp import FastMCP, Context
from pydantic import Field

# 创建MCP Server实例
mcp = FastMCP(
    "weather-advisor",
    version="0.1.0",
    description="天气查询与穿衣建议MCP服务器",
)


# ==================== Primitives 1: Tools ====================

@mcp.tool()
async def get_current_weather(
    city: Annotated[str, Field(description="城市名称，如'北京'、'Shanghai'")],
    units: Annotated[str, Field(description="温度单位：celsius 或 fahrenheit")] = "celsius",
) -> dict:
    """获取指定城市的当前天气信息。

    返回温度、湿度、风速、天气状况等实时数据。
    支持全球主要城市。
    """
    # 使用 wttr.in 免费API（无需API Key）
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://wttr.in/{city}",
            params={"format": "j1"},
            timeout=10.0,
        )
        response.raise_for_status()
        data = response.json()

    current = data["current_condition"][0]
    temp_c = int(current["temp_C"])
    temp_f = int(current["temp_F"])

    return {
        "city": city,
        "temperature": temp_f if units == "fahrenheit" else temp_c,
        "units": units,
        "humidity": current["humidity"],
        "description": current["weatherDesc"][0]["value"],
        "wind_speed_kmh": current["windspeedKmph"],
        "feels_like_c": current["FeelsLikeC"],
        "observation_time": current["observation_time"],
    }


@mcp.tool()
async def get_forecast(
    city: Annotated[str, Field(description="城市名称")],
    days: Annotated[int, Field(description="预报天数（1-3）", ge=1, le=3)] = 3,
) -> list[dict]:
    """获取未来几天的天气预报。

    返回每天的日期、最高/最低温度、天气描述。
    最多支持3天预报。
    """
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://wttr.in/{city}",
            params={"format": "j1"},
            timeout=10.0,
        )
        response.raise_for_status()
        data = response.json()

    forecasts = []
    for day_data in data["weather"][:days]:
        forecasts.append({
            "date": day_data["date"],
            "max_temp_c": day_data["maxtempC"],
            "min_temp_c": day_data["mintempC"],
            "avg_temp_c": day_data["avgtempC"],
            "sun_hours": day_data["sunHour"],
            "uv_index": day_data["uvIndex"],
            "description": day_data["hourly"][4]["weatherDesc"][0]["value"],
        })

    return forecasts


@mcp.tool()
async def get_clothing_advice(
    temperature_c: Annotated[float, Field(description="当前温度（摄氏度）")],
    humidity: Annotated[int, Field(description="当前湿度百分比", ge=0, le=100)] = 50,
    wind_speed_kmh: Annotated[float, Field(description="风速（km/h）")] = 0,
    activity: Annotated[str, Field(description="活动类型：outdoor/commute/exercise/formal")] = "commute",
) -> dict:
    """根据天气条件提供穿衣建议。

    综合考虑温度、湿度、风速和活动类型，
    给出具体的穿衣推荐和注意事项。
    """
    if temperature_c >= 30:
        level = "极热"
        advice = "穿轻薄透气的短袖短裤，注意防晒和补水"
        items = ["防晒霜", "太阳帽", "墨镜", "水壶"]
    elif temperature_c >= 22:
        level = "炎热"
        advice = "短袖T恤搭配薄裤或裙装即可"
        items = ["薄外套（室内空调用）", "防晒用品"]
    elif temperature_c >= 15:
        level = "温暖"
        advice = "长袖衬衫加薄外套，早晚注意加衣"
        items = ["薄外套", "长裤"]
    elif temperature_c >= 5:
        level = "凉爽"
        advice = "穿毛衣或卫衣搭配外套"
        items = ["厚外套", "围巾", "保暖内衣"]
    else:
        level = "寒冷"
        advice = "穿羽绒服或厚棉衣，注意保暖"
        items = ["羽绒服", "帽子", "手套", "保暖内衣", "围巾"]

    # 湿度修正
    if humidity > 80:
        advice += "。湿度较高，建议选择速干面料"
        items.append("雨伞")
    elif humidity < 30:
        advice += "。空气干燥，建议多涂润肤霜"

    # 风速修正
    if wind_speed_kmh > 30:
        advice += "。风力较大，注意防风保暖"
        items.append("防风外套")

    # 活动修正
    activity_tips = {
        "outdoor": "户外活动建议穿运动鞋，携带应急雨具",
        "commute": "通勤建议洋葱式穿法，方便室内脱减",
        "exercise": "运动建议穿排汗速干衣物，运动后及时加衣",
        "formal": "正式场合建议选择商务休闲搭配，注意保暖",
    }

    return {
        "temperature_level": level,
        "clothing_advice": advice,
        "recommended_items": items,
        "activity_tip": activity_tips.get(activity, ""),
    }


# ==================== Primitives 2: Resources ====================

@mcp.resource("weather://cities/supported")
def get_supported_cities() -> str:
    """列出支持的常用城市列表"""
    cities = {
        "中国": ["北京", "上海", "广州", "深圳", "成都", "杭州", "武汉", "南京"],
        "国际": ["Tokyo", "Seoul", "New York", "London", "Paris", "Sydney"],
    }
    import json
    return json.dumps(cities, ensure_ascii=False, indent=2)


@mcp.resource("weather://alerts/{city}")
async def get_weather_alerts(city: str) -> str:
    """获取指定城市的天气预警信息"""
    import json
    # 示例：实际项目中对接气象局API
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://wttr.in/{city}",
            params={"format": "j1"},
            timeout=10.0,
        )
        data = response.json()

    current = data["current_condition"][0]
    alerts = []
    temp = int(current["temp_C"])

    if temp >= 35:
        alerts.append({"type": "高温预警", "level": "橙色", "message": f"当前温度{temp}°C，注意防暑降温"})
    elif temp <= -10:
        alerts.append({"type": "寒潮预警", "level": "蓝色", "message": f"当前温度{temp}°C，注意防寒保暖"})

    if int(current["windspeedKmph"]) >= 60:
        alerts.append({"type": "大风预警", "level": "黄色", "message": "风力较大，减少户外活动"})

    if not alerts:
        alerts.append({"type": "无预警", "level": "正常", "message": f"{city}当前无气象预警"})

    return json.dumps(alerts, ensure_ascii=False, indent=2)


# ==================== Primitives 3: Prompts ====================

@mcp.prompt()
def daily_outfit_prompt(
    city: Annotated[str, Field(description="城市名称")] = "北京",
) -> str:
    """生成每日穿搭建议的提示词模板"""
    return f"""你是一位专业的穿搭顾问。请根据以下信息为用户提供今日穿搭建议：

1. 首先调用 get_current_weather 获取 {city} 的当前天气
2. 然后调用 get_clothing_advice 获取穿搭建议
3. 最后综合天气和穿搭信息，用友好的语气给出建议

请以以下格式输出：
- 🌡️ 今日天气概况
- 👔 穿搭建议（上装、下装、鞋子、配饰）
- ⚠️ 注意事项
- 🎯 今日小贴士"""


@mcp.prompt()
def travel_planning_prompt(
    destination: Annotated[str, Field(description="目的地城市")],
    days: Annotated[int, Field(description="旅行天数")] = 3,
) -> str:
    """生成旅行天气规划提示词模板"""
    return f"""你是一位旅行规划助手。请帮助用户规划前往 {destination} 的 {days} 天旅行：

1. 调用 get_current_weather 获取 {destination} 当前天气
2. 调用 get_forecast 获取未来 {days} 天天气预报
3. 调用 get_clothing_advice 根据天气预报给出每日穿搭建议
4. 调用 get_weather_alerts 查看是否有天气预警

请以以下格式输出：
- 📍 目的地天气概览
- 📅 每日天气详情与穿搭建议
- 🎒 行李清单
- ⚠️ 天气预警与注意事项"""


# ==================== Elicitation：人机交互增强 ====================

@mcp.tool()
async def personalized_weather_report(
    ctx: Context,
) -> dict:
    """生成个性化天气报告。通过Elicitation与用户交互收集偏好。

    该工具演示MCP 2025-11-25规范中的Elicitation能力：
    服务端可以主动向用户请求信息，实现人机协作。
    """
    # 第一步：询问用户的城市偏好
    city_response = await ctx.elicit(
        message="请告诉我您想查看哪个城市的天气？",
        requested_fields={
            "city": {
                "type": "string",
                "description": "城市名称",
                "label": "城市",
            }
        },
    )

    city = city_response.data["city"]

    # 第二步：询问报告偏好
    pref_response = await ctx.elicit(
        message="请选择您需要的报告内容：",
        requested_fields={
            "include_forecast": {
                "type": "boolean",
                "description": "是否包含未来3天预报",
                "label": "包含预报",
            },
            "include_clothing": {
                "type": "boolean",
                "description": "是否包含穿搭建议",
                "label": "包含穿搭建议",
            },
            "temperature_unit": {
                "type": "string",
                "description": "温度单位",
                "label": "温度单位",
                "enum": ["celsius", "fahrenheit"],
            },
        },
    )

    prefs = pref_response.data

    # 获取天气数据
    weather = await get_current_weather(city, prefs.get("temperature_unit", "celsius"))
    result = {"weather": weather}

    if prefs.get("include_forecast"):
        result["forecast"] = await get_forecast(city, 3)

    if prefs.get("include_clothing"):
        result["clothing"] = await get_clothing_advice(
            float(weather["temperature"]),
            int(weather["humidity"]),
            float(weather["wind_speed_kmh"]),
        )

    return result


# ==================== 启动入口 ====================

def main():
    """CLI入口点 - 支持stdio和SSE两种传输模式"""
    mcp.run()


if __name__ == "__main__":
    main()

graph TB
    subgraph "MCP Server 三原语 + Elicitation"
        A[Tool 工具] --> A1[get_current_weather<br/>实时天气查询]
        A --> A2[get_forecast<br/>多日天气预报]
        A --> A3[get_clothing_advice<br/>穿衣建议]
        A --> A4[personalized_weather_report<br/>Elicitation交互式报告]

        B[Resource 资源] --> B1[weather://cities/supported<br/>支持城市列表]
        B --> B2[weather://alerts/城市名<br/>天气预警信息]

        C[Prompt 提示词] --> C1[daily_outfit_prompt<br/>每日穿搭提示词]
        C --> C2[travel_planning_prompt<br/>旅行规划提示词]
    end

    subgraph "外部依赖"
        D[wttr.in API<br/>免费天气数据]
        E[FastMCP 3.2+<br/>MCP框架]
    end

    A1 --> D
    A2 --> D
    A4 --> A1
    A4 --> A2
    A4 --> A3

# 安装MCP Inspector
npx @anthropic-ai/mcp-inspector

# 启动Inspector并加载你的Server
npx @anthropic-ai/mcp-inspector \
  mcp-weather-advisor

"""MCP Weather Advisor 单元测试"""

import pytest
from mcp_weather_advisor.server import mcp


@pytest.fixture
def client():
    """创建MCP测试客户端"""
    from fastmcp import Client
    return Client(mcp)


@pytest.mark.asyncio
async def test_list_tools():
    """验证所有Tool已正确注册"""
    async with Client(mcp) as client:
        tools = await client.list_tools()
        tool_names = [t.name for t in tools]

        assert "get_current_weather" in tool_names
        assert "get_forecast" in tool_names
        assert "get_clothing_advice" in tool_names
        assert "personalized_weather_report" in tool_names


@pytest.mark.asyncio
async def test_get_current_weather():
    """测试天气查询Tool"""
    async with Client(mcp) as client:
        result = await client.call_tool("get_current_weather", {
            "city": "Beijing",
            "units": "celsius",
        })

        assert result is not None
        assert result.data["city"] == "Beijing"
        assert "temperature" in result.data
        assert "humidity" in result.data


@pytest.mark.asyncio
async def test_get_clothing_advice_cold():
    """测试穿衣建议 - 寒冷场景"""
    async with Client(mcp) as client:
        result = await client.call_tool("get_clothing_advice", {
            "temperature_c": -5.0,
            "humidity": 40,
            "activity": "outdoor",
        })

        assert result.data["temperature_level"] == "寒冷"
        assert "羽绒服" in result.data["recommended_items"]


@pytest.mark.asyncio
async def test_get_clothing_advice_hot_humid():
    """测试穿衣建议 - 高温高湿场景"""
    async with Client(mcp) as client:
        result = await client.call_tool("get_clothing_advice", {
            "temperature_c": 35.0,
            "humidity": 90,
            "wind_speed_kmh": 5,
        })

        assert result.data["temperature_level"] == "极热"
        assert "雨伞" in result.data["recommended_items"]


@pytest.mark.asyncio
async def test_list_resources():
    """验证Resource已注册"""
    async with Client(mcp) as client:
        resources = await client.list_resources()
        uris = [str(r.uri) for r in resources]

        assert "weather://cities/supported" in uris


@pytest.mark.asyncio
async def test_list_prompts():
    """验证Prompt已注册"""
    async with Client(mcp) as client:
        prompts = await client.list_prompts()
        prompt_names = [p.name for p in prompts]

        assert "daily_outfit_prompt" in prompt_names
        assert "travel_planning_prompt" in prompt_names

# 运行全部测试
pytest tests/ -v

# 仅运行tool测试
pytest tests/test_server.py::test_get_clothing_advice_cold -v

tests/test_server.py::test_list_tools PASSED
tests/test_server.py::test_get_current_weather PASSED
tests/test_server.py::test_get_clothing_advice_cold PASSED
tests/test_server.py::test_get_clothing_advice_hot_humid PASSED
tests/test_server.py::test_list_resources PASSED
tests/test_server.py::test_list_prompts PASSED

{
  "mcpServers": {
    "weather-advisor": {
      "command": "mcp-weather-advisor",
      "args": [],
      "type": "stdio"
    }
  }
}

{
  "mcpServers": {
    "weather-advisor": {
      "command": "python",
      "args": ["-m", "mcp_weather_advisor.server"],
      "type": "stdio",
      "cwd": "/path/to/mcp-weather-advisor"
    }
  }
}

/mcp

weather-advisor (stdio) — Connected
  Tools: get_current_weather, get_forecast, get_clothing_advice, personalized_weather_report
  Resources: weather://cities/supported, weather://alerts/{city}
  Prompts: daily_outfit_prompt, travel_planning_prompt

帮我查一下北京今天的天气，并告诉我穿什么衣服

FROM python:3.12-slim AS builder

WORKDIR /app

# 安装uv加速依赖安装
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv

COPY pyproject.toml .
COPY src/ src/

RUN uv venv /app/.venv && \
    uv pip install --python /app/.venv/bin/python .

# 运行阶段
FROM python:3.12-slim

RUN useradd -m -s /bin/bash mcp
USER mcp

COPY --from=builder /app/.venv /app/.venv

ENV PATH="/app/.venv/bin:$PATH"

ENTRYPOINT ["mcp-weather-advisor"]

# 构建镜像
docker build -t mcp-weather-advisor:0.1.0 .

# 本地测试运行
docker run --rm mcp-weather-advisor:0.1.0 --help

{
  "mcpServers": {
    "weather-advisor": {
      "command": "docker",
      "args": [
        "run", "--rm",
        "-i",
        "mcp-weather-advisor:0.1.0"
      ],
      "type": "stdio"
    }
  }
}

# docker-compose.mcp.yml
services:
  weather-advisor:
    build: .
    ports:
      - "8080:8080"
    environment:
      - MCP_TRANSPORT=sse
      - MCP_PORT=8080
    healthcheck:
      test: ["CMD", "python", "-c", "import httpx; httpx.get('http://localhost:8080/health')"]
      interval: 30s
      timeout: 5s
      retries: 3

def main():
    """支持stdio和HTTP两种传输模式"""
    import os
    transport = os.environ.get("MCP_TRANSPORT", "stdio")

    if transport == "sse":
        port = int(os.environ.get("MCP_PORT", "8080"))
        mcp.run(transport="sse", port=port)
    else:
        mcp.run()  # 默认stdio模式

# 安装构建工具
pip install build twine

# 构建wheel和source distribution
python -m build

# 检查包质量
twine check dist/*

dist/mcp_weather_advisor-0.1.0-py3-none-any.whl: OK
dist/mcp-weather-advisor-0.1.0.tar.gz: OK

twine upload --repository testpypi dist/* \
  --username __token__ \
  --password <TESTPYPI_TOKEN>

pip install --index-url https://test.pypi.org/simple/ \
  mcp-weather-advisor==0.1.0

twine upload dist/* \
  --username __token__ \
  --password <PYPI_TOKEN>

pip install mcp-weather-advisor

name: Publish to PyPI

on:
  release:
    types: [published]

jobs:
  publish:
    runs-on: ubuntu-latest
    permissions:
      id-token: write  # Trusted Publishing
    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Install build tools
        run: pip install build

      - name: Build package
        run: python -m build

      - name: Publish to PyPI
        uses: pypa/gh-action-pypi-publish@release/v1
        # 使用Trusted Publishing，无需TOKEN

{
  "name": "mcp-weather-advisor",
  "version": "0.1.0",
  "description": "Weather data, forecasts, and clothing recommendations MCP Server",
  "author": {
    "name": "Your Name",
    "url": "https://github.com/yourname"
  },
  "repository": {
    "type": "git",
    "url": "https://github.com/yourname/mcp-weather-advisor"
  },
  "installations": {
    "pip": {
      "package": "mcp-weather-advisor",
      "command": "mcp-weather-advisor"
    },
    "docker": {
      "image": "yourname/mcp-weather-advisor:latest",
      "command": "docker",
      "args": ["run", "--rm", "-i", "yourname/mcp-weather-advisor:latest"]
    }
  },
  "tools": [
    {
      "name": "get_current_weather",
      "description": "Get current weather for a city"
    },
    {
      "name": "get_forecast",
      "description": "Get multi-day weather forecast"
    },
    {
      "name": "get_clothing_advice",
      "description": "Get clothing recommendations based on weather"
    },
    {
      "name": "personalized_weather_report",
      "description": "Interactive personalized weather report via Elicitation"
    }
  ],
  "resources": [
    {
      "uri": "weather://cities/supported",
      "description": "List of supported cities"
    },
    {
      "uri": "weather://alerts/{city}",
      "description": "Weather alerts for a city"
    }
  ],
  "prompts": [
    {
      "name": "daily_outfit_prompt",
      "description": "Daily outfit recommendation prompt template"
    },
    {
      "name": "travel_planning_prompt",
      "description": "Travel weather planning prompt template"
    }
  ]
}

# 使用MCP CLI发布到Registry
npx @anthropic-ai/mcp-registry publish \
  --metadata mcp-metadata.json \
  --package mcp-weather-advisor

@mcp.tool()
async def get_forecast(
    city: Annotated[str, Field(
        description="城市名称",
        min_length=1,
        max_length=100,
    )],
    days: Annotated[int, Field(
        description="预报天数（1-3）",
        ge=1,
        le=3,
    )] = 3,
) -> list[dict]:
    ...

from datetime import datetime, timedelta
from collections import defaultdict

# 简单的内存级限流
_rate_limits: dict[str, list[datetime]] = defaultdict(list)
RATE_LIMIT = 30  # 每分钟30次
RATE_WINDOW = 60  # 60秒窗口


def check_rate_limit(client_id: str) -> bool:
    """检查是否超过限流阈值"""
    now = datetime.now()
    cutoff = now - timedelta(seconds=RATE_WINDOW)

    # 清理过期记录
    _rate_limits[client_id] = [
        t for t in _rate_limits[client_id] if t > cutoff
    ]

    if len(_rate_limits[client_id]) >= RATE_LIMIT:
        return False

    _rate_limits[client_id].append(now)
    return True


@mcp.tool()
async def get_current_weather(
    city: str,
    units: str = "celsius",
    _client_id: str = "default",
) -> dict:
    if not check_rate_limit(_client_id):
        return {"error": "Rate limit exceeded. Try again later."}
    # ... 原有逻辑

# FastMCP 内置OAuth支持
mcp = FastMCP(
    "weather-advisor",
    auth_provider=OAuthProvider(
        issuer="https://your-domain.com",
        client_id="weather-advisor-mcp",
        jwks_uri="https://your-domain.com/.well-known/jwks.json",
    ),
)

flowchart LR
    A[1. 代码开发<br/>FastMCP三原语] --> B[2. 本地调试<br/>MCP Inspector]
    B --> C[3. 单元测试<br/>pytest]
    C --> D[4. Claude Code<br/>集成验证]
    D --> E[5. Docker<br/>容器化]
    E --> F[6. 安全加固<br/>OAuth/限流]
    F --> G[7. PyPI<br/>发布Python包]
    G --> H[8. MCP Registry<br/>注册上架]
    H --> I[9. GitHub Release<br/>自动化发布]

    style A fill:#4CAF50,color:#fff
    style D fill:#2196F3,color:#fff
    style G fill:#FF9800,color:#fff
    style H fill:#9C27B0,color:#fff

# 1. 代码质量检查
ruff check src/ tests/

# 2. 运行全部测试
pytest tests/ -v --tb=short

# 3. 构建包并检查
python -m build && twine check dist/*

# 4. 验证CLI入口点
mcp-weather-advisor --help

# 5. Docker构建测试
docker build -t mcp-weather-advisor:0.1.0 .

# 6. 检查元数据文件
cat mcp-metadata.json | python -m json.tool

# 7. 确认版本号一致
grep "version" pyproject.toml mcp-metadata.json

from functools import lru_cache
from datetime import datetime, timedelta
import json

_cache: dict[str, tuple[datetime, dict]] = {}
CACHE_TTL = timedelta(minutes=15)


async def get_cached_weather(city: str) -> dict:
    """带缓存的天气查询"""
    now = datetime.now()

    if city in _cache:
        cached_time, cached_data = _cache[city]
        if now - cached_time < CACHE_TTL:
            return cached_data

    # 实际请求
    data = await _fetch_weather(city)
    _cache[city] = (now, data)
    return data

@mcp.tool()
async def get_current_weather(city: str, units: str = "celsius") -> dict:
    """获取天气信息 - 带降级策略"""
    try:
        return await _fetch_from_wttr(city, units)
    except httpx.TimeoutException:
        # 主API超时，尝试备用源
        try:
            return await _fetch_from_open_meteo(city, units)
        except Exception:
            return {
                "error": "天气服务暂时不可用",
                "city": city,
                "fallback": True,
            }
    except httpx.HTTPStatusError as e:
        return {
            "error": f"天气查询失败: {e.response.status_code}",
            "city": city,
        }