---

给 Agent 写 MCP 工具，一开始都是查个天气、算个数，几百毫秒就返回。真正接到业务里，很快会遇到跑几秒到几分钟的工具：批量处理一批文件、转码一段视频、跑一轮数据检查。这时候三个问题一起冒出来：

1. 客户端干等，不知道跑到哪了；
2. 不设超时，卡死了也一直等；设了超时，服务端那边到底停没停？
3. 超时之后这条连接还能不能接着用？

这篇用一个最小的批处理工具把这三件事跑一遍：服务端用 ctx.report_progress 报进度，客户端用 progress_callback 收进度、用 read_timeout_seconds 设超时，再实测超时后服务端的取消行为。最后是 3 个我实际跑出来的坑。

环境：Linux，Python 3.13.5，mcp 2.3.0（MCP Python SDK v2）。下文输出都是真实运行结果。

一、准备环境

python3 -m venv .venv && source .venv/bin/activate
pip install "mcp>=2,<3"

二、服务端：一个会报进度的批处理工具

用 asyncio.sleep 模拟每条耗时 0.5 秒的 IO 操作。工具函数多声明一个 ctx: Context 参数，SDK 会自动注入，它不会出现在工具的入参 schema 里。

"""server.py：会跑几秒的批处理工具，边跑边报进度（MCP Python SDK v2）"""
import asyncio, sys, time   # time 给坑 1 的两个工具用
from pydantic import BaseModel
from mcp.server import MCPServer
from mcp.server.mcpserver import Context
from mcp.server.mcpserver.exceptions import ToolError

mcp = MCPServer("batch-demo", version="0.1.0")


def log(msg: str):
    # stdio 模式下 stdout 是协议通道，日志只能写 stderr
    print(f"[server] {msg}", file=sys.stderr, flush=True)


class ItemResult(BaseModel):
    item: str
    chars: int


class BatchResult(BaseModel):
    total: int
    results: list[ItemResult]


@mcp.tool()
async def batch_check(items: list[str], ctx: Context, delay: float = 0.5) -> BatchResult:
    """逐条处理 items（示意：每条 sleep delay 秒模拟耗时），返回处理结果。"""
    if not items:
        raise ToolError("items 不能为空")
    done: list[ItemResult] = []
    total = len(items)
    try:
        for i, it in enumerate(items, 1):
            await asyncio.sleep(delay)          # 模拟耗时的 IO 操作
            done.append(ItemResult(item=it, chars=len(it)))
            await ctx.report_progress(i, total, f"已处理 {it}")
    except asyncio.CancelledError:
        log(f"被取消，已完成 {len(done)}/{total}，在这里做清理")
        raise                                   # 一定要继续往上抛
    log(f"全部完成 {total}/{total}")
    return BatchResult(total=total, results=done)


if __name__ == "__main__":
    mcp.run()   # 默认 stdio

report_progress(progress, total, message) 三个参数：当前值、总量（可选）、一句说明（可选）。总量不知道的时候可以只传 progress，客户端自己决定怎么显示。

三、客户端：收进度

"""client_progress.py"""
import asyncio, sys, time
from mcp import Client, StdioServerParameters

params = StdioServerParameters(command=sys.executable, args=["server.py"])
t0 = time.monotonic()

async def on_progress(progress: float, total: float | None, message: str | None):
    print(f"[{time.monotonic()-t0:4.1f}s] {progress:.0f}/{total:.0f} {message}")

async def main():
    async with Client(params) as client:
        r = await client.call_tool("batch_check", {"items": ["a1", "b22", "c333", "d4444"]},
                                   progress_callback=on_progress)
        print("isError:", r.is_error)
        print("result:", r.structured_content)

asyncio.run(main())

运行结果（计时从启动子进程前开始，所以第一条含服务端启动时间）：

