Python 实战开发 MCP Server：从 NWS API 到 Weather Tools

MCP 从入门到工程实践系列，第 6 篇，共 9 篇。
本文以 MCP 2026-07-28 文档版本中的 Python 示例为基线。

很多人第一次看官方 Weather Server 代码时，会有这些困惑：

• 为什么先定义 HTTP Header？
• httpx2.AsyncClient() 为什么叫“异步客户端”？
• await 到底在等谁？
• response.json() 是不是把内容保存成 JSON 文件？
• 编写者怎么知道响应里有 properties、forecast 和 periods？
• 为什么查一次天气预报，却要调用两次 NWS API？
• @mcp.tool() 生成的是 Tool Schema，还是 JSON-RPC 消息？

这些问题并不全是 MCP 问题。Weather Server 同时使用了四层知识：

业务需求
  ↓
NWS HTTP API Contract
  ↓
Python、字典与 async/await
  ↓
MCP SDK 把函数暴露为 Tool

只盯着最后一层，就会感觉字段和写法都是凭空出现的。本文按开发者真正会采用的顺序，从上游 API 开始拆解。

一、最终要构建什么

Weather Server 对外提供两个 Tool：

Tool	输入	输出
get_alerts	美国州的两位代码，如 CA	当前天气预警
get_forecast	纬度、经度	接下来几个时段的天气预报

内部还有两个普通 Helper：

Helper	作用	是否暴露给 MCP Client
make_nws_request	请求 NWS API，并解析 JSON	否
format_alert	把一条预警格式化成可读文本	否

MCPServer("weather")
  ├─ Helper: make_nws_request
  ├─ Helper: format_alert
  ├─ Tool: get_alerts
  └─ Tool: get_forecast

只有被 @mcp.tool() 注册的函数，才会出现在 tools/list 结果中。普通 Helper 只是 Server 的内部实现。

二、准备 Python 项目

官方 Python 路径要求 Python 3.10+ 和 Python MCP SDK 2.0.0+。使用 uv 可以快速创建环境：

uv init weather
cd weather
uv venv
source .venv/bin/activate
uv add "mcp[cli]"

当前版本示例使用 httpx2，它已经由 SDK 依赖带入，不应把教程中的 Import 擅自改成 httpx。

建立 Server 与常量：

from typing import Any

import httpx2
from mcp.server import MCPServer

mcp = MCPServer("weather")

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

这里有三个关键点：

1. MCPServer("weather") 创建 Server 实例；
2. 后面的 Decorator 会把 Tool 注册到这个实例；
3. NWS_API_BASE 和 USER_AGENT 属于上游 HTTP API 配置，不是 MCP 字段。

旧版或其他 MCP 示例可能出现 FastMCP。学习带版本的官方页面时，应以该页面对应的 SDK API 为准，不要混用不同版本的类名。

三、先理解 HTTP Helper

async def make_nws_request(url: str) -> dict[str, Any] | None:
    """Make a request to the NWS API with proper error handling."""
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "application/geo+json",
    }

    async with httpx2.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

下面逐层解释。

1. 类型标注

url: str 表示函数期望收到字符串 URL。

-> dict[str, Any] | None

表示函数可能返回 Key 为字符串的 Python 字典，也可能返回 None。

类型标注帮助开发者、IDE 和静态检查器理解代码。对注册为 MCP Tool 的函数，SDK 还能利用它生成 Schema；但这个 Helper 没有 @mcp.tool()，不会自动成为 Tool。

2. 为什么先定义请求头

headers = {
    "User-Agent": USER_AGENT,
    "Accept": "application/geo+json",
}

请求头是 HTTP Request 的组成部分：

• User-Agent：请求来自哪个应用；
• Accept：希望 Server 返回 application/geo+json。

为什么需要它们？因为 NWS API 的使用说明和 HTTP Contract 对请求有相应要求，不是因为“写 MCP 必须有 Header”。

换一个 API，Header 可能完全不同：

Authorization: Bearer ...
Content-Type: application/json
Accept-Language: zh-CN

所以这一段属于 HTTP API 基础，而不是 MCP Protocol。

3. AsyncClient 属于什么知识

async with httpx2.AsyncClient() as client:

可以拆成四层：

