让 MCP 工具实时回报进度与日志：FastMCP Context 实战

flowchart LR
    U[用户:分析这批日志] --> CC[Claude Code]
    CC -->|tools/call| T[analyze_log 工具]
    T -->|ctx.report_progress| CC
    T -->|ctx.info 日志| CC
    T -->|ctx.read_resource| R[config 资源]
    R --> T
    T -->|每 1000 行回传进度| CC
    T -->|最终结果| CC
    CC --> U
    style T fill:#fef3c7
    style CC fill:#dbeafe

uv init log-analyzer && cd log-analyzer
uv add fastmcp pytest pytest-asyncio

uv run python -c "import fastmcp; print(fastmcp.__version__)"

from fastmcp import FastMCP, Context

mcp = FastMCP("log-analyzer")


@mcp.tool
async def count_lines(file_path: str, ctx: Context) -> str:
    """逐行统计文件行数，并实时回报进度。"""
    with open(file_path, encoding="utf-8") as f:
        total = sum(1 for _ in f)

    await ctx.info(f"开始统计 {file_path}，共约 {total} 行")
    processed = 0
    with open(file_path, encoding="utf-8") as f:
        for line in f:
            processed += 1
            if processed % 20 == 0:                       # 每 20 行回传一次进度
                await ctx.report_progress(processed, total)
    await ctx.report_progress(total, total)
    return f"完成：{file_path} 共 {processed} 行"


if __name__ == "__main__":
    mcp.run(transport="stdio")

# quick_check.py
import asyncio
from fastmcp import Client
from server import mcp

async def main():
    async with Client(mcp) as client:
        async for msg in client.call_tool("count_lines",
                                          {"file_path": "server.py"},
                                          progress_handler=print):
            pass

asyncio.run(main())

uv run python quick_check.py

log-analyzer/
├── server.py          # MCP 服务器（tool + resource + prompt）
├── sample.log         # 测试用日志
├── test_server.py     # pytest 单测
├── .mcp.json          # Claude Code 接入配置
└── pyproject.toml

from __future__ import annotations
from pathlib import Path
from collections import Counter
from fastmcp import FastMCP, Context

mcp = FastMCP("log-analyzer")

# 配置：通过 resource 暴露，工具内用 ctx.read_resource 读取
SETTINGS = {
    "context_lines": "3",   # ERROR 上下文窗口
    "max_errors": "50",     # 最多抓取多少条 ERROR
}


@mcp.tool
async def analyze_log(file_path: str, ctx: Context) -> str:
    """分析日志文件，统计各级别数量并抓取最近的 ERROR 上下文。

    长时间运行，会通过进度通知实时回报处理进度。
    """
    p = Path(file_path)
    if not p.exists():
        await ctx.warning(f"文件不存在：{file_path}")
        return f"错误：找不到文件 {file_path}"

    # —— 演示在工具内读取兄弟资源 ——
    cfg = await ctx.read_resource("config://analyzer/context_lines")
    context_lines = int(cfg[0].text) if cfg else 3

    total = sum(1 for _ in p.open(encoding="utf-8", errors="replace"))
    await ctx.info(f"开始分析 {file_path}，共 {total} 行，上下文窗口 {context_lines}")

    counter: Counter[str] = Counter()
    error_snippets: list[str] = []
    window: list[str] = []

    with p.open(encoding="utf-8", errors="replace") as f:
        for i, line in enumerate(f, 1):
            window.append(line.rstrip())
            if len(window) > context_lines + 1:
                window.pop(0)

            for level in ("ERROR", "WARN", "INFO", "DEBUG"):
                if level in line:
                    counter[level] += 1
            if "ERROR" in line and len(error_snippets) < 50:
                error_snippets.append("\n".join(window))

            if i % 2000 == 0:                              # 每 2000 行回传进度
                await ctx.report_progress(i, total,
                                          message=f"已处理 {i}/{total}")

    await ctx.report_progress(total, total, message="分析完成")
    await ctx.debug(f"抓取到 {len(error_snippets)} 条 ERROR 片段")

    summary = ", ".join(f"{k}={v}" for k, v in counter.most_common())
    tail = "\n---\n".join(error_snippets[-3:]) or "（无 ERROR）"
    return f"## 统计\n{summary}\n\n## 最近 ERROR 上下文\n{tail}"


