给 MCP Server 加调用链日志和大输出落盘：一个装饰器搞定，附 3 个实测踩坑（Python SDK v2）

中文版正在随流水线自动整理，稍后刷新即可；可先阅读下方英文原文。

本文由 AI图鉴自动抓取并翻译，版权归原作者所有；如需查看原文请点下方链接。

最近 MCP 相关的开源项目里，有两类需求反复出现：一是看清 Agent 到底调了哪个工具、传了什么参数、花了多久；二是工具返回太长，把模型的上下文挤爆。比如让 Agent 在日志里搜 ERROR，一次返回十几万字符，后面的推理基本就废了。

这篇不引入任何框架，用一个 Python 装饰器给 MCP 工具加两层保护：

- 调用链日志：每次调用写一行 JSONL（trace_id、工具名、参数、耗时、成功/失败、输出长度），敏感参数自动打码；

- 大输出落盘：返回文本超过阈值就写到文件，只把「摘要 + 前 10 行 + 怎么继续取」回给模型，再配一个分页读取工具。

最后附 3 个我实际跑出来的坑：装饰器顺序、functools.wraps、stdio 模式下的工作目录。

环境：Linux，Python 3.13.5，mcp 2.3.0。文中输出都是真实运行结果；演示日志是脚本随机生成的示意数据。

一、准备环境和演示数据

python3 -m venv .venv && source .venv/bin/activate

pip install "mcp>=2,<3"

生成一份 2 万行的演示日志：

"""生成一份演示用的假日志（示意数据，非真实业务）"""

import random

from pathlib import Path

random.seed(42)

Path("logs").mkdir(exist_ok=True)

levels = ["INFO"] * 8 + ["WARN", "ERROR"]

with open("logs/app.log", "w", encoding="utf-8") as f:

for i in range(20000):

lv = random.choice(levels)

f.write(f"2026-10-28 10:{i//600%60:02d}:{i%60:02d} {lv} job={i%37} msg=task step {i}\n")

print("ok")

二、核心：guard.py

思路很简单：用 inspect.signature 把实参绑定成字典记下来，调用前后各打一次点；同步、异步工具分开包装；返回值是字符串且超长时，替换成摘要。

"""给 MCP 工具加两层保护：调用链 JSONL 日志 + 大输出落盘"""

import functools, inspect, json, time, uuid

from pathlib import Path

BASE = Path(__file__).resolve().parent # 不依赖启动时的工作目录

TRACE_FILE = BASE / "traces" / "calls.jsonl"

ARTIFACT_DIR = BASE / "artifacts"

MAX_INLINE_CHARS = 2000 # 超过这个长度就落盘，只回摘要

SECRET_KEYS = {"token", "password", "api_key", "secret"}

def _redact(args: dict) -> dict:

out = {}

for k, v in args.items():

if k.lower() in SECRET_KEYS:

out[k] = "***"

else:

s = repr(v)

out[k] = v if len(s) <= 200 else s[:200] + "...(truncated)"

return out

def _write_trace(rec: dict) -> None:

TRACE_FILE.parent.mkdir(parents=True, exist_ok=True)

with TRACE_FILE.open("a", encoding="utf-8") as f:

f.write(json.dumps(rec, ensure_ascii=False, default=str) + "\n")

def _spill(text: str, tool: str) -> str:

"""长文本写文件，返回给模型的是摘要 + 取回方式"""

ARTIFACT_DIR.mkdir(exist_ok=True)

aid = f"{tool}-{uuid.uuid4().hex[:8]}"

(ARTIFACT_DIR / f"{aid}.txt").write_text(text, encoding="utf-8")

lines = text.splitlines()

head = "\n".join(lines[:10])

return (f"[输出共 {len(text)} 字符 / {len(lines)} 行，已存为 artifact_id={aid}]\n"

f"前 10 行：\n{head}\n"

f"需要更多内容请调用 read_artifact(artifact_id='{aid}', offset=10, limit=50)")

def guarded(fn):

"""装饰器：记录调用链；返回值是 str 且过长时落盘"""

sig = inspect.signature(fn)

def _before(args, kwargs):

bound = sig.bind_partial(*args, **kwargs)

return {"trace_id": uuid.uuid4().hex[:12], "tool": fn.__name__,

"args": _redact(dict(bound.arguments)), "t0": time.perf_counter()}

def _after(rec, result=None, err=None):

t0 = rec.pop("t0")

rec["ms"] = round((time.perf_counter() - t0) * 1000, 2)

rec["ok"] = err is None

if err is not None:

rec["error"] = f"{type(err).__name__}: {err}"