• httpx2：Python HTTP Client Library；
• AsyncClient：Library 提供的异步 HTTP Client 类；
• client：创建出来的对象实例；
• async with：异步 Context Manager，用来管理生命周期。

HTTP Client 会持有 Socket、TLS Connection、Connection Pool 等资源：

创建 Client
  ↓
发送一个或多个 Request
  ↓
关闭连接和相关资源

async with 能保证离开代码块时完成清理，即使内部发生异常也不容易遗留连接。

这一段主要涉及：

1. Python Class 与 Object；
2. Context Manager：with / async with；
3. Asyncio：async def / await；
4. HTTP Client Library。

4. await 到底怎样执行

response = await client.get(url)
next_statement()

对当前这个协程来说，必须等 client.get(url) 得到结果以后，才会执行 next_statement()。但等待网络响应期间，Event Loop 可以去运行其他协程，所以不等于冻结整个 Server。

调用 make_nws_request 时也要 await：

data = await make_nws_request(url)

原因是：

1. make_nws_request 由 async def 定义；
2. 调用它会产生 Coroutine；
3. 当前异步函数要用 await 才能取得最终返回值。

5. 状态码与超时

response = await client.get(
    url,
    headers=headers,
    timeout=30.0,
)
response.raise_for_status()

timeout=30.0 限制等待时间。raise_for_status() 检查 HTTP Status：

• 2xx：通常继续；
• 4xx：请求、认证或权限等问题；
• 5xx：上游 Server 故障；
• 出错时抛出异常，随后被 except 捕获。

教程为了简洁，把所有异常都转换为 None：

except Exception:
    return None

生产代码不宜完全吞错，至少应在 stderr 或 Observability 系统中区分 Timeout、HTTP Error、JSON Parse Error，并记录脱敏后的上下文。

6. response.json() 不是生成 JSON 文件

NWS Response Body 在网络上是 Bytes/Text，语法是 JSON：

{
  "features": [
    {
      "properties": {
        "event": "Flood Warning"
      }
    }
  ]
}

return response.json()

这一步把 JSON 文本解析为内存中的 Python 对象：

JSON	Python
object	dict
array	list
string	str
true / false	True / False
null	None

打印时它们很像，但 JSON 文本和 Python dict 不是同一种类型。

如果目标真是读写 .json 文件，Python 标准库常见函数是：

函数	方向
json.loads(text)	JSON 字符串 → Python 对象
json.dumps(obj)	Python 对象 → JSON 字符串
json.load(file)	JSON 文件 → Python 对象
json.dump(obj, file)	Python 对象 → 写入 JSON 文件

response.json() 只解析 HTTP 响应，不会在磁盘创建文件。

四、字段名不是程序员猜出来的

def format_alert(feature: dict) -> str:
    """Format an alert feature into a readable string."""
    props = feature["properties"]
    return f"""
Event: {props.get("event", "Unknown")}
Area: {props.get("areaDesc", "Unknown")}
Severity: {props.get("severity", "Unknown")}
Description: {props.get("description", "No description available")}
Instructions: {props.get("instruction", "No specific instructions provided")}
"""

1. properties 是一个 Key

假设 feature 是：

{
    "properties": {
        "event": "Flood Warning",
        "severity": "Severe",
    }
}

props = feature["properties"]

就是取出 properties 对应的 Value，得到的 props 仍然是一个字典。

properties 不是 Python 保留字，也不是 MCP 固定字段，它来自 NWS GeoJSON Response Schema。

2. .get() 是字典方法

props.get("event", "Unknown")

表示有 event 就返回它，没有则返回默认值 "Unknown"：

写法	Key 不存在时	适用场景
props["event"]	抛出 KeyError	Contract 规定一定存在
props.get("event", "Unknown")	返回默认值	字段可选，展示时允许降级

知道可以使用 .get()，是因为 props 是 Python dict；知道要读取 event，则是因为 NWS API Documentation 定义了这个字段。

3. 正确的 API Client 编写过程

面对陌生 API，合理顺序是：

1. 阅读官方 API Documentation；
2. 查找 OpenAPI、JSON Schema 或字段表；
3. 查看官方 Example Response；
4. 用 curl 或小脚本测试；
5. 区分 Required 与 Optional Field；
6. 再决定使用 []、.get() 或数据模型校验。

