第 12 期给机器人装了日志和告警，半夜挂了能喊出来。但还有个老问题没解决：机器人只会聊它脑子里有的东西。你问它一句"上周那份部署笔记里，令牌桶限流是怎么写的"，它答不上来，开始现编。RAG 能查文档，但每换一个知识库就要改一遍代码。MCP 把这件事变成了标准接口：机器人自己决定什么时候查文件，查哪个文件。

MCP 是模型上下文协议，Anthropic 推的开放标准。说白了就是一套规矩：机器人这边叫 Client，工具那边叫 Server，两边用 JSON-RPC 2.0 通话，走标准输入输出，不用 HTTP，不用装框架。Server 只干一件事：把本地能力包成工具，告诉 Client"我这儿有三个工具，分别是干什么的"。Client 收到用户问题，模型自己判断要不要调工具，调哪个，参数填什么。做完长什么样：kb 目录下放三份笔记，机器人问"令牌桶怎么限流"，Server 把原文找出来交回去，机器人照着原文答，不再现编。适合已经跑起来问答机器人、想给它接本地知识库的人。你不用懂协议细节，照着下面 100 行代码抄就行。

不用 MCP 也能接工具，直接在代码里写个函数调一下就行。那为什么还要多一层协议？直接调函数是写死在代码里的，换个工具就要改机器人；MCP 把工具变成了可插拔的：Server 单独跑，机器人只认 tools/list 返回的清单。今天接三个文件工具，明天再加一个数据库工具，机器人那边一行代码都不用动。工具清单是模型看得见的，模型读着工具的中文说明自己决定调哪个，这才是关键。清单写得越清楚，模型调得越准，工具说明不是给人看的，是给模型看的提示词。

一次完整会话只有 4 步。Client 先发 initialize 握手，Server 回自己的名字和版本号。Client 再发一条 notifications/initialized，意思是"我准备好了"，这条不用回。接着 Client 问 tools/list，Server 把工具清单交出来。再往后，Client 调 tools/call，带上工具名和参数，Server 执行完把结果包成文本返回。数据就这么流：用户问题进来，机器人判断要不要调工具，调了工具读文件，拿原文回去组织答案。Server 和 Client 是两个进程，中间只靠 stdin 和 stdout 传 JSON，一行一条消息。出错的时候回包里没有 result，只有一个 error，带 code 和 message。我的 Server 用了三个码：-32601 是未知方法，-32602 是没有这个工具，-32603 是工具执行失败。Client 收到 error 不能当成正常结果往下传，得先判断有没有 error 字段，有就记下来、换种问法或者直接告诉用户查不到。错误码这层约定写死了，Client 才不用猜"没返回东西到底是空结果还是出事了"。

第一步，目标是把 Server 跑起来，提供三个工具：list_files 列文件，read_file 读全文，search_text 全文搜关键词。代码只用标准库：

import json
import os
import sys

ROOT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "kb")


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


def safe_path(name):
    p = os.path.normpath(os.path.join(ROOT, name))
    if not p.startswith(ROOT):
        raise ValueError("不允许跳出 kb 目录")
    return p


def tool_list_files(_args):
    return {"files": sorted(os.listdir(ROOT))}


def tool_read_file(args):
    name = args["name"]
    with open(safe_path(name), encoding="utf-8") as f:
        return {"name": name, "content": f.read()}


def tool_search_text(args):
    keyword = args["keyword"]
    hits = []
    for name in sorted(os.listdir(ROOT)):
        with open(safe_path(name), encoding="utf-8") as f:
            for i, line in enumerate(f, 1):
                if keyword in line:
                    hits.append({"file": name, "line": i, "text": line.strip()})
    return {"keyword": keyword, "hits": hits}


TOOLS = {
    "list_files": (tool_list_files, "列出 kb 目录下的文件"),
    "read_file": (tool_read_file, "读一个文件全文，参数 name"),
    "search_text": (tool_search_text, "全文搜关键词，参数 keyword"),
}


