用纯 Python 手写一个 MCP 服务器：JSON-RPC 2.0 协议实现与客户端接入
用纯 Python 手写一个 MCP 服务器：JSON-RPC 2.0 协议实现与客户端接入
不依赖官方 MCP SDK，只用 Python 标准库实现一个符合 Model Context Protocol 的服务器：从 JSON-RPC 2.0 报文校验、方法路由、tools/list 与 tools/call、resources/read 的 URI 方案与缓存字段，到版本协商失败时的错误码返回，再到把 stdio 服务器注册进支持 MCP 的编辑器。
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 pytestWindows 下激活命令换成 .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 -vWindows 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 的任意命令或路径拼接，工具的参数校验要在进入业务逻辑之前完成。