使用Python从零实现最基础的MCP协议：服务器与客户端完整示例

MCP协议基础与实现思路

模型上下文协议（MCP, Model Context Protocol）是Anthropic在2024年提出的开放标准，用于标准化AI模型与外部工具和数据源的集成。MCP可类比为AI世界的“USB-C接口”，为大型语言模型（LLM）连接各种工具和数据源提供统一方法。MCP使用JSON-RPC 2.0作为消息格式，在客户端和服务器之间传递请求和响应。

本示例展示如何使用Python实现一个最基础的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服务器实现 (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",
                description="Add 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=add
            ),
            "multiply": Tool(
                name="multiply",
                description="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）。

使用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客户端。客户端将按照顺序演示以下操作：

1. 发现工具：请求获取服务器提供的工具列表（调用tools/list）。
2. 调用工具：调用加法和乘法工具（分别调用tools/call，传入相应参数）。
3. 发现资源：请求获取服务器提供的资源列表（调用resources/list，可选步骤）。
4. 获取资源内容：读取指定资源内容（调用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格式，确保了请求和响应的标准化。

使用说明

要运行该示例，请按照以下步骤操作：

1. 启动服务器：将上述服务器代码保存为mcp_server.py并运行。例如：

   python mcp_server.py

   服务器将启动并监听端口8000。在控制台上可以看到启动日志，例如可用的工具列表打印：

   MCP Server is running at http://localhost:8000
   Available tools: ['add', 'multiply']

2. 运行客户端：在服务器运行的同时，打开另一个终端窗口，将上述客户端代码保存为mcp_client.py并运行：

   python mcp_client.py

   客户端将依次发送JSON-RPC请求到服务器，并在终端打印收到的响应结果。
3. 观察结果：你应当在客户端终端看到类似输出：

   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!

   这些输出对应了我们在客户端代码中的打印语句，验证了各项操作成功执行。
4. 交互验证：你也可以使用其他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协议的核心机制，并为进一步开发打下基础。