def handle(msg):
    method = msg.get("method")
    mid = msg.get("id")
    if method == "initialize":
        return {"jsonrpc": "2.0", "id": mid, "result": {
            "protocolVersion": "2024-11-05",
            "serverInfo": {"name": "file-kb-server", "version": "0.1.0"}}}
    if method == "notifications/initialized":
        return None  # notification：不回包
    if method == "tools/list":
        return {"jsonrpc": "2.0", "id": mid, "result": {"tools": [
            {"name": n, "description": d, "inputSchema": {"type": "object"}}
            for n, (_, d) in TOOLS.items()]}}
    if method == "tools/call":
        name = msg["params"]["name"]
        args = msg["params"].get("arguments", {})
        if name not in TOOLS:
            return {"jsonrpc": "2.0", "id": mid, "error": {
                "code": -32602, "message": f"没有这个工具：{name}"}}
        fn, _ = TOOLS[name]
        try:
            data = fn(args)
        except Exception as e:
            return {"jsonrpc": "2.0", "id": mid, "error": {
                "code": -32603, "message": f"工具执行失败：{e}"}}
        return {"jsonrpc": "2.0", "id": mid, "result": {
            "content": [{"type": "text", "text": json.dumps(
                data, ensure_ascii=False)}]}}
    return {"jsonrpc": "2.0", "id": mid, "error": {
        "code": -32601, "message": f"未知方法 {method}"}}


def main():
    log("file-kb-server 启动，工作目录：", ROOT)
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            msg = json.loads(line)
        except json.JSONDecodeError:
            continue
        resp = handle(msg)
        if resp is not None:
            sys.stdout.write(json.dumps(resp, ensure_ascii=False) + "\n")
            sys.stdout.flush()


if __name__ == "__main__":
    main()

验证很简单，直接往 Server 的 stdin 喂一条 tools/list，看它回什么：

echo '{"jsonrpc":"2.0","id":1,"method":"tools/list"}' | python3 mcp_file_server.py

我本地跑出来的返回里，三个工具都在：list_files、read_file、search_text，每个都带一句中文说明。注意启动日志打在 stderr 里，stdout 里只有 JSON。inputSchema 我先写了个最简的 {“type”: “object”}，意思是"参数是个对象"，没展开每个参数的类型。真要上生产，read_file 的 name 得声明成字符串、必填，search_text 的 keyword 也一样，模型照着 schema 填参数，调错的概率小一半。坑在 safe_path：文件名是用户传进来的，不做校验就能用 …/ 跳出 kb 目录读系统文件。这行校验不能省，工具一多最容易漏的就是这一处。

第二步，目标是写个 Client，把 Server 拉起来走完一次完整会话。代码同样只用标准库：

import json
import os
import subprocess
import sys

server = subprocess.Popen(
    [sys.executable, "mcp_file_server.py"],
    stdin=subprocess.PIPE, stdout=subprocess.PIPE,
    stderr=subprocess.PIPE, text=True,
    cwd=os.path.dirname(os.path.abspath(__file__)))


def send(msg):
    server.stdin.write(json.dumps(msg, ensure_ascii=False) + "\n")
    server.stdin.flush()


def recv():
    line = server.stdout.readline()
    return json.loads(line)


def call(method, mid, params=None):
    msg = {"jsonrpc": "2.0", "id": mid, "method": method}
    if params is not None:
        msg["params"] = params
    send(msg)
    return recv()


r = call("initialize", 1, {"protocolVersion": "2024-11-05"})
print("握手返回：", r["result"]["serverInfo"])
send({"jsonrpc": "2.0", "method": "notifications/initialized"})

r = call("tools/list", 2)
print("工具列表：", [t["name"] for t in r["result"]["tools"]])

r = call("tools/call", 3, {"name": "read_file",
                            "arguments": {"name": "faq.txt"}})
print("read_file 结果：")
print(json.loads(r["result"]["content"][0]["text"])["content"])

r = call("tools/call", 4, {"name": "search_text",
                            "arguments": {"keyword": "令牌桶"}})
print("search_text 命中：",
      json.loads(r["result"]["content"][0]["text"])["hits"])

server.stdin.close()
server.wait(timeout=5)
print("server 退出码：", server.returncode)
print("server 日志（stderr）：", server.stderr.read().strip())

我本地跑出来的结果：握手返回名字 file-kb-server、版本 0.1.0；工具列表三个名字全在；read_file 把 faq.txt 全文打出来了；search_text 搜"令牌桶"命中 1 条，rag_notes.txt 第 3 行；server 退出码 0，日志在 stderr 里。id 这个字段每调一次换一个，从 1 开始往上加，Server 靠它把请求和返回对上号。id 写重了也没事，Client 自己能对上，但别图省事全写 1，并发一上来就分不清谁是谁。坑在 notifications/initialized：它是一条 notification，没有 id，Server 不回包。Client 发完这条不能傻等着读一行回包，否则直接卡死。我的 Client 发完就继续往下走，没去读。还有一个细节：Client 读回包用的是 readline，一行一条消息。Server 回包必须一行写完，不能为了好看打 pretty print 换行，一换行 Client 就只读到半条 JSON，解析直接炸。

