# Python 造日志诊断 MCP：用 Resources + 进度上报把本地日志喂给 Claude ## 概述 本教程带你用 Python 官方 MCP SDK（FastMCP）从零造一台「日志诊断 MCP 服务器」。完成后 Claude Code 既能把任意日志文件作为**资源（Resource）**读取，又能调用 `grep_log`、`summarize_errors` 等工具做正则检索与错误分级统计，长任务还会**实时上报进度**。适合已经写过 MCP 工具、想进一步掌握「动态资源模板 + 上下文进度 + Inspector 调试链路」这一进阶三件套的工程师。 ## 前置条件 - **基础**：会用 Python 写函数、读过一篇 MCP Hello World；装过 Claude Code 并能跑通至少一个 MCP 服务器。 - **环境**：Python 3.10+（用到 `Path.is_relative_to`），Node.js 18+（用于 MCP Inspector）。 - **工具安装**：`pip install "mcp[cli]"` 一步装齐 SDK 与 CLI；Claude Code 已登录可用。 - **可选数据**：任意带文本日志的目录，例如 Nginx `/var/log/nginx`、一个 Spring Boot 应用的 `logs/`，甚至是 Node 程序的 `.log` 文件。 ## 整体架构 Claude Code 与你的服务器之间走 **stdio 传输**：父进程（Claude）拉起子进程（你的 Python 服务器），双方通过 stdin/stdout 收发 JSON-RPC 消息。资源与工具的调用关系如下： ```mermaid flowchart LR U[你在 Claude Code 提问] --> C[Claude Code 客户端] C -- stdio JSON-RPC --> S[logdoctor MCP 服务器] S --> R[Resources: log://name 模板] S --> T1[tail_log 工具] S --> T2[grep_log 工具·带进度] S --> T3[summarize_errors 工具·带进度] R --> F[(本地日志文件)] T1 --> F T2 --> F T3 --> F S -- progress/notification --> C C --> U2[结构化诊断结果回显] ``` 理解这张图的关键点：**Resources 是「可被动的数据」，Tools 是「可被执行的动作」**。日志文件天然适合做成资源——文件不大时让 Claude 直接整篇读；文件巨大时则改用工具做流式处理，避免一次性把几十万行塞进上下文。 ## 第一步：安装与脚手架 创建项目目录并安装依赖： ```bash mkdir logdoctor-mcp && cd logdoctor-mcp python -m venv .venv && source .venv/bin/activate pip install "mcp[cli]" ``` `mcp[cli]` 会同时装上 `mcp` Python 包和命令行工具 `mcp`（含 `mcp run` 与 `mcp dev`）。验证安装： ```bash mcp --version ``` 新建 `server.py`，写入最小骨架： ```python from pathlib import Path import os from mcp.server.fastmcp import FastMCP LOG_DIR = Path(os.environ.get("LOG_DIR", "logs")).resolve() mcp = FastMCP("logdoctor") if __name__ == "__main__": mcp.run(transport="stdio") ``` **预期结果**：`python server.py` 看起来「卡住」——这是正常的，服务器正等待 stdin 上的 JSON-RPC 输入。按 Ctrl+C 退出。 **常见问题**： - *报错 `ModuleNotFoundError: mcp`*：确认虚拟环境已激活，且装的是 `mcp[cli]` 而非某个同名旧包。 - *目录找不到*：先把 `LOG_DIR` 指向一个你确认存在、且里面有 `.log` 文件的目录，例如 `export LOG_DIR=/var/log/nginx`。 ## 第二步：把日志暴露成资源 资源（Resource）让客户端按 URI 读取数据。FastMCP 提供两种写法：**静态资源**（固定 URI）和**资源模板**（带 `{占位符}` 的 URI 模式）。日志文件名千变万化，正是模板的典型场景。 ```python @mcp.resource("logdoctor://index") def list_logs() -> str: """列出 LOG_DIR 下所有日志文件及其大小""" lines = [] for p in sorted(LOG_DIR.glob("*.log")): kb = p.stat().st_size / 1024 lines.append(f"{p.name}\t{kb:.1f} KB") return "\n".join(lines) or "(无日志文件)" @mcp.resource("log://{name}") def read_log(name: str) -> str: """按文件名读取日志（自动截断到末尾 2 万字符以防撑爆上下文）""" safe = (LOG_DIR / name).resolve() if not safe.is_relative_to(LOG_DIR): raise ValueError("非法路径：禁止跳出日志目录") text = safe.read_text(errors="replace") return text[-20000:] if len(text) > 20000 else text ``` 要点： - `@mcp.resource("log://{name}")` 中的占位符 `{name}` 必须与函数参数名**完全一致**，否则模板注册失败、客户端拿不到资源。 - `is_relative_to(LOG_DIR)` 是**关键安全闸**：阻止 `name=../../etc/passwd` 这类路径穿越攻击，把读取范围死死锁在 `LOG_DIR` 之内。 - 自动截断末尾 2 万字符——日志的最新内容通常最有价值，且能保护上下文预算。 **预期结果**：在 MCP Inspector 里点 **List Resources** 能看到 `logdoctor://index`；点进模板，填入 `app.log` 后能读到文件内容。 ## 第三步：写工具，并用 Context 上报进度 资源适合「读」，但当一个日志有几百万行时，把全文塞给模型既慢又贵。这时要用**工具**做服务端预处理。FastMCP 的 `Context` 对象会**按类型提示自动注入**——只要参数里声明 `ctx: Context`，运行时就能拿到它，用来记日志、读别的资源、上报进度。 ```python import re from mcp.server.fastmcp import Context @mcp.tool() def tail_log(name: str, n: int = 50) -> str: """返回指定日志的最后 n 行。""" safe = (LOG_DIR / name).resolve() if not safe.is_relative_to(LOG_DIR): raise ValueError("非法路径") lines = safe.read_text(errors="replace").splitlines() return "\n".join(lines[-n:]) @mcp.tool() def grep_log(name: str, pattern: str, ctx: Context) -> str: """用正则检索日志，边扫边上报进度。返回前 500 条命中。""" safe = (LOG_DIR / name).resolve() if not safe.is_relative_to(LOG_DIR): raise ValueError("非法路径") matched = [] scanned = 0 with safe.open(errors="replace") as f: for i, line in enumerate(f): scanned = i if i % 5000 == 0: ctx.info(f"已扫描 {i} 行…") if re.search(pattern, line): matched.append(line.rstrip()) if len(matched) >= 500: break ctx.info(f"扫描完成，共检查 {scanned + 1} 行，命中 {len(matched)} 条") return "\n".join(matched) @mcp.tool() def summarize_errors(name: str, ctx: Context) -> str: """统计 ERROR/WARN/FATAL/EXCEPTION 出现次数，长任务分批上报进度条。""" safe = (LOG_DIR / name).resolve() if not safe.is_relative_to(LOG_DIR): raise ValueError("非法路径") total = sum(1 for _ in safe.open(errors="replace")) counts: dict[str, int] = {} for i, line in enumerate(safe.open(errors="replace")): for lvl in ("ERROR", "WARN", "FATAL", "EXCEPTION"): if lvl in line: counts[lvl] = counts.get(lvl, 0) + 1 if i % 20000 == 0: ctx.report_progress(i, total) ctx.report_progress(total, total) return "\n".join(f"{k}: {v}" for k, v in sorted(counts.items())) or "(未发现错误关键字)" ``` 关键点解析： - **`ctx.info(msg)`**：发送一条 logging notification，客户端 UI 会显示进度文字。比 `print` 安全得多。 - **`ctx.report_progress(progress, total)`**：发送标准进度通知（分子/分母），Inspector 的进度条会跟着推进。 - **千万不要用 `print()`**：stdio 模式下 stdout 是 JSON-RPC 通道，任何杂乱的 `print` 都会**污染协议**导致连接崩溃。调试输出请一律走 `ctx.info` 或 `logging`（默认输出到 stderr，不会干扰协议）。 ## 第四步：用 MCP Inspector 调试 Inspector 是官方交互式调试工具，能脱离 Claude 直接手动调工具、读资源，是把服务器调通的第一站。两种启动方式等价： ```bash # 方式 A：用 SDK 自带 CLI（会自动打开浏览器） mcp dev server.py # 方式 B：直接用 npx（不依赖 Python CLI） npx @modelcontextprotocol/inspector python server.py ``` 启动后浏览器会打开 `http://localhost:6274`。调试链路： 1. 左侧 **Transport** 选 `STDIO`，确认 Command 为 `python`、Args 为 `server.py`（或完整绝对路径）。 2. 点 **Connect**，右侧应显示服务器名 `logdoctor` 与能力清单。 3. 切到 **Tools** 标签 → 点 `grep_log` → 填 `name=app.log`、`pattern=timeout` → **Run Tool**。注意下方会**实时滚动** `ctx.info` 打出的「已扫描 N 行」，这验证了进度上报链路。 4. 切到 **Resources** 标签 → 列表里点 `logdoctor://index`；再进模板填 `app.log` 验证 `log://{name}`。 ```mermaid sequenceDiagram participant You participant Ins as MCP Inspector participant Srv as logdoctor 服务器 participant F as 日志文件 You->>Ins: Connect(STDIO) Ins->>Srv: initialize Srv-->>Ins: capabilities You->>Ins: Run grep_log(app.log, timeout) Ins->>Srv: tools/call Srv->>F: 流式扫描 Srv-->>Ins: progress(n/total) 每5000行 Ins-->>You: 进度条实时更新 Srv-->>Ins: 命中结果(≤500条) Ins-->>You: 渲染结果 ``` **常见问题**： - *Inspector 一直转圈连不上*：八成是服务器启动报错。在 Inspector 的 **Server Logs** 面板看 stderr；最常见根因是 `LOG_DIR` 不存在或 `import` 路径不对。 - *工具能列出来但一调就报「非法路径」*：模板里 `name` 参数拼到了真实文件名上，但传进来带了前导 `/`。在 Inspector 里直接传 `app.log`（不带斜杠）再试。 - *进度条不动*：确认函数签名里有 `ctx: Context` 且用 `report_progress`。少一个 `ctx` 参数就不会上报。 ## 第五步：接入 Claude Code 调通后，把服务器注册给 Claude Code。命令行方式（推荐用绝对路径，避免工作目录漂移）： ```bash claude mcp add logdoctor -- python /abs/path/logdoctor-mcp/server.py # 想让所有项目共享，加 --scope user claude mcp add logdoctor --scope user -- python /abs/path/logdoctor-mcp/server.py ``` 或在项目根目录放 `.mcp.json`（会被团队共享，提交进 git）： ```json { "mcpServers": { "logdoctor": { "command": "python", "args": ["/abs/path/logdoctor-mcp/server.py"], "env": { "LOG_DIR": "/var/log/myapp" } } } } ``` 注意 `.mcp.json` 里的 `env` 字段——这是把配置（如日志目录）注入服务器的标准方式，比把路径硬编码进代码干净得多。 注册后验证： ```bash claude mcp list # 看到 logdoctor: ✓ Connected ``` 然后在 Claude Code 里自然语言提问即可，模型会自动选资源或工具： > 帮我看一下 app.log 里有没有超时（timeout）相关报错，顺便统计各级别错误数量。 Claude 会先调 `grep_log(app.log, timeout)` 列出命中行，再调 `summarize_errors(app.log)` 给出分级统计；如果你的日志文件不大，它可能直接读 `log://app.log` 资源。**模型自主决定用哪条路径**——这正是 Resources + Tools 双轨设计的价值。 ## 实战案例：Nginx 访问日志 5xx 巡检 假设 `LOG_DIR` 指向一个 Nginx 日志目录，里面有 `access.log`（数百 MB）。完整可跑的 `server.py` 就是上面四步的拼接（约 80 行）。一次真实对话： 1. 你问：**「access.log 最近为什么 5xx 激增？」** 2. Claude 调 `grep_log("access.log", " 5\\d{2} ")`，服务端流式扫描，进度条从 0 走到 total，返回命中行。 3. Claude 进一步调 `summarize_errors` 统计异常关键字，再用 `tail_log` 看最新片段定位时间窗。 4. 因为读的是结构化命中文本（而非原始几百 MB），**上下文消耗只有几 KB**，Claude 能聚焦推理，给出「某上游节点 14:00 起连接超时」的结论。 对比如果用通用文件搜索 MCP（按已覆盖主题里的做法）：它能找到文件，但不懂「日志行」的语义，没有进度上报，也没有错误分级。本教程的差异化正在于此——**资源负责「读」，工具负责「算」，进度负责「让人安心」**。 ## Mermaid 工作流总览 ```mermaid flowchart TD A[pip install mcp-cli] --> B[写 server.py: Resources+Tools+Context] B --> C[mcp dev server.py 起Inspector] C --> D{工具/资源调通?} D -- 否 --> E[查 stderr / 验证路径闸 / 检查参数名] E --> C D -- 是 --> F[claude mcp add 或 写.mcp.json] F --> G[claude mcp list 验证连接] G --> H[自然语言提问, 模型自选资源/工具] ``` ## 延伸阅读 - **采样（Sampling）**：让你的服务器反向调用客户端的 LLM 做摘要/分类，形成「工具+模型」闭环。在工具里 `result = await ctx.sample("把这段日志压缩成一行")`。 - **流式 HTTP 传输**：把同一台服务器从 stdio 切到 `mcp.run(transport="http", host="0.0.0.0", port=8080)`，供远程/多客户端调用，再配合鉴权做成团队级诊断服务。 - **提示词（Prompts）三原语**：把「5xx 巡检」「OOM 追踪」固化成可复用 Prompt，让队友一句话复现你的诊断流程。 - **告警闭环**：再加一个 `write_incident` 工具，让 Claude 检测到 FATAL 时自动写一条工单，从「诊断」升级为「自愈」。 ## 训练营适用性评估 - **Python 后端 / SRE 训练营**：高度契合，4–6 课时。可作为「让 AI 接管运维数据」的实战作业。 - **AI 应用开发训练营**：中等契合，2–3 课时作为 MCP 进阶章节（前提是学员已有 Tools 基础）。 - **零基础训练营**：不推荐直接用，需先补齐 Python 与首个 MCP Hello World。 - **预计教学时长**：讲授 90 分钟 + 动手 120 分钟，学员可带自己公司的真实日志目录，完成度与成就感双高。 ## 参考来源 1. [Resources & Templates — FastMCP 官方文档](https://gofastmcp.com/servers/resources) — 资源与 URI 模板的定义方式与占位符规则。 2. [MCP Context — FastMCP 官方文档](https://gofastmcp.com/servers/context) — Context 自动注入、`ctx.info` 与 `report_progress` 的进度上报机制。 3. [Building Servers — MCP Python SDK 官方文档](https://py.sdk.modelcontextprotocol.io/server/) — FastMCP 服务器搭建、Context 注入与 stdio 运行的权威说明。 4. [MCP Inspector — Model Context Protocol 官方文档](https://modelcontextprotocol.io/docs/tools/inspector) — Inspector 的安装、连接与工具/资源调试流程。 5. [FastMCP in Python: Build a Real MCP Server (2026 Guide)](https://www.danilchenko.dev/posts/fastmcp-mcp-server/) — 2026 年 FastMCP 综合实战范例，覆盖工具/资源/提示词完整三原语。 > [!tip] 51学通信知识星球 > **星主：爱卫生**（微信：gprshome201101）——公众号「51学通信」创办者，19 年通信行业运维与讲师经验，著有《GPRS网络信令实例详解》《EPC原理与实践》《5G核心网原理与实践》三本纸质书（各大电商平台有售）。 > **星球定位**：深耕 5G 核心网、IMS、NFV/AI、云原生等 ICT 领域，主打运维岗位与信令分析。 > **当前规模**：5200 篇主题 / 990 个回答，198 元/年。 > **加入福利**：加入星球后请加爱卫生微信，拉您进知识星球年度陪伴与答疑群——不限量回复您的各种问题，另有群内分享。 > **AI 加持**：51学通信知识星球已作为 skill 接入 WorkBuddy，使用 deepseek-4-flash 在海量星球帖子中精准找贴、找视频、找 FAQ 和案例，做总结、找文档和 PPT 等场景全覆盖。