首页 › tutorials › Python造日志诊断MCP

Python 造日志诊断 MCP：用 Resources + 进度上报把本地日志喂给 Claude

概述

本教程带你用 Python 官方 MCP SDK（FastMCP）从零造一台「日志诊断 MCP 服务器」。完成后 Claude Code 既能把任意日志文件作为资源（Resource）读取，又能调用 grep_log、summarize_errors 等工具做正则检索与错误分级统计，长任务还会实时上报进度。适合已经写过 MCP 工具、想进一步掌握「动态资源模板 + 上下文进度 + Inspector 调试链路」这一进阶三件套的工程师。

前置条件

整体架构

Claude Code 与你的服务器之间走 stdio 传输：父进程（Claude）拉起子进程（你的 Python 服务器），双方通过 stdin/stdout 收发 JSON-RPC 消息。资源与工具的调用关系如下：

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 直接整篇读；文件巨大时则改用工具做流式处理，避免一次性把几十万行塞进上下文。

第一步：安装与脚手架

创建项目目录并安装依赖：

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）。验证安装：

mcp --version

新建 server.py，写入最小骨架：

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 退出。

常见问题：

第二步：把日志暴露成资源

资源（Resource）让客户端按 URI 读取数据。FastMCP 提供两种写法：静态资源（固定 URI）和资源模板（带 {占位符} 的 URI 模式）。日志文件名千变万化，正是模板的典型场景。

@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 Inspector 里点 List Resources 能看到 logdoctor://index；点进模板，填入 app.log 后能读到文件内容。

第三步：写工具，并用 Context 上报进度

资源适合「读」，但当一个日志有几百万行时，把全文塞给模型既慢又贵。这时要用工具做服务端预处理。FastMCP 的 Context 对象会按类型提示自动注入——只要参数里声明 ctx: Context，运行时就能拿到它，用来记日志、读别的资源、上报进度。

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 "(未发现错误关键字)"

关键点解析：

第四步：用 MCP Inspector 调试

Inspector 是官方交互式调试工具，能脱离 Claude 直接手动调工具、读资源，是把服务器调通的第一站。两种启动方式等价：

# 方式 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}。
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: 渲染结果

常见问题：

第五步：接入 Claude Code

调通后，把服务器注册给 Claude Code。命令行方式（推荐用绝对路径，避免工作目录漂移）：

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）：

{
  "mcpServers": {
    "logdoctor": {
      "command": "python",
      "args": ["/abs/path/logdoctor-mcp/server.py"],
      "env": { "LOG_DIR": "/var/log/myapp" }
    }
  }
}

注意 .mcp.json 里的 env 字段——这是把配置（如日志目录）注入服务器的标准方式，比把路径硬编码进代码干净得多。

注册后验证：

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 工作流总览

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[自然语言提问, 模型自选资源/工具]

延伸阅读

训练营适用性评估

参考来源

  1. Resources & Templates — FastMCP 官方文档 — 资源与 URI 模板的定义方式与占位符规则。
  2. MCP Context — FastMCP 官方文档 — Context 自动注入、ctx.info 与 report_progress 的进度上报机制。
  3. Building Servers — MCP Python SDK 官方文档 — FastMCP 服务器搭建、Context 注入与 stdio 运行的权威说明。
  4. MCP Inspector — Model Context Protocol 官方文档 — Inspector 的安装、连接与工具/资源调试流程。
  5. FastMCP in Python: Build a Real MCP Server (2026 Guide) — 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 等场景全覆盖。

51学通信知识星球海报-扫码加入

📎 本分类最新文章