用纯 Python 手写一个 MCP 服务器：JSON-RPC 2.0 协议实现与客户端接入

Model Context Protocol（MCP）把「模型调用外部工具」这件事抽象成一套基于 JSON-RPC 2.0 的通信契约。多数教程会直接引入官方 SDK，几行代码就把服务器跑起来，但这样一来，报文长什么样、方法怎么路由、版本不匹配时谁来兜底，全被封装掉了。这篇教程走另一条路：只用 Python 标准库，从零手写一个 MCP 服务器，把协议层摊开来看。适合已经会用 Python 写脚本、想搞清楚 MCP 通信细节、或者需要把内部数据源接进 AI 客户端的工程师。读完之后，你应该能独立实现一个可被编辑器识别、可被单元测试覆盖的本地 MCP 服务器，并知道哪些字段是协议要求、哪些是工程取舍。

准备工作

先确认环境。这套实现刻意不引入任何第三方 MCP 库，所以依赖极少，但 Python 版本和测试工具还是要提前定好。

• Python 3.9 及以上。实现里用到了 dict[str, Any] 这类内置泛型注解，3.9 起才在运行时可用；如果坚持用更老的版本，需要改成 typing.Dict。
• pytest，用于跑协议一致性测试。版本以你本地安装到的为准，安装前建议先确认当前版本。
• 一个支持 MCP 的客户端，例如具备 MCP 配置能力的编辑器或 IDE。不同客户端对配置文件的路径和字段名可能不同，以该客户端当前文档为准。
• 一个工作目录，用来放服务器代码和测试。

创建虚拟环境并安装测试依赖：

python3 -m venv .venv
source .venv/bin/activate
pip install pytest

Windows 下激活命令换成 .venv\Scripts\activate。虚拟环境的作用不只是隔离依赖，后面配置客户端时还要用到它的解释器绝对路径——这一点很关键，客户端启动服务器进程时不会继承你当前 shell 的环境。

操作步骤

第一步：确定目录结构与测试入口

把服务器主文件命名为 main.py，测试放在同级或子级的 tests/ 目录下。目录大致长这样：

code/
  main.py
  tests/
    test_main.py

这里有一个非常常见的坑：从项目根目录直接执行 pytest，测试文件里的导入会失败，报出 ModuleNotFoundError: No module named 'main'。原因是 pytest 的模块搜索路径以测试文件所在位置为基准，而 main.py 并不在 sys.path 里。

解决办法是显式指定 PYTHONPATH，把它指向 main.py 所在的目录：

PYTHONPATH=code pytest code/tests -v

Windows PowerShell 下的写法是：

$env:PYTHONPATH="code"; pytest code/tests -v

跑通之后，测试输出里应该能看到类似 test_tools_list、test_tools_call、test_resources_read、test_protocol_version_negotiation 这样的用例名，全部通过。测试数量取决于你实现的覆盖面，不必强求某个具体数字，但协议版本协商、工具列表、工具调用、资源读取这四类至少要各有一条用例。

第二步：实现请求校验

MCP 的传输层是 JSON-RPC 2.0，所以进入业务逻辑之前，必须先确认收到的字典是一个合法请求。校验至少要覆盖三点：

• jsonrpc 字段必须等于字符串 "2.0"；
• method 字段必须存在且是字符串；
• id 字段的存在性——通知类消息没有 id，请求类消息必须有。

校验不通过时不要抛异常穿透到主循环，而是构造一个 JSON-RPC 错误响应返回。标准错误码里，-32700 表示解析错误，-32600 表示无效请求，-32601 表示方法不存在，-32602 表示参数无效。这几个是 JSON-RPC 2.0 规范定义的，直接沿用即可。

第三步：写方法路由（dispatch）

路由函数是整个服务器的中枢：读 method 字段，分派到对应处理器，再把结果包成响应。返回类型要允许 None，因为通知类消息不需要响应。

def dispatch(message: dict[str, Any]) -> dict[str, Any] | None:
    validate_request(message)
    method = message["method"]
    params = message.get("params", {})

    if method == "tools/list":
        return complete(message["id"], handle_tools_list(params))
    elif method == "tools/call":
        return complete(message["id"], handle_tools_call(params))
    elif method == "resources/read":
        return complete(message["id"], handle_resources_read(params))
    elif method == "prompts/get":
        return complete(message["id"], handle_prompts_get(params))
    return error(message["id"], -32601, "Method not found")

