在本教程中,我们将构建一个简单的 MCP 天气服务器并将其连接到主机 Claude for Desktop。我们将从基本设置开始,然后逐步进行更复杂的用例。

我们将构建什么

许多 LLM(包括 Claude)目前无法获取天气预报和严重天气警报。让我们使用 MCP 来解决这个问题!

我们将构建一个暴露两个工具的服务器:get-alertsget-forecast。然后我们将服务器连接到 MCP 主机(在本例中为 Claude for Desktop):

服务器可以连接到任何客户端。我们在这里选择了 Claude for Desktop 以简化操作,但我们也有关于构建您自己的客户端以及其他客户端列表的指南。

核心 MCP 概念

MCP 服务器可以提供三种主要类型的功能:

  1. 资源:可以被客户端读取的类似文件的数据(如 API 响应或文件内容)
  2. 工具:可以由 LLM 调用的函数(需要用户批准)
  3. 提示:帮助用户完成特定任务的预编写模板

本教程将主要关注工具。

让我们开始构建我们的天气服务器吧!您可以在此处找到我们将构建的完整代码。

先决知识

本快速入门假设您熟悉:

  • Python
  • LLMs 如 Claude

系统要求

对于 Python,请确保您安装了 Python 3.9 或更高版本。

设置您的环境

首先,让我们安装 uv 并设置我们的 Python 项目和环境:

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

确保在此之后重新启动您的终端,以确保 uv 命令被识别。

现在,让我们创建并设置我们的项目:

# 创建一个新目录用于我们的项目
uv init weather
cd weather

# 创建虚拟环境并激活它
uv venv
source .venv/bin/activate

# 安装依赖项
uv add mcp httpx 

# 删除模板文件
rm hello.py

# 创建我们的文件
mkdir -p src/weather
touch src/weather/__init__.py
touch src/weather/server.py

将此代码添加到 pyproject.toml

...rest of config

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

[project.scripts]
weather = "weather:main"

将此代码添加到 __init__.py

src/weather/__init__.py
from . import server
import asyncio

def main():
    """包的主入口点。"""
    asyncio.run(server.main())

# 可选地在包级别公开其他重要项目
__all__ = ['main', 'server']

现在让我们深入构建您的服务器。

构建您的服务器

导入包

将这些添加到 server.py 的顶部:

from typing import Any
import asyncio
import httpx
from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
import mcp.server.stdio

设置实例

然后初始化服务器实例和 NWS API 的基本 URL:

NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"

server = Server("weather")

实现工具列表

我们需要告诉客户端有哪些工具可用。list_tools() 装饰器注册此处理程序:

@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    """
    列出可用工具。
    每个工具使用 JSON Schema 验证指定其参数。
    """
    return [
        types.Tool(
            name="get-alerts",
            description="获取某个州的天气警报",
            inputSchema={
                "type": "object",
                "properties": {
                    "state": {
                        "type": "string",
                        "description": "两字母州代码(例如 CA,NY)",
                    },
                },
                "required": ["state"],
            },
        ),
        types.Tool(
            name="get-forecast",
            description="获取某个位置的天气预报",
            inputSchema={
                "type": "object",
                "properties": {
                    "latitude": {
                        "type": "number",
                        "description": "位置的纬度",
                    },
                    "longitude": {
                        "type": "number",
                        "description": "位置的经度",
                    },
                },
                "required": ["latitude", "longitude"],
            },
        ),
    ]

这定义了我们的两个工具:get-alertsget-forecast

辅助函数

接下来,让我们添加用于查询和格式化来自国家气象局 API 的数据的辅助函数:

async def make_nws_request(client: httpx.AsyncClient, url: str) -> dict[str, Any] | None:
    """向 NWS API 发出请求并进行适当的错误处理。"""
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "application/geo+json"
    }

    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:
    """将警报功能格式化为简洁的字符串。"""
    props = feature["properties"]
    return (
        f"事件: {props.get('event', '未知')}\n"
        f"区域: {props.get('areaDesc', '未知')}\n"
        f"严重性: {props.get('severity', '未知')}\n"
        f"状态: {props.get('status', '未知')}\n"
        f"标题: {props.get('headline', '无标题')}\n"
        "---"
    )

实现工具执行

工具执行处理程序负责实际执行每个工具的逻辑。让我们添加它:

