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 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",
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 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服務器的利器，提供可視化界面：
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百寶箱