[ 1.0s] 1/4 已处理 a1
[ 1.5s] 2/4 已处理 b22
[ 2.0s] 3/4 已处理 c333
[ 2.5s] 4/4 已处理 d4444
isError: False
result: {'total': 4, 'results': [{'item': 'a1', 'chars': 2}, {'item': 'b22', 'chars': 3}, {'item': 'c333', 'chars': 4}, {'item': 'd4444', 'chars': 5}]}

只要传了 progress_callback，SDK 会自动在请求里带上 progressToken，服务端的 report_progress 就有地方发了；不传的话，服务端调 report_progress 也不会报错，只是没人收。

如果你在做 Agent 的前端，这个回调就是进度条的数据源；如果是纯后台任务，至少把它打进日志，排查「卡住了还是在跑」会省很多事。

四、客户端超时：服务端会不会停？

call_tool 有个 read_timeout_seconds 参数（Client(...) 构造时也能设全局默认值）。我让工具处理 10 条（约 5 秒），超时设 2 秒，然后在超时后立刻再发一个很短的请求，看连接还能不能用。

"""client_compare.py：同样 2 秒超时，对比三种写法"""
import asyncio, sys, time
from mcp import Client, StdioServerParameters

params = StdioServerParameters(command=sys.executable, args=["server.py"])
ITEMS = [f"x{i}" for i in range(10)]   # 10 条 × 0.5 秒 = 5 秒，必然超时

async def run(client, tool):
    t0 = time.monotonic()
    try:
        await client.call_tool(tool, {"items": ITEMS}, read_timeout_seconds=2)
    except Exception as e:
        print(f"{tool}: {time.monotonic()-t0:.1f}s 后 {type(e).__name__}: {e}")
    # 紧接着发一个很短的请求，看连接是不是被堵住
    t1 = time.monotonic()
    await client.call_tool("batch_check", {"items": ["ping"], "delay": 0})
    print(f"{tool}: 下一个请求等了 {time.monotonic()-t1:.1f}s")

async def main():
    async with Client(params) as client:
        for tool in ["batch_check", "batch_check_blocking", "batch_check_thread"]:
            await run(client, tool)
            await asyncio.sleep(0.5)
            print()

asyncio.run(main())

先只看第一种（正常写法 batch_check）的结果：

[server] 被取消，已完成 4/10，在这里做清理
batch_check: 2.0s 后 MCPError: Request 'tools/call' timed out
batch_check: 下一个请求等了 0.0s

结论有三条：

• 客户端 2 秒准时抛 MCPError，提示 Request 'tools/call' timed out；
• 服务端确实停了：工具里的 asyncio.CancelledError 被触发，10 条只做了 4 条。在我这版 SDK 上，客户端超时后会通知服务端取消这次请求，服务端把对应的协程 cancel 掉；
• 连接没断，下一个请求立刻返回。

所以「超时之后要不要重建连接」这个问题，在这种写法下答案是不用。但另外两种写法的结果就不一样了，见下面的踩坑。

五、3 个实测踩坑

坑 1：async 工具里用了阻塞调用，取消和超时都「失灵」

很多现成的库只有同步接口（老的 SDK、图像处理、某些数据库驱动）。直接在 async def 工具里调用，会把整个事件循环堵住。反面教材：

@mcp.tool()
async def batch_check_blocking(items: list[str], ctx: Context, delay: float = 0.5) -> BatchResult:
    """反面教材：在 async 工具里用 time.sleep（阻塞事件循环）。"""
    done: list[ItemResult] = []
    for i, it in enumerate(items, 1):
        time.sleep(delay)                       # ❌ 阻塞调用
        done.append(ItemResult(item=it, chars=len(it)))
        await ctx.report_progress(i, len(items), f"已处理 {it}")
    log(f"blocking 版跑完 {len(done)}/{len(items)}")
    return BatchResult(total=len(items), results=done)

同一个对比脚本的输出：

[server] blocking 版跑完 10/10
batch_check_blocking: 2.0s 后 MCPError: Request 'tools/call' timed out
batch_check_blocking: 下一个请求等了 3.0s

客户端那边照样 2 秒超时，看起来一切正常，但：

