一、MCP是什么? 為什么需要它?

想象一下，你正在開發(fā)一個 AI 編程助手，它需要:

讀取和修改項目文件

查詢數(shù)據(jù)庫Schema

搜索代碼倉庫

執(zhí)行Git操作

傳統(tǒng)做法是為每個數(shù)據(jù)源寫一套專用代碼，不同團隊重復(fù)造輪子。Model Context Protocol(MCP) 就是為了解決這個問題而生的開放標(biāo)準(zhǔn)協(xié)議。

通俗理解: MCP就像是「AI應(yīng)用的USB接口標(biāo)準(zhǔn)」。就像USB讓不同設(shè)備都能接入電腦一樣，MCP讓不同的數(shù)據(jù)源和工具都能以統(tǒng)一方式接入AI應(yīng)用。

實際案例: 在Claude Desktop中，你可以配置多個官方MCP服務(wù)器:

Filesystem服務(wù)器: 安全地讀寫本地文件，有權(quán)限控制

SQLite服務(wù)器: 查詢和分析SQLite數(shù)據(jù)庫，自動生成SQL

GitHub服務(wù)器: 搜索倉庫、創(chuàng)建Issue、管理PR

你的AI應(yīng)用只需實現(xiàn)一個MCP客戶端，就能連接所有服務(wù)器，無需為每個服務(wù)器寫專用代碼。

二、架構(gòu)設(shè)計： 三個角色的分工

MCP采用宿主-客戶端-服務(wù)器三層架構(gòu)，就像一家公司的組織結(jié)構(gòu):

宿主(Host) = 總經(jīng)理

管理所有客戶端

控制安全策略和權(quán)限

負(fù)責(zé)AI模型的調(diào)用

客戶端(Client) = 部門經(jīng)理

客戶端負(fù)責(zé)連接服務(wù)器

負(fù)責(zé)雙方的溝通協(xié)調(diào)

轉(zhuǎn)發(fā)消息和通知

服務(wù)器(Server) = 業(yè)務(wù)專員

提供具體功能(資源、工具、提示模板)

可以是本地程序或遠(yuǎn)程服務(wù)

不知道其他服務(wù)器的存在

三、協(xié)議約定：統(tǒng)一規(guī)范與個性化擴展

每個MCP服務(wù)器提供的工具、資源都不一樣，但它們都遵循相同的MCP協(xié)議規(guī)范。

3.1 協(xié)議的分層設(shè)計

MCP采用 基礎(chǔ)協(xié)議 + 功能擴展 的設(shè)計，就像HTTP協(xié)議一樣:

核心層(所有實現(xiàn)必須支持):

JSON-RPC 2.0消息格式

初始化握手流程(initialize/initialized)

基本錯誤處理

功能層(按需選擇):

Resources、Prompts、Tools(服務(wù)器端)

Roots、Sampling、Elicitation(客戶端)

這樣設(shè)計的好處:

統(tǒng)一的基礎(chǔ)協(xié)議 → 保證互操作性
     +
靈活的功能選擇 → 滿足不同場景需求
     ↓
既標(biāo)準(zhǔn)化又可擴展

3.2 協(xié)議約定的過程

步驟1: 基礎(chǔ)協(xié)議是固定的
所有MCP服務(wù)器和客戶端都遵循相同的JSON-RPC 2.0格式:

// 請求格式(固定)
{
  "jsonrpc": "2.0",      // 必須是2.0
  "id": 1,                // 唯一標(biāo)識
  "method": "方法名",     // 要調(diào)用的方法
  "params": {...}         // 參數(shù)對象
}

// 響應(yīng)格式(固定)
{
  "jsonrpc": "2.0",
  "id": 1,                // 對應(yīng)請求的ID
  "result": {...}         // 成功結(jié)果
  // 或 "error": {...}    // 錯誤信息
}

步驟2: 能力在初始化時協(xié)商

