從零用 Python 實現最基礎的MCP協議 原創
?模型上下文協議（MCP, Model Context Protocol）是 Anthropic 在2024年提出的一種開放標準協議，用于標準化 AI 模型與外部工具和數據源的集成方式。
可以將 MCP 類比為 AI 世界中的 “USB-C 接口”：它為大型語言模型（LLM）連接各種工具和數據源提供了一種統一的方法。
MCP 使用 JSON-RPC 2.0 作為消息格式，在客戶端和服務器之間傳遞請求和響應。
本示例將展示如何使用 Python 實現一個最基礎的 MCP 協議，包括 MCP 服務器和 MCP 客戶端兩部分。
我們將支持
- discovery（發現）
- invoke（調用）
- retrieve（獲取）
等基本操作，并通過一個簡單的“計算器”工具（支持加法和乘法）演示協議的工作原理。
背景和設計概述
MCP 協議采用 ??客戶端-服務器?? 架構。
MCP服務器提供一組工具（tools）或資源（resources），MCP客戶端可以發現服務器提供的功能并進行調用 。
兩者之間通過JSON-RPC進行通信，以標準的請求/響應消息交換指令和數據。
按照 MCP 規范：
??發現（Discovery）???：客戶端能夠查詢服務器，獲取其提供的工具列表、資源列表等。這通常通過調用 ??tools/list??? 或 ??resources/list?? 等方法實現。
??調用（Invoke）???：客戶端可以請求執行服務器上的某個工具功能，例如調用計算器的加法或乘法操作。規范中約定使用 ??tools/call?? 方法來調用指定名稱的工具，并傳遞所需參數。
??獲?。≧etrieve）???：客戶端能夠檢索數據內容，例如獲取某個資源的具體內容。規范中提供了如 ??resources/read?? 等方法用于檢索資源內容。
在簡單工具調用場景下，調用的結果會直接作為響應返回；但對于長任務或資源內容，常采用單獨的檢索步驟獲取結果。
實現思路
我們將使用 Python 標準庫構建一個簡易的 MCP 服務器和客戶端。
服務器會注冊至少一個工具——一個簡單計算器，包含加法和乘法功能。
客戶端可以通過發現操作獲取服務器上可用的工具列表，并通過調用操作執行這些工具函數并得到結果。
為了盡量減少依賴并簡化部署，我們將直接使用 ??Python??? 自帶的 ??HTTPServer??? 來處理 ??JSON-RPC??? 請求，使用內置的 ??json??? 和 ??http.client?? 模塊實現通信。
代碼結構將力求清晰，方便日后擴展更多功能（例如增加新的工具、資源或更復雜的異步處理）。
主要功能模塊：
MCPServer
處理客戶端請求的服務器類。
維護可用的工具和資源列表，能夠響應 ??tools/list???、??tools/call???、??resources/list???、??resources/read??? 等 ??JSON-RPC?? 方法。
MCPClient
客戶端類或腳本，用于連接MCP服務器發送??JSON-RPC??請求。支持發現工具列表、調用指定工具、（可選）獲取資源內容等操作。
通信格式
遵循??JSON-RPC 2.0???規范，每個請求包含 ??jsonrpc??? 版本、請求 id、方法名和參數，響應則包含對應的 id以及結果或錯誤信息。例如，列出工具的請求方法為 ??tools/list??；
調用工具的請求方法為 ??tools/call??，需要提供工具名稱和參數。
本示例中，我們的實現將返回簡化的結果格式，以便直觀展示功能（實際MCP可能返回更結構化的內容，如帶有類型說明的內容列表）。
下面我們分別給出 MCP 服務器和客戶端的代碼實現，并包含必要的注釋和使用說明。
MCP服務器實現 (Server)
首先，實現??MCP???服務器。我們創建一個 ??MCPServer?? 類用于注冊工具和資源，并處理傳入的請求。
然后使用 ??Python??? 的 ??BaseHTTPRequestHandler??? 來構建HTTP接口，使服務器能夠通過HTTP接收 ??JSON-RPC?? 請求。
計算器工具將作為示例工具注冊到服務器上。
import json
from http.server import BaseHTTPRequestHandler, HTTPServer
# 定義一個Tool類，用于存儲工具的元數據和執行函數
class Tool:
def __init__(self, name, description, input_schema, output_schema, func):
self.name = name
self.description = description
self.input_schema = input_schema # JSON Schema for input parameters
self.output_schema = output_schema # JSON Schema for output
self.func = func # Function to execute the tool
# MCP服務器類，維護工具和資源，并處理JSON-RPC請求
class MCPServer:
def __init__(self):
# 注冊工具列表：這里添加一個“計算器”工具，提供加法和乘法功能
def add(a, b): # 簡單加法函數
return a + b
def multiply(a, b): # 簡單乘法函數
return a * b
# 為演示，將加法和乘法作為兩個獨立工具注冊
self.tools = {
"add": Tool(
name="add",
descriptinotallow="Add two numbers",
input_schema={ # 輸入參數的JSON Schema定義
"type": "object",
"properties": {
"a": {"type": "number", "description": "First operand"},
"b": {"type": "number", "description": "Second operand"}
},
"required": ["a", "b"]
},
output_schema={ # 輸出結果的JSON Schema定義
"type": "number"
},
func=add
),
"multiply": Tool(
name="multiply",
descriptinotallow="Multiply two numbers",
input_schema={
"type": "object",
"properties": {
"a": {"type": "number", "description": "First operand"},
"b": {"type": "number", "description": "Second operand"}
},
"required": ["a", "b"]
},
output_schema={
"type": "number"
},
func=multiply
)
}
# 可選：注冊資源列表，這里添加一個簡單文本資源示例
self.resources = {
"mem://greeting": {
"name": "greeting.txt",
"title": "Greeting Message",
"description": "A welcome text message",
"mimeType": "text/plain",
"text": "Hello, welcome to the MCP demo!" # 資源內容
}
}
def handle_request(self, request):
"""
處理JSON-RPC請求，根據method調用對應的處理邏輯。
返回JSON可序列化的響應字典。
"""
# 基礎的JSON-RPC字段解析
jsonrpc = request.get("jsonrpc")
req_id = request.get("id")
method = request.get("method")
params = request.get("params", {})
# 確保符合 JSON-RPC 2.0 協議
if jsonrpc != "2.0" or req_id is None or method is None:
# 返回 JSON-RPC 錯誤：-32600 = Invalid Request
return {
"jsonrpc": "2.0",
"id": req_id,
"error": {"code": -32600, "message": "Invalid Request"}
}
# 處理方法：tools/list, tools/call, resources/list, resources/read
try:
if method == "tools/list":
# 列出可用工具
tools_info = []
for tool in self.tools.values():
tools_info.append({
"name": tool.name,
"description": tool.description,
"inputSchema": tool.input_schema,
"outputSchema": tool.output_schema
})
return {
"jsonrpc": "2.0",
"id": req_id,
"result": {
"tools": tools_info
}
}
elif method == "tools/call":
# 調用指定名稱的工具
name = params.get("name")
arguments = params.get("arguments", {})
if not name or name not in self.tools:
# 工具名稱不存在
return {
"jsonrpc": "2.0",
"id": req_id,
"error": {"code": -32601, "message": f"Tool '{name}' not found"}
}
# 獲取工具并執行
tool = self.tools[name]
try:
# 調用工具函數，傳入參數
result_value = tool.func(**arguments)
except TypeError as e:
# 參數不正確或缺失
return {
"jsonrpc": "2.0",
"id": req_id,
"error": {"code": -32602, "message": f"Invalid parameters: {str(e)}"}
}
except Exception as e:
# 工具執行過程中出現其他異常
return {
"jsonrpc": "2.0",
"id": req_id,
"error": {"code": -32000, "message": f"Tool execution error: {str(e)}"}
}
# 返回執行結果（這里結果直接是數值，符合 outputSchema 定義）
return {
"jsonrpc": "2.0",
"id": req_id,
"result": {
"output": result_value # 將結果包裝在一個字段中
}
}
elif method == "resources/list":
# 列出可用資源
resources_info = []
for uri, res in self.resources.items():
resources_info.append({
"uri": uri,
"name": res.get("name"),
"title": res.get("title"),
"description": res.get("description"),
"mimeType": res.get("mimeType")
})
return {
"jsonrpc": "2.0",
"id": req_id,
"result": {
"resources": resources_info
}
}
elif method == "resources/read":
# 讀取指定URI的資源內容
uri = params.get("uri")
if not uri or uri not in self.resources:
return {
"jsonrpc": "2.0",
"id": req_id,
"error": {"code": -32602, "message": f"Invalid resource URI: {uri}"}
}
res = self.resources[uri]
# 返回資源內容（文本）
return {
"jsonrpc": "2.0",
"id": req_id,
"result": {
"contents": [
{
"uri": uri,
"name": res.get("name"),
"title": res.get("title"),
"mimeType": res.get("mimeType"),
"text": res.get("text")
}
]
}
}
else:
# 未知的方法
return {
"jsonrpc": "2.0",
"id": req_id,
"error": {"code": -32601, "message": f"Method '{method}' not found"}
}
except Exception as e:
# 捕獲任何未預料的異常，返回內部錯誤
return {
"jsonrpc": "2.0",
"id": req_id,
"error": {"code": -32603, "message": f"Internal error: {str(e)}"}
}
# 實例化 MCPServer，全局使用
mcp_server = MCPServer()
# 定義HTTP請求處理器類
class MCPRequestHandler(BaseHTTPRequestHandler):
def do_POST(self):
# 讀取請求體中的JSON數據
content_length = int(self.headers.get('Content-Length', 0))
request_bytes = self.rfile.read(content_length)
try:
request_json = json.loads(request_bytes.decode('utf-8'))
except json.JSONDecodeError:
# 如果不是合法的JSON，返回錯誤
response = {
"jsonrpc": "2.0",
"id": None,
"error": {"code": -32700, "message": "Parse error"}
}
self._send_response(response)
return
# 調用MCPServer處理請求，獲取響應
response = mcp_server.handle_request(request_json)
# 將響應發送回客戶端
self._send_response(response)
def _send_response(self, response_obj):
"""輔助方法：發送JSON響應。"""
response_bytes = json.dumps(response_obj).encode('utf-8')
# 構造HTTP響應頭
self.send_response(200)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(response_bytes)))
self.end_headers()
# 寫出響應內容
self.wfile.write(response_bytes)
self.wfile.flush()
# 啟動 HTTP 服務器（監聽 localhost:8000）
if __name__ == "__main__":
server_address = ('', 8000) # 監聽在所有接口的8000端口
httpd = HTTPServer(server_address, MCPRequestHandler)
print("MCP Server is running at http://localhost:8000")
print("Available tools:", [name for name in mcp_server.tools.keys()])
httpd.serve_forever()上面的服務器代碼說明：
工具注冊：
在 MCPServer.init 中，我們定義了兩個簡單的算術函數 add 和 multiply，并使用它們分別注冊了名為 "add" 和 "multiply" 的 Tool。
每個 Tool 包含名稱、描述、輸入參數模式 (input_schema)、輸出結果模式 (output_schema) 以及執行函數 func。
在輸入/輸出模式中，我們使用了簡化的 JSON Schema 來描述參數（要求提供兩個數字參數 a 和 b）和返回值類型（數字）。
這些元數據會在工具發現時提供給客戶端，方便客戶端或模型了解如何調用該工具。
資源注冊：
為了演示 retrieve 操作，我們可選地注冊了一個簡單資源 mem://greeting，表示一段文本內容（模擬例如文件或數據庫中的數據）。該資源包含一個 URI 標識符和元數據（名稱、描述、MIME類型等），以及文本內容 "Hello, welcome to the MCP demo!"。資源將用于演示客戶端如何列出并讀取服務器上的數據內容。
請求處理：
MCPServer.handle_request 方法根據傳入請求的 method 字段執行
相應邏輯：
- tools/list：返回服務器上所有可用工具的列表，包含每個工具的名稱、描述、輸入/輸出模式等信息。
- tools/call：調用指定的工具。請求需提供參數 name（工具名）和 arguments（參數字典）。服務器檢查工具是否存在，參數是否正確，然后調用對應的函數并獲取結果。
結果在此示例中直接返回為一個簡單數值，包裝在 ??result.output??? 字段中。如出現錯誤（未知工具、參數錯誤、執行異常等），則返回 ??JSON-RPC?? 標準的 error 對象，包括錯誤碼和信息。
resources/list：
返回可用資源列表，每項包含資源的 URI、名稱、描述等元數據。
resources/read：
根據傳入的 uri 參數，返回對應資源的內容。
在響應中，我們將內容放在 contents 列表中，每個元素包含資源的 meta 信息以及實際內容文本。
未知方法：
如果收到非上述定義的方法，返回 JSON-RPC 的 “Method not found” 錯誤（錯誤碼 -32601）。
HTTP服務器：
使用 ??BaseHTTPRequestHandler??? 實現HTTP接口。我們重載了 ??do_POST?? 方法來處理 HTTP POST 請求（JSON-RPC 通常使用 POST 來發送請求）。
服務器從請求體讀取 JSON 數據并解析，然后調用?? mcp_server.handle_request??? 獲得結果，最后通過 ??_send_response?? 返回 JSON 響應。
??HTTPServer??? 監聽在 ??localhost:8000?? 端口，啟動后打印可用工具列表以供參考。
注意，本實現為了簡潔未包含認證、權限控制等機制，也未實現 MCP 規范中的會話管理、通知推送等高級功能。這是一個最基礎的示例，展示了 JSON-RPC 通信和工具/資源調用的基本流程。在實際應用中，可能需要擴展支持例如能力協商（capability negotiation）、異步調用（如長時間運行任務的進度和結果獲取）、安全認證等特性。
MCP客戶端實現 (Client)
接下來，實現與上述服務器交互的 MCP 客戶端?？蛻舳藢凑枕樞蜓菔疽韵虏僮鳎?/p>
- 發現工具：請求獲取服務器提供的工具列表（調用 tools/list）。
- 調用工具：調用加法和乘法工具（分別調用 tools/call，傳入相應參數）。
- 發現資源：請求獲取服務器提供的資源列表（調用 resources/list，可選步驟）。
- 獲取資源內容：讀取指定資源內容（調用 resources/read，示例讀取 mem://greeting 文本）。
我們使用 Python 內置的 ??http.client??? 模塊發送 ??HTTP POST?? 請求，獲取并解析 JSON 響應。以下是客戶端代碼：
import json
import http.client
# MCP客戶端輔助函數：發送JSON-RPC請求并返回解析后的結果或錯誤
def send_json_rpc(method, params=None, request_id=1):
conn = http.client.HTTPConnection("localhost", 8000)
request_obj = {
"jsonrpc": "2.0",
"id": request_id,
"method": method,
"params": params or {}
}
# 將請求對象序列化為JSON并發送
conn.request("POST", "/", body=json.dumps(request_obj), headers={
"Content-Type": "application/json"
})
# 獲取HTTP響應并解析JSON
response = conn.getresponse()
data = response.read().decode('utf-8')
conn.close()
try:
return json.loads(data)
except json.JSONDecodeError:
print("Invalid JSON response:", data)
return None
# 1. 發現可用工具列表
resp = send_json_rpc("tools/list", {}, request_id=1)
print("Available tools:")
if "result" in resp:
for tool in resp["result"]["tools"]:
name = tool["name"]
desc = tool.get("description")
print(f" - {name}: {desc}")
else:
# 錯誤情況
print("Error listing tools:", resp.get("error"))
# 2. 調用加法工具 (例如 3 + 5)
add_params = {"name": "add", "arguments": {"a": 3, "b": 5}}
resp2 = send_json_rpc("tools/call", add_params, request_id=2)
if "result" in resp2:
result_value = resp2["result"].get("output")
print(f"Result of add(3, 5): {result_value}")
else:
print("Error invoking add tool:", resp2.get("error"))
# 3. 調用乘法工具 (例如 4 * 7)
mul_params = {"name": "multiply", "arguments": {"a": 4, "b": 7}}
resp3 = send_json_rpc("tools/call", mul_params, request_id=3)
if "result" in resp3:
result_value = resp3["result"].get("output")
print(f"Result of multiply(4, 7): {result_value}")
else:
print("Error invoking multiply tool:", resp3.get("error"))
# 4. （可選）列出可用資源
resp4 = send_json_rpc("resources/list", {}, request_id=4)
print("Available resources:")
if "result" in resp4:
for res in resp4["result"]["resources"]:
uri = res["uri"]
desc = res.get("description")
print(f" - {uri}: {desc}")
else:
print("Error listing resources:", resp4.get("error"))
# 5. （可選）讀取指定資源內容 (例如 mem://greeting)
res_params = {"uri": "mem://greeting"}
resp5 = send_json_rpc("resources/read", res_params, request_id=5)
if "result" in resp5:
contents = resp5["result"].get("contents", [])
if contents:
text = contents[0].get("text")
print(f"Content of resource mem://greeting: {text}")
else:
print("Error reading resource:", resp5.get("error"))說明：
- send_json_rpc 函數封裝了發送請求和接收響應的過程。它建立HTTP連接到 localhost:8000，發送JSON格式的RPC請求，并返回解析后的Python字典對象。
- 客戶端按照步驟構造請求：
- tools/list 請求沒有額外參數，返回服務器可用工具列表。在示例中，服務器應返回我們注冊的 "add" 和 "multiply" 工具，各自的描述和參數模式也包含在結果中。
- tools/call 請求加法工具，提供參數 {a:3, b:5}。服務器返回結果，客戶端從響應中提取 output 字段即計算結果（期望為8）。類似地調用乘法工具 4*7，應得到28。
- resources/list 請求返回服務器可用資源列表。在我們的示例中，應當包含 mem://greeting 這個資源及其描述信息。
- resources/read 請求讀取 mem://greeting，服務器將返回其內容文本?？蛻舳舜蛴〕鲈撐谋緝热荨?/li>
在打印輸出中，我們對結果做了簡單的格式化。例如：
Available tools:
- add: Add two numbers
- multiply: Multiply two numbers
Result of add(3, 5): 8
Result of multiply(4, 7): 28
Available resources:
- mem://greeting: A welcome text message
Content of resource mem://greeting: Hello, welcome to the MCP demo!這樣，我們就完成了一個基本的 MCP 協議交互流程：客戶端發現服務器的功能，并成功調用工具得到計算結果，還演示了資源的發現和獲取。整個通信過程采用 JSON-RPC 格式，確保了請求和響應的標準化
使用說明
要運行該示例，請按照以下步驟操作：
- 啟動服務器：將上述服務器代碼保存為?
?mcp_server.py?? 并運行。例如:
python mcp_server.py服務器將啟動并監聽端口 8000。在控制臺上可以看到啟動日志，例如可用的工具列表打印:
MCP Server is running at http://localhost:8000
Available tools: ['add', 'multiply']- 運行客戶端：在服務器運行的同時，打開另一個終端窗口，將上述客戶端代碼保存為 mcp_client.py 并運行:
python mcp_client.py客戶端將依次發送 JSON-RPC 請求到服務器，并在終端打印收到的響應結果。
觀察結果：你應當在客戶端終端看到類似輸出：
Available tools:
- add: Add two numbers
- multiply: Multiply two numbers
Result of add(3, 5): 8
Result of multiply(4, 7): 28
Available resources:
- mem://greeting: A welcome text message
Content of resource mem://greeting: Hello, welcome to the MCP demo!這些輸出對應了我們在客戶端代碼中的打印語句，驗證了各項操作成功執行。
交互驗證：你也可以使用其他JSON-RPC客戶端（例如cURL或Postman）手動發送請求進行測試。例如，使用 cURL 獲取工具列表：
curl -X POST http://localhost:8000/ -H "Content-Type: application/json" \
-d '{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}'你將收到JSON響應，包含已注冊工具的詳細信息。
擴展與總結
本示例提供了一個最基礎的 MCP 協議實現框架。通過清晰的結構設計，擴展該實現非常方便：
增加新工具：
可以在服務器的 self.tools 字典中添加新的 Tool 實例，并實現對應的函數。例如，可以添加一個字符串處理工具或數據庫查詢工具?？蛻舳税l現機制使得新工具能夠被動態發現，無需更改客戶端代碼。
增加新資源：
同樣地，可擴充 self.resources 字典來暴露新的資源數據。結合客戶端的資源檢索流程，可以為模型提供更多上下文數據。
復雜功能：
可考慮實現異步調用。如果某工具執行時間較長，可修改 tools/call 的實現為立即返回一個任務ID，然后通過新增的 tools/retrieve（或類似方法）根據ID獲取結果。這類似于拆分調用和結果提取兩個步驟，在長任務場景下很有用。
安全和認證：
實際應用中，應在傳輸層（如HTTPS）以及協議層增加身份驗證和授權機制，確保只有獲得許可的客戶端才能訪問敏感工具或數據。MCP 規范在最新版本中也引入了 ??OAuth2?? 等認證支持。
日志和錯誤處理：
可以擴展錯誤日志記錄、請求計量等，以提高可監控性和可靠性。
通過上述示例，我們驗證了JSON-RPC風格的 MCP 通信模型：客戶端可以動態發現工具并調用之，從而讓LLM應用具備即插即用的擴展能力。
雖然我們的實現是簡化的，但它奠定了構建完整 MCP 服務器-代理系統的基礎結構。
在未來，你可以逐步按照 MCP 規范增加Prompts（預設提示模板）、更多工具類型以及高級特性，將其演變為功能完整的 MCP 服務。希望這個示例能幫助你理解 MCP 協議的核心機制，并為進一步開發打下基礎。
參考文獻：
Anthropic, “Model Context Protocol: Introduction”, 2024
– MCP 協議的簡介，將其比喻為 AI 應用的標準化連接接口。 Anthropic, “Model Context Protocol: Specification”, 2025
– MCP 協議規范概要，定義了通信采用的 JSON-RPC 2.0 消息格式和請求/響應結構。 Anthropic, “Model Context Protocol: Tools”, 2025
– MCP 中工具功能的相關規范片段，包括列出工具 (tools/list) 和調用工具 (tools/call) 的請求格式。 Anthropic, “Model Context Protocol: Resources”, 2025
– MCP 中資源讀取 (resources/read) 操作的請求示例。 Savan Kharod, “What is the Model Context Protocol (MCP)? A Complete Guide”, Treblle Blog, 2025
– 對 MCP 工作方式的通俗解釋，強調了動態工具發現和 JSON-RPC 通信模型的作用。
本文轉載自 ??AI大模型世界??，作者：roclv