不能凭字段名字猜代码，也不能只观察一次响应就假设所有未来响应都完全相同。

五、把 get_alerts 注册成 MCP Tool

@mcp.tool()
async def get_alerts(state: str) -> str:
    """Get weather alerts for a US state.

    Args:
        state: Two-letter US state code (e.g. 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 "Unable to fetch alerts or no alerts found."

    if not data["features"]:
        return "No active alerts for this state."

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

1. Decorator 生成什么

@mcp.tool() 读取函数名、Docstring、state: str 和 -> str。SDK 据此建立 Tool Definition，其中包含类似下面的 Input Schema：

{
  "name": "get_alerts",
  "description": "Get weather alerts for a US state.",
  "inputSchema": {
    "type": "object",
    "properties": {
      "state": {
        "type": "string",
        "description": "Two-letter US state code (e.g. CA, NY)"
      }
    },
    "required": ["state"]
  }
}

这是 Tool Definition 的一部分，不是完整 JSON-RPC Request。

Client 真正执行 Tool 时，SDK 才构造类似：

{
  "jsonrpc": "2.0",
  "id": 8,
  "method": "tools/call",
  "params": {
    "name": "get_alerts",
    "arguments": {
      "state": "CA"
    }
  }
}

Server 作者定义 Python 函数和类型
        ↓
MCP SDK 生成 Tool Definition / Input Schema
        ↓
模型按 Schema 生成本次 Name + Arguments
        ↓
Application 校验并调用 Client SDK
        ↓
SDK 构造 JSON-RPC tools/call

模型通常不生成 Tool Schema，也不需要亲手拼 JSON-RPC 外壳。

2. features 从哪里来

data 是 make_nws_request 返回的字典。NWS Alerts Endpoint 返回 GeoJSON，顶层包含 features：

{
  "features": [
    {
      "properties": {
        "event": "Flood Warning"
      }
    }
  ]
}

data["features"] 是在访问上游响应。format_alert 没有添加 features，它只处理列表中的一项。

3. 三种“没有数据”不同

if not data or "features" not in data:

Python Truthiness 会把 None、{}、[] 和空字符串等视为 False：

data is None / data == {}
  → 请求失败，或没有可用响应

"features" not in data
  → 返回结构不符合预期

data["features"] == []
  → 返回结构合法，只是没有 Active Alert

所以 if not data["features"] 是在区分“请求失败”和“成功请求但没有预警”。

4. List Comprehension 与 join

alerts = [
    format_alert(feature)
    for feature in data["features"]
]

等价于：

alerts = []
for feature in data["features"]:
    alerts.append(format_alert(feature))

最后用 "\n---\n".join(alerts) 把多条字符串连接为一段 Tool Result。

六、get_forecast 为什么要请求两次

完整函数：

@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    """Get weather forecast for a location.

    Args:
        latitude: Latitude of the location
        longitude: Longitude of the location
    """
    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
    points_data = await make_nws_request(points_url)

    if not points_data:
        return "Unable to fetch forecast data for this location."

    forecast_url = points_data["properties"]["forecast"]
    forecast_data = await make_nws_request(forecast_url)

    if not forecast_data:
        return "Unable to fetch detailed forecast."

    periods = forecast_data["properties"]["periods"]
    forecasts = []

    for period in periods[:5]:
        forecast = f"""
{period["name"]}:
Temperature: {period["temperature"]}°{period["temperatureUnit"]}
Wind: {period["windSpeed"]} {period["windDirection"]}
Forecast: {period["detailedForecast"]}
"""
        forecasts.append(forecast)

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

1. 两次请求来自 NWS API 设计

需求是“输入经纬度，得到文字预报”。但 NWS /points/{latitude},{longitude} Endpoint 不直接返回完整预报，而是返回 Grid 信息和 Forecast URL：

latitude + longitude
        ↓
GET /points/{latitude},{longitude}
        ↓
points_data["properties"]["forecast"]
        ↓
GET forecast_url
        ↓
forecast_data["properties"]["periods"]

第一次响应的关键结构类似：

{
  "properties": {
    "forecast": "https://api.weather.gov/gridpoints/..."
  }
}

第二次响应的关键结构类似：

{
  "properties": {
    "periods": [
      {
        "name": "Tonight",
        "temperature": 68,
        "temperatureUnit": "F",
        "windSpeed": "5 mph",
        "windDirection": "NW",
        "detailedForecast": "..."
      }
    ]
  }
}

所以两次调用不是 MCP 的要求，而是 NWS API 的资源关系。

2. 开发者当时的推导逻辑

这段程序不是先写两次请求，再碰巧发现能运行。合理的推导过程是：

1. 明确 Tool 输入是经纬度，输出是可读预报；
2. 阅读 NWS points Endpoint 文档；
3. 发现它返回 properties.forecast URL；
4. 阅读 Forecast Endpoint 文档；
5. 发现时段数组位于 properties.periods；
6. 根据字段说明，选择展示温度、风和详细预报；
7. 对两次网络请求分别处理失败；
8. 只取前五个 Period，控制 Tool Result 大小；
9. 最后把普通 Python 函数注册为 MCP Tool。

也就是说：

先由业务需求和上游 API Contract 决定内部实现，再由 MCP 把这个实现标准化地暴露给 AI Application。

3. 多层字典访问怎样读

forecast_url = points_data["properties"]["forecast"]

可以拆成：

properties = points_data["properties"]
forecast_url = properties["forecast"]

同理，forecast_data["properties"]["periods"] 就是先取 properties 字典，再取其中的 periods 列表。

4. 多行 f-string 怎样读

forecast = f"""
{period["name"]}:
Temperature: {period["temperature"]}°{period["temperatureUnit"]}
Wind: {period["windSpeed"]} {period["windDirection"]}
Forecast: {period["detailedForecast"]}
"""

f"""...""" 是多行 f-string。每个花括号都会求值并插入字符串。

假设当前 period 是：

{
    "name": "Tonight",
    "temperature": 68,
    "temperatureUnit": "F",
    "windSpeed": "5 mph",
    "windDirection": "NW",
    "detailedForecast": "Mostly clear.",
}

最终得到：

Tonight:
Temperature: 68°F
Wind: 5 mph NW
Forecast: Mostly clear.

periods[:5] 是 List Slice，只取前五项；append 把每个格式化字符串加入列表；最后再用 join 合并。

七、完整 Server 代码

把以上部分放在一起：

from typing import Any

import httpx2
from mcp.server import MCPServer

mcp = MCPServer("weather")

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


async def make_nws_request(url: str) -> dict[str, Any] | None:
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "application/geo+json",
    }

    async with httpx2.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:
    props = feature["properties"]
    return f"""