其中 complete 负责拼出 {"jsonrpc": "2.0", "id": ..., "result": ...} 这样的成功响应，error 负责拼出带 error 对象的失败响应。把这两个构造函数抽出来，后面所有处理器都只关心业务数据，不用重复拼协议外壳。

MCP 的三大要素是 Tools / Resources / Prompts，分别对应「模型可以主动调用的动作」「模型可以读取的数据」「预置的提示模板」。路由表里把这四类方法（含 tools/list）都列出来，服务器才算完整。

第四步：实现 tools/call

工具调用是模型真正「动手」的地方。处理器从 params 里取出 arguments，按工具名分派到具体实现。下面是一个笔记检索工具的例子，数据源先用内存字典模拟：

def exec_notes_search(arguments: dict[str, Any]) -> list[dict[str, str]]:
    query = arguments.get("query", "").lower()
    results = []
    for note in NOTES_STORE.values():
        if query in note["title"].lower() or query in note["content"].lower():
            results.append({"id": note["id"], "title": note["title"]})
    return results

注意返回值的形状。工具调用的结果不是裸数据，而是包在 content 数组里的内容块，每个块有 type 和 text 两个字段。文本块里的 text 通常是序列化后的 JSON 字符串。响应里还要带一个 isError 布尔字段，用来区分「工具执行成功但结果为空」和「工具执行失败」这两种情况——模型客户端会据此决定是继续推理还是重试。

一个典型的工具调用响应长这样：

{
  "id": 5,
  "jsonrpc": "2.0",
  "result": {
    "content": [
      {
        "text": "[{\"id\": \"note-1\", \"title\": \"MCP overview\"}]",
        "type": "text"
      }
    ],
    "isError": false,
    "resultType": "complete"
  }
}

参数缺失时不要静默返回空列表。更稳妥的做法是把 query 视为必填，缺失就返回 -32602 参数错误，让调用方知道是自己传错了，而不是数据源里真的没有匹配项。

第五步：实现 resources/read 与 URI 方案

资源读取和工具调用的区别在于，资源用 URI 定位，而不是用参数。自定义 URI 方案是 MCP 里很实用的一招，比如用 notes://note-1 表示某一条笔记。服务器收到 URI 后，剥掉方案前缀拿到真实 ID，再去数据源里取。

def handle_resources_read(params: dict[str, Any]) -> dict[str, Any]:
    uri = params.get("uri", "")
    note_id = uri.replace("notes://", "")
    note = NOTES_STORE[note_id]
    return {
        "contents": [
            {
                "uri": uri,
                "mimeType": "text/markdown",
                "text": f"# {note['title']}\n\n{note['content']}",
            }
        ],
        "ttlMs": 5000,
    }

两个细节值得单独说：

• mimeType 告诉客户端这段内容该怎么解析。返回 Markdown 就写 text/markdown，返回纯文本写 text/plain，返回 JSON 写 application/json。写错不会报错，但客户端可能渲染成一片乱码。
• ttlMs 是缓存存活时间，单位毫秒。上面写 5000 表示这份内容 5 秒内可以被客户端复用，不必反复回服务器拉取。对于读多写少、内容变化不频繁的数据源，显式给出这个字段能显著降低 I/O 压力。如果数据实时性要求高，就把它设小，或者干脆不返回该字段。

另外，URI 不存在时应当返回资源未找到的错误，而不是让 KeyError 冒到主循环里把进程打挂。一个健壮的服务器不应该因为客户端传了个不存在的 ID 就整体退出。

第六步：处理协议版本协商

客户端在初始化时会声明自己期望的协议版本。服务器如果支持，就正常继续；如果不支持，必须明确拒绝，而不是硬着头皮按老版本处理——那会导致后续字段解析全乱。

拒绝的方式是返回一个 JSON-RPC 错误对象，其中 code 用协议约定的版本不支持错误码，data 里带上客户端请求的版本和服务器实际支持的版本列表，方便对方排查：

{
  "id": 8,
  "jsonrpc": "2.0",
  "error": {
    "code": -32022,
    "message": "Unsupported protocol version",
    "data": {
      "requested": "2027-01-01",
      "supported": ["2026-07-28"]
    }
  }
}

