MCP Server 实战:stdout 污染与工具描述陷阱
MCP Server 实战:stdout 污染与工具描述陷阱
一、问题:装饰器写得出来,工具却调不通
MCP(Model Context Protocol)如今被主流客户端广泛支持,Python SDK 也把服务端封装得很薄:一个 @mcp.tool() 装饰器,函数就变成了模型可以调用的工具。很多人第一次跑通 Demo 后,会得出"这东西半小时就能接完"的结论。
但真正接入业务时,故障往往不在协议本身,而在两个被忽略的边界上:进程的标准输出和工具的语义描述。前者会让连接直接断掉且报错信息毫无指向性,后者会让模型频繁选错工具或传错参数,表现为"模型有点笨",实际上是你没把工具当成给模型看的 API 来设计。
下面用一个可运行的日志查询 Server 串起完整方案,再逐个讲清楚踩过的坑。
二、方案:一个最小但完整的 MCP Server
依赖只有官方 SDK(pip install mcp),代码保存为 server.py。示例数据是虚构的,替换成你自己的数据源即可。
import json
import logging
import sys
from mcp.server.fastmcp import Context, FastMCP
# 关键:日志一律写 stderr,绝不碰 stdout
logging.basicConfig(
level=logging.INFO,
stream=sys.stderr,
format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
log = logging.getLogger("log-query")
mcp = FastMCP("log-query")
SAMPLE_LOGS = [
{"service": "order", "level": "ERROR", "msg": "db connect timeout after 3s"},
{"service": "order", "level": "INFO", "msg": "order 1024 created"},
{"service": "payment", "level": "ERROR", "msg": "callback signature invalid"},
]
@mcp.tool()
async def query_logs(
ctx: Context,
service: str,
level: str = "ERROR",
limit: int = 20,
) -> str:
"""按服务名与级别查询最近的日志,结果按时间倒序返回。
参数约定:
service —— 服务名,精确匹配,取值见 config://services 资源;
level —— 只能是 DEBUG / INFO / WARN / ERROR,大小写敏感;
limit —— 返回条数,1~50,超出会被夹到边界值,默认 20。
当没有匹配日志时返回 not_found 提示,而不是空数组;
单次结果体积有上限,需要更多时请缩小 level 或分页查询。
"""
await ctx.info(f"query service={service} level={level} limit={limit}")
if level not in {"DEBUG", "INFO", "WARN", "ERROR"}:
raise ValueError(f"level 非法:{level},只接受 DEBUG/INFO/WARN/ERROR")
if not service:
raise ValueError("service 不能为空")
limit = max(1, min(limit, 50))
rows = [r for r in SAMPLE_LOGS if r["service"] == service and r["level"] == level][:limit]
log.info("hit %d rows", len(rows))
if not rows:
return json.dumps(
{"result": "not_found", "service": service, "level": level},
ensure_ascii=False,
)
return json.dumps(
{"count": len(rows), "has_more": False, "items": rows},
ensure_ascii=False,
indent=2,
)
@mcp.resource("config://services")
def list_services() -> str:
"""返回当前可查询的服务清单。"""
return "\n".join(sorted({r["service"] for r in SAMPLE_LOGS}))
if __name__ == "__main__":
mcp.run() # 默认 stdio 传输
客户端侧只需要一段配置,注意 command 必须指向装了 mcp 的那个解释器,路径用绝对路径:
{
"mcpServers": {
"log-query": {
"command": "/usr/bin/python3",
"args": ["/abs/path/server.py"]
}
}
}
调试时不必每次都重启客户端,用官方 Inspector 直接对接:npx @modelcontextprotocol/inspector python server.py。它会把 tools/list 的原始返回、每次 tools/call 的入参和结果都摊开,排查"模型到底传了什么"最有效。
三、坑:四个真实会卡住你的地方
坑一:stdout 是协议通道,不是日志通道。
stdio 传输下,JSON-RPC 消息就是通过 stdin/stdout 以"一行一条"的方式收发的。任何一次 print()、第三方库的启动 banner、tqdm 进度条、甚至某些库默认的 logging handler,都会往 stdout 插一行非 JSON 文本。客户端解析失败后通常只会报一句 JSON 语法错误然后断开,不会告诉你"是你在第 37 行 print 了东西"。上面代码里 logging.basicConfig(stream=sys.stderr) 不是可选项,而是硬约束。如果某个依赖不受控地写 stdout,稳妥做法是在启动后立刻把 sys.stdout 重定向到 stderr,再让 SDK 从原始 fd 读。
坑二:工具描述是提示词,不是代码注释。
tools/list 里的 description 是模型选择工具、填写参数的唯一依据。写"查询日志"四个字,模型就无法判断 service 要不要精确匹配、level 是否大小写敏感、limit 上限多少。上面把单位、取值范围、越界行为、空结果的返回形态全写进 docstring,模型的表现会明显稳定。反过来,工具数量多的时候,名字也要有区分度——query_logs 和 search_logs 在描述模糊时几乎等价,模型会随机挑一个。
坑三:返回值直接进上下文,必须自己定预算。
tools/call 的文本结果会被原样塞进模型上下文。一个"把最近 1000 条错误日志都返回"的实现,会把上下文挤满,还让模型在长文本里找不到答案。正确做法是服务端截断并显式给出 count、has_more,把"要不要继续取"的决策权交给模型,而不是把全量数据丢过去。示例里返回 JSON 而不是裸文本,也是为了让模型更容易判断条数。
坑四:长任务不报进度,只能等超时。
规范提供了 notifications/progress,前提是客户端在请求的 _meta 里带了 progressToken,服务端再用 ctx.report_progress(...) 上报。如果不实现,一个跑三分钟的任务在客户端看来就是"没有任何响应"。这类任务还应该在描述里写明预期耗时,让模型决定是否值得发起——否则它会为了一个闲聊问题去调用一个重任务。
四、结论
MCP 把"工具接入"的工程成本压得很低,但压低的是连接成本,不是设计成本。真正决定一个 Server 好不好用的是三件事:stdout 是否干净、描述是否把参数约束说清楚、返回体积是否在上下文预算之内。这三件事都不在协议文档的显眼位置,却几乎决定了第一次接入是半小时搞定还是排查一整天。
把工具当成给模型看的公开 API 来设计,比把它当成给自己写的函数,要重要得多。
- 点赞
- 收藏
- 关注作者
评论(0)