// 客戶端發(fā)起初始化
{
  "method": "initialize",
  "params": {
    "protocolVersion": "2024-11-05",
    "capabilities": {
      "sampling": {},       // 我支持LLM采樣
      "roots": {}           // 我支持根目錄
    },
    "clientInfo": {"name": "MyClient", "version": "1.0"}
  }
}

// 服務(wù)器響應(yīng)
{
  "result": {
    "protocolVersion": "2024-11-05",
    "capabilities": {
      "tools": {},          // 我提供工具
      "resources": {}       // 我提供資源
    },
    "serverInfo": {"name": "SQLiteServer", "version": "2.0"}
  }
}

協(xié)商完成后，雙方都知道對方支持什么功能，只使用交集部分。

步驟3: 方法名稱是標(biāo)準(zhǔn)化的
MCP規(guī)范定義了標(biāo)準(zhǔn)方法名:

步驟4: 具體內(nèi)容是個性化的
雖然方法名固定，但每個服務(wù)器返回的具體數(shù)據(jù)不同:

// SQLite服務(wù)器的工具
{
  "tools": [
    {"name": "query", "description": "執(zhí)行SQL查詢"},
    {"name": "list_tables", "description": "列出所有表"}
  ]
}

// Filesystem服務(wù)器的工具
{
  "tools": [
    {"name": "read_file", "description": "讀取文件"},
    {"name": "write_file", "description": "寫入文件"},
    {"name": "search_files", "description": "搜索文件"}
  ]
}

3.3 協(xié)議發(fā)現(xiàn)機制

客戶端如何知道服務(wù)器有哪些工具?

第一步：列舉

客戶端 → 服務(wù)器: {"method": "tools/list"}
服務(wù)器 → 客戶端: {
  "tools": [
    {
      "name": "query",
      "description": "執(zhí)行SQL查詢",
      "inputSchema": {           // JSON Schema定義輸入格式
        "type": "object",
        "properties": {
          "sql": {"type": "string"}
        }
      }
    }
  ]
}

第二步：調(diào)用

客戶端 → 服務(wù)器: {
  "method": "tools/call",
  "params": {
    "name": "query",           // 使用第一步獲得的工具名
    "arguments": {"sql": "SELECT * FROM users"}
  }
}

關(guān)鍵點:通過JSON Schema，客戶端知道如何正確調(diào)用工具，無需硬編碼。

四、協(xié)議基礎(chǔ)：如何通信?

MCP基于JSON-RPC 2.0構(gòu)建，這是一個成熟的遠(yuǎn)程過程調(diào)用協(xié)議。理解這一層對掌握MCP至關(guān)重要。

4.1 JSON-RPC 2.0基礎(chǔ)

消息類型

MCP中有三種基本消息類型。
1. 請求(Request) - 期待響應(yīng)

{
  "jsonrpc": "2.0",           // 協(xié)議版本,必須是"2.0"
  "id": 1,                     // 請求唯一標(biāo)識(字符串或數(shù)字)
  "method": "tools/list",     // 要調(diào)用的方法名
  "params": {                  // 可選的參數(shù)對象
    "cursor": "page2"
  }
}

2. 響應(yīng)(Response) - 對請求的回復(fù)

// 成功響應(yīng)
{
  "jsonrpc": "2.0",
  "id": 1,                     // 必須與請求的id相同
  "result": {                  // 成功結(jié)果
    "tools": [
      {"name": "query", "description": "執(zhí)行查詢"}
    ]
  }
}

// 錯誤響應(yīng)
{
  "jsonrpc": "2.0",
  "id": 1,
  "error": {                   // 錯誤對象
    "code": -32602,            // 錯誤碼(整數(shù))
    "message": "參數(shù)無效",      // 錯誤描述
    "data": {                  // 可選的額外信息
      "field": "cursor",
      "reason": "格式錯誤"
    }
  }
}

3. 通知(Notification) - 單向消息,無需響應(yīng)

{
  "jsonrpc": "2.0",
  "method": "notifications/resources/updated",  // 通知方法名
  "params": {                                   // 通知參數(shù)
    "uri": "file:///project/data.json"
  }
  // 注意:沒有id字段
}

