一、MCP是什麼? 為什麼需要它?
想象一下,你正在開發一個 AI 編程助手,它需要:
- 讀取和修改項目文件
- 查詢數據庫Schema
- 搜索代碼倉庫
- 執行Git操作
傳統做法是為每個數據源寫一套專用代碼,不同團隊重複造輪子。Model Context Protocol(MCP) 就是為了解決這個問題而生的開放標準協議。
通俗理解: MCP就像是「AI應用的USB接口標準」。就像USB讓不同設備都能接入電腦一樣,MCP讓不同的數據源和工具都能以統一方式接入AI應用。
實際案例: 在Claude Desktop中,你可以配置多個官方MCP服務器:
- Filesystem服務器: 安全地讀寫本地文件,有權限控制
- SQLite服務器: 查詢和分析SQLite數據庫,自動生成SQL
- GitHub服務器: 搜索倉庫、創建Issue、管理PR
你的AI應用只需實現一個MCP客户端,就能連接所有服務器,無需為每個服務器寫專用代碼。
二、架構設計: 三個角色的分工
MCP採用宿主-客户端-服務器三層架構,就像一家公司的組織結構:
宿主(Host) = 總經理
- 管理所有客户端
- 控制安全策略和權限
- 負責AI模型的調用
客户端(Client) = 部門經理
- 客户端負責連接服務器
- 負責雙方的溝通協調
- 轉發消息和通知
服務器(Server) = 業務專員
- 提供具體功能(資源、工具、提示模板)
- 可以是本地程序或遠程服務
- 不知道其他服務器的存在
三、協議約定:統一規範與個性化擴展
每個MCP服務器提供的工具、資源都不一樣,但它們都遵循相同的MCP協議規範。
3.1 協議的分層設計
MCP採用 基礎協議 + 功能擴展 的設計,就像HTTP協議一樣:
核心層(所有實現必須支持) :
- JSON-RPC 2.0消息格式
- 初始化握手流程(initialize/initialized)
- 基本錯誤處理
功能層(按需選擇) :
- Resources、Prompts、Tools(服務器端)
- Roots、Sampling、Elicitation(客户端)
這樣設計的好處:
統一的基礎協議 → 保證互操作性
+
靈活的功能選擇 → 滿足不同場景需求
↓
既標準化又可擴展
3.2 協議約定的過程
步驟1: 基礎協議是固定的\
所有MCP服務器和客户端都遵循相同的JSON-RPC 2.0格式:
// 請求格式(固定)
{
"jsonrpc": "2.0", // 必須是2.0
"id": 1, // 唯一標識
"method": "方法名", // 要調用的方法
"params": {...} // 參數對象
}
// 響應格式(固定)
{
"jsonrpc": "2.0",
"id": 1, // 對應請求的ID
"result": {...} // 成功結果
// 或 "error": {...} // 錯誤信息
}
步驟2: 能力在初始化時協商
// 客户端發起初始化
{
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {
"sampling": {}, // 我支持LLM採樣
"roots": {} // 我支持根目錄
},
"clientInfo": {"name": "MyClient", "version": "1.0"}
}
}
// 服務器響應
{
"result": {
"protocolVersion": "2024-11-05",
"capabilities": {
"tools": {}, // 我提供工具
"resources": {} // 我提供資源
},
"serverInfo": {"name": "SQLiteServer", "version": "2.0"}
}
}
協商完成後,雙方都知道對方支持什麼功能,只使用交集部分。
步驟3: 方法名稱是標準化的\
MCP規範定義了標準方法名:
| 功能 | 方法名 | 説明 |
|---|---|---|
| 列出資源 | resources/list |
固定方法名 |
| 讀取資源 | resources/read |
固定方法名 |
| 列出工具 | tools/list |
固定方法名 |
| 調用工具 | tools/call |
固定方法名 |
| 列出提示 | prompts/list |
固定方法名 |
| 獲取提示 | prompts/get |
固定方法名 |
步驟4: 具體內容是個性化的\
雖然方法名固定,但每個服務器返回的具體數據不同:
// SQLite服務器的工具
{
"tools": [
{"name": "query", "description": "執行SQL查詢"},
{"name": "list_tables", "description": "列出所有表"}
]
}
// Filesystem服務器的工具
{
"tools": [
{"name": "read_file", "description": "讀取文件"},
{"name": "write_file", "description": "寫入文件"},
{"name": "search_files", "description": "搜索文件"}
]
}
3.3 協議發現機制
客户端如何知道服務器有哪些工具?
第一步:列舉
客户端 → 服務器: {"method": "tools/list"}
服務器 → 客户端: {
"tools": [
{
"name": "query",
"description": "執行SQL查詢",
"inputSchema": { // JSON Schema定義輸入格式
"type": "object",
"properties": {
"sql": {"type": "string"}
}
}
}
]
}
第二步:調用
客户端 → 服務器: {
"method": "tools/call",
"params": {
"name": "query", // 使用第一步獲得的工具名
"arguments": {"sql": "SELECT * FROM users"}
}
}
關鍵點:通過JSON Schema,客户端知道如何正確調用工具,無需硬編碼。
四、協議基礎:如何通信?
MCP基於JSON-RPC 2.0構建,這是一個成熟的遠程過程調用協議。理解這一層對掌握MCP至關重要。
4.1 JSON-RPC 2.0基礎
消息類型
MCP中有三種基本消息類型。\
1. 請求(Request) - 期待響應
{
"jsonrpc": "2.0", // 協議版本,必須是"2.0"
"id": 1, // 請求唯一標識(字符串或數字)
"method": "tools/list", // 要調用的方法名
"params": { // 可選的參數對象
"cursor": "page2"
}
}
2. 響應(Response) - 對請求的回覆
// 成功響應
{
"jsonrpc": "2.0",
"id": 1, // 必須與請求的id相同
"result": { // 成功結果
"tools": [
{"name": "query", "description": "執行查詢"}
]
}
}
// 錯誤響應
{
"jsonrpc": "2.0",
"id": 1,
"error": { // 錯誤對象
"code": -32602, // 錯誤碼(整數)
"message": "參數無效", // 錯誤描述
"data": { // 可選的額外信息
"field": "cursor",
"reason": "格式錯誤"
}
}
}
3. 通知(Notification) - 單向消息,無需響應
{
"jsonrpc": "2.0",
"method": "notifications/resources/updated", // 通知方法名
"params": { // 通知參數
"uri": "file:///project/data.json"
}
// 注意:沒有id字段
}
標準錯誤碼
MCP使用JSON-RPC 2.0的標準錯誤碼:
| 錯誤碼 | 含義 | 説明 |
|---|---|---|
| -32700 | Parse error | JSON解析錯誤 |
| -32600 | Invalid Request | 無效的請求格式 |
| -32601 | Method not found | 方法不存在 |
| -32602 | Invalid params | 參數無效 |
| -32603 | Internal error | 服務器內部錯誤 |
| -32002 | Resource not found | 資源未找到(MCP擴展) |
4.2 能力協商詳解
能力協商是MCP連接建立的第一步,決定了整個會話中可用的功能。
初始化流程詳解
階段1: 客户端發起初始化
{
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05", // 客户端支持的協議版本
"capabilities": { // 客户端能力聲明
"roots": { // 支持根目錄
"listChanged": true // 支持根目錄變更通知
},
"sampling": {}, // 支持LLM採樣
"elicitation": {}, // 支持用户詢問
"experimental": { // 實驗性功能
"customFeature": {} // 自定義功能
}
},
"clientInfo": { // 客户端信息
"name": "MyAIApp", // 程序名(必填)
"version": "1.2.0", // 版本號(必填)
"title": "我的AI應用" // 顯示名稱(可選)
}
}
}
階段2: 服務器響應能力
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"protocolVersion": "2024-11-05", // 服務器選擇的協議版本
"capabilities": { // 服務器能力聲明
"resources": { // 支持資源
"subscribe": true, // 支持資源訂閲
"listChanged": true // 支持資源列表變更通知
},
"tools": { // 支持工具
"listChanged": true
},
"prompts": { // 支持提示模板
"listChanged": false // 不支持列表變更通知
},
"logging": {} // 支持日誌輸出
},
"serverInfo": { // 服務器信息
"name": "sqlite-mcp-server",
"version": "2.1.0",
"title": "SQLite MCP服務器"
},
"instructions": "此服務器提供SQLite數據庫訪問能力" // 可選的使用説明
}
}
階段3: 客户端確認就緒
{
"jsonrpc": "2.0",
"method": "notifications/initialized" // 無id,這是通知
}
協議版本協商規則
客户端請求版本: "2024-11-05"
↓
服務器支持?
↙ ↘
支持 不支持
↓ ↓
返回相同版本 返回服務器支持的最新版本
↓ ↓
協商成功 客户端檢查是否支持
↙ ↘
支持 不支持
↓ ↓
協商成功 斷開連接
實際示例:
// 場景1: 版本匹配
客户端: "protocolVersion": "2024-11-05"
服務器: "protocolVersion": "2024-11-05" ✅ 成功
// 場景2: 服務器版本更新
客户端: "protocolVersion": "2024-06-01"
服務器: "protocolVersion": "2024-11-05"
→ 客户端檢查是否支持2024-11-05 → 如果不支持則斷開
// 場景3: 客户端版本更新
客户端: "protocolVersion": "2025-01-01"
服務器: "protocolVersion": "2024-11-05"
→ 客户端檢查是否支持2024-11-05 → 如果支持則降級使用
能力交集計算
初始化後,雙方只能使用共同支持的能力:
客户端能力: {roots, sampling, elicitation}
服務器能力: {resources, tools, prompts}
↓
可用功能集合
├─ 客户端 → 服務器: resources, tools, prompts
└─ 服務器 → 客户端: roots, sampling, elicitation
示例:
# 客户端代碼示例
if server_capabilities.get("tools"):
# 服務器支持工具,可以調用
tools = await session.list_tools()
else:
# 服務器不支持工具,跳過
print("服務器不提供工具功能")
if client_capabilities.get("sampling"):
# 客户端支持採樣,服務器可以請求
# (服務器端會檢查這個能力)
pass
4.3 連接生命週期深入
完整的消息時序圖
客户端 服務器
│ │
│ 1. initialize (請求) │
├──────────────────────────────────────>│
│ {protocolVersion, capabilities} │
│ │
│ 2. initialize (響應) │
│<──────────────────────────────────────┤
│ {protocolVersion, capabilities} │
│ │
│ 3. initialized (通知) │
├──────────────────────────────────────>│
│ │
│═══════════ 正常操作階段 ════════════ │
│ │
│ 4. tools/list (請求) │
├──────────────────────────────────────>│
│ │
│ 5. tools/list (響應) │
│<──────────────────────────────────────┤
│ {tools: [...]} │
│ │
│ 6. tools/call (請求) │
├──────────────────────────────────────>│
│ {name: "query", arguments: {...}} │
│ │
│ 7. notifications/progress (通知) │
│<──────────────────────────────────────┤
│ {progress: 50, total: 100} │
│ │
│ 8. tools/call (響應) │
│<──────────────────────────────────────┤
│ {content: [...]} │
│ │
│ 9. notifications/resources/updated │
│<──────────────────────────────────────┤
│ {uri: "file://..."} │
│ │
│═══════════ 關閉階段 ═══════════ │
│ │
│ 10. 關閉stdin │
├─────────────X │
│ │
│ 服務器退出
初始化前的限制
在initialized通知發送前:
客户端只能發送:
- ✅
initialize請求 - ✅
ping請求(用於保活) - ❌ 其他任何請求
服務器只能發送:
- ✅
initialize響應 - ✅
ping請求 - ✅
logging通知(日誌) - ❌ 其他任何消息
違反限制的後果:
// 客户端在初始化前調用tools/list
請求: {"method": "tools/list"}
響應: {
"error": {
"code": -32600,
"message": "會話未初始化"
}
}
超時和重試機制
請求超時:
import asyncio
# 設置30秒超時
try:
result = await asyncio.wait_for(
session.call_tool("slow_operation", {}),
timeout=30.0
)
except asyncio.TimeoutError:
# 發送取消通知
await session.send_notification(
"notifications/cancelled",
{"requestId": "123", "reason": "超時"}
)
進度通知重置超時:
# 當收到進度通知時,可以重置超時計時器
timeout = 30 # 基礎超時
max_timeout = 300 # 最大超時(5分鐘)
while True:
try:
msg = await wait_for_message(timeout)
if msg.method == "notifications/progress":
# 收到進度,重置超時
timeout = 30
except TimeoutError:
# 超時處理
break
4.4 傳輸方式對比
stdio傳輸詳解
優點:
- ✅ 簡單直接,適合本地開發
- ✅ 進程隔離,安全性好
- ✅ 自動管理生命週期
- ✅ 無需網絡配置
缺點:
- ❌ 只能本地使用
- ❌ 不支持多客户端
- ❌ 調試相對困難
消息格式:
消息1\n
消息2\n
消息3\n
每個JSON對象佔一行,以\n分隔。
HTTP傳輸詳解
架構:
┌─────────┐ HTTP POST ┌─────────┐
│ ├──────────────────────────>│ │
│ 客户端 │ 請求/通知/響應(JSON-RPC) │ 服務器 │
│ │<──────────────────────────┤ │
└─────────┘ HTTP 響應/SSE流 └─────────┘
(application/json 或
text/event-stream)
發送消息(POST) :
POST /mcp HTTP/1.1
Host: localhost:8080
Content-Type: application/json
Accept: application/json, text/event-stream
Mcp-Session-Id: abc123
{"jsonrpc":"2.0","id":1,"method":"tools/list"}
立即響應(JSON) :
HTTP/1.1 200 OK
Content-Type: application/json
{"jsonrpc":"2.0","id":1,"result":{"tools":[...]}}
流式響應(SSE) :
HTTP/1.1 200 OK
Content-Type: text/event-stream
Mcp-Session-Id: abc123
id: 1
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progress":25}}
id: 2
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progress":50}}
id: 3
data: {"jsonrpc":"2.0","id":1,"result":{"content":[...]}}
接收服務器消息(GET) :
GET /mcp HTTP/1.1
Host: localhost:8080
Accept: text/event-stream
Mcp-Session-Id: abc123
Last-Event-ID: 42
會話管理:
# 服務器端設置會話ID
@app.post("/mcp")
async def handle_mcp(request):
if request.method == "initialize":
session_id = generate_session_id()
return Response(
content=json.dumps(result),
headers={"Mcp-Session-Id": session_id}
)
# 客户端後續請求攜帶會話ID
@client.request
async def send_request(method, params):
headers = {}
if self.session_id:
headers["Mcp-Session-Id"] = self.session_id
return await http.post(
"/mcp",
json={"jsonrpc": "2.0", "method": method, "params": params},
headers=headers
)
斷線重連:
async def connect_sse(last_event_id=None):
headers = {"Accept": "text/event-stream"}
if last_event_id:
headers["Last-Event-ID"] = last_event_id
async with httpx.stream("GET", "/mcp", headers=headers) as stream:
async for line in stream.aiter_lines():
if line.startswith("id:"):
last_event_id = line[3:].strip()
elif line.startswith("data:"):
data = json.loads(line[5:])
yield data, last_event_id
4.5 實際通信示例
讓我們看一個完整的SQLite查詢場景:
// 1. 列出工具
客户端 → 服務器:
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/list"
}
服務器 → 客户端:
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"tools": [
{
"name": "query",
"description": "執行SQL查詢",
"inputSchema": {
"type": "object",
"properties": {
"sql": {"type": "string"}
},
"required": ["sql"]
}
}
]
}
}
// 2. 調用查詢工具
客户端 → 服務器:
{
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {
"name": "query",
"arguments": {
"sql": "SELECT COUNT(*) FROM users WHERE active = 1"
},
"_meta": {
"progressToken": "query-123" // 請求進度通知
}
}
}
// 3. 服務器發送進度(異步通知)
服務器 → 客户端:
{
"jsonrpc": "2.0",
"method": "notifications/progress",
"params": {
"progressToken": "query-123",
"progress": 50,
"total": 100,
"message": "正在掃描users表..."
}
}
// 4. 返回查詢結果
服務器 → 客户端:
{
"jsonrpc": "2.0",
"id": 2,
"result": {
"content": [
{
"type": "text",
"text": "查詢結果: 1,234個活躍用户"
}
],
"isError": false
}
}
// 5. 如果查詢出錯
服務器 → 客户端(錯誤情況):
{
"jsonrpc": "2.0",
"id": 2,
"error": {
"code": -32603,
"message": "SQL語法錯誤",
"data": {
"sql": "SELECT COUNT(*) FROM users WHERE active = 1",
"error": "near "WHERE": syntax error",
"position": 35
}
}
}
這就是MCP通信的完整過程!通過JSON-RPC 2.0,客户端和服務器可以進行結構化、類型安全的通信。
五、服務器能力:三種核心功能
MCP服務器可以提供三種功能。
5.1 Resources(資源):應用決定用什麼
資源就是數據,比如文件內容、數據庫記錄、API響應。
誰控制: 應用程序決定把哪些資源提供給AI
如何使用:
// 列出所有可用資源
{"method": "resources/list"}
// 讀取某個資源
{
"method": "resources/read",
"params": {"uri": "file:///project/main.py"}
}
資源URI示例:
file:///project/src/main.py- 文件db://schema/users- 數據庫表結構git://commits/main- Git提交歷史https://api.example.com/data- Web API
訂閲變更: 可以訂閲資源,當它變化時自動收到通知。
實際案例: Filesystem服務器暴露資源
{
"uri": "file:///Users/alice/project/src/main.py", // Python源文件
"name": "main.py", // 文件名
"mimeType": "text/x-python", // 文件類型
"text": "import os\ndef main()..." // 文件內容
}
客户端AI可以讀取這個資源,理解代碼結構後提供重構建議或生成測試。
5.2 Prompts(提示模板):用户選擇用什麼
什麼是Prompt?
Prompt就像是「對話模板」或「快捷指令」,把常用的複雜指令預設好,用户一鍵調用。用生活中的例子類比,就像微信的「快捷回覆」或IDE中的「代碼片段(Snippet)」。
為什麼需要Prompt?\
場景1:沒有Prompt時
用户每次都要輸入:
"請分析這個Git倉庫最近一週的提交,統計:
1. 總提交次數
2. 每個作者的貢獻
3. 修改的主要文件
4. 是否有破壞性變更
請用表格格式輸出"
場景2:有Prompt後
用户只需:
1. 點擊 "/analyze-commits" 命令
2. 選擇分支 "main"
3. AI自動執行完整分析
Prompt的數據結構
定義一個Prompt:
{
"name": "analyze_commits", // Prompt的唯一標識
"title": "提交歷史分析", // 用户界面顯示的名稱
"description": "分析Git提交併生成報告", // 功能説明
"arguments": [ // 需要的參數列表
{
"name": "branch", // 參數名
"description": "要分析的分支名", // 參數説明
"required": true // 是否必填
},
{
"name": "since", // 時間範圍
"description": "起始日期(如:7 days ago)",
"required": false // 可選參數
},
{
"name": "author", // 作者過濾
"description": "只看某個作者的提交",
"required": false
}
]
}
實際使用示例
步驟1: 列出所有可用的Prompt
// 客户端請求
{
"jsonrpc": "2.0",
"id": 1,
"method": "prompts/list"
}
// 服務器響應
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"prompts": [
{
"name": "analyze_commits",
"title": "📊 提交歷史分析",
"description": "分析指定分支的提交歷史,生成統計報告",
"arguments": [
{"name": "branch", "required": true},
{"name": "since", "required": false}
]
},
{
"name": "review_code",
"title": "🔍 代碼審查",
"description": "對代碼進行質量審查和改進建議",
"arguments": [
{"name": "file_path", "required": true},
{"name": "focus", "required": false}
]
},
{
"name": "explain_error",
"title": "🐛 錯誤診斷",
"description": "解釋錯誤信息並提供修復建議",
"arguments": [
{"name": "error_message", "required": true}
]
}
]
}
}
步驟2: 用户在界面上看到這些選項
━━━━━━━━━━━━━━━━━━━━━━━━━━━━
可用命令:
📊 /analyze-commits
分析指定分支的提交歷史
🔍 /review-code
對代碼進行質量審查
🐛 /explain-error
解釋錯誤信息並修復
━━━━━━━━━━━━━━━━━━━━━━━━━━━━
步驟3: 用户選擇並填寫參數
用户輸入: /analyze-commits
系統彈窗:
┌─────────────────────────┐
│ 提交歷史分析 │
├─────────────────────────┤
│ 分支名 *: [main ] │
│ 時間範圍: [7 days ago] │
│ 作者: [ ] │
│ │
│ [取消] [確定] │
└─────────────────────────┘
步驟4: 獲取完整的Prompt內容
// 客户端請求
{
"jsonrpc": "2.0",
"id": 2,
"method": "prompts/get",
"params": {
"name": "analyze_commits", // 使用哪個模板
"arguments": { // 用户填寫的參數
"branch": "main",
"since": "7 days ago"
}
}
}
// 服務器響應 - 返回完整的對話消息
{
"jsonrpc": "2.0",
"id": 2,
"result": {
"description": "分析main分支最近7天的提交",
"messages": [ // 發送給AI的完整對話
{
"role": "user", // 用户角色
"content": {
"type": "text",
"text": "請分析main分支在過去7天的Git提交歷史。\n\n需要統計:\n1. 總提交次數\n2. 每個作者的貢獻次數和代碼行數\n3. 主要修改的文件列表\n4. 是否包含breaking changes\n5. 提交消息的規範性\n\n請用表格格式輸出結果,並在最後給出改進建議。"
}
}
]
}
}
步驟5: 客户端將消息發送給AI
用户看到AI正在分析...
AI返回:
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
📊 main分支提交分析報告(最近7天)
總提交次數: 23
作者貢獻統計:
┌────────────┬────────┬──────────┐
│ 作者 │ 提交數 │ 代碼行數 │
├────────────┼────────┼──────────┤
│ Alice │ 12 │ +543/-89│
│ Bob │ 8 │ +234/-45│
│ Charlie │ 3 │ +123/-12│
└────────────┴────────┴──────────┘
主要修改文件:
- src/api/users.py (8次修改)
- src/models/user.py (5次修改)
- tests/test_user.py (4次修改)
Breaking Changes: 無
提交規範性: 良好 (91%符合Conventional Commits)
改進建議:
1. 建議增加單元測試覆蓋率
2. 部分提交消息過於簡短
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Prompt的內容類型
Prompt消息中可以包含多種內容:\
1. 純文本
{
"role": "user",
"content": {
"type": "text",
"text": "請審查這段代碼..."
}
}
2. 嵌入圖片
{
"role": "user",
"content": {
"type": "image",
"data": "base64-encoded-image-data", // 圖片數據
"mimeType": "image/png" // 圖片類型
}
}
3. 嵌入資源(引用MCP資源)
{
"role": "user",
"content": {
"type": "resource",
"resource": {
"uri": "file:///project/src/user.py", // 資源URI
"mimeType": "text/x-python",
"text": "class User:\n def __init__..." // 資源內容
}
}
}
4. 多輪對話
{
"messages": [
{
"role": "user",
"content": {"type": "text", "text": "我想優化這段代碼"}
},
{
"role": "assistant", // AI的回覆
"content": {"type": "text", "text": "請提供代碼內容"}
},
{
"role": "user",
"content": {
"type": "resource", // 用户提供代碼
"resource": {...}
}
}
]
}
Prompt vs Tool vs Resource 對比
特性 Prompt Tool Resource
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
控制者 用户主動選擇 AI自動決定 應用程序控制
觸發方式 用户點擊命令 AI判斷需要調用 應用自動附加
/analyze 或用户選擇
返回內容 對話消息 執行結果 數據內容
(給AI的指令) (函數返回值) (上下文信息)
典型用途 工作流模板 執行操作 提供背景信息
快捷指令 查詢數據 文件內容
示例 代碼審查模板 執行SQL查詢 項目README
錯誤診斷嚮導 發送郵件 數據庫Schema
用户感知 ✅ 明顯 ❓ 可能不知道 ❓ 透明的
(用户點擊) (AI決定) (自動加載)
Prompt是預設的對話模板,通過參數化實現靈活應用,提升用户體驗,並能與MCP其他能力組合形成完整工作流。
代碼實現示例
# 服務器端:註冊Prompt
@server.list_prompts()
async def list_prompts():
return [
Prompt(
name="analyze_commits",
title="📊 提交歷史分析",
description="分析Git提交歷史並生成統計報告",
arguments=[
{"name": "branch", "description": "分支名", "required": True},
{"name": "since", "description": "時間範圍", "required": False}
]
)
]
@server.get_prompt()
async def get_prompt(name: str, arguments: dict):
if name == "analyze_commits":
branch = arguments["branch"]
since = arguments.get("since", "7 days ago")
# 構建完整的提示消息
prompt_text = f"""
請分析{branch}分支在{since}的Git提交歷史。
需要統計:
1. 總提交次數
2. 每個作者的貢獻
3. 主要修改的文件
4. 是否有breaking changes
請用表格格式輸出。
"""
return {
"messages": [
{
"role": "user",
"content": {"type": "text", "text": prompt_text}
}
]
}
# 客户端:使用Prompt
async def use_prompt(session, prompt_name, arguments):
# 獲取Prompt內容
prompt = await session.get_prompt(
name=prompt_name,
arguments=arguments
)
# 將消息發送給AI
for message in prompt.messages:
ai_response = await send_to_ai(message)
print(ai_response)
5.3 Tools(工具):AI 自己決定用什麼
Tool就是可執行的函數,比如查詢數據庫、調用API、寫文件。
誰控制:AI模型根據對話內容自己決定調用哪個工具
如何使用:
// 列出可用工具
{"method": "tools/list"}
// AI調用工具
{
"method": "tools/call",
"params": {
"name": "get_weather",
"arguments": {"city": "北京"}
}
}
返回結果:
{
"content": [{
"type": "text",
"text": "北京天氣:晴,温度22°C"
}],
"isError": false
}
5.4 其他功能
補全
MCP提供標準化的參數自動補全功能,支持為提示和資源URI提供上下文相關的建議,實現類似IDE的交互體驗。服務器通過聲明completions能力,支持對ref/prompt和ref/resource兩種引用類型的補全,每次最多返回100個按相關性排序的建議值,並可通過completion/complete請求獲取補全結果。
日誌
MCP提供結構化日誌消息傳遞機制,允許服務器向客户端發送包含嚴重性級別、可選記錄器名稱和任意JSON可序列化數據的日誌通知。服務器需聲明logging能力,支持遵循RFC 5424標準的日誌級別(從debug到emergency),客户端可通過logging/setLevel請求配置最低日誌級別,服務器通過notifications/message通知發送日誌消息。
分頁
MCP支持對可能返回大量結果集的列表操作進行分頁處理,使用基於不透明遊標的分頁模型而非數字頁碼。服務器在響應中包含當前頁結果和可選的nextCursor字段(表示更多結果存在),客户端可通過在請求中包含遊標繼續分頁。支持分頁的操作包括resources/list、resources/templates/list、prompts/list和tools/list,客户端必須將遊標視為不透明令牌。
六、客户端能力:反向請求
客户端不僅接收服務器的數據,也可以提供能力給服務器使用:
6.1 Sampling(採樣):服務器請求客户端調用AI
場景: 服務器在處理任務時,需要AI幫忙分析中間結果。
如何使用:
{
"method": "sampling/createMessage",
"params": {
"messages": [{
"role": "user",
"content": {"type": "text", "text": "這個數據正常嗎?"}
}],
"modelPreferences": {
"hints": [{"name": "claude-3-sonnet"}], // 建議用的模型
"intelligencePriority": 0.8, // 要求智能程度
"speedPriority": 0.5 // 速度要求
}
}
}
實際案例:Filesystem服務器在搜索大量文件時,請求AI判斷哪些文件最相關。
6.2 Roots(目錄):告訴服務器工作範圍
場景: 讓服務器知道可以訪問哪些目錄。
如何使用:
{
"method": "roots/list"
}
返回:
{
"roots": [{
"uri": "file:///home/user/project",
"name": "我的項目"
}]
}
服務器知道只能在這個目錄裏操作,保護其他文件安全。
6.3 Elicitation(引導):服務器向用户詢問信息
場景: 服務器需要用户提供額外信息才能繼續。
如何使用:
{
"method": "elicitation/create",
"params": {
"message": "請提供您的GitHub用户名",
"requestedSchema": {
"type": "object",
"properties": {
"username": {"type": "string"}
}
}
}
}
用户響應:
{
"action": "accept", // 或"decline"拒絕, "cancel"取消
"content": {
"username": "octocat"
}
}
實際案例: Git服務器需要知道提交信息格式,彈窗問用户:"請選擇提交規範:Conventional Commits/Angular/Custom?"
七、完整實戰:從零構建天氣查詢MCP
下面讓我們從頭到尾構建一個完整的MCP系統,包含服務器和客户端。
7.1 需求分析
目標: 構建一個天氣查詢MCP服務器,提供:
- 資源: 城市列表
- 工具: 查詢天氣、獲取預報
- 提示: 天氣分析模板
7.2 服務器實現(Python)
第一步: 安裝MCP SDK
pip install mcp
第二步: 創建服務器 (weather\_server.py)
from mcp.server import Server
from mcp.types import Resource, Tool, Prompt, TextContent
import mcp.server.stdio
import httpx
# 創建MCP服務器實例
server = Server("weather-server")
# 1. 定義資源:支持的城市列表
@server.list_resources()
async def list_resources():
"""返回可用的資源列表"""
return [
Resource(
uri="weather://cities",
name="支持的城市列表",
mimeType="application/json",
description="查詢天氣支持的所有城市"
)
]
@server.read_resource()
async def read_resource(uri: str):
"""讀取具體資源內容"""
if uri == "weather://cities":
cities = ["北京", "上海", "廣州", "深圳", "杭州"]
return {
"contents": [{
"uri": uri,
"mimeType": "application/json",
"text": str(cities)
}]
}
# 2. 定義工具:天氣查詢
@server.list_tools()
async def list_tools():
"""返回可用的工具列表"""
return [
Tool(
name="get_current_weather",
description="獲取指定城市的當前天氣",
inputSchema={
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名稱,如'北京'"
}
},
"required": ["city"]
}
),
Tool(
name="get_forecast",
description="獲取未來3天天氣預報",
inputSchema={
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名稱"},
"days": {"type": "number", "description": "預報天數(1-3)", "default": 3}
},
"required": ["city"]
}
)
]
@server.call_tool()
async def call_tool(name: str, arguments: dict):
"""執行工具調用"""
if name == "get_current_weather":
city = arguments["city"]
# 實際項目中這裏調用真實的天氣API
# 示例:使用模擬數據
weather_data = {
"city": city,
"temperature": 22,
"condition": "晴",
"humidity": 45
}
return {
"content": [{
"type": "text",
"text": f"{city}當前天氣:\n温度: {weather_data['temperature']}°C\n天氣: {weather_data['condition']}\n濕度: {weather_data['humidity']}%"
}]
}
elif name == "get_forecast":
city = arguments["city"]
days = arguments.get("days", 3)
# 模擬預報數據
forecast = f"{city}未來{days}天預報:\n第1天: 晴,20-25°C\n第2天: 多雲,18-23°C\n第3天: 小雨,16-20°C"
return {
"content": [{"type": "text", "text": forecast}]
}
# 3. 定義提示模板:天氣分析
@server.list_prompts()
async def list_prompts():
"""返回可用的提示模板"""
return [
Prompt(
name="analyze_weather",
description="分析天氣趨勢並給出建議",
arguments=[
{"name": "city", "description": "城市名稱", "required": True}
]
)
]
@server.get_prompt()
async def get_prompt(name: str, arguments: dict):
"""獲取提示模板內容"""
if name == "analyze_weather":
city = arguments["city"]
return {
"messages": [
{
"role": "user",
"content": {
"type": "text",
"text": f"請分析{city}的天氣情況,並給出出行建議。包括:\n1. 温度是否適宜\n2. 是否需要帶傘\n3. 穿衣建議"
}
}
]
}
# 啓動服務器
if __name__ == "__main__":
# 使用stdio傳輸(本地)
mcp.server.stdio.run_stdio_server(server)
7.3 配置服務器(Claude Desktop)
創建配置文件 ~/Library/Application Support/Claude/claude_desktop_config.json:
{
"mcpServers": {
"weather": {
"command": "python",
"args": ["/path/to/weather_server.py"]
}
}
}
7.4 客户端實現(Python)
如果要自己實現客户端:
from mcp import ClientSession
from mcp.client.stdio import stdio_client
import asyncio
async def main():
# 連接到服務器
async with stdio_client(
command="python",
args=["/path/to/weather_server.py"]
) as (read, write):
async with ClientSession(read, write) as session:
# 1. 初始化連接
await session.initialize()
print("✅ 連接成功!")
# 2. 列出可用資源
resources = await session.list_resources()
print(f"\n📁 可用資源: {len(resources.resources)}")
for r in resources.resources:
print(f" - {r.name}: {r.uri}")
# 3. 讀取城市列表資源
cities_resource = await session.read_resource(
uri="weather://cities"
)
print(f"\n🌍 城市列表: {cities_resource.contents[0].text}")
# 4. 列出可用工具
tools = await session.list_tools()
print(f"\n🔧 可用工具: {len(tools.tools)}")
for t in tools.tools:
print(f" - {t.name}: {t.description}")
# 5. 調用工具查詢天氣
result = await session.call_tool(
name="get_current_weather",
arguments={"city": "北京"}
)
print(f"\n🌤️ 查詢結果:\n{result.content[0].text}")
# 6. 獲取預報
forecast = await session.call_tool(
name="get_forecast",
arguments={"city": "上海", "days": 3}
)
print(f"\n📅 天氣預報:\n{forecast.content[0].text}")
# 7. 列出提示模板
prompts = await session.list_prompts()
print(f"\n💡 提示模板: {len(prompts.prompts)}")
for p in prompts.prompts:
print(f" - {p.name}: {p.description}")
# 8. 獲取提示內容
prompt = await session.get_prompt(
name="analyze_weather",
arguments={"city": "廣州"}
)
print(f"\n📝 生成的提示:\n{prompt.messages[0]['content']['text']}")
if __name__ == "__main__":
asyncio.run(main())
7.5 運行效果
$ python weather_client.py
✅ 連接成功!
📁 可用資源: 1
- 支持的城市列表: weather://cities
🌍 城市列表: ['北京', '上海', '廣州', '深圳', '杭州']
🔧 可用工具: 2
- get_current_weather: 獲取指定城市的當前天氣
- get_forecast: 獲取未來3天天氣預報
🌤️ 查詢結果:
北京當前天氣:
温度: 22°C
天氣: 晴
濕度: 45%
📅 天氣預報:
上海未來3天預報:
第1天: 晴,20-25°C
第2天: 多雲,18-23°C
第3天: 小雨,16-20°C
💡 提示模板: 1
- analyze_weather: 分析天氣趨勢並給出建議
📝 生成的提示:
請分析廣州的天氣情況,並給出出行建議。包括:
1. 温度是否適宜
2. 是否需要帶傘
3. 穿衣建議
八、其他部分:MCP基礎協議的另一半
8.1 授權(Authorization)
MCP授權規範定義了基於HTTP傳輸的安全授權機制,使MCP客户端能夠代表資源所有者向受限制的MCP服務器發出請求。該規範基於OAuth 2.1及相關標準,實現了授權服務器發現、動態客户端註冊和訪問令牌管理。例如,客户端通過resource參數明確指定目標MCP服務器(如https://mcp.example.com),服務器則驗證令牌是否專門為其頒發,確保令牌不會被誤用於其他服務,從而防止"令牌傳遞"安全漏洞。
8.2 取消(Cancellation)
MCP取消機制允許通過通知消息中止正在進行的請求,任何一方都可以發送notifications/cancelled通知來終止先前發出的請求。例如,當用户取消長時間運行的操作時,客户端可以發送包含請求ID和可選原因的取消通知,接收方應停止處理、釋放資源且不發送響應。該機制考慮了網絡延遲導致的競態條件,允許接收方在請求已完成或無法取消時忽略通知,同時建議雙方記錄取消原因以便調試。
{
"method": "notifications/cancelled",
"params": {
"requestId": "123",
"reason": "用户取消"
}
}
8.3 Ping機制
MCP提供了可選的ping機制,允許任何一方驗證對方是否仍然響應且連接存活。該機制通過簡單的請求/響應模式實現,例如客户端發送{"jsonrpc":"2.0","id":"123","method":"ping"},服務器必須立即響應{"jsonrpc":"2.0","id":"123","result":{}}。如果在合理超時時間內未收到響應,發送方可以將連接視為陳舊並終止連接或嘗試重新連接。實現應定期發送ping以檢測連接健康狀況,但應避免過度ping以減少網絡開銷。
8.4 進度跟蹤(Progress)
MCP支持通過通知消息對長時間運行的操作進行可選的進度跟蹤。請求方可以在請求元數據中包含唯一的progressToken(如字符串"task123")來接收進度更新,接收方則可以發送包含進度值、可選總值和消息的notifications/progress通知。例如,文件上傳操作可以發送{"progress":50,"total":100,"message":"正在上傳文件..."}來指示完成百分比。進度值必須隨每個通知遞增,雙方應實現速率限制以防止消息氾濫,並在操作完成後停止發送進度通知。
{
"method": "notifications/progress",
"params": {
"progressToken": "task123",
"progress": 50, // 當前進度
"total": 100, // 總量
"message": "正在上傳文件..."
}
}
九、安全實踐:必須重視
9.1 核心原則
1. 用户同意優先
- 所有數據訪問必須經用户明確同意
- 所有工具調用前必須讓用户確認
2. 數據隱私保護
- 服務器只能看到必要的信息
- 完整對話歷史保留在宿主,不發給服務器
3. 工具安全
- 工具代表代碼執行,必須謹慎
- 顯示工具要做什麼,讓用户批准
4. 輸入驗證
- 服務器必須驗證所有輸入
- 客户端必須驗證工具返回的結果
9.2 實際建議
服務器開發者:
- 驗證所有輸入參數
- 實現訪問控制和速率限制
- 記錄操作日誌供審計
客户端開發者:
- 顯示清晰的權限請求界面
- 在調用工具前展示參數
- 實現工具調用超時機制
十、MCP生態:誰開發客户端?
關鍵認知: 在MCP生態中,客户端通常不是由下游開發者開發的,而是內置在AI應用平台中。
開發者開發MCP服務器
↓
配置到AI平台(Claude/Cursor等)
↓
AI平台內置的MCP客户端自動連接
對於軟件開發者來説,在MCP生態中的位置如下。
角色定位:
┌─────────────────────────────────────────┐
│ AI平台開發者(Anthropic, Cursor等) │
│ ──────────────────────────────── │
│ 職責: │
│ ✅ 開發MCP客户端SDK │
│ ✅ 集成到自己的AI應用中 │
│ ✅ 提供配置界面 │
│ ✅ 管理MCP服務器生命週期 │
│ ✅ 處理AI與MCP的交互邏輯 │
└─────────────────────────────────────────┘
↓ 提供平台
┌─────────────────────────────────────────┐
│ MCP服務器開發者(你、我、社區) │
│ ──────────────────────────────── │
│ 職責: │
│ ✅ 開發MCP服務器 │
│ ✅ 實現Resources/Tools/Prompts │
│ ✅ 編寫使用文檔 │
│ ✅ 發佈到npm/PyPI │
│ ❌ 不需要開發客户端 │
│ ❌ 不需要關心AI如何調用 │
└─────────────────────────────────────────┘
↓ 使用服務
┌─────────────────────────────────────────┐
│ 最終用户(開發者、分析師等) │
│ ──────────────────────────────── │
│ 職責: │
│ ✅ 安裝需要的MCP服務器 │
│ ✅ 配置到AI平台 │
│ ✅ 使用AI完成任務 │
│ ❌ 不需要寫代碼 │
└─────────────────────────────────────────┘
後記
MCP 讓 AI 應用開發變得更簡單、更安全、更強大。它不是銀彈,但為構建可靠的AI系統提供了堅實基礎。本文全部內容基於提示編寫,歡迎交流討論!
參考文獻
- MCP官方規範: https://modelcontextprotocol.io/specification/2025-06-18
- JSON-RPC 2.0: https://www.jsonrpc.org/