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

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

讀取和修改項(xiàng)目文件

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

搜索代碼倉(cāng)庫(kù)

執(zhí)行Git操作

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

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

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

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

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

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

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

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

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

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

管理所有客戶端

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

JSON-RPC 2.0消息格式

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

基本錯(cuò)誤處理

功能層(按需選擇):

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

Roots、Sampling、Elicitation(客戶端)

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

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

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

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

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

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

步驟2: 能力在初始化時(shí)協(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é)商完成后，雙方都知道對(duì)方支持什么功能，只使用交集部分。

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

步驟4: 具體內(nèi)容是個(gè)性化的
雖然方法名固定，但每個(gè)服務(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": "寫(xiě)入文件"},
    {"name": "search_files", "description": "搜索文件"}
  ]
}

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

客戶端如何知道服務(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)鍵點(diǎn):通過(guò)JSON Schema，客戶端知道如何正確調(diào)用工具，無(wú)需硬編碼。

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

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

4.1 JSON-RPC 2.0基礎(chǔ)

消息類(lèi)型

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

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

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

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

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

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

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

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

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

4.2 能力協(xié)商詳解

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

初始化流程詳解

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

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "initialize",
  "params": {
    "protocolVersion": "2024-11-05",  // 客戶端支持的協(xié)議版本
    "capabilities": {                  // 客戶端能力聲明
      "roots": {                       // 支持根目錄
        "listChanged": true            // 支持根目錄變更通知
      },
      "sampling": {},                  // 支持LLM采樣
      "elicitation": {},               // 支持用戶詢問(wèn)
      "experimental": {                // 實(shí)驗(yàn)性功能
        "customFeature": {}            // 自定義功能
      }
    },
    "clientInfo": {                    // 客戶端信息
      "name": "MyAIApp",               // 程序名(必填)
      "version": "1.2.0",              // 版本號(hào)(必填)
      "title": "我的AI應(yīng)用"             // 顯示名稱(chē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ù)庫(kù)訪問(wèn)能力"  // 可選的使用說(shuō)明
  }
}

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

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

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

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

實(shí)際示例:

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

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

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

能力交集計(jì)算

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

客戶端能力: {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ù)器不支持工具,跳過(guò)
    print("服務(wù)器不提供工具功能")

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

4.3 連接生命周期深入

完整的消息時(shí)序圖

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

	

	初始化前的限制

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

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

	? initialize請(qǐng)求

	? ping請(qǐng)求(用于?；?

	? 其他任何請(qǐng)求

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

	? initialize響應(yīng)

	? ping請(qǐng)求

	? logging通知(日志)

	? 其他任何消息

	違反限制的后果:

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


	

	超時(shí)和重試機(jī)制

	請(qǐng)求超時(shí):

	
import asyncio

# 設(shè)置30秒超時(shí)
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": "超時(shí)"}
    )


	

	進(jìn)度通知重置超時(shí):

	
# 當(dāng)收到進(jìn)度通知時(shí),可以重置超時(shí)計(jì)時(shí)器
timeout = 30  # 基礎(chǔ)超時(shí)
max_timeout = 300  # 最大超時(shí)(5分鐘)

while True:
    try:
        msg = await wait_for_message(timeout)
        if msg.method == "notifications/progress":
            # 收到進(jìn)度,重置超時(shí)
            timeout = 30
    except TimeoutError:
        # 超時(shí)處理
        break


	

	4.4 傳輸方式對(duì)比

	stdio傳輸詳解

	優(yōu)點(diǎn):

	? 簡(jiǎn)單直接,適合本地開(kāi)發(fā)

	? 進(jìn)程隔離,安全性好

	? 自動(dòng)管理生命周期

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

	缺點(diǎn):

	? 只能本地使用

	? 不支持多客戶端

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

	消息格式:

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


	

	每個(gè)JSON對(duì)象占一行,以n分隔。

	HTTP傳輸詳解

	架構(gòu):

	
┌─────────┐         HTTP POST         ┌─────────┐
│         ├──────────────────────────>│         │
│ 客戶端  │  請(qǐng)求/通知/響應(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


	

	會(huì)話管理:

	
# 服務(wù)器端設(shè)置會(huì)話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ù)請(qǐng)求攜帶會(huì)話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 實(shí)際通信示例

	讓我們看一個(gè)完整的SQLite查詢場(chǎng)景:

	
// 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"  // 請(qǐng)求進(jìn)度通知
    }
  }
}

// 3. 服務(wù)器發(fā)送進(jìn)度(異步通知)
服務(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個(gè)活躍用戶"
      }
    ],
    "isError": false
  }
}

// 5. 如果查詢出錯(cuò)
服務(wù)器 → 客戶端(錯(cuò)誤情況):
{
  "jsonrpc": "2.0",
  "id": 2,
  "error": {
    "code": -32603,
    "message": "SQL語(yǔ)法錯(cuò)誤",
    "data": {
      "sql": "SELECT COUNT(*) FROM users WHERE active = 1",
      "error": "near "WHERE": syntax error",
      "position": 35
    }
  }
}


	

	這就是MCP通信的完整過(guò)程!通過(guò)JSON-RPC 2.0，客戶端和服務(wù)器可以進(jìn)行結(jié)構(gòu)化、類(lèi)型安全的通信。

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

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

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

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

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

	如何使用:

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

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


	

	資源URI示例:

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

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

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

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

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

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

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


	

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

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

	什么是Prompt?

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

	為什么需要Prompt?

	場(chǎng)景1:沒(méi)有Prompt時(shí)

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


	

	場(chǎng)景2:有Prompt后

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


	

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

	定義一個(gè)Prompt:

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


	

	實(shí)際使用示例

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

	
// 客戶端請(qǐng)求
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "prompts/list"
}

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


審核編輯 黃宇