標(biāo)準(zhǔn)錯誤碼

MCP使用JSON-RPC 2.0的標(biāo)準(zhǔn)錯誤碼:

4.2 能力協(xié)商詳解

能力協(xié)商是MCP連接建立的第一步，決定了整個會話中可用的功能。

初始化流程詳解

階段1: 客戶端發(fā)起初始化

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "initialize",
  "params": {
    "protocolVersion": "2024-11-05",  // 客戶端支持的協(xié)議版本
    "capabilities": {                  // 客戶端能力聲明
      "roots": {                       // 支持根目錄
        "listChanged": true            // 支持根目錄變更通知
      },
      "sampling": {},                  // 支持LLM采樣
      "elicitation": {},               // 支持用戶詢問
      "experimental": {                // 實驗性功能
        "customFeature": {}            // 自定義功能
      }
    },
    "clientInfo": {                    // 客戶端信息
      "name": "MyAIApp",               // 程序名(必填)
      "version": "1.2.0",              // 版本號(必填)
      "title": "我的AI應(yīng)用"             // 顯示名稱(可選)
    }
  }
}

階段2: 服務(wù)器響應(yīng)能力

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "protocolVersion": "2024-11-05",  // 服務(wù)器選擇的協(xié)議版本
    "capabilities": {                  // 服務(wù)器能力聲明
      "resources": {                   // 支持資源
        "subscribe": true,             // 支持資源訂閱
        "listChanged": true            // 支持資源列表變更通知
      },
      "tools": {                       // 支持工具
        "listChanged": true
      },
      "prompts": {                     // 支持提示模板
        "listChanged": false           // 不支持列表變更通知
      },
      "logging": {}                    // 支持日志輸出
    },
    "serverInfo": {                    // 服務(wù)器信息
      "name": "sqlite-mcp-server",
      "version": "2.1.0",
      "title": "SQLite MCP服務(wù)器"
    },
    "instructions": "此服務(wù)器提供SQLite數(shù)據(jù)庫訪問能力"  // 可選的使用說明
  }
}

階段3: 客戶端確認(rèn)就緒

{
  "jsonrpc": "2.0",
  "method": "notifications/initialized"  // 無id,這是通知
}

協(xié)議版本協(xié)商規(guī)則

客戶端請求版本: "2024-11-05"
         ↓
    服務(wù)器支持?
    ↙        ↘
  支持        不支持
   ↓            ↓
返回相同版本  返回服務(wù)器支持的最新版本
   ↓            ↓
協(xié)商成功    客戶端檢查是否支持
              ↙        ↘
           支持        不支持
            ↓            ↓
         協(xié)商成功     斷開連接

實際示例:

// 場景1: 版本匹配
客戶端: "protocolVersion": "2024-11-05"
服務(wù)器: "protocolVersion": "2024-11-05"  ? 成功

// 場景2: 服務(wù)器版本更新
客戶端: "protocolVersion": "2024-06-01"
服務(wù)器: "protocolVersion": "2024-11-05"  
→ 客戶端檢查是否支持2024-11-05 → 如果不支持則斷開

// 場景3: 客戶端版本更新
客戶端: "protocolVersion": "2025-01-01"
服務(wù)器: "protocolVersion": "2024-11-05"  
→ 客戶端檢查是否支持2024-11-05 → 如果支持則降級使用

能力交集計算

初始化后,雙方只能使用共同支持的能力:

客戶端能力: {roots, sampling, elicitation}
服務(wù)器能力: {resources, tools, prompts}
         ↓
   可用功能集合
   ├─ 客戶端 → 服務(wù)器: resources, tools, prompts
   └─ 服務(wù)器 → 客戶端: roots, sampling, elicitation

示例:

# 客戶端代碼示例
if server_capabilities.get("tools"):
    # 服務(wù)器支持工具,可以調(diào)用
    tools = await session.list_tools()
