别再手动查天气了！用Python和MCP给Claude做个专属天气助手（附完整代码）
打造智能天气助手：Python与MCP的无缝集成实战
每次出门前都要打开天气应用查看预报？在不同工具间反复切换查询天气信息？这种低效操作已经成为过去式。今天我们将用Python和MCP协议为Claude打造一个专属天气助手，让你直接通过对话获取精准天气信息。这个项目不仅适合Python中级开发者练手，更能让你深入理解AI助手的扩展机制。
1. 环境准备与项目初始化
在开始编码前，我们需要搭建好开发环境。这个天气助手项目基于Python 3.10+，使用MCP 1.2.0协议实现与Claude的通信。
核心依赖工具：
- Python 3.10+（推荐使用pyenv管理多版本）
- UV包管理器（替代pip的现代工具）
- FastMCP框架（简化MCP服务器开发）
- HTTPX库（异步HTTP客户端）
# 创建项目目录并初始化虚拟环境 mkdir weather_assistant && cd weather_assistant python -m venv .venv source .venv/bin/activate # Linux/Mac # .venv\Scripts\activate # Windows # 安装核心依赖 uv install "mcp[cli]" httpx python-dotenv提示：建议使用
.env文件管理API密钥等敏感信息，避免硬编码在代码中
项目结构设计遵循模块化原则：
weather_assistant/ ├── .env # 环境变量配置 ├── config.py # 配置文件 ├── services/ # 服务模块 │ ├── weather.py # 天气数据获取 │ └── mcp_server.py # MCP服务主文件 └── tools/ # 辅助工具 └── formatters.py # 数据格式化2. MCP服务器核心架构设计
MCP（Model Context Protocol）协议的核心价值在于标准化AI模型与外部服务的交互方式。我们的天气服务器需要实现两个核心功能：
- 天气预报查询：根据经纬度获取未来5天预报
- 天气警报查询：按地区代码获取极端天气预警
# services/mcp_server.py from typing import Optional import httpx from mcp.server.fastmcp import FastMCP from config import settings from tools.formatters import format_forecast mcp = FastMCP("weather_assistant") @mcp.tool() async def get_forecast(lat: float, lon: float) -> str: """获取指定位置的天气预报""" async with httpx.AsyncClient() as client: try: response = await client.get( f"{settings.WEATHER_API}/forecast", params={"lat": lat, "lon": lon}, timeout=10.0 ) return format_forecast(response.json()) except httpx.RequestError: return "无法获取天气预报数据"关键设计考量：
- 异步IO处理避免阻塞主线程
- 完善的错误处理机制
- 数据格式化提升可读性
- 类型注解增强代码可维护性
3. 天气数据获取与处理
我们选择心知天气API作为数据源，它提供全球范围的天气数据且支持中文响应。相比美国国家气象局(NWS)API，更适合国内开发者使用。
API响应示例：
{ "results": [{ "location": {"name": "北京"}, "daily": [{ "date": "2023-07-15", "text_day": "多云", "high": "32", "low": "25", "wind_direction": "东南风" }] }] }数据处理函数需要将原始JSON转换为易读的自然语言：
# tools/formatters.py def format_forecast(data: dict) -> str: """格式化天气预报数据""" if not data.get('results'): return "未找到该位置的天气信息" location = data['results'][0]['location']['name'] daily = data['results'][0]['daily'][:5] # 取最近5天 forecast_lines = [f"{location}未来5天天气预报："] for day in daily: forecast_lines.append( f"{day['date']}: {day['text_day']}, " f"气温{day['low']}~{day['high']}℃, " f"{day['wind_direction']}" ) return "\n".join(forecast_lines)4. Claude客户端集成实战
完成服务器开发后，需要配置Claude桌面客户端识别我们的天气服务。不同操作系统的配置路径有所差异：
| 操作系统 | 配置文件路径 |
|---|---|
| macOS | ~/Library/Application Support/Claude/config.json |
| Windows | %APPDATA%\Claude\config.json |
| Linux | ~/.config/claude/config.json |
配置文件示例：
{ "mcpServers": { "weather": { "command": "python", "args": [ "-m", "uvicorn", "services.mcp_server:mcp", "--reload" ], "environment": { "WEATHER_API_KEY": "your_api_key_here" } } } }常见集成问题排查：
服务未显示：
- 检查配置文件路径是否正确
- 确认Claude版本支持MCP插件
- 查看服务启动日志是否有错误
权限问题：
chmod +x services/mcp_server.py # 确保文件有执行权限网络连接问题：
# 测试API连通性 async def test_connection(): async with httpx.AsyncClient() as client: try: await client.get("https://api.seniverse.com/v3") return True except httpx.ConnectError: return False
5. 功能扩展与优化思路
基础功能实现后，可以考虑以下增强功能：
5.1 多数据源容灾
async def get_weather_multi_sources(lat, lon): """从多个数据源获取天气，提高可靠性""" sources = [SeniverseAPI(), OpenWeatherMap(), AccuWeather()] for source in sources: try: return await source.get_forecast(lat, lon) except APIError: continue raise AllAPIsDownError()5.2 天气缓存机制
from datetime import datetime, timedelta from cachetools import TTLCache weather_cache = TTLCache(maxsize=1000, ttl=timedelta(minutes=30)) async def get_forecast_with_cache(lat, lon): cache_key = f"{lat},{lon}" if cache_key in weather_cache: return weather_cache[cache_key] data = await get_weather(lat, lon) weather_cache[cache_key] = data return data5.3 用户偏好记忆
user_preferences = { "user123": { "location": (39.9042, 116.4074), "units": "metric", "alerts_threshold": "orange" } } async def get_personalized_weather(user_id): prefs = user_preferences.get(user_id, {}) forecast = await get_forecast(*prefs.get('location', (0,0))) return adjust_units(forecast, prefs.get('units'))6. 性能监控与日志记录
生产环境部署需要完善的监控体系：
# 监控装饰器示例 def monitor_performance(func): async def wrapper(*args, **kwargs): start = time.time() try: result = await func(*args, **kwargs) duration = time.time() - start log_metric(func.__name__, "success", duration) return result except Exception as e: log_metric(func.__name__, "failure", 0) raise e return wrapper # 应用监控 @monitor_performance @mcp.tool() async def get_alerts(region: str): # 原有实现...日志配置建议：
import logging from logging.handlers import RotatingFileHandler logging.basicConfig( handlers=[ RotatingFileHandler('weather.log', maxBytes=1e6, backupCount=3), logging.StreamHandler() ], level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' )7. 安全加固措施
对外提供服务必须考虑安全性：
7.1 输入验证
from pydantic import BaseModel, confloat class Coordinates(BaseModel): lat: confloat(ge=-90, le=90) lon: confloat(ge=-180, le=180) @mcp.tool() async def get_forecast(lat: float, lon: float): coords = Coordinates(lat=lat, lon=lon) # 处理逻辑...7.2 API限流
from fastapi import Request from slowapi import Limiter from slowapi.util import get_remote_address limiter = Limiter(key_func=get_remote_address) @mcp.tool() @limiter.limit("10/minute") async def get_alerts(request: Request, region: str): # 实现...7.3 敏感数据保护
import ssl from httpx import AsyncClient async with AsyncClient( verify=ssl.CERT_REQUIRED, timeout=10.0, limits=httpx.Limits(max_connections=100) ) as client: # 安全请求...8. 实际应用场景演示
完成所有开发后，让我们看看如何在Claude中使用这个天气助手：
用户：上海明天会下雨吗？ Claude: [调用天气助手] 上海明日天气预报： 2023-07-16: 雷阵雨转多云, 气温26~32℃, 东南风3-4级 建议携带雨具出门高级查询示例：
用户：对比北京和广州未来三天的天气 Claude: [并行调用两地天气接口] 北京 vs 广州未来3天天气预报： 北京： 7/16: 晴, 28~35℃ 7/17: 多云, 27~34℃ 7/18: 雷阵雨, 26~32℃ 广州： 7/16: 暴雨, 25~29℃ 7/17: 中雨, 26~30℃ 7/18: 阵雨, 27~31℃企业级应用场景：
- 物流公司整合到运输调度系统
- 外卖平台为骑手提供实时天气预警
- 旅游APP的行程推荐引擎
- 智能家居的自动化控制条件
开发这类AI扩展服务时，最常遇到的坑是异步IO的异常处理不完善导致服务假死。我在初期版本中就遇到过因为未设置HTTP超时，导致整个服务卡在某个外部API调用上的情况。现在所有外部调用都明确设置了超时限制，并添加了熔断机制，当连续失败达到阈值时会自动切换备用数据源。