MCP協議之MCP簡述 原創
背景
隨著AI Agent在2025年的火爆,與之相關的MCP協議也越來越受到開發者的重視,本文將結合示例深入了解MCP協議的原理架構以及應用方法。
(What)MCP協議是什么
MCP(Model Context Protocol)是一種專為AI Agent設計的標準化協議,旨在解決AI模型與外部數據、工具之間的集成難題。其核心定位是成為AI領域的“通用接口”,類似于物理世界中的USB-C標準,為不同AI系統提供安全、無縫、可擴展的數據交換能力。
(Why)為什么要使用MCP協議
MCP的作用主要有三點:
- 消除對接碎片化。就像早期USB-C標準沒有誕生之前,我們的手機、電腦設備不得不面臨五花八門的插口問題,現在大模型在與各家服務商進行API調用的時候,如果沒有統一的協議,那么開發者和使用者不得不面臨類似的問題。通過MCP協議,開發者和AI模型可以輕松地集成,實現數據交換和交互。
- 在不同 LLM 提供商和供應商之間切換的靈活性
- 在您的基礎設施中保護數據的最佳實踐
備注:以上兩點在MCP官網有提到,但目前接觸不夠,暫未有深入體會。
(How)如何實現一個MCP協議
MCP協議的架構構成主要主要由5部分組成:
- MCP主機(MCP Hosts)如Claude Desktop、IDE 或 AI 工具等想要通過 MCP 訪問數據的程序
- MCP客戶端(MCP Clients)與服務器保持 1:1 連接的協議客戶端
- MCP服務器(MCP Servers)通過標準化的模型上下文協議暴露特定功能的輕量級程序
- 本地數據源(Local Data Sources)MCP 服務器可以安全訪問的計算機文件、數據庫和服務
- 遠程服務(Remote Services)MCP 服務器可以連接的通過互聯網提供的外部系統(例如通過 API)
接下來,我們實現一個??MCPServer?
??,這個??Server?
??可以通過??Web API?
?訪問遠程的服務器以獲取天氣信息。
1. 準備環境:安裝??uv?
?
# Mac下使用curl命令安裝
curl -LsSf https://astral.sh/uv/install.sh | sh
# Window下使用PowerShell命令安裝
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
運行結果:
備注:如果提示 ?
?url: (7) Failed to connect to github.com port 443 after 93 ms: Couldn't connect to server?
? 可能需要科學上網或者過一段時間再試。
2. 創建項目
# 為我們的項目創建一個新目錄
uv init weather
cd weather
# 創建虛擬環境并激活它
uv venv
source .venv/bin/activate
# 安裝依賴
uv add "mcp[cli]" httpx
# 創建我們的服務器文件
touch weather.py
運行結果:
3. 實現weather.py的代碼
3.1 導入包并設置實例
from typing import Any, Dict
import httpx
from mcp.server.fastmcp import FastMCP
# 初始化FastMCP服務器
mcp = FastMCP("weather")
# 常量
BAIDU_API_BASE = "https://api.map.baidu.com/weather/v1/"
BAIDU_API_KEY = "8HkEwz5h********"
3.2 實現輔助函數
# 城市與行政區ID映射表
WEATHER_DISTRICT_ID = {
"北京": "110100",
"上海": "310000",
"廣州": "440100",
"深圳": "440300",
# 可以根據需要添加更多城市
}
asyncdefmake_baidu_request(district_id: str) -> Dict[str, Any] | None:
"""向百度天氣API發出GET請求,處理錯誤并返回JSON響應"""
params = {
"district_id": district_id,
"data_type": "now",
"ak": BAIDU_API_KEY
}
asyncwith httpx.AsyncClient() as client:
try:
response = await client.get(BAIDU_API_BASE, params=params, timeout=30.0)
response.raise_for_status()
return response.json()
except Exception:
returnNone
defformat_weather(data: Dict) -> str:
"""將天氣數據格式化為可讀字符串"""
location = data["result"]["location"]
now = data["result"]["now"]
returnf"""
城市: {location['city']}
天氣狀況: {now['text']}
溫度: {now['temp']}°C
體感溫度: {now['feels_like']}°C
濕度: {now['rh']}%
風力: {now['wind_class']}
風向: {now['wind_dir']}
更新時間: {now['uptime']}
"""
defget_district_id(city: str) -> str | None:
"""根據城市名稱獲取對應的行政區ID"""
return WEATHER_DISTRICT_ID.get(city)
3.3 實現工具執行
@mcp.tool()
asyncdefget_weather(city: str) -> str:
"""獲取指定城市的當前天氣
Args:
city: 城市名稱
"""
district_id = get_district_id(city)
ifnot district_id:
returnf"未找到{city}對應的行政區ID。"
data = await make_baidu_request(district_id)
ifnot data or data.get("status") != 0:
return"無法獲取天氣信息。"
return format_weather(data)
3.4 實現入口函數
if __name__ == "__main__":
# 初始化并運行服務器
mcp.run(transport='stdio')
完整代碼
from typing importAny, Dict
import httpx
from mcp.server.fastmcp import FastMCP
# 初始化FastMCP服務器
mcp = FastMCP("weather")
# 常量
BAIDU_API_BASE = "https://api.map.baidu.com/weather/v1/"
BAIDU_API_KEY = "8HkEwz5h********"
# 城市與行政區ID映射表
WEATHER_DISTRICT_ID = {
"北京": "110100",
"上海": "310000",
"廣州": "440100",
"深圳": "440300",
# 可以根據需要添加更多城市
}
asyncdefmake_baidu_request(district_id: str) -> Dict[str, Any] | None:
"""向百度天氣API發出GET請求,處理錯誤并返回JSON響應"""
params = {
"district_id": district_id,
"data_type": "now",
"ak": BAIDU_API_KEY
}
asyncwith httpx.AsyncClient() as client:
try:
response = await client.get(BAIDU_API_BASE, params=params, timeout=30.0)
response.raise_for_status()
return response.json()
except Exception:
returnNone
defformat_weather(data: Dict) -> str:
"""將天氣數據格式化為可讀字符串"""
location = data["result"]["location"]
now = data["result"]["now"]
returnf"""
城市: {location['city']}
天氣狀況: {now['text']}
溫度: {now['temp']}°C
體感溫度: {now['feels_like']}°C
濕度: {now['rh']}%
風力: {now['wind_class']}
風向: {now['wind_dir']}
更新時間: {now['uptime']}
"""
defget_district_id(city: str) -> str | None:
"""根據城市名稱獲取對應的行政區ID"""
return WEATHER_DISTRICT_ID.get(city)
@mcp.tool()
asyncdefget_weather(city: str) -> str:
"""獲取指定城市的當前天氣
Args:
city: 城市名稱
"""
district_id = get_district_id(city)
ifnot district_id:
returnf"未找到{city}對應的行政區ID。"
data = await make_baidu_request(district_id)
ifnot data or data.get("status") != 0:
return"無法獲取天氣信息。"
return format_weather(data)
if __name__ == "__main__":
# 初始化并運行服務器
mcp.run(transport='stdio')
?
?BAIDU_API_KEY?
?? 需要訪問 ??http://lbsyun.baidu.com?
? 注冊獲取。
4. 啟動服務器
通過??uv run weather.py?
?啟動服務。
5. 測試服務
命令行下運行如下命令安裝inspector。
npx @modelcontextprotocol/inspector
運行結果:
6. 調試服務
- 安裝完畢后,在瀏覽器中打開?
?http://localhost:5173/?
?。 - 頁面輸入調試命令:
- command: uv
- Arguments: --directory /Users/deadwalk/Code/ai_proj_agent/weather run weather.py
備注:
- ?
?/Users/deadwalk/Code/ai_proj_agent/weather?
?對應創建的工程目錄,請根據自己的情況進行修改。
3. 點擊Connect按鈕,確認服務可以正常連接;
4. 在右側Tools點擊List Tools->Weather->輸入參數"北京"->Run Tool,可以看到正常獲得北京的天氣情況,此時代表mcp-server-weather可以正常運行了。
運行結果:
7. 集成到其他應用中
7.1 在cherry-studio中集成服務
- 打開cherry-studio的設置->MCP服務器->編輯JSON,添加如下mcp服務設置:
{
"mcpServers":{
"weather":{
"command":"/Users/deadwalk/.local/bin/uv",
"args":[
"--directory",
"/Users/deadwalk/Code/ai_proj_agent/weather",
"run",
"weather.py"
]
}
}
}
備注:
- ?
?/Users/deadwalk/.local/bin/uv?
?? 對應??uv?
?可執行文件的完整路徑,可以通過MacOS/Linux上運行which uv或在Windows上運行where uv來獲取此路徑。
2.添加成功之后,啟用??weather?
??服務和??deepseek-chat?
?模型,提問大模型:北京今天的天氣是多少?運行結果:
從圖中可以看到,大模型調用了我們之前封裝的mcp-server-weather服務,并成功獲取了北京的天氣。
7.2 在cursor中集成服務
- 我們也可以在cursor中添加對應的服務
- 配置完畢MCPserver之后,啟用weather服務(weather左側的綠點點亮),然后在對話框中詢問北京的天氣情況即可獲得查詢結果。
運行結果:
總結
- MCP是一套服務間通信協議,它通過統一的協議,解決了大模型與工具間繁瑣的適配通信問題。
- MCP的構成包括:MCP Host、MCP Client、MCP Server、Local Data Source、Remote Data Source。
- 封裝MCP-server-weather服務時,需要在工具函數上添加@mcp.tool()裝飾器。
- 通過MCP協議,我們可以封裝各種服務提供給Cursor、Cherry-Studio、甚至我們自己開發的Agent使用,從而使得LLM+Agent的功能更加強大。
本文轉載自公眾號一起AI技術 作者:熱情的Dongming
