MCP 服务端从零写：协议细节与三个取舍

一、问题不在"接不上"，在"接得乱"

给大模型接工具，最容易的做法是在 prompt 里塞一段 JSON Schema，让模型输出函数调用。项目里接三个模型厂商，就有三套 schema 方言、三套错误约定。MCP（Model Context Protocol）想解决的就是这件事：把"模型怎么发现工具、怎么调用工具、怎么拿回结果"抽成一层协议，客户端和服务端各自实现一次。

但真正动手时，问题往往不出在协议本身，而出在实现细节：调试用的 print 把 stdout 弄脏、工具描述写得像给人看的说明书、把业务异常当成协议错误返回。下面用一个零依赖的 stdio 服务端把这些点摊开。

二、把协议跑起来：一个零依赖实现

stdio 传输的规则很朴素：客户端启动服务端进程，双方通过 stdin/stdout 交换 JSON-RPC 2.0 消息，每条消息一行，行内不能出现裸换行。服务端只要保证"读一行、解析、回一行"就够了，不需要任何框架。

import hashlib
import json
import os
import sys

PROTOCOL_VERSIONS = ("2025-06-18", "2025-03-26", "2024-11-05")
SERVER_INFO = {"name": "file-inspect", "version": "0.1.0"}


def file_sha256(arguments):
    path = arguments["path"]
    size = os.path.getsize(path)
    digest = hashlib.sha256()
    with open(path, "rb") as handle:
        for chunk in iter(lambda: handle.read(1 << 20), b""):
            digest.update(chunk)
    return f"{digest.hexdigest()}  {size} bytes"


def dir_usage(arguments):
    root = arguments["path"]
    top_n = int(arguments.get("top_n", 5))
    entries = []
    for current, _dirs, files in os.walk(root):
        for name in files:
            full = os.path.join(current, name)
            try:
                entries.append((os.path.getsize(full), full))
            except OSError:
                continue
    entries.sort(reverse=True)
    lines = [f"{size:>12}  {path}" for size, path in entries[:top_n]]
    return "\n".join(lines) if lines else "(empty)"


TOOLS = {
    "file_sha256": {
        "description": "计算单个文件或模块的 SHA-256 摘要及字节数。输入必须是已存在的文件路径，不接受目录或通配符。",
        "inputSchema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "文件路径，支持绝对或相对路径"}
            },
            "required": ["path"],
        },
        "handler": file_sha256,
    },
    "dir_usage": {
        "description": "扫描目录并返回按体积倒序排列的最大文件列表，用于定位大文件。会递归遍历子目录，跳过无权限项。",
        "inputSchema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "待扫描的目录路径"},
                "top_n": {
                    "type": "integer",
                    "description": "返回条数，默认 5，范围 1-50",
                    "minimum": 1,
                    "maximum": 50,
                },
            },
            "required": ["path"],
        },
        "handler": dir_usage,
    },
}


def text_result(text):
    return {"content": [{"type": "text", "text": text}], "isError": False}


def error_result(text):
    return {"content": [{"type": "text", "text": text}], "isError": True}


def handle(request):
    method = request.get("method")
    params = request.get("params") or {}
    if method == "initialize":
        requested = params.get("protocolVersion")
        version = requested if requested in PROTOCOL_VERSIONS else PROTOCOL_VERSIONS[0]
        return {
            "protocolVersion": version,
            "capabilities": {"tools": {"listChanged": False}},
            "serverInfo": SERVER_INFO,
        }
    if method == "tools/list":
        return {
            "tools": [
                {"name": name, "description": spec["description"], "inputSchema": spec["inputSchema"]}
                for name, spec in TOOLS.items()
            ]
        }
    if method == "tools/call":
        name = params.get("name")
        spec = TOOLS.get(name)
        if spec is None:
            return error_result(f"unknown tool: {name}")
        try:
            return text_result(spec["handler"](params.get("arguments") or {}))
        except Exception as exc:
            return error_result(f"{type(exc).__name__}: {exc}")
    if method == "ping":
        return {}
    raise KeyError(method)


def main():
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            request = json.loads(line)
        except json.JSONDecodeError as exc:
            reply = {"jsonrpc": "2.0", "id": None, "error": {"code": -32700, "message": str(exc)}}
            sys.stdout.write(json.dumps(reply) + "\n")
            sys.stdout.flush()
            continue
        if "id" not in request:
            continue
        try:
            result = handle(request)
            reply = {"jsonrpc": "2.0", "id": request["id"], "result": result}
        except KeyError:
            reply = {
                "jsonrpc": "2.0",
                "id": request["id"],
                "error": {"code": -32601, "message": "Method not found"},
            }
        except Exception as exc:
            reply = {
                "jsonrpc": "2.0",
                "id": request["id"],
                "error": {"code": -32603, "message": str(exc)},
            }
        sys.stdout.write(json.dumps(reply, ensure_ascii=False) + "\n")
        sys.stdout.flush()


