最近这段时间接手的项目里，有一半都在给 AI 客户端挂工具。MCP 这个词从去年年末还只是个新鲜玩意儿，到现在基本成了默认选项。官方 SDK 确实省事，几行代码就能起一个服务器，但第一次通读协议文档的时候我心里一直有个疙瘩：客户端和服务端之间到底传了些什么东西？真的只是发几段 JSON 吗？
后来索性把 SDK 撇开，只用 Python 标准库从头写了一个。整套东西两百行不到，跑通之后很多之前含糊的地方一下就清楚了——比如为什么你 print 一句调试信息就会把整个服务搞崩，比如工具执行报错到底该走哪条返回路径。这篇文章把这个过程完整记下来，附上两个真能用的工具。
stdio 传输层其实就一句话
MCP 支持好几种传输方式，但绝大多数桌面客户端用的都是 stdio：客户端把你的脚本当成子进程拉起来，往它的标准输入里写消息，从标准输出里读回复。没有端口，没有 HTTP 请求头，没有握手包，就是两根管子。
管子里面流的东西，本质上是 JSON-RPC 2.0 协议的消息，而且约定好了一行一条。也就是说，标准输出的每一行都必须是一个完整、可解析的 JSON 对象，除此之外什么都不能有。
一条请求长这样：
{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"read_file","arguments":{"path":"a.txt"}}}对应的响应必须带回同一个 id：
{"jsonrpc":"2.0","id":1,"result":{"content":[{"type":"text","text":"..."}],"isError":false}}规则不多，但每条都得守住：带 id 的请求必须回复，没有 id 的通知绝对不能回复；方法名不存在时返回错误码 -32601；服务端内部异常返回 -32603。
先把消息循环搭起来
我习惯先把最外层的骨架写完再填内容，这样调试的时候能立刻看到原始报文。下面这段是整个服务器的主循环，只做三件事：读一行、解析、分发。
import json
import sys
def send(obj):
sys.stdout.write(json.dumps(obj, ensure_ascii=False) + "n")
sys.stdout.flush()
def handle(msg):
if "id" not in msg:
return # 通知，不回
sys.stderr.write(f"收到 {msg.get('method')}n")
send({"jsonrpc": "2.0", "id": msg["id"], "result": {}})
def main():
while True:
line = sys.stdin.readline()
if not line:
break
line = line.strip()
if not line:
continue
try:
msg = json.loads(line)
except json.JSONDecodeError:
sys.stderr.write(f"无法解析：{line[:120]}n")
continue
handle(msg)
if __name__ == "__main__":
main()注意 send 里那句 flush()。stdout 默认带缓冲，客户端在等你回应的时候你还在攒缓冲区，表现就是“请求发出去了但一直没反应”，这个坑我在一个线上脚本里踩过一次，排查了快半小时。
initialize 不只是握个手
客户端启动子进程之后发的第一条消息一定是 initialize。这一步除了打招呼，还顺带完成了两件实事：协议版本的协商，以及双方能力集的交换。
def handle_initialize(params):
return {
"protocolVersion": params.get("protocolVersion", "2024-11-05"),
"capabilities": {"tools": {"listChanged": False}},
"serverInfo": {"name": "mini-mcp", "version": "0.1.0"},
}protocolVersion 建议原样回显客户端给的值，除非你明确知道自己不支持。乱回一个版本号，部分客户端会直接断开连接。capabilities 里声明了 tools，等于告诉对方“我这边有工具可以调”，客户端才会接着发 tools/list。
紧跟在 initialize 后面的是 notifications/initialized。这条没有 id，是通知，收到之后什么都不用做——真的什么都不用做，下面会讲为什么这个“不回复”很关键。
tools/list 与 tools/call
握手完成后，正常流程是这样走的：客户端先拉工具清单，把每个工具的名字、描述、参数结构塞进模型的上下文；模型决定调用某个工具时，客户端再发 tools/call。
def describe_tools():
return [
{k: v for k, v in item.items() if k != "fn"}
for item in REGISTRY.values()
]
def call_tool(params):
name = params.get("name", "")
args = params.get("arguments") or {}
entry = REGISTRY.get(name)
if entry is None:
return {
"content": [{"type": "text", "text": f"没有名为 {name} 的工具"}],
"isError": True,
}
try:
text = entry["fn"](**args)
except Exception as exc:
return {
"content": [{"type": "text", "text": f"{type(exc).__name__}: {exc}"}],
"isError": True,
}
return {"content": [{"type": "text", "text": text}], "isError": False}这里有个设计上的分界，特别容易搞混：工具自己执行失败，和请求本身格式错误，是两回事。
如果模型让你读一个不存在的文件，这是工具执行失败，应该返回一条正常的结果，只是把 isError 置为 true，错误文本放进 content 里。客户端会把这段文本当成工具输出喂回给模型，模型看到“文件不存在：a.txt”，大概率会换个路径再试一次。
如果返回的是 JSON-RPC 层面的 error，那客户端会认为这条消息在协议层就坏了，模型根本看不到错误内容，于是它卡在那里，不知道该干嘛。我最早的版本全用了后一种写法，结果工具连续失败时模型的表现非常呆，改成 isError 之后明显好转。
填两个真能用的工具
骨架搭好，接下来往里塞东西。用装饰器注册是最省事的，工具函数保持纯 Python，跟协议完全解耦。
import os
import sqlite3
from pathlib import Path
BASE_DIR = Path(os.environ.get("MCP_BASE_DIR", ".")).resolve()
DB_PATH = os.environ.get("MCP_DB", "demo.db")
REGISTRY = {}
def register(name, description, schema):
def deco(fn):
REGISTRY[name] = {
"name": name,
"description": description,
"inputSchema": schema,
"fn": fn,
}
return fn
return deco第一个工具是文件阅读器。既然要交给模型去调，路径校验就必须自己做，不能指望客户端拦着。
@register(
"read_file",
"读取工作目录下的文本文件，最多返回前 200 行。",
{
"type": "object",
"properties": {
"path": {"type": "string", "description": "相对于工作目录的文件路径"},
},
"required": ["path"],
},
)
def read_file(path):
target = (BASE_DIR / path).resolve()
if not target.is_relative_to(BASE_DIR):
raise ValueError("路径越界，拒绝访问")
if not target.is_file():
raise FileNotFoundError(f"找不到文件：{path}")
lines = target.read_text(encoding="utf-8", errors="replace").splitlines()
body = "n".join(lines[:200])
if len(lines) > 200:
body += f"n...（共 {len(lines)} 行，已截断）"
return body or "文件是空的"is_relative_to 是 Python 3.9 才有的方法，比字符串前缀匹配可靠得多——后者在 /data 和 /data-backup 这种名字上会误判。截断也很必要，一份几千行的日志原样丢进上下文，token 直接烧掉一大半，模型反而抓不住重点。
第二个工具查 SQLite。只读模式打开，连驱动层都帮你把写操作挡掉。
@register(
"query_db",
"对内置的 SQLite 数据库执行只读 SELECT 查询，最多返回 50 行。",
{
"type": "object",
"properties": {
"sql": {"type": "string", "description": "一条 SELECT 语句"},
},
"required": ["sql"],
},
)
def query_db(sql):
if not sql.strip().lower().startswith("select"):
raise ValueError("只接受 SELECT 语句")
if not os.path.exists(DB_PATH):
raise FileNotFoundError(f"数据库不存在：{DB_PATH}")
conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True)
try:
cur = conn.execute(sql)
cols = [d[0] for d in cur.description]
rows = cur.fetchmany(50)
finally:
conn.close()
if not rows:
return "查询成功，但没有匹配的数据"
lines = [" | ".join(cols)]
lines += [" | ".join("NULL" if v is None else str(v) for v in row) for row in rows]
return "n".join(lines)输出用竖线分隔的纯文本，不用 JSON。原因是模型读表格型文本比读嵌套 JSON 稳，字段多的时候也不容易看串行。
几个把我卡住的细节
代码到这里就完整了，但真正让它稳定跑起来的是下面这五条。
第一，stdout 是协议通道，任何多余输出都是灾难。我调试时在工具函数里留了一句 print("查到了", rows)，客户端立刻抛 Unexpected token '查'。它按行读取 stdout，每一行都试图当 JSON 解析，你哪怕输出一个空行都会让它崩。所有日志走 sys.stderr，客户端不会读它，但你在终端里能看到。
第二，不要给 json.dumps 加 indent。缩进之后一条消息变成好几行，按行切分的逻辑直接乱套。要给人看的美化版本，请写到 stderr 的日志里。
第三，通知真的不能回。我见过有人收到 notifications/initialized 之后条件反射地返回 {"result":{}}，客户端会因为找不到对应的请求而报错。判断方法就是看 "id" in msg，这个字段的存在与否比任何方法名前缀都可靠。
第四，错误码别乱造。JSON-RPC 预留了几个标准码：-32700 解析错误、-32600 请求无效、-32601 方法不存在、-32603 内部错误。方法不存在的时候返回什么都不会生效，但返回正确的码至少能让客户端日志好看一点。
第五，小心阻塞。整个服务器是单线程顺序处理的，一个工具跑了三十秒，后面所有请求都排队等着。要跑长任务，要么开线程，要么利用 MCP 的进度通知机制把状态推回给客户端——但这两件事都超出了“先跑通”的范围，我一般是等到确实卡住了再加。
手动测一遍，再接到客户端上
不用急着找客户端，直接用 shell 把消息喂进去就能验证。标准输入是文件，服务器会逐行读、逐行回。
python3 server.py <<'EOF'
{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"tester","version":"0.1"}}}
{"jsonrpc":"2.0","method":"notifications/initialized"}
{"jsonrpc":"2.0","id":2,"method":"tools/list"}
{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"read_file","arguments":{"path":"README.md"}}}
EOF正常的输出应该是三行响应，加上 stderr 里几行日志。第二行对应的通知没有任何回应，如果你看到四条输出，说明哪里多回了。
确认没问题，就可以挂到客户端上了。绝大多数客户端用的是同一套配置格式，键名是 mcpServers：
{
"mcpServers": {
"mini-mcp": {
"command": "python3",
"args": ["/绝对路径/server.py"],
"env": {
"MCP_BASE_DIR": "/Users/me/notes",
"MCP_DB": "/Users/me/notes/app.db"
}
}
}
}路径一定写绝对路径。客户端拉起子进程时的工作目录跟你终端里完全不是一回事，用相对路径会得到一个“文件不存在”的诡异结果，而且你会以为是代码问题。
接下来还能往哪走
这套骨架已经能撑起日常用的工具了。再往后，比较自然的扩展方向有三个：一是给工具加进度通知，长任务在 tools/call 的 _meta.progressToken 里带个令牌，服务端定期发 notifications/progress，客户端就能显示进度条；二是实现 resources 和 prompts 两种能力，前者适合暴露大量只读数据，后者适合把固定工作流封装成模板；三是换成 streamable HTTP 传输，让服务能跑在远端，不再是本机子进程。
不过说到底，协议层的东西就这么多。真正决定一个 MCP 服务器好不好用的是工具本身的设计：参数是不是够少、描述是不是够准、出错时给出的信息是不是足够让模型自我修正。这些跟写普通 Python 函数没什么区别，只是现在你的调用方从人变成了模型而已。