spilled = False

if isinstance(result, str):

rec["out_chars"] = len(result)

if len(result) > MAX_INLINE_CHARS:

result, spilled = _spill(result, fn.__name__), True

rec["spilled"] = spilled

rec["ts"] = time.strftime("%Y-%m-%dT%H:%M:%S%z")

_write_trace(rec)

return result

if inspect.iscoroutinefunction(fn):

@functools.wraps(fn)

async def wrapper(*args, **kwargs):

rec = _before(args, kwargs)

try:

result = await fn(*args, **kwargs)

except Exception as e:

_after(rec, err=e)

raise

return _after(rec, result)

else:

@functools.wraps(fn)

def wrapper(*args, **kwargs):

rec = _before(args, kwargs)

try:

result = fn(*args, **kwargs)

except Exception as e:

_after(rec, err=e)

raise

return _after(rec, result)

return wrapper

def read_artifact_text(artifact_id: str, offset: int = 0, limit: int = 50) -> str:

# 只允许读 artifacts 目录下的文件，防止路径穿越

if not artifact_id.replace("-", "").replace("_", "").isalnum():

raise ValueError("非法 artifact_id")

p = ARTIFACT_DIR / f"{artifact_id}.txt"

if not p.exists():

raise ValueError(f"artifact 不存在：{artifact_id}")

lines = p.read_text(encoding="utf-8").splitlines()

limit = max(1, min(limit, 200))

chunk = lines[offset:offset + limit]

more = offset + limit < len(lines)

return "\n".join(chunk) + (f"\n[还有 {len(lines) - offset - limit} 行]" if more else "\n[已到末尾]")

几个设计点：

- trace 写 JSONL：一行一条，追加写，grep/jq/pandas 都能直接用，不需要起任何服务；

- 参数打码 + 截断：参数名命中 token/password/api_key/secret 的一律写 ***，超过 200 字符的截断，避免日志里出现密钥或者巨大的输入；

- 落盘只处理 str 返回：如果你的工具返回的是 Pydantic 结构化结果，这个装饰器不会动它，需要的话可以自己加一个序列化后判断长度的分支；

- read_artifact 限制 id 字符和单次最多 200 行，防止被拿来读任意文件或一次又读爆上下文。

三、Server：server.py

一个「日志检索」Server，包含同步工具、异步工具，以及取回落盘内容的 read_artifact：

"""日志检索 MCP Server：演示调用链日志 + 大输出落盘"""

import asyncio

from pathlib import Path

from mcp.server import MCPServer

from mcp.server.mcpserver.exceptions import ToolError

from guard import guarded, read_artifact_text

mcp = MCPServer("log-search", version="0.1.0")

LOG_DIR = (Path(__file__).resolve().parent / "logs")

@mcp.tool()

@guarded

def search_log(filename: str, keyword: str, max_hits: int = 5000) -> str:

"""在 logs/ 目录下的日志文件里按关键词检索，返回匹配行（行号: 内容）。"""

p = (LOG_DIR / filename).resolve()

if LOG_DIR not in p.parents or not p.is_file():

raise ToolError(f"文件不存在或不在 logs/ 下：{filename}")

hits = []

with p.open(encoding="utf-8", errors="replace") as f:

for no, line in enumerate(f, 1):

if keyword in line:

hits.append(f"{no}: {line.rstrip()}")

if len(hits) >= max_hits:

break

return "\n".join(hits) if hits else "没有匹配行"

@mcp.tool()

@guarded

async def slow_count(filename: str) -> str:

"""异步工具示例：统计日志行数（故意 sleep 模拟慢 IO）。"""

await asyncio.sleep(0.3)

p = (LOG_DIR / filename).resolve()

if LOG_DIR not in p.parents or not p.is_file():

raise ToolError(f"文件不存在：{filename}")

return f"{filename} 共 {sum(1 for _ in p.open(encoding='utf-8'))} 行"

@mcp.tool()

def read_artifact(artifact_id: str, offset: int = 0, limit: int = 50) -> str:

"""分页读取之前落盘的大输出。"""

try:

return read_artifact_text(artifact_id, offset, limit)

except ValueError as e:

raise ToolError(str(e))

if __name__ == "__main__":

mcp.run() # 默认 stdio

注意 search_log 里做了路径校验：解析后的路径必须在 logs/ 目录下，否则直接抛 ToolError。给 Agent 用的文件类工具，这一步不能省。

四、跑起来看效果

用 SDK 自带的内存内 Client 直接连 Server 对象测试，不用起进程：

import asyncio, json

from mcp import Client

from server import mcp

async def main():