if __name__ == "__main__":
    main()

客户端侧的生命周期是四步：initialize 协商版本与能力，notifications/initialized 通知（无 id，服务端不回），tools/list 拿工具清单，tools/call 执行。代码里 initialize 会回显客户端请求的版本，只有请求版本不在支持列表时才降级到自己最新的那个——这个细节后面单独说。

不装客户端也能验：把几条消息用 printf 灌进 stdin，看 stdout 逐行返回。

printf '%s\n' \
  '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"smoke","version":"0"}}}' \
  '{"jsonrpc":"2.0","id":2,"method":"tools/list"}' \
  '{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"file_sha256","arguments":{"path":"server.py"}}}' \
  | python3 server.py

三、三个真实取舍

1. stdout 是协议通道，不是日志通道

这是第一个坑，也是新手最常踩的。调试时顺手 print("here")，客户端那行就解析失败；更麻烦的是按行帧的协议里，一行脏输出会把后续消息的边界搞乱，报错位置和真正的错位相距很远。

规矩只有一条：stdout 只写 JSON-RPC 消息，所有日志、异常栈、进度信息一律走 stderr。用 logging.basicConfig(stream=sys.stderr) 是最省事的做法。同样要注意第三方库在 import 时往 stdout 打 banner，这类目录往往藏在依赖深处，一旦出现只能靠 2>/dev/null 之外的手段逐个排查。

2. 工具描述是给模型看的，边界比功能重要

写 file_sha256 的描述时，我特意加了一句"输入必须是已存在的文件路径，不接受目录或通配符"。原因很实在：只写功能时，模型会把目录也传进来，服务端抛 IsADirectoryError，模型看到报错再换参数重试，一次调用变成三次，延迟和 token 都翻倍。把否定条件写进描述、把取值范围写进 inputSchema（top_n 的 minimum/maximum 就是干这个的），能让第一次调用就落在合法区间。

另一个取舍是工具粒度。一个大而全的"文件分析"工具参数多、分支多，模型容易漏参数；拆成小工具组合更灵活，但工具一多，tools/list 的返回本身就开始吃上下文，模型选错的概率也随之上升。实际做法是单服务端控制在 5 到 10 个工具，超了就按领域拆成多个服务端，由客户端按需挂载，而不是塞进一个"超级服务端"。

3. 业务失败用 isError，协议失败才用 error

这是最容易写错的一处。tools/call 有两种失败，必须分开处理：

• 协议层失败（方法不存在、参数不是合法 JSON-RPC）：返回 error 对象，用 -32601、-32602 这类标准码。
• 工具执行失败（文件不存在、权限不足）：正常返回 result，把错误说明放进 content，isError 置 true。

不能反过来。模型能看到的是 result.content 里的文本，而多数客户端会把 JSON-RPC 的 error 当作传输异常直接往上抛，模型根本看不到。把 FileNotFoundError 放进 isError，模型读到"文件不存在"可以换个路径重试；塞进 error 对象，这个回合就直接断了，调用方只能看到一句没有上下文的内部错误。代码里 handle 把所有执行异常统一转成 isError: true，就是为了保住这条重试链路。

四、还有两个容易漏的点

版本协商不要硬编码最新版。服务端声明自己支持的版本集合，客户端请求哪个就尽量回哪个，交集为空才降级到最新版。直接写死 2025-06-18，会让仍停留在 2024-11-05 的客户端在握手阶段失败，而且失败信息通常只有一行版本不匹配。

返回结果要有体积上限。dir_usage 里 top_n 上限 50 不是随手写的：一个目录扫出几万条路径扔进上下文，token 直接爆掉，模型也读不完。凡是返回列表的工具，都应该有默认截断，并在结果里注明"还有多少条未显示"，让模型自己决定要不要缩小范围再查。

五、结论

MCP 的价值不在于多了一套 API，而在于把工具接入从"每个模型厂商各写一遍"变成"服务端写一次、客户端适配一次"。但协议只是骨架，真正决定体验的是三件事：stdout 保持干净、工具描述带上边界、错误分层清楚。这三点和用不用官方 SDK 无关——用高层框架照样会踩，只是踩的时候错得更靠后、更难定位。

上面的代码 python3 server.py 就能跑，接任何支持 stdio 的 MCP 客户端都行。往下比较自然的两步是补上 resources 和 prompts 两个原语，以及把长任务改成 notifications/progress 的进度通知，让客户端知道服务端在干活，而不是干等一个超时。