Auto-Finance/scripts/gen_mcp_docs.py

"""从运行中的 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_id>/，阶段由产物推导。",
        ["case_create", "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,
        "",
        "<!-- 本节由 scripts/gen_mcp_docs.py 从运行时 schema 生成，别手改；改完工具跑 make mcp-docs -->",
        "",
        "共 " + 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())