從零用 Python 實現最基礎的MCP協議原創

sulu637

發布于 2025-7-16 07:31

瀏覽

0收藏

?模型上下文協議（MCP, Model Context Protocol）是 Anthropic 在2024年提出的一種開放標準協議，用于標準化 AI 模型與外部工具和數據源的集成方式。

可以將 MCP 類比為 AI 世界中的 “USB-C 接口”：它為大型語言模型（LLM）連接各種工具和數據源提供了一種統一的方法。

MCP 使用 JSON-RPC 2.0 作為消息格式，在客戶端和服務器之間傳遞請求和響應。

從零用 Python 實現最基礎的MCP協議-AI.x社區

本示例將展示如何使用 Python 實現一個最基礎的 MCP 協議，包括 MCP 服務器和 MCP 客戶端兩部分。

我們將支持

discovery（發現）
invoke（調用）
retrieve（獲取）

等基本操作，并通過一個簡單的“計算器”工具（支持加法和乘法）演示協議的工作原理。

背景和設計概述

MCP 協議采用 ??客戶端-服務器?? 架構。

MCP服務器提供一組工具（tools）或資源（resources），MCP客戶端可以發現服務器提供的功能并進行調用。

兩者之間通過JSON-RPC進行通信，以標準的請求/響應消息交換指令和數據。

按照 MCP 規范：

??發現（Discovery）???：客戶端能夠查詢服務器，獲取其提供的工具列表、資源列表等。這通常通過調用 ??tools/list??? 或 ??resources/list?? 等方法實現。

??調用（Invoke）???：客戶端可以請求執行服務器上的某個工具功能，例如調用計算器的加法或乘法操作。規范中約定使用 ??tools/call?? 方法來調用指定名稱的工具，并傳遞所需參數。

??獲取（Retrieve）???：客戶端能夠檢索數據內容，例如獲取某個資源的具體內容。規范中提供了如 ??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 客戶端。客戶端將按照順序演示以下操作：

發現工具：請求獲取服務器提供的工具列表（調用 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，服務器將返回其內容文本。客戶端打印出該文本內容。

在打印輸出中，我們對結果做了簡單的格式化。例如：

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 實例，并實現對應的函數。例如，可以添加一個字符串處理工具或數據庫查詢工具。客戶端發現機制使得新工具能夠被動態發現，無需更改客戶端代碼。

增加新資源：

同樣地，可擴充 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

?著作權歸作者所有，如需轉載，請注明出處，否則將追究法律責任

標簽

贊

回復

舉報

回復

51CTO

51CTO博客

51CTO學堂

從零用 Python 實現最基礎的MCP協議原創

背景和設計概述

按照 MCP 規范：

實現思路

主要功能模塊：

MCPServer

MCPClient

通信格式

MCP服務器實現 (Server)

工具注冊：

資源注冊：

請求處理：

相應邏輯：

resources/list：

resources/read：

未知方法：

HTTP服務器：

MCP客戶端實現 (Client)

說明：

使用說明

擴展與總結

增加新工具：

增加新資源：

復雜功能：

安全和認證：

日志和錯誤處理：

參考文獻：

目錄

51CTO

51CTO博客

51CTO學堂

從零用 Python 實現最基礎的MCP協議 原創

背景和設計概述

按照 MCP 規范：

實現思路

主要功能模塊：

MCPServer

MCPClient

通信格式

MCP服務器實現 (Server)

工具注冊：

資源注冊：

請求處理：

相應邏輯：

resources/list：

resources/read：

未知方法：

HTTP服務器：

MCP客戶端實現 (Client)

說明：

使用說明

擴展與總結

增加新工具：

增加新資源：

復雜功能：

安全和認證：

日志和錯誤處理：

參考文獻：

目錄

從零用 Python 實現最基礎的MCP協議原創