"""从运行中的 MCP 服务器生成 docs/MCP.md 的「工具参考」一节。 这一节占文档大半，手工维护一定会跟代码漂移，所以由运行时 schema 生成： 起 stdio 子进程 -> list_tools -> 按分组渲染 -> 从「# 工具参考」起到文件末尾整段替换。 用法： uv run python scripts/gen_mcp_docs.py # 重新生成并写回 uv run python scripts/gen_mcp_docs.py --check # 只比对；有漂移退出码 1 uv run python scripts/gen_mcp_docs.py --diff # 比对并打印差异前 30 处 新增工具时必须在 GROUPS 里登记它属于哪一组，否则本脚本报错退出——这是刻意的闸门： 加了工具却不进文档，比不进更糟。 本文件刻意不写反引号与反斜杠（都用 chr 生成），因为它是从别的脚本里拼出来写盘的。 """ from __future__ import annotations import argparse import asyncio import json import sys from pathlib import Path from typing import Any sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from mcp.client.stdio import StdioServerParameters # noqa: E402 from mcp import Client # noqa: E402 ROOT = Path(__file__).resolve().parents[1] DOC = ROOT / "docs" / "MCP.md" HEADING = "# 工具参考" PREFIX = "mcp__arb__" BACKTICK = chr(96) NL = chr(10) # 分组：标题 -> (组说明, 工具顺序)。顺序就是文档里的出现顺序。 GROUPS: list[tuple[str, str, list[str]]] = [ ( "会话与材料", "每个报销任务是一个会话目录 cases//，阶段由产物推导。", ["case_create", "case_upload_link", "case_list", "case_status", "case_add_files", "case_inbox", "case_reset"], ), ( "抽取交接", "渲染页面与读文本层由宿主 Agent 自己判断，本服务不调用任何模型。", [ "case_render_pages", "case_read_text", "extraction_spec", "case_save_extraction", "case_patch_extraction", ], ), ( "匹配与校验", "金额匹配是确定性的；完整性的裁判是规则文件，不是宿主。", ["match_payments", "info_skeleton", "save_info", "validate_info", "rules_get"], ), ( "文档产物", "需要 Windows + Excel（COM）。Word 出库单已下线，只剩申购单。", ["doc_requisition"], ), ( "浏览器分步填报", "常驻会话加分步执行；提交由人在财务系统里做，工具不含提交动作。" "断线用 browser_open_form 按单据编号找回原单续填，别重开新单。", [ "browser_open", "browser_login", "browser_new_form", "browser_open_form", "browser_fill", "browser_page_state", "browser_steps", "browser_close", ], ), ] def type_of(spec: dict[str, Any]) -> str: """参数类型：anyOf 用斜杠连接，保持 schema 里的原始顺序。""" if "anyOf" in spec: return "/".join(str(part.get("type", "?")) for part in spec["anyOf"] if isinstance(part, dict)) raw = spec.get("type") if isinstance(raw, list): return "/".join(str(item) for item in raw) return str(raw or "?") def render_tool(name: str, tool: Any) -> str: lines = ["### " + BACKTICK + PREFIX + name + BACKTICK, ""] description = (getattr(tool, "description", "") or "").rstrip() body = description.splitlines() if body: lines.append(body[0].strip()) rest = [line.strip() for line in body[1:]] while rest and not rest[0]: rest.pop(0) if rest: lines.append("") lines.extend(" " + line for line in rest) lines.append("") schema = getattr(tool, "input_schema", None) or {} props = schema.get("properties") or {} required = set(schema.get("required") or []) lines.append("| 参数 | 类型 | 必填 | 默认 | 说明 |") lines.append("|------|------|------|------|------|") if not props: lines.append("| （无参数） | — | — | — | — |") for pname, spec in props.items(): spec = spec if isinstance(spec, dict) else {} default = json.dumps(spec["default"], ensure_ascii=False) if "default" in spec else "—" note_lines = str(spec.get("description", "")).replace("|", "/").splitlines() note = (note_lines[0].strip() if note_lines else "") or "—" need = "是" if pname in required else "否" cell = BACKTICK + pname + BACKTICK lines.append("| " + cell + " | " + type_of(spec) + " | " + need + " | " + default + " | " + note + " |") return NL.join(lines) async def fetch_tools() -> dict[str, Any]: params = StdioServerParameters(command=sys.executable, args=["-m", "src.mcp.server"], cwd=str(ROOT)) async with Client(params) as client: listed = await client.list_tools() tools = list(getattr(listed, "tools", listed)) return {t.name: t for t in tools} def build_section(live: dict[str, Any]) -> tuple[str, list[str]]: """渲染整节；返回正文与「登记表与线上不一致」清单。""" problems: list[str] = [] planned = [n for _, _, names in GROUPS for n in names] dup = sorted({n for n in planned if planned.count(n) > 1}) if dup: problems.append("GROUPS 里重复登记：" + ", ".join(dup)) missing = sorted(set(live) - set(planned)) if missing: problems.append("工具已上线但没进文档（去 GROUPS 登记它）：" + ", ".join(missing)) ghost = sorted(set(planned) - set(live)) if ghost: problems.append("文档登记了但线上没有的工具：" + ", ".join(ghost)) if problems: return "", problems out = [ HEADING, "", "", "", "共 " + str(len(live)) + " 个工具，在宿主里都带 " + BACKTICK + PREFIX + BACKTICK + " 前缀。", ] for title, intro, names in GROUPS: out += ["", "## " + title, "", intro] for n in names: out += ["", render_tool(n, live[n])] return NL.join(out).rstrip() + NL, problems def split_doc(text: str) -> tuple[str, str]: idx = text.find(NL + HEADING) if idx < 0: raise SystemExit("docs/MCP.md 里找不到标题：" + HEADING) return text[: idx + 1], text[idx + 1 :] def show_diff(old: str, new: str, limit: int = 30) -> None: a, b = old.splitlines(), new.splitlines() shown = 0 for i in range(max(len(a), len(b))): x = a[i] if i < len(a) else "<无>" y = b[i] if i < len(b) else "<无>" if x != y: print(" 旧| " + x[:110]) print(" 新| " + y[:110]) shown += 1 if shown >= limit: return def main() -> int: parser = argparse.ArgumentParser(description="生成 docs/MCP.md 的工具参考一节") parser.add_argument("--check", action="store_true", help="只比对，不写文件") parser.add_argument("--diff", action="store_true", help="比对并打印差异") args = parser.parse_args() live = asyncio.run(fetch_tools()) section, problems = build_section(live) if problems: for item in problems: print("登记不一致：" + item, file=sys.stderr) return 2 head, old = split_doc(DOC.read_text(encoding="utf-8")) if section == old: print("文档已与代码同源（" + str(len(live)) + " 个工具）") return 0 if args.check or args.diff: print("docs/MCP.md 的工具参考与线上 schema 不一致：跑 make mcp-docs 重新生成", file=sys.stderr) if args.diff: show_diff(old, section) return 1 DOC.write_text(head + section, encoding="utf-8") print("已重新生成工具参考：" + str(len(live)) + " 个工具，" + str(len(section.splitlines())) + " 行") return 0 if __name__ == "__main__": raise SystemExit(main())