添加审计日志、模式校验、超时与调用限制

目标

给上一个实验中加固过的服务器加上生产环境运行所需的四样东西。记录每次调用的审计日志、执行前的参数 schema 校验、按工具设置的执行时间上限、对破坏性工具的调用次数限制。并且还要写一个读取并汇总该日志的工具。

为什么重要

出了事故之后的第一个问题是“谁、在什么时候、用什么参数调用了什么”。没有审计日志，就回答不了这个问题，于是同样的事故会再次发生。规范写道，服务器必须验证所有工具输入并限制调用频率（MUST），客户端应对工具调用设置超时，并为审计记录使用情况（SHOULD）。把这些句子原样写成代码，就是本实验。尤其是超时，不能只指望客户端——服务器如果不自己中断，一个卡住的工具就会占住整个会话。

步骤

保存 /root/mcp/audit/seed.sql，并加载到 /root/mcp/audit/shop.db。customers 表要有 5 行，orders 表要有 8 行。
创建 /root/mcp/audit/server.py。它要回答 initialize，并在 tools/list 中给出 list_customers、count_orders、delete_order（需要 confirm）三个工具。数据库路径为 MCP_DB（默认 /root/mcp/audit/shop.db）。
把每一次 tools/call 都以一行 JSON 记录到 /root/mcp/audit/audit.jsonl（要能通过环境变量 MCP_AUDIT_LOG 更改）。键为 ts、tool、arguments、ok、duration_ms，如果工具给出了 isError，则 ok 为 false。
在执行工具之前，用 inputSchema 检查参数。缺少 required 项，或者 type 不符（例如 status 传入数字、id 传入字符串）时，不执行，以 JSON-RPC 错误 -32602 回答。
增加工具 slow_report（参数 seconds，integer），并把单个工具的执行时间限制为 2 秒（环境变量 MCP_TOOL_TIMEOUT，默认 2）。超出时，以带有 isError: true 和 timeout 的文本回答，并在审计日志中记为 ok: false。seconds: 1 必须正常结束。
在 initialize 的 capabilities 中声明 logging，并在每次工具调用时，在响应之前把 notifications/message 通知（level 取 RFC 5424 级别之一，另有 logger、data）通过 stdout 发出。失败的调用使用 error 级别。
让 delete_order 在一个会话中最多只允许 3 次。第四次调用不删除，以带有 isError: true 和 rate limit 的文本回答。
创建 /root/mcp/audit/summary.py。读取 audit.jsonl，把每个工具的 {"calls": n, "failed": m} 写入 /root/mcp/audit/summary.json（如果有环境变量 MCP_SUMMARY_OUT，就写到该路径）。真实的 audit.jsonl 里必须已经积累了至少五行调用记录。

参考

审计的一行，只要用 open(path, "a") 追加 json.dumps({...}) 即可。ts 用 datetime.now(timezone.utc).isoformat()，duration_ms 是 time.monotonic() 的差值。
schema 检查只看 required 和 properties[*].type 两项就足够了。Python 里 bool 是 int 的子类型，所以在 integer 的位置传入 true 的情况要另外拦截。
执行时间上限用 signal.signal(signal.SIGALRM, ...) 和 signal.alarm(초)（占位符为秒数）来设置，结束后用 signal.alarm(0) 解除。服务器是单线程的，这种方式最简单。
日志通知不是响应，而是通知——不要放入 id。级别是 debug、info、notice、warning、error、critical、alert、emergency 之一。
评分器会用学员数据库的临时副本（MCP_DB）和临时审计日志（MCP_AUDIT_LOG）来测试破坏性调用。只有第 8 步查看学员真实的 audit.jsonl。
常见错误 1：没有把超时的调用记入审计日志。恰恰是失败的调用才应该记录。
常见错误 2：把 schema 校验失败以 isError: true 返回。参数有误的请求属于在执行之前就被拒绝的协议错误。

创建商店数据库

保存 /root/mcp/audit/seed.sql，并加载到 /root/mcp/audit/shop.db。customers 表要有 5 行，orders 表要有 8 行。