@mcp.resource("config://analyzer/{key}")
def get_setting(key: str) -> str:
    """读取分析器运行配置（context_lines / max_errors）。"""
    return SETTINGS.get(key, "未设置")


@mcp.prompt
def summarize_errors(log: str) -> str:
    """让 Claude 总结日志中的错误模式与可能根因。"""
    return (
        "你是一名资深 SRE。请分析下面日志的 ERROR 上下文，"
        "归纳错误模式、给出 3 条最可能的根因假设，并按优先级排序：\n\n"
        f"{log}"
    )


if __name__ == "__main__":
    mcp.run(transport="stdio")

uv run python -c "
import random
lv = ['INFO','INFO','INFO','WARN','ERROR','DEBUG']
with open('sample.log','w') as f:
    for i in range(12000):
        lvl = random.choice(lv)
        f.write(f'2026-06-19 10:{i%60:02d}:{i%60:02d} [{lvl}] event #{i} msg payload here\n')
print('sample.log 已生成 12000 行')
"

printf '2026-06-19 11:00:00 [INFO] connection established\n2026-06-19 11:00:01 [ERROR] NullPointerException at com.app.Handler\n2026-06-19 11:00:02 [ERROR] DatabaseTimeout after 30s\n' >> sample.log

uv run fastmcp run server.py

import pytest
from fastmcp import Client
from server import mcp


@pytest.fixture
def client():
    return Client(mcp)               # 进程内内存传输


@pytest.mark.asyncio
async def test_lists_three_primitives(client):
    async with client as c:
        tools = await c.list_tools()
        resources = await c.list_resources()
        prompts = await c.list_prompts()
    assert any(t.name == "analyze_log" for t in tools)
    assert any(r.uri.startswith("config://analyzer") for r in resources)
    assert any(p.name == "summarize_errors" for p in prompts)


@pytest.mark.asyncio
async def test_analyze_log_counts_levels(client):
    async with client as c:
        result = await c.call_tool("analyze_log", {"file_path": "sample.log"})
    text = result.content[0].text
    assert "统计" in text
    assert "ERROR=" in text              # 确实统计到了 ERROR


@pytest.mark.asyncio
async def test_analyze_log_missing_file(client):
    async with client as c:
        result = await c.call_tool("analyze_log", {"file_path": "nope.log"})
    assert "找不到文件" in result.content[0].text


@pytest.mark.asyncio
async def test_progress_is_reported(client):
    progress = []
    async with client as c:
        await c.call_tool(
            "analyze_log",
            {"file_path": "sample.log"},
            progress_handler=lambda p, t, m=None: progress.append((p, t)),
        )
    assert progress, "应当至少回传一次进度"
    assert progress[-1][0] == progress[-1][1]   # 最后一次 = 100%


@pytest.mark.asyncio
async def test_resource_read(client):
    async with client as c:
        content = await c.read_resource("config://analyzer/context_lines")
    assert content[0].text == "3"

uv run pytest -v

npx @modelcontextprotocol/inspector uv run server.py

claude mcp add log-analyzer -- uv run --directory $(pwd) server.py

{
  "mcpServers": {
    "log-analyzer": {
      "command": "uv",
      "args": ["run", "--directory", "/绝对路径/log-analyzer", "server.py"],
      "env": { "PYTHONUNBUFFERED": "1" }
    }
  }
}

claude mcp list

claude mcp get log-analyzer

flowchart TB
    subgraph 开发期
        A[写 server.py<br/>tool/resource/prompt] --> B[uv run pytest<br/>进程内单测]
        B --> C[npx inspector<br/>手动验证进度]
    end
    subgraph 接入期
        C --> D[claude mcp add<br/>或写 .mcp.json]
        D --> E[claude mcp list<br/>确认 connected]
    end
    subgraph 运行期
        E --> F[Claude 调用 analyze_log]
        F --> G[ctx.report_progress<br/>终端进度条]
        F --> H[ctx.info/debug<br/>结构化日志回流]
        F --> I[ctx.read_resource<br/>读 config]
        G --> J[用户实时可见]
        H --> J
        I --> K[最终 Markdown 结果]
        K --> L[summarize_errors prompt 归因]
    end
    style F fill:#fef3c7
    style G fill:#dcfce7
    style H fill:#dcfce7