MCP 服务端手写记：stdout 污染与错误码混用

给编码 Agent 加一个"在仓库里按关键词找代码"的能力，最省事的做法是自己写一个 MCP 服务端：不用引入框架，不依赖任何 SDK，一个 Python 文件加标准库就够。协议本身确实薄——JSON-RPC 2.0，一行一条消息。但真实折腾的时间几乎都不在协议上，而在几个边界上：stdout 被调试信息污染、工具执行失败被错报成协议错误、返回结果把上下文撑爆。下面按"协议怎么走通、几个坑怎么踩的、最后怎么收"来讲，代码在本地跑通过。

一、问题：Agent 需要一个能收敛输出的检索工具

把 grep 直接塞给模型，最典型的失败是返回几千行匹配，模型的上下文被一次工具调用占满，后面全是噪声。所以工具要满足三点：参数校验在服务端做、结果有上限且明确告诉模型"被截断了"、失败信息要以模型能读懂的形式返回，而不是让它看到一个连接错误。

MCP 的 stdio 传输用换行分隔的 JSON-RPC 消息：每条消息一行，行内不能有裸换行。这里最容易和 LSP 混淆——LSP 用 Content-Length 头，MCP 的 stdio 不用。生命周期是 initialize → notifications/initialized → tools/list → tools/call，通知（没有 id 的消息）不需要回包，回了反而是协议违规。

还有一个细节值得较真：版本协商。客户端在 initialize 里报一个 protocolVersion，如果服务端支持这个版本，就原样回；不支持则回自己支持的最新版本，由客户端决定是否继续。直接报错拒绝握手，会让很多宿主认为服务端不可用。

二、方案：一个文件的服务端

核心只有三件事：读一行、分发、写一行。完整实现如下，保存为 server.py 即可运行。

#!/usr/bin/env python3
import json
import sys
import threading
import traceback
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

PROTOCOL_VERSION = "2025-06-18"
SUPPORTED = {"2024-11-05", "2025-03-26", "2025-06-18"}
TEXT_SUFFIX = {".py", ".md", ".ts", ".js", ".yaml", ".yml", ".json", ".go", ".java"}
WRITE_LOCK = threading.Lock()
POOL = ThreadPoolExecutor(max_workers=4)


def log(msg):
    print(f"[repo-search] {msg}", file=sys.stderr, flush=True)


def send(payload):
    line = json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
    with WRITE_LOCK:
        sys.stdout.write(line + "\n")
        sys.stdout.flush()


TOOLS = [{
    "name": "search_text",
    "description": "在目录内的文本文件中搜索关键词，返回 文件:行号: 内容，超出上限会被截断",
    "inputSchema": {
        "type": "object",
        "properties": {
            "path": {"type": "string", "description": "要搜索的绝对目录路径"},
            "keyword": {"type": "string", "description": "关键词，区分大小写"},
            "limit": {"type": "integer", "minimum": 1, "maximum": 200, "default": 50},
        },
        "required": ["path", "keyword"],
    },
}]


def search_text(args):
    root = Path(args["path"])
    if not root.is_dir():
        raise ValueError(f"不是有效目录: {root}")
    keyword = args["keyword"]
    limit = min(int(args.get("limit", 50)), 200)
    hits, total = [], 0
    for p in sorted(root.rglob("*")):
        if not p.is_file() or p.suffix not in TEXT_SUFFIX:
            continue
        try:
            with p.open(errors="ignore") as f:
                for no, line in enumerate(f, 1):
                    if keyword in line:
                        total += 1
                        if len(hits) < limit:
                            hits.append(f"{p}:{no}: {line.rstrip()[:200]}")
        except OSError as e:
            log(f"跳过 {p}: {e}")
    if not hits:
        return "未找到匹配。"
    tail = "" if total <= limit else f"\n…… 共 {total} 条命中，已截断到前 {limit} 条"
    return "\n".join(hits) + tail


def check_args(args):
    if not isinstance(args, dict):
        raise TypeError("arguments 必须是对象")
    for key in TOOLS[0]["inputSchema"]["required"]:
        if not args.get(key):
            raise ValueError(f"缺少必填参数: {key}")
    return args


def handle(msg):
    method, mid = msg.get("method"), msg.get("id")
    params = msg.get("params") or {}
    if method == "initialize":
        asked = params.get("protocolVersion")
        return {"jsonrpc": "2.0", "id": mid, "result": {
            "protocolVersion": asked if asked in SUPPORTED else PROTOCOL_VERSION,
            "capabilities": {"tools": {"listChanged": False}},
            "serverInfo": {"name": "repo-search", "version": "0.1.0"},
        }}
    if method == "tools/list":
        return {"jsonrpc": "2.0", "id": mid, "result": {"tools": TOOLS}}
    if method == "tools/call":
        if params.get("name") != "search_text":
            return {"jsonrpc": "2.0", "id": mid,
                    "error": {"code": -32602, "message": f"未知工具: {params.get('name')}"}}
        try:
            text = search_text(check_args(params.get("arguments")))
            return {"jsonrpc": "2.0", "id": mid, "result": {
                "content": [{"type": "text", "text": text}], "isError": False}}
        except ValueError as e:
            return {"jsonrpc": "2.0", "id": mid, "result": {
                "content": [{"type": "text", "text": f"参数不合法: {e}"}], "isError": True}}
        except Exception as e:
            traceback.print_exc(file=sys.stderr)
            return {"jsonrpc": "2.0", "id": mid, "result": {
                "content": [{"type": "text", "text": f"工具内部错误: {e}"}], "isError": True}}
    return {"jsonrpc": "2.0", "id": mid,
            "error": {"code": -32601, "message": f"不支持的方法: {method}"}}


