博客 / 詳情

返回

最新MCP規範解讀,看這篇就夠了!

一、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/promptref/resource兩種引用類型的補全,每次最多返回100個按相關性排序的建議值,並可通過completion/complete請求獲取補全結果。

日誌

MCP提供結構化日誌消息傳遞機制,允許服務器向客户端發送包含嚴重性級別、可選記錄器名稱和任意JSON可序列化數據的日誌通知。服務器需聲明logging能力,支持遵循RFC 5424標準的日誌級別(從debug到emergency),客户端可通過logging/setLevel請求配置最低日誌級別,服務器通過notifications/message通知發送日誌消息。

分頁

MCP支持對可能返回大量結果集的列表操作進行分頁處理,使用基於不透明遊標的分頁模型而非數字頁碼。服務器在響應中包含當前頁結果和可選的nextCursor字段(表示更多結果存在),客户端可通過在請求中包含遊標繼續分頁。支持分頁的操作包括resources/listresources/templates/listprompts/listtools/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系統提供了堅實基礎。本文全部內容基於提示編寫,歡迎交流討論!

參考文獻

  1. MCP官方規範: https://modelcontextprotocol.io/specification/2025-06-18
  2. JSON-RPC 2.0: https://www.jsonrpc.org/
user avatar u_16213461 頭像 ftkj_2018 頭像 hezuidekuaizi 頭像 lzfhope 頭像 xingxingshangdesigua 頭像 edonsoft 頭像 u_16018702 頭像
7 位用戶收藏了這個故事!

發佈 評論

Some HTML is okay.