MCP之weather server demo

基于 MCP（Model Context Protocol）协议的天气服务服务器，用于提供美国天气预警和天气预报
https://github.com/modelcontextprotocol/quickstart-resources/blob/main/weather-server-python/weather.py

通过集成美国国家气象局（NWS）的 API，提供两个核心功能：

• 获取美国各州的天气预警信息
• 根据经纬度获取具体地点的天气预报

from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP# Initialize FastMCP server
# 初始化 MCP 服务器
# 使用 FastMCP 创建了一个名为 "weather" 的 MCP 服务器实例
# FastMCP 是 MCP 协议的快速实现，用于简化服务器开发
mcp = FastMCP("weather")# Constants
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"# 网络请求工具函数
# 封装了对 NWS API 的异步 HTTP 请求
# 设置了必要的请求头（符合 NWS API 要求）
# 包含错误处理，请求失败时返回 None
# 使用 httpx.AsyncClient 进行异步网络请求
async def make_nws_request(url: str) -> dict[str, Any] | None:"""Make a request to the NWS API with proper error handling."""# 告诉 NWS API 服务器：“我需要地理空间格式的数据”。# NWS（美国国家气象局）的 API 可能支持多种返回格式（如普通 JSON、XML、GeoJSON）# 通过 Accept 头明确要求返回 GeoJSON，确保服务器返回的是包含地理坐标（如预警区域的多边形范围、预报地点的经纬度）的结构化数据# 而非纯文本描述# 便于后续代码解析：GeoJSON 结构固定，代码可以直接通过 feature["geometry"]["coordinates"] 提取坐标# 通过 feature["properties"] 提取预警等级、温度等信息，无需处理格式不一致的问题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# 数据格式化函数
# 将天气预警的原始 JSON 数据转换为易读的字符串格式
# 提取关键信息：事件类型、影响区域、严重程度、描述和指导说明
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')}
"""# MCP 工具函数 - 获取天气预警（weather alert）
# 被 @mcp.tool() 装饰，表明这是一个可供 MCP 客户端调用的工具
# 功能：根据美国州的两字母代码（如 CA 表示加利福尼亚州）获取该州的天气预警
# 调用 NWS API 的 /alerts/active/area/{state} 端点
# 处理返回数据，格式化后返回给客户端
@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 工具函数 - 用于根据经纬度获取天气预报
# 先调用 /points/{latitude},{longitude} 端点获取该坐标对应的预报网格信息
# 再从返回结果中获取具体的预报数据 URL 并请求详细预报
# 返回未来 5 个时段的天气预报，包括温度、风向风力和详细描述
@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:"""Get weather forecast for a location.Args:latitude: Latitude of the locationlongitude: Longitude of the location"""# First get the forecast grid endpointpoints_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."# Get the forecast URL from the points responseforecast_url = points_data["properties"]["forecast"]forecast_data = await make_nws_request(forecast_url)if not forecast_data:return "Unable to fetch detailed forecast."# Format the periods into a readable forecastperiods = forecast_data["properties"]["periods"]forecasts = []for period in periods[:5]:  # Only show next 5 periodsforecast = f"""
{period['name']}:
Temperature: {period['temperature']}°{period['temperatureUnit']}
Wind: {period['windSpeed']} {period['windDirection']}
Forecast: {period['detailedForecast']}
"""forecasts.append(forecast)return "\n---\n".join(forecasts)# 启动 MCP 服务器
# 使用标准输入输出（stdio）作为传输方式，这是 MCP 协议常用的简单传输方式
if __name__ == "__main__":# Initialize and run the servermcp.run(transport='stdio')

协议与交互

该服务器遵循 MCP 协议规范：
声明了自身支持的功能（通过 @mcp.tool() 装饰的函数）
客户端可以通过 MCP 协议发现并调用这些天气相关的工具
所有交互，基于 JSON-RPC（MCP 协议的基础）

总结

其他遵循 MCP 协议的客户端可以轻松发现并使用这些功能，体现了 MCP 协议 “像 USB-C 接口一样标准化连接” 的设计理念

附录

english

latitude 纬度
longitude 经度
NWS 美国国家气象局，National Weather Service

http传输格式

application/geo+json 是一种标准化的地理空间数据交换格式，本质是在 JSON（JavaScript Object Notation）基础上，通过特定的结构和规则定义地理信息，属于 IETF（互联网工程任务组）正式标准化的媒体类型（RFC 7946 规范），全称为 GeoJSON。

核心本质：JSON + 地理空间规则
普通 JSON 仅用于存储键值对、列表等通用数据
GeoJSON 在 JSON 语法框架内，强制规定地理要素（如点、线、面）的描述结构，让机器和系统能统一识别 “地理数据”，而非普通文本。

简单来说：GeoJSON = JSON 语法 + 地理空间语义
它解决了不同系统间地理数据交换的 “格式混乱” 问题（比如避免甲系统用 “lat/lng” 表示经纬度，乙系统用 “x/y” 导致无法兼容）

GeoJSON 的顶层是一个 JSON 对象，必须包含 type 字段（定义数据类型），且根据类型包含对应的地理信息字段。
最常用的类型分为两类：

基础地理要素（Feature）

单个地理实体（如一个气象站点、一条河流、一片预警区域）的完整描述，是 GeoJSON 中最常用的单元，结构固定：

{"type": "Feature",  // 固定值，标识这是一个地理要素"geometry": {       // 核心：地理形状（必选）"type": "Point",  // 形状类型（如点、线、面）"coordinates": [-77.0369, 38.9072]  // 坐标（经纬度，注意顺序：[经度, 纬度]）},"properties": {     // 附加信息（可选，存储非地理属性）"name": "华盛顿特区",  // 比如地点名称"temperature": 25,    // 比如该地点的温度"alert_level": "严重" // 比如该区域的预警等级},"id": "loc001"      // 可选：要素唯一标识
}

地理要素集合（FeatureCollection）

多个 Feature 的组合（如一个州的所有气象预警区域、多个城市的预报点），结构：

{"type": "FeatureCollection",  // 固定值，标识这是要素集合"features": [                 // 数组：包含多个 Feature 对象{"type": "Feature", "geometry": {...}, "properties": {...}},{"type": "Feature", "geometry": {...}, "properties": {...}}]
}

常见的 geometry.type（地理形状类型）

类型（Type） 描述 坐标格式示例 应用场景
Point 单点（如气象站） [经度, 纬度] 标记具体地点（如预报的经纬度位置）
LineString 线段（如河流、道路） [[经1,纬1], [经2,纬2], …] 描述线性地理实体（如台风路径）
Polygon 多边形（如区域） [[[经1,纬1], [经2,纬2], …]] 描述面状区域（如天气预警覆盖范围）
MultiPoint 多个点 [[经1,纬1], [经2,纬2], …] 标记多个分散地点（如多个观测站）