• 服务端没停，10 条全跑完了（取消信号根本没机会被处理）；
• 后面那个本该 0 秒返回的请求，等了 3 秒，因为服务端事件循环一直被占着。

在 Agent 场景里这很要命：模型以为工具失败了，可能马上重试，结果服务端同时跑着两份；如果工具有副作用（写文件、发请求），就会重复执行。

解法：把同步调用丢进线程，事件循环保持畅通。

@mcp.tool()
async def batch_check_thread(items: list[str], ctx: Context, delay: float = 0.5) -> BatchResult:
    """正解：阻塞的同步函数丢进线程跑，事件循环保持畅通。"""
    def work(it: str) -> ItemResult:
        time.sleep(delay)                       # 假装是一个只能同步调用的库
        return ItemResult(item=it, chars=len(it))

    done: list[ItemResult] = []
    try:
        for i, it in enumerate(items, 1):
            done.append(await asyncio.to_thread(work, it))
            await ctx.report_progress(i, len(items), f"已处理 {it}")
    except asyncio.CancelledError:
        log(f"thread 版被取消，已完成 {len(done)}/{len(items)}")
        raise
    return BatchResult(total=len(items), results=done)

[server] thread 版被取消，已完成 3/10
batch_check_thread: 2.0s 后 MCPError: Request 'tools/call' timed out
batch_check_thread: 下一个请求等了 0.0s

取消生效、下一个请求不受影响。注意一个细节：asyncio.to_thread 取消的是「等待」，已经在线程里跑的那一条不会被强行杀掉，它会跑完这一条再停。所以每个同步步骤尽量切小，副作用操作要能重复执行（幂等）。

坑 2：返回 dict，structured_content 是 None

我第一版工具的返回值类型写的是 -> dict，客户端拿到：

isError: False
result: None

数据其实在 r.content[0].text 里，是一段 JSON 字符串，但 structured_content 是空的。在我这版 SDK 上，裸 dict 返回类型没有生成结构化输出，只给了文本。改成上面的 Pydantic 模型 -> BatchResult 之后，structured_content 就是正常的字典了，客户端和模型都能按字段取，不用自己再 json.loads。

建议：凡是给程序消费的工具结果，都用 Pydantic 模型（或 TypedDict）声明返回类型。

坑 3：捕获了 CancelledError 却没有往上抛

服务端想在取消时做清理（删临时文件、记录做到第几条），很自然会写 except asyncio.CancelledError。关键是最后那个 raise 不能省。这是 asyncio 的通用规矩：吞掉 CancelledError，上层就收不到取消信号，任务可能被当成正常结束继续往下走（这一条是通用写法建议，文中没有单独做对照实验）。清理逻辑如果比较多，用 try/finally 更稳，正常结束和被取消都会执行。

另外，工具主动报错用 ToolError，客户端拿到的是 isError=True 的正常结果，而不是异常。实测：

True [TextContent(type='text', text='Error executing tool batch_check: items 不能为空', ...)]

这和超时不一样：超时在客户端是抛 MCPError 异常，业务错误是 isError 结果。Agent 循环里两种都要处理，前者通常可以重试，后者一般应该把错误文本回给模型，让它改参数。

六、小结

• 长任务工具：服务端 ctx.report_progress，客户端 progress_callback，进度条和日志都有了；
• 每个工具调用都设 read_timeout_seconds，别让 Agent 无限等待；
• 在 Python SDK v2 上，超时会触发服务端取消，前提是工具代码里没有阻塞事件循环的调用；同步库一律 asyncio.to_thread；
• 取消时的清理放在 except CancelledError: ... raise 或 finally 里；
• 返回值用 Pydantic 模型，拿到的是结构化结果。

完整代码就是文中的 server.py（把坑 1 里的两个工具也加进去）、client_progress.py、client_compare.py 三个文件，复制下来就能跑。

---

作者在做短视频内容自动化相关的产品，日常和 MCP 工具打交道，这里记录的是实际开发中踩过的坑。