打造智能天气助手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服务器核心架构设计MCPModel 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}, timeout10.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.jsonWindows%APPDATA%\Claude\config.jsonLinux~/.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 False5. 功能扩展与优化思路基础功能实现后可以考虑以下增强功能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(maxsize1000, ttltimedelta(minutes30)) 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, maxBytes1e6, backupCount3), logging.StreamHandler() ], levellogging.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, le90) lon: confloat(ge-180, le180) mcp.tool() async def get_forecast(lat: float, lon: float): coords Coordinates(latlat, lonlon) # 处理逻辑...7.2 API限流from fastapi import Request 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_alerts(request: Request, region: str): # 实现...7.3 敏感数据保护import ssl from httpx import AsyncClient async with AsyncClient( verifyssl.CERT_REQUIRED, timeout10.0, limitshttpx.Limits(max_connections100) ) 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调用上的情况。现在所有外部调用都明确设置了超时限制并添加了熔断机制当连续失败达到阈值时会自动切换备用数据源。