def dispatch(msg):
    if not isinstance(msg, dict) or msg.get("jsonrpc") != "2.0":
        send({"jsonrpc": "2.0", "id": None,
              "error": {"code": -32600, "message": "无效请求"}})
        return
    if "method" not in msg:
        return
    if "id" not in msg:
        log(f"收到通知: {msg.get('method')}")
        return
    try:
        send(handle(msg))
    except Exception:
        traceback.print_exc(file=sys.stderr)
        send({"jsonrpc": "2.0", "id": msg.get("id"),
              "error": {"code": -32603, "message": "内部错误"}})


def main():
    log("server started")
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            msg = json.loads(line)
        except json.JSONDecodeError as e:
            send({"jsonrpc": "2.0", "id": None,
                  "error": {"code": -32700, "message": f"解析失败: {e}"}})
            continue
        POOL.submit(dispatch, msg)

    POOL.shutdown(wait=True)
    log("stdin closed, bye")


if __name__ == "__main__":
    main()

用管道就能验证，不需要任何依赖：

import json, subprocess, sys
p = subprocess.Popen([sys.executable, "server.py"], stdin=subprocess.PIPE,
                     stdout=subprocess.PIPE, text=True, bufsize=1)
def rpc(obj):
    p.stdin.write(json.dumps(obj) + "\n"); p.stdin.flush()
rpc({"jsonrpc": "2.0", "id": 1, "method": "initialize",
     "params": {"protocolVersion": "2024-11-05", "capabilities": {}}})
print(p.stdout.readline())          # 回 {"protocolVersion": "2024-11-05", ...}
rpc({"jsonrpc": "2.0", "method": "notifications/initialized"})
rpc({"jsonrpc": "2.0", "id": 2, "method": "tools/call",
     "params": {"name": "search_text", "arguments": {"path": ".", "keyword": "def "}}})
print(p.stdout.readline())

这段 description 和 inputSchema 不是写给文档的，而是模型每一轮都能看到的 prompt：它直接决定模型传不传 limit、用绝对路径还是相对路径。所以"超出上限会被截断"必须写在描述里，不能藏着；limit 既给 minimum/maximum 又给默认值，前者是给模型看的约束，后者是服务端自己的兜底——客户端的 schema 校验未必真被执行，服务端必须再 clamp 一次，这里是 200。

三、坑：三个真实翻车点

1. stdout 是协议通道，一行 print 就能断链。 我加过一行 print(f"debug: 收到 {method}") 调试，客户端读到的第一行从 JSON 变成了 debug: 收到 tools/list，json.loads 直接抛 Expecting value。更麻烦的是字节流从此错位，后面每条消息都对不上号，现象看起来像"服务端挂死"，其实服务端还在正常输出。所以日志一律走 stderr，json.dumps 也绝对不能带 indent——格式化成多行，等于把一条消息劈成几条非法帧。同理，ensure_ascii=False 是可以的，中文原样输出不会引入换行；但要小心别手工拼接字符串，工具返回内容里的 \n 会由 json.dumps 转义成 \\n，这才是单行帧能成立的原因。

2. 工具执行失败不等于协议错误。 目录不存在、缺必填参数，这类问题我第一版返回的是 JSON-RPC error（-32602）。结果客户端把整个调用当协议级失败，模型看不到"不是有效目录: /nope"这句话，也就无法自己改参数重试，会话经常直接卡住。正确分层是：参数和运行时错误放在 result 里，isError: true，错误文本放 content——这段文本会回灌给模型，它下一轮就能自我修正。error 只留给真正的协议问题：-32601 方法不存在、-32602 未知工具名、-32700 解析失败。改完之后实测返回的正文是 参数不合法: 不是有效目录: /nope，这段文本会进入模型下一轮的上下文，它才有机会自己换个路径重试。

3. 并发处理和结果截断。 每个请求丢进线程池，好处是慢查询不堵住读循环，代价是响应乱序——协议允许乱序（客户端按 id 匹配），但 stdout 的写入必须加锁，否则多个线程的 write 交错，一行会被切成两段，客户端解析直接崩。另外我这个版本把 initialize 也丢进了池，靠的是客户端"握手完成才发下一个请求"的时序；要更严格就把握手同步处理完再进池。还有一个更隐蔽的错：notifications/initialized 这类消息没有 id，一定不能回包。我早期图省事统一走 send(handle(msg))，给通知也回了一条 id 为空的 result，客户端收到无法匹配的响应后直接断开，日志里只留下一句"连接关闭"。结果截断则是另一道防线：limit 硬上限 200，超出时在末尾补一句"共 N 条命中，已截断到前 200 条"，让模型自己缩小范围，而不是悄悄吞掉。截断不可怕，不告诉模型被截断了才可怕。

四、结论

MCP 的协议层很薄，一个下午能写完；真正花时间的是"边界怎么定义"。给三个可以直接照抄的判断标准：stdout 只写协议帧，日志全走 stderr；工具错误放 isError，协议错误放 error；任何可能很大的返回都设上限，并在文本里显式声明截断。做到这三条，这个服务端挂到 Claude Code、Cherry Studio 这类宿主上基本不会再出"莫名其妙连不上"的问题。