async with Client(mcp) as c:

tools = await c.list_tools()

for t in tools.tools:

print(t.name, json.dumps(t.input_schema.get("properties", {}), ensure_ascii=False))

r = await c.call_tool("search_log", {"filename": "app.log", "keyword": "ERROR"})

text = r.content[0].text

print("\n--- search_log 返回给模型的内容 ---\n" + text)

aid = text.split("artifact_id=")[1].split("]")[0]

r2 = await c.call_tool("read_artifact", {"artifact_id": aid, "offset": 10, "limit": 3})

print("\n--- read_artifact ---\n" + r2.content[0].text)

r3 = await c.call_tool("search_log", {"filename": "../server.py", "keyword": "import"})

print("\n--- 路径穿越 ---\n", r3.is_error, r3.content[0].text)

r4 = await c.call_tool("slow_count", {"filename": "app.log"})

print("\n--- slow_count ---\n", r4.content[0].text)

asyncio.run(main())

运行结果（截取）：

search_log {"filename": {"title": "Filename", "type": "string"}, "keyword": {"title": "Keyword", "type": "string"}, "max_hits": {"default": 5000, "title": "Max Hits", "type": "integer"}}

slow_count {"filename": {"title": "Filename", "type": "string"}}

read_artifact {"artifact_id": {"title": "Artifact Id", "type": "string"}, "offset": {"default": 0, "title": "Offset", "type": "integer"}, "limit": {"default": 50, "title": "Limit", "type": "integer"}}

--- search_log 返回给模型的内容 ---

[输出共 117400 字符 / 2003 行，已存为 artifact_id=search_log-b120b6c5]

前 10 行：

10: 2026-10-28 10:00:09 ERROR job=9 msg=task step 9

18: 2026-10-28 10:00:17 ERROR job=17 msg=task step 17

26: 2026-10-28 10:00:25 ERROR job=25 msg=task step 25

42: 2026-10-28 10:00:41 ERROR job=4 msg=task step 41

52: 2026-10-28 10:00:51 ERROR job=14 msg=task step 51

54: 2026-10-28 10:00:53 ERROR job=16 msg=task step 53

73: 2026-10-28 10:00:12 ERROR job=35 msg=task step 72

92: 2026-10-28 10:00:31 ERROR job=17 msg=task step 91

105: 2026-10-28 10:00:44 ERROR job=30 msg=task step 104

107: 2026-10-28 10:00:46 ERROR job=32 msg=task step 106

需要更多内容请调用 read_artifact(artifact_id='search_log-b120b6c5', offset=10, limit=50)

--- read_artifact ---

120: 2026-10-28 10:00:59 ERROR job=8 msg=task step 119

124: 2026-10-28 10:00:03 ERROR job=12 msg=task step 123

147: 2026-10-28 10:00:26 ERROR job=35 msg=task step 146

[还有 1990 行]

--- 路径穿越 ---

True Error executing tool search_log: 文件不存在或不在 logs/ 下：../server.py

--- slow_count ---

app.log 共 20000 行

可以看到：

- 工具 schema 正常，参数名、类型、默认值都在，说明装饰器没把签名弄丢；

- search_log 实际匹配了 2003 行、117400 字符，模型只拿到了几百字的摘要和取回方式；

- 路径穿越被拦下，返回 is_error=True；

- 异步工具正常工作。

traces/calls.jsonl 的内容：

{"trace_id": "630b7778e0c0", "tool": "search_log", "args": {"filename": "app.log", "keyword": "ERROR", "max_hits": 5000}, "ms": 2.03, "ok": true, "out_chars": 117400, "spilled": true, "ts": "2026-10-28T11:08:21+0800"}

{"trace_id": "036c7276bd8b", "tool": "search_log", "args": {"filename": "../server.py", "keyword": "import", "max_hits": 5000}, "ms": 0.13, "ok": false, "error": "ToolError: 文件不存在或不在 logs/ 下：../server.py", "spilled": false, "ts": "2026-10-28T11:08:21+0800"}

{"trace_id": "8151a9bd99e4", "tool": "slow_count", "args": {"filename": "app.log"}, "ms": 301.53, "ok": true, "out_chars": 17, "spilled": false, "ts": "2026-10-28T11:08:22+0800"}

再写个十几行的汇总脚本，按工具看调用次数、失败数、落盘次数和耗时：

"""汇总 traces/calls.jsonl：每个工具的调用次数、失败数、落盘次数、平均/最大耗时"""

import json, statistics

from collections import defaultdict

from guard import TRACE_FILE

stat = defaultdict(lambda: {"n": 0, "fail": 0, "spill": 0, "ms": []})