else:
    # 服務(wù)器不支持工具,跳過
    print("服務(wù)器不提供工具功能")

if client_capabilities.get("sampling"):
    # 客戶端支持采樣,服務(wù)器可以請求
    # (服務(wù)器端會檢查這個能力)
    pass

4.3 連接生命周期深入

完整的消息時序圖

客戶端                                            服務(wù)器
  │                                              │
  │  1. initialize (請求)                         │
  ├──────────────────────────────────────>│
  │     {protocolVersion, capabilities}          │
  │                                              │
  │  2. initialize (響應(yīng))                         │
  ││
  │                                              │
  │═══════════ 正常操作階段 ════════════        │
  │                                              │
  │  4. tools/list (請求)                         │
  ├──────────────────────────────────────>│
  │                                              │
  │  5. tools/list (響應(yīng))                         │
  ││
  │     {name: "query", arguments: {...}}        │
  │                                              │
  │  7. notifications/progress (通知)             │
  │

	

	初始化前的限制

	在initialized通知發(fā)送前:

	客戶端只能發(fā)送:

	? initialize請求

	? ping請求(用于?；?

	? 其他任何請求

	服務(wù)器只能發(fā)送:

	? initialize響應(yīng)

	? ping請求

	? logging通知(日志)

	? 其他任何消息

	違反限制的后果:

	
// 客戶端在初始化前調(diào)用tools/list
請求: {"method": "tools/list"}
響應(yīng): {
  "error": {
    "code": -32600,
    "message": "會話未初始化"
  }
}


	

	超時和重試機制

	請求超時:

	
import asyncio

# 設(shè)置30秒超時
try:
    result = await asyncio.wait_for(
        session.call_tool("slow_operation", {}),
        timeout=30.0
    )
except asyncio.TimeoutError:
    # 發(fā)送取消通知
    await session.send_notification(
        "notifications/cancelled",
        {"requestId": "123", "reason": "超時"}
    )


	

	進度通知重置超時:

	
# 當(dāng)收到進度通知時,可以重置超時計時器
timeout = 30  # 基礎(chǔ)超時
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傳輸詳解

	優(yōu)點:

	? 簡單直接,適合本地開發(fā)

	? 進程隔離,安全性好

	? 自動管理生命周期

	? 無需網(wǎng)絡(luò)配置

	缺點:

	? 只能本地使用

	? 不支持多客戶端

	? 調(diào)試相對困難

	消息格式:

	
消息1n
消息2n
消息3n


	

	每個JSON對象占一行,以n分隔。

	HTTP傳輸詳解

	架構(gòu):

	
┌─────────┐         HTTP POST         ┌─────────┐
│         ├──────────────────────────>│         │
│ 客戶端  │  請求/通知/響應(yīng)(JSON-RPC) │ 服務(wù)器  │
│         │

	

	發(fā)送消息(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"}


	

	立即響應(yīng)(JSON):

	
HTTP/1.1 200 OK
Content-Type: application/json

{"jsonrpc":"2.0","id":1,"result":{"tools":[...]}}


	

	流式響應(yīng)(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":[...]}}


	

	接收服務(wù)器消息(GET):

	
GET /mcp HTTP/1.1
Host: localhost:8080
Accept: text/event-stream
Mcp-Session-Id: abc123
Last-Event-ID: 42


	

	會話管理:

	
# 服務(wù)器端設(shè)置會話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}
        )

# 客戶端后續(xù)請求攜帶會話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. 列出工具
客戶端 → 服務(wù)器:
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/list"
}

服務(wù)器 → 客戶端:
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "tools": [
      {
        "name": "query",
        "description": "執(zhí)行SQL查詢",
        "inputSchema": {
          "type": "object",
          "properties": {
            "sql": {"type": "string"}
          },
          "required": ["sql"]
        }
      }
    ]
  }
}

