别再手动查天气了！用Python和MCP给Claude做个专属天气插件（附完整代码）

用Python和MCP为Claude打造智能天气查询系统1. 为什么需要为AI助手开发天气插件每天早上起床第一件事是什么对很多人来说查看天气预报已经成了固定流程。但手动打开天气应用或网页查询实在不够优雅——尤其是当你已经习惯用AI助手处理各种事务时。想象这样的场景你正在规划周末的户外活动直接问Claude这周末纽约的天气怎么样它不仅能告诉你温度、降水概率还能主动提醒周六下午有雷暴警报建议调整出行时间。这种无缝体验才是智能助手应有的样子。传统AI助手的局限在于它们通常是封闭系统无法实时获取外部数据。而MCP(Model Context Protocol)就像给AI装上了万能接口让它能自由调用各种工具和服务。通过开发一个MCP天气服务器我们实际上是在扩展Claude的感官能力。2. MCP技术架构解析2.1 MCP的核心设计理念MCP的巧妙之处在于它采用了客户端-服务器的松耦合架构[Claude客户端] ←MCP协议→ [天气服务器] ←HTTP→ [国家气象局API]这种设计带来三个关键优势数据隔离敏感API密钥和原始数据保留在服务器端协议统一不同工具都通过相同方式与AI交互动态扩展新增功能只需部署新服务器无需修改客户端2.2 天气服务器的功能规划我们的服务器将提供两个核心工具工具名称输入参数输出内容数据源get_forecast纬度、经度5天天气预报(温度、风速等)国家气象局点预报APIget_alerts州代码(如CA/NY)当前天气警报(暴雨、飓风等)国家气象局警报API3. 开发环境准备3.1 基础工具安装确保你的系统已准备好以下组件# 检查Python版本(需要3.10) python --version # 安装uv工具包管理器和虚拟环境 curl -LsSf https://astral.sh/uv/install.sh | sh # 创建项目目录 mkdir weather-plugin cd weather-plugin uv venv source .venv/bin/activate3.2 依赖库安装项目需要以下Python包uv add mcp[cli] httpx关键依赖说明mcp[cli]: MCP协议的Python实现包含服务器框架httpx: 异步HTTP客户端用于调用天气APIpython-dotenv: 管理环境变量(可选用于存储API密钥)4. 核心代码实现4.1 服务器基础框架创建weather.py文件并初始化MCP服务器from typing import Any import httpx from mcp.server.fastmcp import FastMCP # 初始化天气服务器实例 mcp FastMCP(weather) # 国家气象局API配置 NWS_API_BASE https://api.weather.gov USER_AGENT claude-weather-plugin/1.0 (youremail.com)4.2 API请求封装实现带错误处理的通用请求函数async def fetch_weather_data(url: str) - dict[str, Any] | None: headers { User-Agent: USER_AGENT, Accept: application/geojson } try: async with httpx.AsyncClient() as client: response await client.get(url, headersheaders, timeout15.0) response.raise_for_status() return response.json() except httpx.HTTPStatusError as e: print(fAPI请求失败: {e.response.status_code}) except Exception as e: print(f网络错误: {str(e)}) return None4.3 天气预报工具实现mcp.tool() async def get_forecast(lat: float, lng: float) - str: 获取指定位置的5天天气预报 参数: lat: 纬度(如37.7749) lng: 经度(如-122.4194) # 获取点位数据 points_url f{NWS_API_BASE}/points/{lat},{lng} points_data await fetch_weather_data(points_url) if not points_data: return 无法获取该位置的气象数据 # 获取详细预报 forecast_url points_data[properties][forecast] forecast_data await fetch_weather_data(forecast_url) if not forecast_data: return 无法获取详细天气预报 # 格式化输出 forecasts [] for period in forecast_data[properties][periods][:10]: # 获取5天数据(每天早晚两个时段) forecasts.append( f{period[name]}:\n f 温度: {period[temperature]}°{period[temperatureUnit]}\n f 风速: {period[windSpeed]} {period[windDirection]}\n f 预报: {period[shortForecast]} ) return 未来5天天气预报:\n\n \n\n.join(forecasts)4.4 天气警报工具实现mcp.tool() async def get_alerts(state: str) - str: 获取指定州的活跃天气警报 参数: state: 两位州代码(如CA/NY) alerts_url f{NWS_API_BASE}/alerts/active/area/{state} alerts_data await fetch_weather_data(alerts_url) if not alerts_data or not alerts_data.get(features): return f{state}州当前没有活跃天气警报 alert_messages [] for alert in alerts_data[features]: props alert[properties] alert_messages.append( f⚠️ {props[event]}警报\n f地区: {props[areaDesc]}\n f严重程度: {props[severity]}\n f有效期: {props[effective]} 至 {props[expires]}\n f详情: {props[description]} ) return f{state}州当前天气警报:\n\n \n\n----------\n.join(alert_messages)4.5 服务器启动配置if __name__ __main__: # 启动MCP服务器 mcp.run( transportstdio, log_levelinfo )5. 部署与集成5.1 本地测试运行uv run weather.py测试工具是否正常工作# 在另一个终端测试 from mcp.client import Client async with Client(transportstdio, command[python, weather.py]) as client: print(await client.tools[get_forecast](lat37.7749, lng-122.4194)) # 旧金山天气预报5.2 Claude桌面端配置创建或修改Claude配置文件{ mcpServers: { weather: { command: python, args: [/path/to/weather.py], environment: { PYTHONPATH: /path/to/project } } } }配置完成后重启Claude你应该能在工具面板看到新增的天气查询功能。6. 高级功能扩展6.1 添加位置缓存为避免频繁查询相同位置可以添加本地缓存from datetime import datetime, timedelta from functools import lru_cache lru_cache(maxsize100) async def cached_forecast(lat: float, lng: float) - str: # 缓存有效期为30分钟 return await get_forecast(lat, lng)6.2 多语言支持扩展服务器支持多语言响应def translate_weather(forecast: str, lang: str en) - str: # 实现简单的翻译逻辑或调用翻译API pass mcp.tool() async def get_forecast_i18n(lat: float, lng: float, lang: str en) - str: forecast await get_forecast(lat, lng) return translate_weather(forecast, lang)6.3 预警自动推送通过Webhook实现主动预警import asyncio from fastapi import FastAPI app FastAPI() app.on_event(startup) async def startup_event(): asyncio.create_task(alert_monitor()) async def alert_monitor(): while True: alerts await check_all_states_alerts() if alerts: send_webhook_notification(alerts) await asyncio.sleep(300) # 每5分钟检查一次7. 性能优化与监控7.1 异步批处理优化多个位置的并发查询async def batch_forecast(locations: list[tuple[float, float]]) - dict: async with httpx.AsyncClient() as client: tasks [get_forecast(lat, lng) for lat, lng in locations] return await asyncio.gather(*tasks, return_exceptionsTrue)7.2 Prometheus监控添加性能指标收集from prometheus_client import start_http_server, Counter REQUEST_COUNTER Counter(weather_requests, API请求统计, [endpoint]) mcp.tool() async def get_forecast(lat: float, lng: float) - str: REQUEST_COUNTER.labels(endpointforecast).inc() # ...原有逻辑...启动监控服务器start_http_server(8000) # 指标暴露在8000端口8. 安全加固措施8.1 输入验证防止非法参数注入from pydantic import confloat mcp.tool() async def safe_forecast( lat: confloat(ge-90, le90), lng: confloat(ge-180, le180) ) - str: # 确保经纬度在合法范围内 return await get_forecast(lat, lng)8.2 速率限制防止API滥用from slowapi import Limiter from slowapi.util import get_remote_address limiter Limiter(key_funcget_remote_address) mcp.tool() limiter.limit(10/minute) async def get_forecast(lat: float, lng: float) - str: # ...原有逻辑...8.3 敏感数据过滤清理API响应中的敏感信息def sanitize_alert(alert: dict) - dict: return { k: v for k, v in alert.items() if k in [event, areaDesc, severity] }