把 supported 做成一个列表而不是单个字符串，是为了将来同时兼容多个版本时不用改客户端。版本号本身会随协议演进而变化，实现时应当把它抽成常量，不要散落在各处硬编码。

第七步：用 demo 模式观察原始报文

调试协议实现时，最有价值的工具是「把收发双方的原始 JSON 打印出来」。给 main.py 加一个 --demo 命令行开关，让它依次构造几条请求、调用 dispatch、再把请求和响应成对打印。这样不用接客户端就能看到完整报文，排查字段拼写错误非常高效。

demo 至少要覆盖四条路径：工具列表、工具调用、资源读取、版本协商失败。前三条验证正常流程，第四条验证错误分支——错误分支往往才是真正出问题的地方。

第八步：接入客户端

服务器实现了 stdio（标准输入输出）传输，就可以被支持 MCP 的客户端直接拉起。配置的核心是两件事：用哪个解释器、跑哪个脚本。

{
  "mcpServers": {
    "local-notes-server": {
      "command": "/path/to/.venv/bin/python3",
      "args": ["/path/to/code/main.py"]
    }
  }
}

两个路径都必须是绝对路径。客户端启动子进程时工作目录不确定，用相对路径几乎必然失败。解释器要指向虚拟环境里的那个，而不是系统 Python——否则依赖装在哪就白装了。

配置写好后重启客户端，在工具列表里应该能看到 notes_search 之类的条目。如果看不到，先确认脚本单独执行不报错，再检查路径拼写。

一个完整示例

把上面的片段串成一个可运行的最小服务器。以下代码只依赖标准库：

import json
import sys
from typing import Any

PROTOCOL_VERSION = "2026-07-28"

NOTES_STORE = {
    "note-1": {"id": "note-1", "title": "MCP overview", "content": "MCP 基于 JSON-RPC 2.0。"},
    "note-2": {"id": "note-2", "title": "Tools vs Resources", "content": "工具是动作，资源是数据。"},
}

TOOLS = [
    {
        "name": "notes_search",
        "description": "按关键词检索本地笔记",
        "inputSchema": {
            "type": "object",
            "properties": {"query": {"type": "string"}},
            "required": ["query"],
        },
    }
]


def complete(msg_id: Any, result: Any) -> dict[str, Any]:
    return {"jsonrpc": "2.0", "id": msg_id, "result": result}


def error(msg_id: Any, code: int, message: str, data: Any = None) -> dict[str, Any]:
    payload = {"code": code, "message": message}
    if data is not None:
        payload["data"] = data
    return {"jsonrpc": "2.0", "id": msg_id, "error": payload}


def validate_request(message: dict[str, Any]) -> None:
    if message.get("jsonrpc") != "2.0":
        raise ValueError("invalid jsonrpc version")
    if not isinstance(message.get("method"), str):
        raise ValueError("missing method")


def handle_tools_list(params: dict[str, Any]) -> dict[str, Any]:
    return {"tools": TOOLS}


def exec_notes_search(arguments: dict[str, Any]) -> list[dict[str, str]]:
    query = arguments.get("query", "").lower()
    results = []
    for note in NOTES_STORE.values():
        if query in note["title"].lower() or query in note["content"].lower():
            results.append({"id": note["id"], "title": note["title"]})
    return results


def handle_tools_call(params: dict[str, Any]) -> dict[str, Any]:
    name = params.get("name")
    arguments = params.get("arguments", {})
    if name != "notes_search":
        return {"content": [{"type": "text", "text": "unknown tool"}], "isError": True}
    if not arguments.get("query"):
        return {"content": [{"type": "text", "text": "query is required"}], "isError": True}
    hits = exec_notes_search(arguments)
    return {
        "content": [{"type": "text", "text": json.dumps(hits, ensure_ascii=False)}],
        "isError": False,
        "resultType": "complete",
    }


def handle_resources_read(params: dict[str, Any]) -> dict[str, Any]:
    uri = params.get("uri", "")
    note_id = uri.replace("notes://", "")
    note = NOTES_STORE.get(note_id)
    if note is None:
        raise KeyError(note_id)
    return {
        "contents": [
            {
                "uri": uri,
                "mimeType": "text/markdown",
                "text": f"# {note['title']}\n\n{note['content']}",
            }
        ],
        "ttlMs": 5000,
    }


