手写 MCP Server：三个坑与可跑代码

MCP（Model Context Protocol）解决的是一个很具体的问题：让大模型知道“有哪些工具可用、参数长什么样、调用结果是什么”。它由 Anthropic 在 2024 年底提出，底层就是 JSON-RPC 2.0，官方也提供了 Python 和 TypeScript SDK。

但我在实际项目里越来越多地选择不用 SDK。原因不是“手写更优雅”，而是两个很现实的约束：第一，内部系统往往只需要暴露两三个工具，引入一套 SDK 和它的传输封装，收益不明显；第二，线上出问题时，你必须知道客户端和服务端之间到底传了什么字节，而这一层被框架包住之后，排查成本会陡增。

这篇文章用一个 150 行的 stdio server 把 initialize → tools/list → tools/call 三步走通，再讲三个我真实踩过的坑。代码在 Python 3.10+ 下可直接运行，不依赖任何第三方库。

一、先想清楚传输方式

MCP 目前有两类传输：stdio 和 Streamable HTTP。

• stdio：客户端把 server 作为子进程拉起，一行一条 JSON-RPC 消息，进程退出即会话结束。适合本机工具、IDE 插件、CLI。
• Streamable HTTP：server 独立部署，多个客户端共享，需要处理会话标识、SSE 流、鉴权和限流。

很多教程一上来就写 HTTP 版本，结果一半时间花在会话管理上。我的建议是：先把 stdio 版本跑通，协议语义完全一样，等确实需要远程共享了再换传输层。本文只做 stdio。

二、服务端：只实现必需的方法