sqlite3 可以用 sqlite3 shop.db < seed.sql 整个执行文件。用 Python 的话是 sqlite3.connect(...).executescript(open(...).read())。如果向已有的数据库再次加载，会报表已存在的错误，所以先把它删掉。

搭起基础服务器

创建 /root/mcp/audit/server.py。它要回答 initialize，并在 tools/list 中给出 list_customers、count_orders、delete_order（需要 confirm）三个工具。数据库路径为 MCP_DB（默认 /root/mcp/audit/shop.db）。

只要从上一个实验的 v3 中去掉白名单即可（本实验会把三个工具全部打开）。delete_order 在 confirm 不是 true 时不删除。

记录每一次调用

把每一次 tools/call 都以一行 JSON 记录到 /root/mcp/audit/audit.jsonl（要能通过环境变量 MCP_AUDIT_LOG 更改）。键为 ts、tool、arguments、ok、duration_ms，如果工具给出了 isError，则 ok 为 false。

包住执行工具的那个函数即可——记录开始时间，根据结果的 isError 确定 ok，再追加一行。一行的形式是 {"ts": "2026-01-01T00:00:00+00:00", "tool": "get_weather", "arguments": {"location": "Seoul"}, "ok": true, "duration_ms": 12, "error": null}。评分器会把 MCP_AUDIT_LOG 设为临时路径，在两次调用（一次正常、一次未知状态值）之后，检查是否有两行。

执行前检查参数

在执行工具之前，用 inputSchema 检查参数。缺少 required 项，或者 type 不符（例如 status 传入数字、id 传入字符串）时，不执行，以 JSON-RPC 错误 -32602 回答。

写一个小函数，把 schema 的 required 列表和 properties 的 type 对应到 Python 类型即可（string→str、integer→int、boolean→bool）。检查失败发生在工具执行之前，所以是协议错误（-32602，Invalid params）。

给工具执行时间设上限

增加工具 slow_report（参数 seconds，integer），并把单个工具的执行时间限制为 2 秒（环境变量 MCP_TOOL_TIMEOUT，默认 2）。超出时，以带有 isError: true 和 timeout 的文本回答，并在审计日志中记为 ok: false。seconds: 1 必须正常结束。

设置 signal.alarm(TOOL_TIMEOUT)，在 SIGALRM 处理函数里抛出异常，即使在 time.sleep 期间也能脱身。结束后 signal.alarm(0)。评分器会发送 seconds: 6，并测量 5 秒内是否收到 isError 响应。

通过协议发送日志

在 initialize 的 capabilities 中声明 logging，并在每次工具调用时，在响应之前把 notifications/message 通知（level 取 RFC 5424 级别之一，另有 logger、data）通过 stdout 发出。失败的调用使用 error 级别。

通知形如 {"jsonrpc":"2.0","method":"notifications/message","params":{"level":"info","logger":"...","data":{...}}}，没有 id。在写响应之前先写一行即可。与 stderr 日志不同，这是客户端以结构化方式接收的日志。

给删除类工具限制次数

让 delete_order 在一个会话中最多只允许 3 次。第四次调用不删除，以带有 isError: true 和 rate limit 的文本回答。

会话就是一个进程，所以用模块级全局字典按工具统计调用次数即可。评分器会用 confirm=true 对临时副本数据库调用四次，检查行数是否恰好少了 3 行。

读取审计日志并汇总

创建 /root/mcp/audit/summary.py。读取 audit.jsonl，把每个工具的 {"calls": n, "failed": m} 写入 /root/mcp/audit/summary.json（如果有环境变量 MCP_SUMMARY_OUT，就写到该路径）。真实的 audit.jsonl 里必须已经积累了至少五行调用记录。

逐行 json.loads，按 tool 计数，ok 为假就累加 failed。让脚本把日志路径作为第一个参数接收，评分器就可以用临时日志来验算。如果前面的步骤都运行过，audit.jsonl 里已经有好几行了。