第三步，目标是看一眼最常见的翻车现场。把 Server 里那行 log 的 stderr 改成 stdout，别的什么都不动，再跑 Client。日志那行"file-kb-server 启动，工作目录：……"就混进了 stdout，Client 读到的第一行不再是 JSON，直接报错：Expecting value: line 1 column 1 (char 0)。这个报错我本地复现过，一模一样。修法只有一条：Server 里所有 print、所有日志，全部写 stderr。调第三方库的时候也要盯着，有些库默认往 stdout 打日志，照样会把协议冲垮。判断标准就一个：stdout 里除了 JSON-RPC 消息，什么都不能有。

第四步，目标是把工具串起来，让机器人真的用它回答问题。模型判断调哪个工具那一步，在真实机器人里由模型完成，这里先用关键词规则代替，只演示"调工具、取原文"的流程：

for question, keyword in [("令牌桶怎么限流？", "令牌桶"),
                          ("问答机器人怎么加记忆？", "记忆")]:
    r = call(2, "tools/call", {"name": "search_text",
                                "arguments": {"keyword": keyword}})
    hits = json.loads(r["result"]["content"][0]["text"])["hits"]
    print(f"问题：{question}")
    print(f"命中 {len(hits)} 条，答案来源：{hits[0]['file']} 第 {hits[0]['line']} 行")
    print("原文：", hits[0]["text"])

跑出来的结果：问"令牌桶怎么限流"，命中 1 条，来源 rag_notes.txt 第 3 行，原文"令牌桶限流：60 行代码，调用次数超了直接拒绝"；问"问答机器人怎么加记忆"，命中 2 条，取第一条 faq.txt 第 1 行。答案的每一句都有出处，机器人不用再现编。坑在"命中 0 条"的时候：search_text 返回空列表，机器人不能硬答，得回一句"知识库里没找到"，再把问题记下来。这一步不做，机器人迟早在没命中的时候开始编。还有一个坑在 read_file：文件一大，全文返回能把上下文撑爆。我的三份笔记都很小，全文返回没问题，真上了几万字的文档，read_file 得加分页参数，一次只返回前 50 行，翻页再接着读。工具返回的文本越短，机器人答得越稳，这笔账得在设计工具的时候就先算好。

避坑清单单独列一遍，都是上面踩过的。stdout 只走 JSON-RPC 消息，日志、报错、打印全部写 stderr。notification 不回包，Client 发完不要等。tools/call 的参数要校验，文件名先过 safe_path，关键词空字符串直接返回空结果。调用要设超时，Server 卡住了 Client 不能无限等，5 秒没回就杀掉重拉。工具一多先写 tools/list 自查，清单里没有的工具，模型再怎么想调也调不到。工具说明要写给模型看：名字说清动作，说明说清输入，参数名别用 a、b 这种缩写，模型看不懂缩写，调错参数的概率直线上升。还有一个坑是我写第一版的时候真踩到的：tools/call 里直接拿工具名去查表，模型一旦报个不存在的工具名，Server 抛 KeyError，整个进程退出码 1，直接没了，连句回包都没有。我本地复现过，调 delete_all，Server 一声不吭就死了。现在的版本先查名字在不在，不在回 -32602，执行出错回 -32603，Server 全程活着。工具一多，模型调错名字是迟早的事，Server 得先学会说"没有这个工具"，而不是直接崩。

照着打勾的清单：kb 目录建好，三份笔记放进去；Server 单独跑一次，tools/list 能返回三个工具；Client 跑通，握手、列工具、读文件、搜关键词四步全过；故意把日志打到 stdout，确认报错复现过、也确认修好了；再跑一遍问答演示，答案能指到具体文件和行号。

下期预告：第 14 期给机器人做套考试题，离线评测集加自动打分，改了提示词先跑一遍题再上线。这个系列在专栏《Python实战》里连着更新，从第 1 期手搓 RAG 检索引擎一路写到上线运维，关注后专栏里能连着看。

这篇建议收藏。MCP 这套东西看着是协议，其实就四步加一条铁律，收藏这篇，下次给机器人接工具的时候照着抄，半小时就能跑起来。

你给自己的机器人接过外部工具吗？接的是什么，本地文件、数据库，还是别的 API？

署名：Cyrus