MCP实战入门：让AI模型获取实时天气信息
本文将带您了解大模型上下文协议(Model Context Protocol, 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的局限
传统方式中，开发者需要：
- 解析模型输出
- 识别API调用意图
- 手动构造API请求
- 将结果返回给模型
这种方式不仅繁琐，而且容易出错，特别是当需要处理多个API调用或复杂逻辑时。
Function Calling的进步与局限
Function Calling（如OpenAI的函数调用或Anthropic的Tool Use）是一种改进，模型可以直接输出结构化的JSON来表示函数调用意图。但它仍有局限：
- 输出格式不稳定，需要额外验证和错误处理
- 安全边界模糊，需要开发者自行实现安全措施
- 缺乏标准化，不同模型实现差异大
MCP的优势
MCP通过标准化协议解决了上述问题：
- 标准接口：提供统一的工具定义和调用方式
- 安全隔离：工具在沙箱环境中执行，减少安全风险
- 双向通信：模型和工具可以进行实时交互
- 环境感知：工具可以访问完整上下文
- 简化开发：开发者只需实现工具逻辑，协议处理由MCP框架管理
MCP天气工具实战项目
下面，我们将通过一个实际项目，展示如何使用MCP创建一个天气信息工具，让AI模型能够查询实时天气数据。
项目介绍
这是一个基于MCP的天气工具演示项目，通过和风天气API获取实时天气数据，提供以下功能：
- 天气预警查询：获取指定城市的天气灾害预警信息
- 天气预报查询：获取指定城市未来几天的天气预报
项目架构
项目分为三个主要部分：
┌─────────────┐ stdio ┌──────────────┐
│ │◄────────────►│ │
│ MCP 客户端 │ │ MCP 服务器 │
│ │ │ │
└─────────────┘ └──────────────┘
▲
│
调试 │
│
▼
┌─────────────┐
│ │
│MCP Inspector│
│ │
└─────────────┘
- MCP服务器：提供天气工具的核心实现
- MCP客户端：连接服务器，发送工具调用请求
- MCP Inspector：用于调试和测试服务器
环境准备
开始前，我们需要准备以下环境：
- Python 3.10.12 或更高版本
- NodeJS 22.14.0+ 和 NPM 10.9.2+（用于MCP Inspector）
- 和风天气API Key和API Host（注册地址[1]）【具体流程参考文末】
实战步骤
第一步：创建MCP服务器
MCP服务器是提供工具功能的核心部分。以下是实现天气服务器的核心代码：
import os
import json
import httpx
import asyncio
from dotenv import load_dotenv
from 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",
description="获取指定位置的天气灾害预警",
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",
description="获取指定位置的天气预报",
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 asyncio
import json
import os
import signal
import subprocess
import sys
from asyncio import create_subprocess_exec
from asyncio.subprocess import PIPE
from modelcontextprotocol.client import create_client, ClientConfig
from 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服务器的利器，提供可视化界面：
- 安装Inspector：
npm install -g @modelcontextprotocol/inspector
- 启动Inspector：
mcp dev server/weather_server.py
- 在浏览器访问 http://localhost:6274
Inspector界面可以让您直观地查看工具定义、测试调用并查看结果。
- 查看可用工具及其描述
查看可用工具及其描述
- 查询北京未来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
能见度: 25km
MCP的进阶应用
MCP不仅限于天气查询，还可以实现：
- 文件操作：读写文件、处理上传文件
- 数据库交互：查询和修改数据库
- 多媒体处理：处理图像、音频、视频
- 复杂工作流：多工具链式调用
MCP开发最佳实践
-
工具设计：
-
- 单一职责：每个工具只做一件事
- 明确参数：详细描述每个参数的用途
- 健壮错误处理：优雅处理各类异常情况
-
安全考虑：
-
- 输入验证：使用JSON Schema验证输入
- 权限控制：限制工具访问范围
- 资源限制：防止资源滥用
-
调试技巧：
-
- 使用MCP Inspector可视化调试
- 日志记录：添加详细日志
- 参数测试：测试边界条件和异常输入
结语
MCP为AI模型与外部系统的交互提供了标准化、安全、高效的解决方案。通过本文的天气工具实战项目，您已经掌握了MCP的基本应用。随着大模型应用的普及，MCP将在AI工具链开发中扮演越来越重要的角色。
希望这篇入门指南能帮助您开始MCP之旅，构建更强大、更安全的AI应用。欢迎在评论区分享您的想法和实践经验！
如何学习大模型 AI ？
由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。
但是具体到个人，只能说是：
“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。
这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。
我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。
我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。
第一阶段（10天）：初阶应用
该阶段让大家对大模型 AI有一个最前沿的认识，对大模型 AI 的理解超过 95% 的人，可以在相关讨论时发表高级、不跟风、又接地气的见解，别人只会和 AI 聊天，而你能调教 AI，并能用代码将大模型和业务衔接。
- 大模型 AI 能干什么？
- 大模型是怎样获得「智能」的？
- 用好 AI 的核心心法
- 大模型应用业务架构
- 大模型应用技术架构
- 代码示例：向 GPT-3.5 灌入新知识
- 提示工程的意义和核心思想
- Prompt 典型构成
- 指令调优方法论
- 思维链和思维树
- Prompt 攻击和防范
- …
第二阶段（30天）：高阶应用
该阶段我们正式进入大模型 AI 进阶实战学习，学会构造私有知识库，扩展 AI 的能力。快速开发一个完整的基于 agent 对话机器人。掌握功能最强的大模型开发框架，抓住最新的技术进展，适合 Python 和 JavaScript 程序员。
- 为什么要做 RAG
- 搭建一个简单的 ChatPDF
- 检索的基础概念
- 什么是向量表示（Embeddings）
- 向量数据库与向量检索
- 基于向量检索的 RAG
- 搭建 RAG 系统的扩展知识
- 混合检索与 RAG-Fusion 简介
- 向量模型本地部署
- …
第三阶段（30天）：模型训练
恭喜你，如果学到这里，你基本可以找到一份大模型 AI相关的工作，自己也能训练 GPT 了！通过微调，训练自己的垂直大模型，能独立训练开源多模态大模型，掌握更多技术方案。
到此为止，大概2个月的时间。你已经成为了一名“AI小子”。那么你还想往下探索吗？
- 为什么要做 RAG
- 什么是模型
- 什么是模型训练
- 求解器 & 损失函数简介
- 小实验2：手写一个简单的神经网络并训练它
- 什么是训练/预训练/微调/轻量化微调
- Transformer结构简介
- 轻量化微调
- 实验数据集的构建
- …
第四阶段（20天）：商业闭环
对全球大模型从性能、吞吐量、成本等方面有一定的认知，可以在云端和本地等多种环境下部署大模型，找到适合自己的项目/创业方向，做一名被 AI 武装的产品经理。
- 硬件选型
- 带你了解全球大模型
- 使用国产大模型服务
- 搭建 OpenAI 代理
- 热身：基于阿里云 PAI 部署 Stable Diffusion
- 在本地计算机运行大模型
- 大模型的私有化部署
- 基于 vLLM 部署大模型
- 案例：如何优雅地在阿里云私有部署开源大模型
- 部署一套开源 LLM 项目
- 内容安全
- 互联网信息服务算法备案
- …
学习是一个过程，只要学习就会有挑战。天道酬勤，你越努力，就会成为越优秀的自己。
如果你能在15天内完成所有的任务，那你堪称天才。然而，如果你能完成 60-70% 的内容，你就已经开始具备成为一名大模型 AI 的正确特征了。
这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】
欢迎加入 MCP 技术社区！与志同道合者携手前行，一同解锁 MCP 技术的无限可能！
更多推荐
22
0- 0
已为社区贡献74条内容
所有评论(0)