1. 项目概述
最近在开发一个需要集成天气查询功能的MCP项目时,遇到了一个有趣的问题:如何让MCP模型能够方便地调用外部天气数据?经过一番探索,我决定用Python搭建一个轻量级的天气查询服务器。这个方案不仅解决了我的实际问题,还让我对Python的标准库http.server有了更深入的理解。
这个天气查询服务器的核心功能很简单:提供一个HTTP接口,当MCP模型需要查询深圳天气时,只需向这个服务器发送请求,就能获取结构化的天气数据。我选择了OpenWeatherMap作为数据源,因为它提供免费的API调用额度,数据质量也不错。整个实现只用了不到100行Python代码,却解决了模型与外部服务交互的大问题。
2. 核心设计与实现思路
2.1 为什么选择Python标准库
在技术选型时,我考虑过Flask、FastAPI等流行的Web框架,但最终决定使用Python自带的http.server模块。这个选择基于几个实际考量:
- 依赖最小化:项目环境越简单越好,标准库意味着零额外依赖
- 功能足够:我们只需要处理简单的GET请求,http.server完全够用
- 学习成本低:团队成员都熟悉标准库,无需额外学习框架API
提示:对于生产环境的高并发场景,建议还是使用专业Web框架。但在我们的MCP集成场景下,轻量级方案更合适。
2.2 接口设计原则
为了让MCP模型方便调用,我遵循了几个关键设计原则:
- RESTful风格:使用标准的HTTP GET方法和资源路径
- 单一职责:每个接口只做一件事(这里只提供深圳天气查询)
- 结构化响应:统一返回JSON格式,包含完整的错误处理
接口路径设计为/weather/shenzhen,这样既清晰表达了功能,又保留了扩展性(未来可以添加其他城市)。
3. 详细实现步骤
3.1 环境准备与API配置
首先需要注册OpenWeatherMap账号并获取API Key:
- 访问 OpenWeatherMap官网 注册账号
- 在个人面板中找到API Keys选项卡
- 创建新的API Key(免费版足够使用)
- 将获得的Key替换代码中的
YOUR_API_KEY
注意:免费API有调用频率限制(每分钟60次),生产环境需要考虑购买付费套餐或添加缓存机制。
3.2 核心代码解析
让我们深入看看关键部分的实现逻辑:
class WeatherHandler(http.server.BaseHTTPRequestHandler): def do_GET(self): if self.path == "/weather/shenzhen": try: # 构建API请求参数 params = { "q": "Shenzhen,CN", "appid": API_KEY, "units": "metric", # 使用摄氏度 "lang": "zh_cn" # 中文描述 } url = WEATHER_API_URL + "?" + urllib.parse.urlencode(params) # 发送请求并处理响应 with urllib.request.urlopen(url) as response: data = json.loads(response.read().decode()) # 提取关键天气数据 weather_info = { "city": data.get("name"), "temperature": data.get("main", {}).get("temp"), "humidity": data.get("main", {}).get("humidity"), "weather": data.get("weather", [{}])[0].get("description"), "wind_speed": data.get("wind", {}).get("speed"), "pressure": data.get("main", {}).get("pressure"), "timestamp": data.get("dt") } # 返回成功响应 self.send_response(200) self.send_header("Content-type", "application/json") self.end_headers() self.wfile.write(json.dumps(weather_info, ensure_ascii=False).encode("utf-8"))这段代码有几个值得注意的技术点:
- 参数编码:使用
urllib.parse.urlencode正确处理URL参数 - 防御性编程:通过
.get()方法避免KeyError异常 - 字符编码:
ensure_ascii=False确保中文正常显示 - 错误处理:完整的try-except块捕获各种异常情况
3.3 服务器启动与测试
启动服务器非常简单:
def run_server(port=8000): handler = WeatherHandler with socketserver.TCPServer(("", port), handler) as httpd: print(f"天气查询服务器运行在 http://localhost:{port}") print(f"查询深圳天气: http://localhost:{port}/weather/shenzhen") httpd.serve_forever() if __name__ == "__main__": run_server()启动后,可以通过浏览器或curl测试接口:
curl http://localhost:8000/weather/shenzhen正常响应示例如下:
{ "city": "Shenzhen", "temperature": 28.5, "humidity": 78, "weather": "多云", "wind_speed": 3.6, "pressure": 1012, "timestamp": 1689153600 }4. 关键技术与优化建议
4.1 性能优化技巧
虽然这个实现已经很简洁,但在实际使用中我发现了几处可以优化的地方:
- 连接复用:频繁创建HTTP连接开销大,可以考虑使用
requests.Session() - 响应缓存:天气数据不需要实时更新,可以添加内存缓存(如30秒)
- 异步处理:使用async/await避免阻塞主线程
一个简单的缓存实现示例:
from time import time class CachedWeather: def __init__(self): self.cache = {} self.cache_time = 30 # 30秒缓存 def get_weather(self, params): cache_key = str(params) now = time() if cache_key in self.cache: data, timestamp = self.cache[cache_key] if now - timestamp < self.cache_time: return data # 实际API调用 url = WEATHER_API_URL + "?" + urllib.parse.urlencode(params) with urllib.request.urlopen(url) as response: data = json.loads(response.read().decode()) self.cache[cache_key] = (data, now) return data4.2 安全增强措施
在生产环境中使用时,还需要考虑以下安全因素:
- API密钥保护:不要硬编码在代码中,使用环境变量或配置文件
- 输入验证:虽然当前只有固定路径,但应该添加路径白名单
- HTTPS支持:考虑使用http.server.HTTPServer基础类
- 速率限制:防止API被滥用
5. 常见问题与解决方案
在实际部署过程中,我遇到了几个典型问题,这里分享解决方法:
5.1 API调用失败排查
问题现象:服务器返回500错误,日志显示"URLError: <urlopen error [SSL: CERTIFICATE_VERIFY_FAILED]"
原因分析:本地Python环境缺少SSL根证书
解决方案:
- 安装certifi包:
pip install certifi - 修改请求代码:
import ssl ssl._create_default_https_context = ssl._create_unverified_context5.2 中文乱码问题
问题现象:返回的天气描述显示为Unicode编码
原因分析:JSON序列化时未正确处理非ASCII字符
解决方案:
json.dumps(weather_info, ensure_ascii=False).encode("utf-8")5.3 性能瓶颈
问题现象:连续请求时响应变慢
原因分析:每次请求都新建连接,没有复用
优化方案:
- 使用requests库替代urllib
- 实现连接池或缓存机制
6. 扩展思路与应用场景
这个基础实现可以进一步扩展为更强大的工具:
6.1 多城市支持
通过路径参数动态指定城市:
# 匹配 /weather/<city> 形式的路径 match = re.match(r"/weather/(\w+)", self.path) if match: city = match.group(1) # 使用city变量构建API请求6.2 历史数据查询
扩展接口支持查询历史天气:
/weather/shenzhen/history?days=76.3 与MCP深度集成
将天气服务注册为MCP工具:
mcp.register_tool( name="weather_query", description="查询指定城市天气", parameters={ "city": {"type": "string", "description": "城市名称"} }, execute=query_weather )这个天气查询服务器虽然简单,但体现了几个重要的设计理念:单一职责、接口标准化、轻量级实现。在实际项目中,这种"小而美"的服务往往比大而全的系统更实用。