@server.call_tool()
async def handle_call_tool(
    name: str, arguments: dict | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    """
    处理工具执行请求。
    工具可以获取天气数据并通知客户端更改。
    """
    if not arguments:
        raise ValueError("缺少参数")
  
    if name == "get-alerts":
        state = arguments.get("state")
        if not state:
            raise ValueError("缺少 state 参数")

        # 将 state 转换为大写以确保格式一致
        state = state.upper()
        if len(state) != 2:
            raise ValueError("state 必须是两字母代码(例如 CA,NY)")

        async with httpx.AsyncClient() as client:
            alerts_url = f"{NWS_API_BASE}/alerts?area={state}"
            alerts_data = await make_nws_request(client, alerts_url)

            if not alerts_data:
                return [types.TextContent(type="text", text="无法检索警报数据")]

            features = alerts_data.get("features", [])
            if not features:
                return [types.TextContent(type="text", text=f"{state} 没有活动警报")]

            # 将每个警报格式化为简洁的字符串
            formatted_alerts = [format_alert(feature) for feature in features[:20]] # 仅取前 20 个警报
            alerts_text = f"{state} 的活动警报:\n\n" + "\n".join(formatted_alerts)

            return [
                types.TextContent(
                    type="text",
                    text=alerts_text
                )
            ]
    elif name == "get-forecast":
        try:
            latitude = float(arguments.get("latitude"))
            longitude = float(arguments.get("longitude"))
        except (TypeError, ValueError):
            return [types.TextContent(
                type="text",
                text="无效的坐标。请提供有效的纬度和经度数字。"
            )]
            
        # 基本坐标验证
        if not (-90 <= latitude <= 90)not (-180 <= longitude <= 180):
            return [types.TextContent(
                type="text",
                text="无效的坐标。纬度必须在 -90 到 90 之间,经度在 -180 到 180 之间。"
            )]

        async with httpx.AsyncClient() as client:
            # 首先获取网格点
            lat_str = f"{latitude}"
            lon_str = f"{longitude}"
            points_url = f"{NWS_API_BASE}/points/{lat_str},{lon_str}"
            points_data = await make_nws_request(client, points_url)

            if not points_data:
                return [types.TextContent(type="text", text=f"无法检索坐标的网格点数据:{latitude}, {longitude}。此位置可能不受 NWS API 支持(仅支持美国位置)。")]

            # 从响应中提取预报 URL
            properties = points_data.get("properties", {})
            forecast_url = properties.get("forecast")
            
            if not forecast_url:
                return [types.TextContent(type="text", text="无法从网格点数据中获取预报 URL")]

            # 获取预报
            forecast数据 = await make_nws_request(client, forecast_url)
            
            if not forecast数据:
                return [types.TextContent(type="text", text="无法检索预报数据")]

            # 格式化预报期
            periods = forecast数据.get("properties", {}).get("periods", [])
            if not periods:
                return [types.TextContent(type="text", text="没有可用的预报期")]

            # 将每个期格式化为简洁的字符串
            formatted_forecast = []
            for period in periods:
                forecast_text = (
                    f"{period.get('name', '未知')}:\n"
                    f"温度: {period.get('temperature', '未知')}°{period.get('temperatureUnit', 'F')}\n"
                    f"风: {period.get('windSpeed', '未知')} {period.get('windDirection', '')}\n"
                    f"{period.get('shortForecast', '无可用预报')}\n"
                    "---"
                )
                formatted_forecast.append(forecast_text)

            forecast_text = f"{latitude}, {longitude} 的预报:\n\n" + "\n".join(formatted_forecast)

            return [types.TextContent(
                type="text",
                text=forecast_text
            )]
    else:
        raise ValueError(f"未知工具: {name}")

运行服务器

最后,实现运行服务器的主函数:

async def main():
    # 使用 stdin/stdout 流运行服务器
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="weather",
                server_version="0.1.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                ),
            ),
        )

# 如果您想连接到自定义客户端,这是必需的
if __name__ == "__main__":
    asyncio.run(main())

您的服务器已完成!运行 uv run src/weather/server.py 以确认一切正常。

现在让我们从现有的 MCP 主机 Claude for Desktop 测试您的服务器。

使用 Claude for Desktop 测试您的服务器

Claude for Desktop 尚不支持 Linux。Linux 用户可以继续构建客户端教程,以构建一个连接到我们刚刚构建的服务器的 MCP 客户端。

首先,确保您已安装 Claude for Desktop。您可以在此处安装最新版本。 如果您已经有 Claude for Desktop,请确保它已更新到最新版本。

我们需要为您想要使用的 MCP 服务器配置 Claude for Desktop。为此,请在文本编辑器中打开您的 Claude for Desktop 应用配置 ~/Library/Application Support/Claude/claude_desktop_config.json。如果文件不存在,请确保创建它。

例如,如果您安装了 VS Code

code ~/Library/Application\ Support/Claude/claude_desktop_config.json

然后,您将在 mcpServers 键中添加您的服务器。只有在至少一个服务器配置正确时,Claude for Desktop 中的 MCP UI 元素才会显示。

在本例中,我们将添加我们的单个天气服务器,如下所示:

Python
{
    "mcpServers": {
        "weather": {
            "command": "uv",
            "args": [
                "--directory",
                "/ABSOLUTE/PATH/TO/PARENT/FOLDER/weather",
                "run",
                "weather"
            ]
        }
    }
}

确保传递给服务器的路径是绝对路径。

这告诉 Claude for Desktop:

  1. 有一个名为 “weather” 的 MCP 服务器
  2. 通过运行 uv --directory /ABSOLUTE/PATH/TO/PARENT/FOLDER/weather run weather 启动它

保存文件,并重新启动 Claude for Desktop

使用命令进行测试

让我们确保 Claude for Desktop 正在拾取我们在 weather 服务器中公开的两个工具。您可以通过查找锤子 图标来执行此操作:

点击锤子图标后,您应该会看到列出的两个工具:

如果您的服务器未被 Claude for Desktop 拾取,请转到故障排除部分以获取调试提示。

如果锤子图标已显示,您现在可以通过在 Claude for Desktop 中运行以下命令来测试您的服务器:

  • 萨克拉门托的天气怎么样?
  • 德克萨斯州的活动天气警报是什么?

由于这是美国国家气象局,因此查询仅适用于美国位置。

背后的工作原理

当您提出问题时:

  1. 客户端将您的问题发送给 Claude
  2. Claude 分析可用工具并决定使用哪些工具
  3. 客户端通过 MCP 服务器执行所选工具
  4. 结果返回给 Claude
  5. Claude 形成自然语言响应
  6. 响应显示给您!

故障排除

有关更高级的故障排除,请查看我们的调试 MCP指南

下一步