for line in TRACE_FILE.read_text(encoding="utf-8").splitlines():

r = json.loads(line)

s = stat[r["tool"]]

s["n"] += 1

s["fail"] += not r["ok"]

s["spill"] += r.get("spilled", False)

s["ms"].append(r["ms"])

print(f"{'tool':<12}{'calls':>6}{'fail':>6}{'spill':>6}{'avg_ms':>9}{'max_ms':>9}")

for tool, s in sorted(stat.items()):

print(f"{tool:<12}{s['n']:>6}{s['fail']:>6}{s['spill']:>6}"

f"{statistics.mean(s['ms']):>9.2f}{max(s['ms']):>9.2f}")

tool calls fail spill avg_ms max_ms

search_log 2 1 1 1.08 2.03

slow_count 1 0 0 301.53 301.53

Agent 跑长任务出问题时，先看这张表：哪个工具失败多、哪个工具慢、哪个工具老在输出超长内容，基本就能定位。

五、踩坑记录

坑 1：装饰器顺序写反，日志一条都没有

我一开始顺手写成了这样：

@guarded

@mcp.tool()

def wrong_order(x: int) -> str:

return "x" * x

调用 wrong_order(x=5000)，返回了完整的 5000 个字符，trace 文件新增 0 行：

wrong_order 返回长度: 5000

新增 trace 行数: 0

原因：@mcp.tool() 在注册时就把它收到的那个函数存进了工具表。写反之后，注册的是原始函数，guarded 包出来的新函数根本没人调用。

解决：@mcp.tool() 必须在最外层，@guarded 紧贴函数定义。

坑 2：包装函数不用 functools.wraps，工具名和参数全乱了

一个「看起来没问题」的包装器：

def naive(fn):

def wrapper(*args, **kwargs):

return fn(*args, **kwargs)

return wrapper

挂上去之后 list_tools 的结果：

工具名: ['wrong_order', 'wrapper']

no_wraps schema: {"type": "object", "properties": {"args": {"title": "args", "type": "string"}, "kwargs": {"title": "kwargs", "type": "string"}}, "required": ["args", "kwargs"], "title": "wrapperArguments"}

工具名变成了 wrapper，参数变成了两个必填的字符串 args、kwargs，按正常参数调用直接校验失败：

Error executing tool wrapper: 2 validation errors for wrapperArguments

args

Field required [type=missing, input_value={'x': 1}, input_type=dict]

原因：SDK 是从函数的 __name__、签名和类型注解生成工具名和 JSON Schema 的。functools.wraps 会复制 __name__、__doc__、__annotations__，并设置 __wrapped__，inspect.signature 会顺着 __wrapped__ 找到原函数的签名。

解决：所有包装 MCP 工具的装饰器都要加 @functools.wraps(fn)；异步工具的包装函数也要是 async def，不然 await 的结果会是协程对象。

坑 3：stdio 模式下相对路径跟着客户端的工作目录走

最初 guard.py 和 server.py 里写的是相对路径 Path("traces/calls.jsonl")、Path("logs")。在项目目录里测试一切正常，换一个目录用 stdio 方式启动 Server：

import asyncio, sys

from mcp import Client, StdioServerParameters

SERVER = "/path/to/server.py"

params = StdioServerParameters(command=sys.executable, args=[SERVER])

async def main():

async with Client(params) as c:

r = await c.call_tool("slow_count", {"filename": "app.log"})

print(r.is_error, r.content[0].text)

asyncio.run(main())

cd /tmp/otherdir && python /path/to/client_stdio.py

结果：

True Error executing tool slow_count: 文件不存在：app.log

而且 /tmp/otherdir 下面多出了一个 traces/ 目录：日志写到了别处。

原因：stdio 模式下 Server 是客户端拉起的子进程，工作目录继承自客户端。各种 MCP 客户端的启动目录并不一定是你的项目目录。

解决：所有路径都基于 Path(__file__).resolve().parent 拼，也就是现在代码里的 BASE 和 LOG_DIR。改完后在 /tmp/otherdir 再跑一次：

False app.log 共 20000 行

trace 也老老实实写回了项目目录下的 traces/calls.jsonl。

六、还能往哪走

- 阈值按工具配置：检索类工具 2000 字符可能都嫌多，生成类工具可以放宽，把 MAX_INLINE_CHARS 改成装饰器参数即可；

- artifact 清理：落盘文件按时间定期删，或者在 trace 里记录 artifact_id，任务结束后统一清；

- 把 trace_id 透

Auto-fetched and translated by AI Atlas. All rights belong to the original author.