MCP 的方法不少，但一个能用的 server 只需要四个：initialize、tools/list、tools/call、ping。其余方法在收到时回 -32601（method not found）即可，客户端不会因为你没实现 resources/* 就拒绝连接——前提是你在 capabilities 里没有声明它。

#!/usr/bin/env python3
"""最小 MCP Server：stdio 传输，只暴露一个工具。"""
import json
import sys
from pathlib import Path

PROTOCOL_VERSIONS = {"2024-11-05", "2025-03-26", "2025-06-18"}
LATEST = "2025-06-18"
ROOT = (Path(__file__).parent / "sandbox").resolve()

TOOLS = [{
    "name": "count_lines",
    "description": "统计 sandbox 目录内某个文本文件的行数。",
    "inputSchema": {
        "type": "object",
        "properties": {
            "path": {"type": "string", "description": "相对 sandbox 的路径，如 notes.md"}
        },
        "required": ["path"],
        "additionalProperties": False,
    },
}]


def log(*args):
    print("[server]", *args, file=sys.stderr, flush=True)


def send(msg):
    # stdio 传输：一行一条 JSON-RPC 消息，消息内不能有换行
    sys.stdout.write(json.dumps(msg, ensure_ascii=False) + "\n")
    sys.stdout.flush()


def send_error(msg_id, code, message):
    send({"jsonrpc": "2.0", "id": msg_id, "error": {"code": code, "message": message}})


def resolve(path_str):
    p = (ROOT / path_str).resolve()
    # 关键：resolve 之后再比较前缀，防 ../ 和符号链接穿越
    if p != ROOT and ROOT not in p.parents:
        raise ValueError(f"path escapes sandbox: {path_str}")
    return p


def call_tool(name, args):
    if name != "count_lines":
        raise KeyError(name)
    if not isinstance(args, dict) or "path" not in args:
        raise ValueError("missing required argument: path")
    path = resolve(args["path"])
    text = path.read_text(encoding="utf-8")
    return [{"type": "text", "text": f"{path.name}: {len(text.splitlines())} 行"}]


def handle(msg):
    method = msg.get("method")
    msg_id = msg.get("id")
    is_notification = "id" not in msg

    if method == "initialize":
        asked = msg["params"].get("protocolVersion", LATEST)
        version = asked if asked in PROTOCOL_VERSIONS else LATEST
        return {
            "protocolVersion": version,
            "capabilities": {"tools": {"listChanged": False}},
            "serverInfo": {"name": "line-counter", "version": "0.1.0"},
        }
    if method == "ping":
        return {}
    if method == "tools/list":
        return {"tools": TOOLS}
    if method == "tools/call":
        params = msg.get("params", {})
        try:
            content = call_tool(params.get("name"), params.get("arguments", {}))
            return {"content": content, "isError": False}
        except Exception as exc:  # 工具执行失败走 result.isError，而不是 JSON-RPC error
            log("tool failed:", exc)
            return {"content": [{"type": "text", "text": str(exc)}], "isError": True}
    if is_notification:
        return None
    raise LookupError(method)


def main():
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            msg = json.loads(line)
        except json.JSONDecodeError:
            send_error(None, -32700, "parse error")
            continue
        if not isinstance(msg, dict):
            send_error(None, -32600, "invalid request")
            continue
        try:
            result = handle(msg)
        except LookupError:
            if "id" in msg:
                send_error(msg["id"], -32601, f"method not found: {msg.get('method')}")
            continue
        except (KeyError, TypeError) as exc:
            if "id" in msg:
                send_error(msg["id"], -32602, f"invalid params: {exc}")
            continue
        except Exception as exc:
            log("unhandled:", exc)
            if "id" in msg:
                send_error(msg["id"], -32603, "internal error")
            continue
        if result is not None and "id" in msg:
            send({"jsonrpc": "2.0", "id": msg["id"], "result": result})


if __name__ == "__main__":
    main()

有两个设计点值得说明。

协议版本不是常量。 客户端在 initialize 里报出它想用的版本，服务端应当回一个自己支持的版本；如果双方没有交集，客户端会主动断开。写死一个版本号，会在客户端升级后直接连不上。

工具执行失败不要返回 JSON-RPC error。 -32602 这类错误码表示“请求本身不合法”，客户端可能不会把它喂给模型；而文件不存在、参数类型不对这类业务失败，应该放在 result 里，用 isError: true 和 content 表达。模型看到的是文本，它才有机会自我纠正。这个区分在协议规范里写得很清楚，但手写时最容易混淆。

三、客户端：用子进程验证

#!/usr/bin/env python3
"""配套最小客户端：用子进程拉起 server，走同一套 JSON-RPC。"""
import json
import subprocess
import sys
from pathlib import Path

HERE = Path(__file__).parent


class MCPClient:
    def __init__(self):
        self.proc = subprocess.Popen(
            [sys.executable, str(HERE / "server.py")],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=None,
            text=True,
            bufsize=1,
        )
        self._id = 0

    def _send(self, msg):
        self.proc.stdin.write(json.dumps(msg, ensure_ascii=False) + "\n")
        self.proc.stdin.flush()

    def notify(self, method, params=None):
        msg = {"jsonrpc": "2.0", "method": method}
        if params is not None:
            msg["params"] = params
        self._send(msg)

    def request(self, method, params=None):
        self._id += 1
        msg = {"jsonrpc": "2.0", "id": self._id, "method": method}
        if params is not None:
            msg["params"] = params
        self._send(msg)
        while True:
            line = self.proc.stdout.readline()
            if not line:
                raise RuntimeError("server 提前退出")
            resp = json.loads(line)
            if resp.get("id") == self._id:
                if "error" in resp:
                    raise RuntimeError(resp["error"])
                return resp["result"]

    def close(self):
        self.proc.stdin.close()
        self.proc.wait(timeout=5)


def main():
    client = MCPClient()
    init = client.request("initialize", {
        "protocolVersion": "2025-06-18",
        "capabilities": {},
        "clientInfo": {"name": "demo-client", "version": "0.1.0"},
    })
    print("协议版本:", init["protocolVersion"], "服务端:", init["serverInfo"]["name"])
    client.notify("notifications/initialized")

    tools = client.request("tools/list")["tools"]
    print("工具:", [t["name"] for t in tools])

    ok = client.request("tools/call", {"name": "count_lines", "arguments": {"path": "notes.md"}})
    print("正常调用:", ok["content"][0]["text"])

    bad = client.request("tools/call", {"name": "count_lines", "arguments": {"path": "../server.py"}})
    print("越界调用 isError =", bad["isError"], "|", bad["content"][0]["text"])

    client.close()


if __name__ == "__main__":
    main()

建一个 sandbox/notes.md 放三行文本，运行结果是：

协议版本: 2025-06-18 服务端: line-counter
工具: ['count_lines']
正常调用: notes.md: 3 行
越界调用 isError = True | path escapes sandbox: ../server.py

注意 notifications/initialized 必须发。省略它，规范的客户端会认为握手没完成，后续请求可能被拒绝——虽然很多宽松实现不校验。

四、三个真实的坑

坑一：stdout 被日志污染。 stdio 传输把 stdout 完全让给协议，任何一句 print("debug") 都会让客户端解析失败，而且报错位置通常离真正的 bug 很远。更麻烦的是，有些库默认往 stdout 打日志，你甚至不知道是谁写的。我的做法是：server 入口处就把日志函数固定到 sys.stderr，并在 Code Review 里禁止 server 模块出现裸 print。

坑二：给通知回了响应。 通知（notification）没有 id，按 JSON-RPC 规范绝对不能回。我对早期版本做压测时，客户端偶发地把某个响应错配到另一个请求上，最后发现是 server 对 notifications/cancelled 回了一条 id: null 的消息，把客户端的请求队列打乱了。判断逻辑其实就一行："id" not in msg 就是通知，处理完直接 continue，不发包。

坑三：路径穿越与“工具描述即提示词”。 上面 resolve() 里先 resolve 再判断 ROOT in p.parents，是因为只做字符串 startswith 会被 .. 和符号链接绕过。另一个常被忽视的点是 description 字段：它会被原样放进模型的上下文，等于一段你不完全可控的提示词。所以描述只写“这个工具做什么、参数含义”，不要塞用户输入，也不要写“忽略前面所有指令”之类的内容。同理，inputSchema 里加 "additionalProperties": false 很有必要——模型很爱自作主张多传几个字段，严格一点能让错误在参数校验阶段就暴露。

还有两个我没在代码里处理、但上线前必须补的能力：一是并发，当前的 for line in sys.stdin 是单线程阻塞的，一个慢工具会卡住整个会话，而 notifications/cancelled 也只有在你读到它之后才能生效；二是超时，客户端会自己计时，服务端最好也设一个上界，避免子进程僵死。这两点用 SDK 会省事很多，这也是我唯一建议直接上 SDK 的场景。

五、结论

手写 MCP Server 的价值不在于省一个依赖，而在于你清楚每一层在做什么：谁能写 stdout、哪个消息该回、错误该放在哪一层。150 行的代价是你要自己补并发、超时和会话管理；收益是排查问题时不用猜框架。

我的取舍是：本机、工具少、需要深度排查 → 手写 stdio 版本；远程部署、多客户端、要鉴权和流式 → 直接用官方 SDK 的 Streamable HTTP 实现。 两者协议语义一致，迁移时工具定义可以原样搬过去。