def dispatch(message: dict[str, Any]) -> dict[str, Any] | None:
    validate_request(message)
    method = message["method"]
    params = message.get("params", {})
    if method == "tools/list":
        return complete(message["id"], handle_tools_list(params))
    if method == "tools/call":
        return complete(message["id"], handle_tools_call(params))
    if method == "resources/read":
        try:
            return complete(message["id"], handle_resources_read(params))
        except KeyError:
            return error(message["id"], -32602, "Resource not found")
    if method == "initialize":
        requested = params.get("protocolVersion")
        if requested != PROTOCOL_VERSION:
            return error(
                message["id"],
                -32022,
                "Unsupported protocol version",
                {"requested": requested, "supported": [PROTOCOL_VERSION]},
            )
        return complete(message["id"], {"protocolVersion": PROTOCOL_VERSION})
    return error(message["id"], -32601, "Method not found")


def serve() -> None:
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            message = json.loads(line)
        except json.JSONDecodeError:
            response = error(None, -32700, "Parse error")
        else:
            response = dispatch(message)
        if response is not None:
            sys.stdout.write(json.dumps(response, ensure_ascii=False) + "\n")
            sys.stdout.flush()


if __name__ == "__main__":
    serve()

配套的测试可以这样写，直接调用 dispatch，不经过标准输入输出：

import main


def test_tools_list():
    resp = main.dispatch({"jsonrpc": "2.0", "id": 1, "method": "tools/list"})
    assert resp["result"]["tools"][0]["name"] == "notes_search"


def test_tools_call():
    resp = main.dispatch({
        "jsonrpc": "2.0",
        "id": 2,
        "method": "tools/call",
        "params": {"name": "notes_search", "arguments": {"query": "mcp"}},
    })
    assert resp["result"]["isError"] is False


def test_resources_read():
    resp = main.dispatch({
        "jsonrpc": "2.0",
        "id": 3,
        "method": "resources/read",
        "params": {"uri": "notes://note-1"},
    })
    assert resp["result"]["contents"][0]["mimeType"] == "text/markdown"


def test_protocol_version_negotiation():
    resp = main.dispatch({
        "jsonrpc": "2.0",
        "id": 4,
        "method": "initialize",
        "params": {"protocolVersion": "2027-01-01"},
    })
    assert resp["error"]["code"] == -32022

运行方式：

PYTHONPATH=code pytest code/tests -v

四条用例全绿，说明正常路径和错误路径都符合预期。之后把 main.py 的绝对路径填进客户端配置，重启客户端即可在工具列表里看到 notes_search。

注意事项

关于无状态设计。 上面这个服务器是刻意做成无状态的：每次请求自带全部上下文，服务器不保存会话。这样做的好处是数据源可以随时替换——把内存里的 NOTES_STORE 换成关系型数据库、缓存或对象存储，协议层代码一行都不用改。反过来说，如果你确实需要跨请求保持状态，就必须自己引入会话标识和生命周期管理，复杂度会明显上升。

关于缓存字段。 ttlMs 只是给客户端的建议值，客户端可以忽略它。不要把它当成强一致性的保证，也不要指望靠它解决数据过期问题——真正的失效逻辑仍然要在数据源那一侧做。

关于错误码。 JSON-RPC 2.0 的标准错误码范围是 -32768 到 -32000，其中 -32000 到 -32099 留给实现自定义。版本不支持用的 -32022 就落在这个区间里。自定义错误码不要越界，否则可能和标准码冲突。

关于测试路径。 ModuleNotFoundError: No module named 'main' 是这套结构里最容易踩的坑。除了设置 PYTHONPATH，另一种做法是在 tests/ 下放一个空的 __init__.py 并调整导入方式，但显式指定路径更直观，也更不容易在 CI 上出意外。

关于版本与配额。 协议版本号、客户端支持的配置字段、以及各家客户端对 MCP 的具体支持程度都在持续变化，本文涉及的版本标识仅用于说明协商机制本身。实际接入前请以对应客户端与协议规范的当前信息为准。

关于安全边界。 stdio 传输意味着客户端会以你的身份启动一个本地进程。服务器代码里不要执行来自 arguments 的任意命令或路径拼接，工具的参数校验要在进入业务逻辑之前完成。