Event: {props.get("event", "Unknown")}
Area: {props.get("areaDesc", "Unknown")}
Severity: {props.get("severity", "Unknown")}
Description: {props.get("description", "No description available")}
Instructions: {props.get("instruction", "No specific instructions provided")}
"""


@mcp.tool()
async def get_alerts(state: str) -> str:
    """Get weather alerts for a US state.

    Args:
        state: Two-letter US state code (e.g. 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 "Unable to fetch alerts or no alerts found."

    if not data["features"]:
        return "No active alerts for this state."

    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:
    """Get weather forecast for a location.

    Args:
        latitude: Latitude of the location
        longitude: Longitude of the location
    """
    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
    points_data = await make_nws_request(points_url)

    if not points_data:
        return "Unable to fetch forecast data for this location."

    forecast_url = points_data["properties"]["forecast"]
    forecast_data = await make_nws_request(forecast_url)

    if not forecast_data:
        return "Unable to fetch detailed forecast."

    periods = forecast_data["properties"]["periods"]
    forecasts = []

    for period in periods[:5]:
        forecast = f"""
{period["name"]}:
Temperature: {period["temperature"]}°{period["temperatureUnit"]}
Wind: {period["windSpeed"]} {period["windDirection"]}
Forecast: {period["detailedForecast"]}
"""
        forecasts.append(forecast)

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


def main() -> None:
    mcp.run(transport="stdio")


if __name__ == "__main__":
    main()

八、启动 stdio Server

mcp.run(transport="stdio") 表示从 stdin 读取 MCP 消息，并向 stdout 写协议消息。它通常由 Host 启动和管理。

直接运行：

uv run weather.py

终端看起来没有输出，往往只是 Server 正在等待 Client，并不代表失败。不要向它手工输入自然语言。

stdio 最重要的日志规则

stdout 是协议通道，只能写 MCP/JSON-RPC 数据。

不要随意执行：

print("server started")

这可能污染协议流。普通日志应写 stderr，例如使用 Python logging；生产环境还应添加结构化日志、Correlation ID 和脱敏。

九、连接到 Host

以能够启动本地 stdio Server 的 Host 为例，配置思路是：

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

Host 读取配置
  ↓
启动 uv run weather.py
  ↓
创建对应 MCP Client 连接
  ↓
调用 tools/list
  ↓
发现 get_alerts 与 get_forecast
  ↓
模型需要时建议 Tool Use
  ↓
Host 发出 MCP tools/call

必须换成实际绝对路径，并在修改配置后完全重启 Host。

十、怎样测试这台 Server

推荐从内向外分层测试：

1. 单独测试 NWS API：确认机器能访问 NWS，Header、州代码和坐标合法。NWS 服务主要覆盖美国地区。
2. 测试 Helper：检查 make_nws_request 是否能返回字典，关键字段是否存在。
3. 使用 MCP Inspector：查看 tools/list、Input Schema，手工执行 Tool 并检查 Result。
4. 接入真实 Host：确认模型能选择正确 Tool 和参数，权限界面合理，结果能被自然语言解释。

调试 Helper 时，注意不要在正式 stdio 协议运行期间把调试文本写入 stdout。

十一、教学代码的生产边界

生产 Server 通常还应补充：

• 复用 HTTP Client，减少重复建连；
• 在 Shutdown 时正确关闭 Client；
• 校验 Latitude、Longitude 与州代码；
• 区分 Timeout、DNS、4xx、5xx 与 JSON Parse Error；
• 避免 except Exception 完全吞错；
• 为 Rate Limit 加退避和重试策略；
• 缓存适合缓存的数据；
• 限制 Tool Result 大小；
• 明确“当前没有预警”不是系统错误；
• 为外部响应结构变化增加 Schema Validation；
• 对日志中的 URL、参数与用户数据脱敏。

十二、常见误区

误区 1：这些全都是 MCP 知识

不是。Header、Status Code 和 JSON 属于 HTTP；AsyncClient、await、字典和 f-string 属于 Python；Tool 注册、Schema、Transport 和协议交互才属于 MCP。

误区 2：response.json() 会生成 JSON 文件

不会。它只把 HTTP Response Body 的 JSON 解析成 Python 对象。

误区 3：properties 和 forecast 是 MCP 固定字段

不是。它们来自 NWS API Contract。换一个上游 API，字段结构也会变化。

误区 4：模型创建 Tool Schema

不是。Server 作者和 MCP SDK 定义 Tool Schema；模型只按 Schema 生成本次调用的 Name 与 Arguments。

误区 5：一次用户问题只能调用一次 API

不是。MCP Tool 内部可以调用零次、一次或多次上游 API。这里的两次调用由 NWS API 设计决定。

误区 6：await 会冻结整个 Server

当前协程会等待结果，但 Event Loop 可以运行其他协程。真正会阻塞 Event Loop 的通常是同步、耗时的 I/O 或 CPU 工作。

误区 7：没有 Active Alert 就是程序出错

features: [] 是合法业务结果。它与网络失败、响应结构错误需要分别表达。

十三、总结

Weather Server 的完整思路是：

阅读 NWS API Documentation
        ↓
确定 Endpoint、Header、Response Schema
        ↓
用异步 HTTP Helper 获取并解析 JSON
        ↓
用普通 Python 处理 dict/list 和格式化
        ↓
用 @mcp.tool() 注册业务入口
        ↓
SDK 生成 Tool Definition
        ↓
通过 stdio 暴露给 MCP Host

真正重要的不是背下 properties["forecast"]，而是掌握推导方法：

字段来自 API Contract，控制流来自业务需求，MCP 负责把已经实现好的能力以标准形式交给 AI Application。

下一篇将站到另一端，完整实现 Python MCP Client，并拆解从 tools/list、模型 tool_use 到 MCP tools/call、再到 tool_result 的闭环。

参考资料

• https://modelcontextprotocol.io/docs/2026-07-28/develop/build-server
• https://modelcontextprotocol.io/docs/2026-07-28/tools/debugging
• National Weather Service API Documentation