一、MCP 简介

MCP 是一个开放协议，它为应用程序向 LLM 提供上下文的方式进行了标准化。你可以将 MCP 想象成 AI 应用程序的 USB-C 接口。就像 USB-C 为设备连接各种外设和配件提供了标准化的方式一样，MCP 为 AI 模型连接各种数据源和工具提供了标准化的接口。

1.1 MCP 通用架构

MCP 核心采用客户端-服务器架构，主机应用可以连接多个服务器：

MCP Hosts: 如 Claude Desktop、IDE 或 AI 工具，希望通过 MCP 访问数据的程序

MCP Clients: 维护与服务器一对一连接的协议客户端

MCP Servers: 轻量级程序，通过标准的 Model Context Protocol 提供特定能力

本地数据源: MCP 服务器可安全访问的计算机文件、数据库和服务

远程服务: MCP 服务器可连接的互联网上的外部系统（如通过 APIs）

二、MCP 天气服务需求

许多 LLM 目前不具备获取天气预报和恶劣天气警报的能力。因此我们将构建一个服务器，它暴露两个 tools：get-alerts 和 get-forecast。然后，我们将服务器连接到一个 MCP host（在本例中是 Claude for Desktop）：

三、环境准备

3.1 版本

• Python 3.10 或更高
• Python MCP SDK 1.2.0 或更高

3.2 环境安装

• 安装 uv

curl -LsSf https://astral.sh/uv/install.sh | sh

• 初始化项目

# 为我们的项目创建一个新 directory
uv init weather
cd weather

• 创建虚拟环境

# 创建 virtual environment 并激活它

uv venv
source .venv/bin/activate

• 安装依赖

# 安装 dependencies

uv add "mcp[cli]" httpx

• 创建项目模块文件

# 创建我们的 server file

touch weather.py

如果使用 Pycharm 开发，此时可以看到如下图示：

四、构建 MCP 天气服务

4.1 导入 packages 并设置 instance

将如下代码添加到 weather.py 文件的顶部：

# -*- coding:utf-8 -*-

# 导入依赖
from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP

# 初始化 FastMCP server
mcp = FastMCP("weather")

# 设置常量
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"

FastMCP class 使用 Python type hints 和 docstrings 来自动生成 tool definitions，从而轻松创建和维护 MCP tools。

4.2 添加 Helper functions

async def make_nws_request(url: str) -> dict[str, Any] | None:
    """向 NWS API 发送请求，并进行适当的错误处理。"""
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "application/geo+json"
    }
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(url, headers=headers, timeout=30.0)
            response.raise_for_status()
            return response.json()
        except Exception:
            return None


def format_alert(feature: dict) -> str:
    """将警报 feature 格式化为可读的字符串。"""
    props = feature["properties"]
    return f"""
事件: {props.get('event', 'Unknown')}
区域: {props.get('areaDesc', 'Unknown')}
严重性: {props.get('severity', 'Unknown')}
描述: {props.get('description', 'No description available')}
指示: {props.get('instruction', 'No specific instructions provided')}
"""

4.3 实现 tool execution

@mcp.tool()
async def get_alerts(state: str) -> str:
    """获取美国州的天气警报。

    Args:
        state: 两个字母的美国州代码（例如 CA、NY）
    """
    url = f"{NWS_API_BASE}/alerts/active/area/{state}"
    data = await make_nws_request(url)

    if not data or "features" not in data:
        return "无法获取警报或未找到警报。"

    if not data["features"]:
        return "该州没有活跃的警报。"

    alerts = [format_alert(feature) for feature in data["features"]]
    return "\n---\n".join(alerts)


@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    """获取某个位置的天气预报。

    Args:
        latitude: 位置的纬度
        longitude: 位置的经度
    """
    # 首先获取预报网格 endpoint
    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
    points_data = await make_nws_request(points_url)

    if not points_data:
        return "无法获取此位置的预报数据。"

    # 从 points response 中获取预报 URL
    forecast_url = points_data["properties"]["forecast"]
    forecast_data = await make_nws_request(forecast_url)

    if not forecast_data:
        return "无法获取详细预报。"

    # 将 periods 格式化为可读的预报
    periods = forecast_data["properties"]["periods"]
    forecasts = []
    for period in periods[:5]:  # 仅显示接下来的 5 个 periods
        forecast = f"""
{period['name']}:
温度: {period['temperature']}°{period['temperatureUnit']}
风: {period['windSpeed']} {period['windDirection']}
预报: {period['detailedForecast']}
"""
        forecasts.append(forecast)

    return "\n---\n".join(forecasts)

4.4 添加 main 入口函数

if __name__ == '__main__':
    mcp.run(transport='stdio')

4.5 运行 server

mcp dev weather.py

4.6 测试 server

访问链接 http://127.0.0.1:6274 后在 tools 页面可以看到如下图中的两个工具