MCP实战基础：如何利用AI模型实时获取和分析天气数据
本文将带您了解大模型上下文协议(Model Context Protocol, MCP)，并通过一个获取实时天气信息的实战项目，手把手教您如何实现AI模型与外部工具的无缝交互。
什么是Model Context Protocol (MCP)?
Model Context Protocol (MCP) 是一种开放协议，专为大语言模型（如 Claude、ChatGPT等）设计，允许这些模型与外部系统安全地交互。简单来说，MCP 提供了一种标准化的方式，让AI模型能够：
• 调用外部工具和API
• 访问实时数据和信息
• 获取环境上下文
• 执行复杂操作
通过MCP，AI模型不再局限于训练数据，而是可以获取实时信息、控制外部系统，并执行各种实用任务。
为什么需要MCP？对比传统方法
在AI应用开发中，有几种主流方式让模型与外部世界交互：
特性 | 传统REST API | Function Calling | MCP |
实现方式 | 直接调用HTTP接口 | 模型输出JSON格式工具调用 | 标准化协议 |
安全性 | 中等（需手动处理） | 中等（解析不稳定） | 高（沙箱隔离） |
集成难度 | 高（需自行实现） | 中等（需处理解析错误） | 低（标准接口） |
交互方式 | 异步、单向 | 半同步 | 同步、双向 |
上下文感知 | 无 | 有限 | 完整 |
适用场景 | 简单集成 | 单次调用 | 复杂工具链 |
传统REST API的局限
传统方式中，开发者需要：
1. 解析模型输出
2. 识别API调用意图
3. 手动构造API请求
4. 将结果返回给模型
这种方式不仅繁琐，而且容易出错，特别是当需要处理多个API调用或复杂逻辑时。
Function Calling的进步与局限
Function Calling（如OpenAI的函数调用或Anthropic的Tool Use）是一种改进，模型可以直接输出结构化的JSON来表示函数调用意图。但它仍有局限：
1. 输出格式不稳定，需要额外验证和错误处理
2. 安全边界模糊，需要开发者自行实现安全措施
3. 缺乏标准化，不同模型实现差异大
MCP的优势
MCP通过标准化协议解决了上述问题：
1.标准接口：提供统一的工具定义和调用方式
2.安全隔离：工具在沙箱环境中执行，减少安全风险
3.双向通信：模型和工具可以进行实时交互
4.环境感知：工具可以访问完整上下文
5.简化开发：开发者只需实现工具逻辑，协议处理由MCP框架管理
MCP天气工具实战项目
下面，我们将通过一个实际项目，展示如何使用MCP创建一个天气信息工具，让AI模型能够查询实时天气数据。
项目介绍
这是一个基于MCP的天气工具演示项目，通过和风天气API获取实时天气数据，提供以下功能：
1.天气预警查询：获取指定城市的天气灾害预警信息
2.天气预报查询：获取指定城市未来几天的天气预报
项目架构
项目分为三个主要部分：
┌─────────────┐ stdio ┌──────────────┐│ │◄────────────►│ ││ MCP 客户端 │ │ MCP 服务器 ││ │ │ │└─────────────┘ └──────────────┘ ▲ │ 调试 │ │ ▼ ┌─────────────┐ │ │ │MCP Inspector│ │ │ └─────────────┘1.MCP服务器：提供天气工具的核心实现
2.MCP客户端：连接服务器，发送工具调用请求
3.MCP Inspector：用于调试和测试服务器
环境准备
开始前，我们需要准备以下环境：
• Python 3.10.12 或更高版本
• NodeJS 22.14.0+ 和 NPM 10.9.2+（用于MCP Inspector）
• 和风天气API Key和API Host（注册地址[1]）【具体流程参考文末】
实战步骤
第一步：创建MCP服务器
MCP服务器是提供工具功能的核心部分。以下是实现天气服务器的核心代码：
import osimport jsonimport httpximport asynciofrom dotenv import load_dotenvfrom modelcontextprotocol.server import ( create_server, ServerConfig, tools, JsonSchema,)# 加载环境变量load_dotenv()API_KEY = os.getenv("QWEATHER_API_KEY")API_HOST = os.getenv("QWEATHER_API_HOST", "https://XXX.qweather.com")# 定义天气预警工具@tools.tool( name="get_weather_warning", descriptinotallow="获取指定位置的天气灾害预警", parameters=JsonSchema( { "type": "object", "properties": { "location": { "type": "string", "description": "城市ID或经纬度坐标（经度,纬度）\n例如：101010100（北京）或 116.41,39.92\", }, }, "required": ["location"], } ),)asyncdefget_weather_warning(location: str) -> str: """ 获取指定位置的天气灾害预警 参数: location: 城市ID或经纬度坐标（经度,纬度） 例如：101010100（北京）或 116.41,39.92 返回: 格式化的预警信息字符串 """ asyncwith httpx.AsyncClient() as client: response = await client.get( f"{API_HOST}/v7/warning/now", params={ "location": location, "key": API_KEY, "lang": "zh", }, ) data = response.json() if data["code"] != "200": returnf"获取天气预警失败: {data[code]}" warnings = data.get("warning", []) ifnot warnings: return"当前没有天气预警信息" result = [] for warning in warnings: result.append( f"预警ID: {warning[id]}\n" f"标题: {warning[title]}\n" f"发布时间: {warning[pubTime]}\n" f"开始时间: {warning[startTime]}\n" f"结束时间: {warning[endTime]}\n" f"预警类型: {warning[typeName]}\n" f"预警等级: {warning[severityName]} ({warning[level]})\n" f"发布单位: {warning[sender]}\n" f"状态: {warning[status]}\n" f"详细信息: {warning[text]}" ) return"\n\n".join(result)# 定义天气预报工具@tools.tool( name="get_daily_forecast", descriptinotallow="获取指定位置的天气预报", parameters=JsonSchema( { "type": "object", "properties": { "location": { "type": "string", "description": "城市ID或经纬度坐标（经度,纬度）\n例如：101010100（北京）或 116.41,39.92\", }, "days": { "type": "integer", "description": "预报天数，可选值为 3、7、10、15、30，默认为 3", "enum": [3, 7, 10, 15, 30], "default": 3, }, }, "required": ["location"], } ),)asyncdefget_daily_forecast(location: str, days: int = 3) -> str: """ 获取指定位置的天气预报 参数: location: 城市ID或经纬度坐标（经度,纬度） 例如：101010100（北京）或 116.41,39.92 days: 预报天数，可选值为 3、7、10、15、30，默认为 3 返回: 格式化的天气预报字符串 """ # 根据天数选择API版本 version = "3d"if days == 3else"7d"if days == 7else"10d"if days in [10, 15, 30] else"3d" asyncwith httpx.AsyncClient() as client: response = await client.get( f"{API_HOST}/v7/weather/{version}", params={ "location": location, "key": API_KEY, "lang": "zh", }, ) data = response.json() if data["code"] != "200": returnf"获取天气预报失败: {data[code]}" daily = data.get("daily", []) ifnot daily: return"无法获取天气预报信息" result = [] for day in daily[:days]: # 限制天数 result.append( f"日期: {day[fxDate]}\n" f"日出: {day[sunrise]} 日落: {day[sunset]}\n" f"最高温度: {day[tempMax]}°C 最低温度: {day[tempMin]}°C\n" f"白天天气: {day[textDay]} 夜间天气: {day[textNight]}\n" f"白天风向: {day[windDirDay]} {day[windScaleDay]}级 ({day[windSpeedDay]}km/h)\n" f"夜间风向: {day[windDirNight]} {day[windScaleNight]}级 ({day[windSpeedNight]}km/h)\n" f"相对湿度: {day[humidity]}%\n" f"降水量: {day[precip]}mm\n" f"紫外线指数: {day[uvIndex]}\n" f"能见度: {day[vis]}km" ) return"\n\n---\n\n".join(result)# 主函数asyncdefmain(): config = ServerConfig() server = create_server(config) # 注册工具 server.register_tool(get_weather_warning) server.register_tool(get_daily_forecast) # 启动服务器 await server.serve()if __name__ == "__main__": asyncio.run(main())第二步：实现MCP客户端
MCP客户端用于连接服务器并调用工具。以下是客户端的实现：
import asyncioimport jsonimport osimport signalimport subprocessimport sysfrom asyncio import create_subprocess_execfrom asyncio.subprocess import PIPEfrom modelcontextprotocol.client import create_client, ClientConfigfrom modelcontextprotocol.protocol.tool_schemas import ToolSchema# 启动服务器进程asyncdefstart_server_process(): server_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "server", "weather_server.py") returnawait create_subprocess_exec( sys.executable, server_path, stdin=PIPE, stdout=PIPE, stderr=PIPE )# 主函数asyncdefmain(): print("启动 MCP 服务器进程...") server_process = await start_server_process() # 配置客户端 config = ClientConfig() client = create_client(config) try: # 连接到服务器 await client.connect_process(server_process) # 获取可用工具 tools = await client.get_tools() print(f"已连接到服务器，可用工具: {len(tools)}") # 显示工具信息 for tool in tools: print(f" - {tool.name}: \n{tool.description}\n") print("使用 help 查看帮助，使用 exit 退出\n") # 命令行交互循环 whileTrue: user_input = input("> ").strip() if user_input.lower() == "exit": break elif user_input.lower() == "help": print("\n可用命令:") print(" help - 显示此帮助信息") print(" list - 列出可用工具") print(" call <工具名> <参数JSON> - 调用工具") print(" exit - 退出程序") print("\n示例:") print( call get_weather_warning {"location": "101010100"}) print(" call get_daily_forecast 116.41,39.92") print(" call get_daily_forecast 101010100 7") print() elif user_input.lower() == "list": for tool in tools: print(f" - {tool.name}: {tool.description[:50]}...") print() elif user_input.lower().startswith("call "): # 解析命令 parts = user_input[5:].strip().split(" ", 1) iflen(parts) < 1: print("错误: 需要指定工具名称") continue tool_name = parts[0] # 查找工具 tool = next((t for t in tools if t.name == tool_name), None) ifnot tool: print(f"错误: 找不到工具 {tool_name}\") continue # 解析参数 args = {} iflen(parts) > 1: arg_text = parts[1].strip() # 简单参数处理 ifnot arg_text.startswith("{"): # 简单模式: call get_daily_forecast 101010100 7 simple_args = arg_text.split(" ") # 检查是否为天气预报工具 if tool_name == "get_daily_forecast": iflen(simple_args) >= 1: args["location"] = simple_args[0] iflen(simple_args) >= 2: try: args["days"] = int(simple_args[1]) except ValueError: print("错误: days 参数必须是整数") continue elif tool_name == "get_weather_warning": iflen(simple_args) >= 1: args["location"] = simple_args[0] else: print("错误: 简单参数模式仅支持预定义工具") continue else: # JSON模式: call get_weather_warning {"location": "101010100"} try: args = json.loads(arg_text) except json.JSONDecodeError: print("错误: 无效的JSON参数") continue print("正在调用工具...\n") try: # 调用工具 result = await client.call_tool(tool_name, args) print("结果:") print(result) print() except Exception as e: print(f"错误: {str(e)}") print() else: print("未知命令，使用 help 查看帮助") print() finally: # 关闭连接和进程 await client.close() if server_process.returncode isNone: server_process.terminate() try: await asyncio.wait_for(server_process.wait(), timeout=5.0) except asyncio.TimeoutError: server_process.kill()if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: print("\n已退出")第三步：使用MCP Inspector调试
首先在
.env文件配置QWEATHER_API_KEY和 QWEATHER_API_KEY
MCP Inspector是调试MCP服务器的利器，提供可视化界面：
1. 安装Inspector：
npm install -g @modelcontextprotocol/inspector2. 启动Inspector：
mcp dev server/weather_server.py3. 在浏览器访问 http://localhost:6274
Inspector界面可以让您直观地查看工具定义、测试调用并查看结果。
查看可用工具及其描述
查看可用工具及其描述
查询北京未来3天天气
查询北京未来3天天气
查询北京灾害预警
查询北京未来3天天气
查询北京灾害预警
查询北京灾害预警
提示：MCP Inspector 提供了更直观的界面来测试和调试 MCP 服务器，特别适合开发和调试复杂工具。
实际效果展示
首先在
.env文件配置QWEATHER_API_KEY和 QWEATHER_API_KEY
启动程序python client/mcp_client.py
天气预警查询
> call get_weather_warning {"location": "101010100"}正在调用工具...结果:预警ID: 10123020120230713145500551323468标题: 杭州市气象台发布高温黄色预警[III级/较重]发布时间: 2023-07-13T14:55+08:00开始时间: 2023-07-13T14:55+08:00结束时间: 2023-07-14T14:55+08:00预警类型: 高温预警等级: Moderate (Yellow)发布单位: 杭州市气象台状态: active详细信息: 杭州市气象台2023年07月13日14时55分发布高温黄色预警信号：预计未来24小时内最高气温将达到37℃以上，请注意防暑降温。天气预报查询
> call get_daily_forecast 101010100 3正在调用工具...结果:日期: 2023-07-13日出: 04:54 日落: 19:44最高温度: 32°C 最低温度: 22°C白天天气: 多云 夜间天气: 阴白天风向: 东南风 3级 (19km/h)夜间风向: 东南风 3级 (16km/h)相对湿度: 75%降水量: 0mm紫外线指数: 7能见度: 25km---日期: 2023-07-14日出: 04:55 日落: 19:43最高温度: 33°C 最低温度: 23°C白天天气: 多云 夜间天气: 阴白天风向: 东南风 3级 (21km/h)夜间风向: 东风 3级 (15km/h)相对湿度: 72%降水量: 0mm紫外线指数: 8能见度: 25km---日期: 2023-07-15日出: 04:56 日落: 19:43最高温度: 34°C 最低温度: 23°C白天天气: 多云 夜间天气: 多云白天风向: 东南风 3级 (18km/h)夜间风向: 东风 3级 (14km/h)相对湿度: 70%降水量: 0mm紫外线指数: 9能见度: 25kmMCP的进阶应用
MCP不仅限于天气查询，还可以实现：
1.文件操作：读写文件、处理上传文件
2.数据库交互：查询和修改数据库
3.多媒体处理：处理图像、音频、视频
4.复杂工作流：多工具链式调用
MCP开发最佳实践
1.工具设计：
• 单一职责：每个工具只做一件事
• 明确参数：详细描述每个参数的用途
• 健壮错误处理：优雅处理各类异常情况
2.安全考虑：
• 输入验证：使用JSON Schema验证输入
• 权限控制：限制工具访问范围
• 资源限制：防止资源滥用
3.调试技巧：
• 使用MCP Inspector可视化调试
• 日志记录：添加详细日志
• 参数测试：测试边界条件和异常输入
结语
MCP为AI模型与外部系统的交互提供了标准化、安全、高效的解决方案。通过本文的天气工具实战项目，您已经掌握了MCP的基本应用。随着大模型应用的普及，MCP将在AI工具链开发中扮演越来越重要的角色。
希望这篇入门指南能帮助您开始MCP之旅，构建更强大、更安全的AI应用。欢迎在评论区分享您的想法和实践经验！
附录
和风天气 API 注册与使用
要使用本项目，需要先注册和风天气开发者账号并获取 API Key：
1.注册和风天气开发者账号：
• 访问和风天气开发服务[2]
• 点击"注册"，按照提示完成账号注册
2.创建项目并获取 API Key：
• 登录开发者控制台
• 点击"项目管理" -> "创建项目"
• 填写项目名称、创建凭据
• 创建成功后，在项目详情页可以获取 API Key
和风天气API Key
3.开发者的API Host：
• 登录开发者控制台
• 点击"头像" -> "设置"，或直接访问https://console.qweather.com/setting?lang=zh
• 查看API Host
和风天气API Host
4.API 使用说明：
• 免费版API有调用次数限制，详情请参考和风天气定价页面[3]
• 支持通过城市ID或经纬度坐标查询天气信息
• 城市ID可通过和风天气城市查询API[4]获取
参考资源
• MCP官方文档：https://modelcontextprotocol.io/
• MCP快速入门：https://modelcontextprotocol.io/quickstart/server
• 项目源码：https://github.com/FlyAIBox/mcp-in-action/tree/qweather_0.1/mcp_demo
• 和风天气API：https://dev.qweather.com/
引用链接
[1] 注册地址: https://dev.qweather.com/
[2] 和风天气开发服务: https://dev.qweather.com/
[3] 和风天气定价页面: https://dev.qweather.com/docs/pricing/
[4] 和风天气城市查询API: https://dev.qweather.com/docs/api/geoapi/
本文转载自 萤火AI百宝箱，作者： 萤火AI百宝箱
