MCP實(shí)戰(zhàn)入門(mén)：讓AI模型獲取實(shí)時(shí)天氣信息
本文將帶您了解大模型上下文協(xié)議(Model Context Protocol, MCP)，并通過(guò)一個(gè)獲取實(shí)時(shí)天氣信息的實(shí)戰(zhàn)項(xiàng)目，手把手教您如何實(shí)現(xiàn)AI模型與外部工具的無(wú)縫交互。
什么是Model Context Protocol (MCP)?
Model Context Protocol (MCP) 是一種開(kāi)放協(xié)議，專(zhuān)為大語(yǔ)言模型（如 Claude、ChatGPT等）設(shè)計(jì)，允許這些模型與外部系統(tǒng)安全地交互。簡(jiǎn)單來(lái)說(shuō)，MCP 提供了一種標(biāo)準(zhǔn)化的方式，讓AI模型能夠：
? 調(diào)用外部工具和API
? 訪(fǎng)問(wèn)實(shí)時(shí)數(shù)據(jù)和信息
? 獲取環(huán)境上下文
? 執(zhí)行復(fù)雜操作
通過(guò)MCP，AI模型不再局限于訓(xùn)練數(shù)據(jù)，而是可以獲取實(shí)時(shí)信息、控制外部系統(tǒng)，并執(zhí)行各種實(shí)用任務(wù)。
為什么需要MCP？對(duì)比傳統(tǒng)方法
在A(yíng)I應(yīng)用開(kāi)發(fā)中，有幾種主流方式讓模型與外部世界交互：
特性 | 傳統(tǒng)REST API | Function Calling | MCP |
實(shí)現(xiàn)方式 | 直接調(diào)用HTTP接口 | 模型輸出JSON格式工具調(diào)用 | 標(biāo)準(zhǔn)化協(xié)議 |
安全性 | 中等（需手動(dòng)處理） | 中等（解析不穩(wěn)定） | 高（沙箱隔離） |
集成難度 | 高（需自行實(shí)現(xiàn)） | 中等（需處理解析錯(cuò)誤） | 低（標(biāo)準(zhǔn)接口） |
交互方式 | 異步、單向 | 半同步 | 同步、雙向 |
上下文感知 | 無(wú) | 有限 | 完整 |
適用場(chǎng)景 | 簡(jiǎn)單集成 | 單次調(diào)用 | 復(fù)雜工具鏈 |
傳統(tǒng)REST API的局限
傳統(tǒng)方式中，開(kāi)發(fā)者需要：
1. 解析模型輸出
2. 識(shí)別API調(diào)用意圖
3. 手動(dòng)構(gòu)造API請(qǐng)求
4. 將結(jié)果返回給模型
這種方式不僅繁瑣，而且容易出錯(cuò)，特別是當(dāng)需要處理多個(gè)API調(diào)用或復(fù)雜邏輯時(shí)。
Function Calling的進(jìn)步與局限
Function Calling（如OpenAI的函數(shù)調(diào)用或Anthropic的Tool Use）是一種改進(jìn)，模型可以直接輸出結(jié)構(gòu)化的JSON來(lái)表示函數(shù)調(diào)用意圖。但它仍有局限：
1. 輸出格式不穩(wěn)定，需要額外驗(yàn)證和錯(cuò)誤處理
2. 安全邊界模糊，需要開(kāi)發(fā)者自行實(shí)現(xiàn)安全措施
3. 缺乏標(biāo)準(zhǔn)化，不同模型實(shí)現(xiàn)差異大
MCP的優(yōu)勢(shì)
MCP通過(guò)標(biāo)準(zhǔn)化協(xié)議解決了上述問(wèn)題：
1.標(biāo)準(zhǔn)接口：提供統(tǒng)一的工具定義和調(diào)用方式
2.安全隔離：工具在沙箱環(huán)境中執(zhí)行，減少安全風(fēng)險(xiǎn)
3.雙向通信：模型和工具可以進(jìn)行實(shí)時(shí)交互
4.環(huán)境感知：工具可以訪(fǎng)問(wèn)完整上下文
5.簡(jiǎn)化開(kāi)發(fā)：開(kāi)發(fā)者只需實(shí)現(xiàn)工具邏輯，協(xié)議處理由MCP框架管理
MCP天氣工具實(shí)戰(zhàn)項(xiàng)目
下面，我們將通過(guò)一個(gè)實(shí)際項(xiàng)目，展示如何使用MCP創(chuàng)建一個(gè)天氣信息工具，讓AI模型能夠查詢(xún)實(shí)時(shí)天氣數(shù)據(jù)。
項(xiàng)目介紹
這是一個(gè)基于MCP的天氣工具演示項(xiàng)目，通過(guò)和風(fēng)天氣API獲取實(shí)時(shí)天氣數(shù)據(jù)，提供以下功能：
1.天氣預(yù)警查詢(xún)：獲取指定城市的天氣災(zāi)害預(yù)警信息
2.天氣預(yù)報(bào)查詢(xún)：獲取指定城市未來(lái)幾天的天氣預(yù)報(bào)
項(xiàng)目架構(gòu)
項(xiàng)目分為三個(gè)主要部分：
┌─────────────┐ stdio ┌──────────────┐
│ │?────────────?│ │
│ MCP 客戶(hù)端 │ │ MCP 服務(wù)器 │
│ │ │ │
└─────────────┘ └──────────────┘
▲
│
調(diào)試 │
│
▼
┌─────────────┐
│ │
│MCP Inspector│
│ │
└─────────────┘1.MCP服務(wù)器：提供天氣工具的核心實(shí)現(xiàn)
2.MCP客戶(hù)端：連接服務(wù)器，發(fā)送工具調(diào)用請(qǐng)求
3.MCP Inspector：用于調(diào)試和測(cè)試服務(wù)器
環(huán)境準(zhǔn)備
開(kāi)始前，我們需要準(zhǔn)備以下環(huán)境：
? Python 3.10.12 或更高版本
? NodeJS 22.14.0+ 和 NPM 10.9.2+（用于MCP Inspector）
? 和風(fēng)天氣API Key和API Host（注冊(cè)地址[1]）【具體流程參考文末】
實(shí)戰(zhàn)步驟
第一步：創(chuàng)建MCP服務(wù)器
MCP服務(wù)器是提供工具功能的核心部分。以下是實(shí)現(xiàn)天氣服務(wù)器的核心代碼：
import os
import json
import httpx
import asyncio
from dotenv import load_dotenv
from modelcontextprotocol.server import (
create_server,
ServerConfig,
tools,
JsonSchema,
)
# 加載環(huán)境變量
load_dotenv()
API_KEY = os.getenv("QWEATHER_API_KEY")
API_HOST = os.getenv("QWEATHER_API_HOST", "https://XXX.qweather.com")
# 定義天氣預(yù)警工具
@tools.tool(
name="get_weather_warning",
descriptinotallow="獲取指定位置的天氣災(zāi)害預(yù)警",
parameters=JsonSchema(
{
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "城市ID或經(jīng)緯度坐標(biāo)（經(jīng)度,緯度）\n例如：'101010100'（北京）或 '116.41,39.92'",
},
},
"required": ["location"],
}
),
)
asyncdefget_weather_warning(location: str) -> str:
"""
獲取指定位置的天氣災(zāi)害預(yù)警
參數(shù):
location: 城市ID或經(jīng)緯度坐標(biāo)（經(jīng)度,緯度）
例如：'101010100'（北京）或 '116.41,39.92'
返回:
格式化的預(yù)警信息字符串
"""
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"獲取天氣預(yù)警失敗: {data['code']}"
warnings = data.get("warning", [])
ifnot warnings:
return"當(dāng)前沒(méi)有天氣預(yù)警信息"
result = []
for warning in warnings:
result.append(
f"預(yù)警ID: {warning['id']}\n"
f"標(biāo)題: {warning['title']}\n"
f"發(fā)布時(shí)間: {warning['pubTime']}\n"
f"開(kāi)始時(shí)間: {warning['startTime']}\n"
f"結(jié)束時(shí)間: {warning['endTime']}\n"
f"預(yù)警類(lèi)型: {warning['typeName']}\n"
f"預(yù)警等級(jí): {warning['severityName']} ({warning['level']})\n"
f"發(fā)布單位: {warning['sender']}\n"
f"狀態(tài): {warning['status']}\n"
f"詳細(xì)信息: {warning['text']}"
)
return"\n\n".join(result)
# 定義天氣預(yù)報(bào)工具
@tools.tool(
name="get_daily_forecast",
descriptinotallow="獲取指定位置的天氣預(yù)報(bào)",
parameters=JsonSchema(
{
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "城市ID或經(jīng)緯度坐標(biāo)（經(jīng)度,緯度）\n例如：'101010100'（北京）或 '116.41,39.92'",
},
"days": {
"type": "integer",
"description": "預(yù)報(bào)天數(shù)，可選值為 3、7、10、15、30，默認(rèn)為 3",
"enum": [3, 7, 10, 15, 30],
"default": 3,
},
},
"required": ["location"],
}
),
)
asyncdefget_daily_forecast(location: str, days: int = 3) -> str:
"""
獲取指定位置的天氣預(yù)報(bào)
參數(shù):
location: 城市ID或經(jīng)緯度坐標(biāo)（經(jīng)度,緯度）
例如：'101010100'（北京）或 '116.41,39.92'
days: 預(yù)報(bào)天數(shù)，可選值為 3、7、10、15、30，默認(rèn)為 3
返回:
格式化的天氣預(yù)報(bào)字符串
"""
# 根據(jù)天數(shù)選擇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"獲取天氣預(yù)報(bào)失敗: {data['code']}"
daily = data.get("daily", [])
ifnot daily:
return"無(wú)法獲取天氣預(yù)報(bào)信息"
result = []
for day in daily[:days]: # 限制天數(shù)
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"白天風(fēng)向: {day['windDirDay']} {day['windScaleDay']}級(jí) ({day['windSpeedDay']}km/h)\n"
f"夜間風(fēng)向: {day['windDirNight']} {day['windScaleNight']}級(jí) ({day['windSpeedNight']}km/h)\n"
f"相對(duì)濕度: {day['humidity']}%\n"
f"降水量: {day['precip']}mm\n"
f"紫外線(xiàn)指數(shù): {day['uvIndex']}\n"
f"能見(jiàn)度: {day['vis']}km"
)
return"\n\n---\n\n".join(result)
# 主函數(shù)
asyncdefmain():
config = ServerConfig()
server = create_server(config)
# 注冊(cè)工具
server.register_tool(get_weather_warning)
server.register_tool(get_daily_forecast)
# 啟動(dòng)服務(wù)器
await server.serve()
if __name__ == "__main__":
asyncio.run(main())第二步：實(shí)現(xiàn)MCP客戶(hù)端
MCP客戶(hù)端用于連接服務(wù)器并調(diào)用工具。以下是客戶(hù)端的實(shí)現(xiàn)：
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
# 啟動(dòng)服務(wù)器進(jìn)程
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
)
# 主函數(shù)
asyncdefmain():
print("啟動(dòng) MCP 服務(wù)器進(jìn)程...")
server_process = await start_server_process()
# 配置客戶(hù)端
config = ClientConfig()
client = create_client(config)
try:
# 連接到服務(wù)器
await client.connect_process(server_process)
# 獲取可用工具
tools = await client.get_tools()
print(f"已連接到服務(wù)器，可用工具: {len(tools)}")
# 顯示工具信息
for tool in tools:
print(f" - {tool.name}: \n{tool.description}\n")
print("使用 'help' 查看幫助，使用 'exit' 退出\n")
# 命令行交互循環(huá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 <工具名> <參數(shù)JSON> - 調(diào)用工具")
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("錯(cuò)誤: 需要指定工具名稱(chēng)")
continue
tool_name = parts[0]
# 查找工具
tool = next((t for t in tools if t.name == tool_name), None)
ifnot tool:
print(f"錯(cuò)誤: 找不到工具 '{tool_name}'")
continue
# 解析參數(shù)
args = {}
iflen(parts) > 1:
arg_text = parts[1].strip()
# 簡(jiǎn)單參數(shù)處理
ifnot arg_text.startswith("{"):
# 簡(jiǎn)單模式: call get_daily_forecast 101010100 7
simple_args = arg_text.split(" ")
# 檢查是否為天氣預(yù)報(bào)工具
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("錯(cuò)誤: days 參數(shù)必須是整數(shù)")
continue
elif tool_name == "get_weather_warning":
iflen(simple_args) >= 1:
args["location"] = simple_args[0]
else:
print("錯(cuò)誤: 簡(jiǎn)單參數(shù)模式僅支持預(yù)定義工具")
continue
else:
# JSON模式: call get_weather_warning {"location": "101010100"}
try:
args = json.loads(arg_text)
except json.JSONDecodeError:
print("錯(cuò)誤: 無(wú)效的JSON參數(shù)")
continue
print("正在調(diào)用工具...\n")
try:
# 調(diào)用工具
result = await client.call_tool(tool_name, args)
print("結(jié)果:")
print(result)
print()
except Exception as e:
print(f"錯(cuò)誤: {str(e)}")
print()
else:
print("未知命令，使用 'help' 查看幫助")
print()
finally:
# 關(guān)閉連接和進(jìn)程
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調(diào)試
首先在?
?.env???文件配置??QWEATHER_API_KEY???和 ??QWEATHER_API_KEY??
MCP Inspector是調(diào)試MCP服務(wù)器的利器，提供可視化界面：
1. 安裝Inspector：
npm install -g @modelcontextprotocol/inspector2. 啟動(dòng)Inspector：
mcp dev server/weather_server.py3. 在瀏覽器訪(fǎng)問(wèn) http://localhost:6274
Inspector界面可以讓您直觀(guān)地查看工具定義、測(cè)試調(diào)用并查看結(jié)果。
查看可用工具及其描述
查看可用工具及其描述
查詢(xún)北京未來(lái)3天天氣
查詢(xún)北京未來(lái)3天天氣
查詢(xún)北京災(zāi)害預(yù)警
查詢(xún)北京未來(lái)3天天氣
查詢(xún)北京災(zāi)害預(yù)警
查詢(xún)北京災(zāi)害預(yù)警
提示：MCP Inspector 提供了更直觀(guān)的界面來(lái)測(cè)試和調(diào)試 MCP 服務(wù)器，特別適合開(kāi)發(fā)和調(diào)試復(fù)雜工具。
實(shí)際效果展示
首先在?
?.env???文件配置??QWEATHER_API_KEY???和 ??QWEATHER_API_KEY??
啟動(dòng)程序??python client/mcp_client.py??
天氣預(yù)警查詢(xún)
> call get_weather_warning {"location": "101010100"}
正在調(diào)用工具...
結(jié)果:
預(yù)警ID: 10123020120230713145500551323468
標(biāo)題: 杭州市氣象臺(tái)發(fā)布高溫黃色預(yù)警[III級(jí)/較重]
發(fā)布時(shí)間: 2023-07-13T14:55+08:00
開(kāi)始時(shí)間: 2023-07-13T14:55+08:00
結(jié)束時(shí)間: 2023-07-14T14:55+08:00
預(yù)警類(lèi)型: 高溫
預(yù)警等級(jí): Moderate (Yellow)
發(fā)布單位: 杭州市氣象臺(tái)
狀態(tài): active
詳細(xì)信息: 杭州市氣象臺(tái)2023年07月13日14時(shí)55分發(fā)布高溫黃色預(yù)警信號(hào)：預(yù)計(jì)未來(lái)24小時(shí)內(nèi)最高氣溫將達(dá)到37℃以上，請(qǐng)注意防暑降溫。天氣預(yù)報(bào)查詢(xún)
> call get_daily_forecast 101010100 3
正在調(diào)用工具...
結(jié)果:
日期: 2023-07-13
日出: 04:54 日落: 19:44
最高溫度: 32°C 最低溫度: 22°C
白天天氣: 多云 夜間天氣: 陰
白天風(fēng)向: 東南風(fēng) 3級(jí) (19km/h)
夜間風(fēng)向: 東南風(fēng) 3級(jí) (16km/h)
相對(duì)濕度: 75%
降水量: 0mm
紫外線(xiàn)指數(shù): 7
能見(jiàn)度: 25km
---
日期: 2023-07-14
日出: 04:55 日落: 19:43
最高溫度: 33°C 最低溫度: 23°C
白天天氣: 多云 夜間天氣: 陰
白天風(fēng)向: 東南風(fēng) 3級(jí) (21km/h)
夜間風(fēng)向: 東風(fēng) 3級(jí) (15km/h)
相對(duì)濕度: 72%
降水量: 0mm
紫外線(xiàn)指數(shù): 8
能見(jiàn)度: 25km
---
日期: 2023-07-15
日出: 04:56 日落: 19:43
最高溫度: 34°C 最低溫度: 23°C
白天天氣: 多云 夜間天氣: 多云
白天風(fēng)向: 東南風(fēng) 3級(jí) (18km/h)
夜間風(fēng)向: 東風(fēng) 3級(jí) (14km/h)
相對(duì)濕度: 70%
降水量: 0mm
紫外線(xiàn)指數(shù): 9
能見(jiàn)度: 25kmMCP的進(jìn)階應(yīng)用
MCP不僅限于天氣查詢(xún)，還可以實(shí)現(xiàn)：
1.文件操作：讀寫(xiě)文件、處理上傳文件
2.數(shù)據(jù)庫(kù)交互：查詢(xún)和修改數(shù)據(jù)庫(kù)
3.多媒體處理：處理圖像、音頻、視頻
4.復(fù)雜工作流：多工具鏈?zhǔn)秸{(diào)用
MCP開(kāi)發(fā)最佳實(shí)踐
1.工具設(shè)計(jì)：
? 單一職責(zé)：每個(gè)工具只做一件事
? 明確參數(shù)：詳細(xì)描述每個(gè)參數(shù)的用途
? 健壯錯(cuò)誤處理：優(yōu)雅處理各類(lèi)異常情況
2.安全考慮：
? 輸入驗(yàn)證：使用JSON Schema驗(yàn)證輸入
? 權(quán)限控制：限制工具訪(fǎng)問(wèn)范圍
? 資源限制：防止資源濫用
3.調(diào)試技巧：
? 使用MCP Inspector可視化調(diào)試
? 日志記錄：添加詳細(xì)日志
? 參數(shù)測(cè)試：測(cè)試邊界條件和異常輸入
結(jié)語(yǔ)
MCP為AI模型與外部系統(tǒng)的交互提供了標(biāo)準(zhǔn)化、安全、高效的解決方案。通過(guò)本文的天氣工具實(shí)戰(zhàn)項(xiàng)目，您已經(jīng)掌握了MCP的基本應(yīng)用。隨著大模型應(yīng)用的普及，MCP將在A(yíng)I工具鏈開(kāi)發(fā)中扮演越來(lái)越重要的角色。
希望這篇入門(mén)指南能幫助您開(kāi)始MCP之旅，構(gòu)建更強(qiáng)大、更安全的AI應(yīng)用。歡迎在評(píng)論區(qū)分享您的想法和實(shí)踐經(jīng)驗(yàn)！
附錄
和風(fēng)天氣 API 注冊(cè)與使用
要使用本項(xiàng)目，需要先注冊(cè)和風(fēng)天氣開(kāi)發(fā)者賬號(hào)并獲取 API Key：
1.注冊(cè)和風(fēng)天氣開(kāi)發(fā)者賬號(hào)：
? 訪(fǎng)問(wèn)和風(fēng)天氣開(kāi)發(fā)服務(wù)[2]
? 點(diǎn)擊"注冊(cè)"，按照提示完成賬號(hào)注冊(cè)
2.創(chuàng)建項(xiàng)目并獲取 API Key：
? 登錄開(kāi)發(fā)者控制臺(tái)
? 點(diǎn)擊"項(xiàng)目管理" -> "創(chuàng)建項(xiàng)目"
? 填寫(xiě)項(xiàng)目名稱(chēng)、創(chuàng)建憑據(jù)
? 創(chuàng)建成功后，在項(xiàng)目詳情頁(yè)可以獲取 API Key
和風(fēng)天氣API Key
3.開(kāi)發(fā)者的API Host：
? 登錄開(kāi)發(fā)者控制臺(tái)
? 點(diǎn)擊"頭像" -> "設(shè)置"，或直接訪(fǎng)問(wèn)https://console.qweather.com/setting?lang=zh
? 查看API Host
和風(fēng)天氣API Host
4.API 使用說(shuō)明：
? 免費(fèi)版API有調(diào)用次數(shù)限制，詳情請(qǐng)參考和風(fēng)天氣定價(jià)頁(yè)面[3]
? 支持通過(guò)城市ID或經(jīng)緯度坐標(biāo)查詢(xún)天氣信息
? 城市ID可通過(guò)和風(fēng)天氣城市查詢(xún)API[4]獲取
參考資源
? MCP官方文檔：https://modelcontextprotocol.io/
? MCP快速入門(mén)：https://modelcontextprotocol.io/quickstart/server
? 項(xiàng)目源碼：https://github.com/FlyAIBox/mcp-in-action/tree/qweather_0.1/mcp_demo
? 和風(fēng)天氣API：https://dev.qweather.com/
引用鏈接
??[1] 注冊(cè)地址: https://dev.qweather.com/??
??[2]??? 和風(fēng)天氣開(kāi)發(fā)服務(wù): ??https://dev.qweather.com/??
??[3]??? 和風(fēng)天氣定價(jià)頁(yè)面: ??https://dev.qweather.com/docs/pricing/??
??[4]??? 和風(fēng)天氣城市查詢(xún)API: ???https://dev.qweather.com/docs/api/geoapi/???
本文轉(zhuǎn)載自 ??螢火AI百寶箱???，作者： 螢火AI百寶箱