// 2. 調(diào)用查詢工具
客戶端 → 服務(wù)器:
{
  "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. 服務(wù)器發(fā)送進度(異步通知)
服務(wù)器 → 客戶端:
{
  "jsonrpc": "2.0",
  "method": "notifications/progress",
  "params": {
    "progressToken": "query-123",
    "progress": 50,
    "total": 100,
    "message": "正在掃描users表..."
  }
}

// 4. 返回查詢結(jié)果
服務(wù)器 → 客戶端:
{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "查詢結(jié)果: 1,234個活躍用戶"
      }
    ],
    "isError": false
  }
}

// 5. 如果查詢出錯
服務(wù)器 → 客戶端(錯誤情況):
{
  "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，客戶端和服務(wù)器可以進行結(jié)構(gòu)化、類型安全的通信。

	五、服務(wù)器能力：三種核心功能

	MCP服務(wù)器可以提供三種功能。

	5.1 Resources(資源)：應(yīng)用決定用什么

	資源就是數(shù)據(jù)，比如文件內(nèi)容、數(shù)據(jù)庫記錄、API響應(yīng)。

	誰控制: 應(yīng)用程序決定把哪些資源提供給AI

	如何使用:

	
// 列出所有可用資源
{"method": "resources/list"}

// 讀取某個資源
{
  "method": "resources/read",
  "params": {"uri": "file:///project/main.py"}
}


	

	資源URI示例:

	file:///project/src/main.py - 文件

	db://schema/users - 數(shù)據(jù)庫表結(jié)構(gòu)

	git://commits/main - Git提交歷史

	https://api.example.com/data - Web API

	訂閱變更: 可以訂閱資源,當(dāng)它變化時自動收到通知。

	實際案例: Filesystem服務(wù)器暴露資源

	
{
  "uri": "file:///Users/alice/project/src/main.py",  // Python源文件
  "name": "main.py",                                  // 文件名
  "mimeType": "text/x-python",                        // 文件類型
  "text": "import osndef main()..."                  // 文件內(nèi)容
}


	

	客戶端AI可以讀取這個資源，理解代碼結(jié)構(gòu)后提供重構(gòu)建議或生成測試。

	5.2 Prompts(提示模板)：用戶選擇用什么

	什么是Prompt?

	Prompt就像是「對話模板」或「快捷指令」，把常用的復(fù)雜指令預(yù)設(shè)好，用戶一鍵調(diào)用。用生活中的例子類比，就像微信的「快捷回復(fù)」或IDE中的「代碼片段(Snippet)」。

	為什么需要Prompt?

	場景1:沒有Prompt時

	
用戶每次都要輸入:
"請分析這個Git倉庫最近一周的提交,統(tǒng)計:
1. 總提交次數(shù)
2. 每個作者的貢獻
3. 修改的主要文件
4. 是否有破壞性變更
請用表格格式輸出"


	

	場景2:有Prompt后

	
用戶只需:
1. 點擊 "/analyze-commits" 命令
2. 選擇分支 "main"
3. AI自動執(zhí)行完整分析


	

	Prompt的數(shù)據(jù)結(jié)構(gòu)

	定義一個Prompt:

	
{
  "name": "analyze_commits",              // Prompt的唯一標(biāo)識
  "title": "提交歷史分析",                  // 用戶界面顯示的名稱
  "description": "分析Git提交并生成報告",    // 功能說明
  "arguments": [                          // 需要的參數(shù)列表
    {
      "name": "branch",                   // 參數(shù)名
      "description": "要分析的分支名",      // 參數(shù)說明
      "required": true                    // 是否必填
    },
    {
      "name": "since",                    // 時間范圍
      "description": "起始日期(如:7 days ago)",
      "required": false                   // 可選參數(shù)
    },
    {
      "name": "author",                   // 作者過濾
      "description": "只看某個作者的提交",
      "required": false
    }
  ]
}


	

	實際使用示例

	步驟1: 列出所有可用的Prompt

	
// 客戶端請求
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "prompts/list"
}

// 服務(wù)器響應(yīng)
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "prompts": [
      {
        "name": "analyze_commits",
        "title": "


審核編輯 黃宇