用假 MCP 服务器离线测试 Agent 的故障处理

MCP 规范只用了短短一句话，建议客户端为工具调用设置超时。演示时，这句话很容易一眼带过。等客户把 Agent 接入真实的 CRM，情况就变了：对方的 API 有速率限制，偶尔停机维护，后端团队还可能不打招呼就把某个数据字段改了名。

本指南用 Python 搭一个假 MCP 服务器。它暴露与真实系统相同的工具名，但会按你设定的脚本出错。整套测试在笔记本上就能跑，无需联网，每次结果都一样。

客户想知道：自己的系统出故障时，Agent 会怎么做。手里有一套故障测试，就能当着客户的面跑一遍来回答，而不只是口头保证。

写代码之前，先弄清两种错误通道

MCP 规范（2025-06-18 版）为工具定义了两种报错机制。第一种是 JSON-RPC 协议错误，例如调用不存在的工具会返回错误码 -32602。第二种是工具执行错误：结果照常返回，但带有 isError: true。API 错误和业务逻辑错误都属于这一类。

两类错误从不同通道到达，Agent 就应该为它们各设一个分支。比较合理的分工是：把执行错误交给模型，由它决定重试还是换个做法；把协议错误记为集成缺陷，交给人来修。这是设计建议，并非规范要求。

规范里还现成给了一个限流类执行错误的例子，文本是“Failed to fetch weather data: API rate limit exceeded”。可以直接借用这句话，作为假 429 消息的模板。

需要准备什么？

需要 Python 3.10 或更高版本、pytest 及一个支持异步测试的插件，以及 FastMCP 库。官方 MCP Python SDK 只需带类型注解和 docstring 的函数就能定义服务器，每个工具用装饰器注册。FastMCP 还能把服务器包进一个 Client，在同一进程里测试，不必另开 MCP Inspector。

下面的代码为讲清思路做了精简。类名、参数和异常映射在不同版本间可能有差异，请对照所装 SDK 版本的文档核对。

第 1 步：写一个从环境变量读取故障脚本的服务器

原则是脚本必须确定。不要让故障以 10% 的概率随机发生，而是传入一串固定的故障序列，例如 429,ok：第一次调用被限流，第二次成功。这样，今天通过的测试明天照样通过。

# fake_crm.py（精简版）
import asyncio, os
from fastmcp import FastMCP
from fastmcp.exceptions import ToolError

mcp = FastMCP("fake-crm")
PLAN = [f for f in os.environ.get("FAULTS", "").split(",") if f]
state = {"n": 0}

def next_fault() -> str:
    i = state["n"]; state["n"] += 1
    return PLAN[i] if i < len(PLAN) else "ok"

@mcp.tool
async def get_order(order_id: str) -> dict:
    """按 ID 获取订单。"""
    fault = next_fault()
    if fault == "429":
        raise ToolError("API 速率超限。retry-after: 2")
    if fault == "429cap":
        raise ToolError("API 速率超限：已达到每月支出上限")
    if fault == "hang":
        await asyncio.sleep(3600)
    if fault == "drift":
        return {"orderId": order_id, "total": "12.5"}
    return {"order_id": order_id, "status": "shipped", "total": 12.5}

PLAN 在导入时从 FAULTS 读取一次。测试中用 monkeypatch 覆盖它并重置计数器，每个测试就有各自的脚本，无需重启进程。

**检查：**新建 test_fake_crm.py，写第一个“正常路径”测试，脚本留空，然后运行 pytest -q test_fake_crm.py。

# test_fake_crm.py（精简版）
import pytest
from fastmcp import Client
from fake_crm import mcp, state

@pytest.mark.asyncio
async def test_no_fault(monkeypatch):
    monkeypatch.setattr("fake_crm.PLAN", [])
    state["n"] = 0
    async with Client(mcp) as client:
        r = await client.call_tool("get_order", {"order_id": "A1"})
        assert not r.is_error

这个测试必须通过，并返回 status 为 shipped 的订单。这一步没跑通，就不要往下走。

第 2 步：为什么需要两种 429？

这里的错误模式借鉴自 Claude API 的文档，那是一个真实的 HTTP API。普通 429 的 retry-after 头会告诉你要等几秒，提前重试只会再失败。

Claude API 的速率限制采用令牌桶机制：额度逐步回填，而不是在固定时间点重置。但因达到某一档位的支出上限而触发的 429 不带 retry-after，在恢复访问之前，重试多少次都不会成功。

需要说明：假服务器里的工具并不返回 HTTP 429，也没有任何头部。它只是把这两种行为模式照搬进一个文本形式的 ToolError，把 retry-after: 2 写在错误消息里。如果客户的真实后端是 HTTP，你代码中调用 API 的那一层会看到真实的头部，那一层也应该 mock。

所以假服务器里既有 429，也有 429cap。好的 Agent 遇到前者会等待后重试；遇到后者则停下来告知用户，而不是徒劳地重试。在同一个文件里加上这个测试：

@pytest.mark.asyncio
async def test_429_is_execution_error(monkeypatch):
    monkeypatch.setattr("fake_crm.PLAN", ["429", "ok"])
    state["n"] = 0
    async with Client(mcp) as client:
        r = await client.call_tool("get_order", {"order_id": "A1"},
                                   raise_on_error=False)
        assert r.is_error  # 走 isError 通道，不是 JSON-RPC 错误

**检查：**结果必须带错误标记，而不是抛出协议异常。如果客户端抛了异常，说明你把两种错误通道混在了一起。

第 3 步：模拟挂起的工具，然后取消请求

hang 场景模拟一个永远不响应的 API。规范建议客户端设置超时，因此测试必须断言 Agent 会在有限时间内放弃。

import asyncio

@pytest.mark.asyncio
async def test_hung_tool_is_cut_off(monkeypatch):
    monkeypatch.setattr("fake_crm.PLAN", ["hang"])
    state["n"] = 0
    async with Client(mcp) as client:
        with pytest.raises(asyncio.TimeoutError):
            await asyncio.wait_for(
                client.call_tool("get_order", {"order_id": "A1"}), timeout=5)

超时之后，还有一个容易忽略的细节。规范中关于取消的部分规定，收到取消通知的一方应停止处理，且不再为被取消的请求发送响应。规范也承认存在竞态：取消通知到达之前，响应可能已经发出。

再写一个服务器场景：“刚被取消就迟到地返回结果”，检查 Agent 是否忽略这个结果，而不是把它挂到另一轮对话上。

第 4 步：协议错误与 schema 漂移

协议错误最简单：调用 get_ordr（故意拼错）。按规范，预期结果是一个 JSON-RPC 错误，例如 -32602。检查你的 SDK 是否这样映射，以及 Agent 是否把它记为集成错误，而不是塞进提示词。

schema 漂移更难发现。drift 场景把 order_id 改名为 orderId，并把 total 从数字改成字符串。这个结果以成功调用的形式到达 Agent，既没有 isError，也没有 JSON-RPC 错误，两条错误通道都不经过。

如果工具声明了 outputSchema，规范要求服务器返回符合该 schema 的结果，并建议客户端做校验。假服务器故意违反这条规则，看客户端能否发现错误，还是 Agent 会悄无声息地把字符串“12.5”加进发票。

规范还提供了一种通告工具列表变化的机制：服务器声明 listChanged 能力，随后发出 notifications/tools/list_changed 通知。加一个在会话中途发出该通知的场景，然后检查 Agent 在下一次调用工具前是否重新调用了 tools/list。各个 SDK 发送通知的方式不同，这部分取决于你选的库。

最常见的三个重试错误

第一个是重试叠加重试。官方 Claude SDK 默认会对临时错误重试 2 次，采用指数退避，并遵守 retry-after。

算一笔账：你的 Agent 循环自己重试 3 次，加上首次共调用 4 次；每次调用，SDK 最多发出 3 个请求。一个 429 就可能变成 12 个请求，涌向一个本已过载的 API。

第二个错误是按固定间隔重试，不读 retry-after。第三个是把所有 429 归入同一个处理分支，导致支出上限已到时 Agent 仍在无休止地重试。你的故障集还应包含 504 timeout_error 和 529 overloaded_error，这是 Claude API 文档中列出的、429 之外的两种临时错误。

这项技能在客户项目中用在哪里？

设想你在一家物流公司的第一周。你还拿不到生产环境 API 的权限，预发环境每天只开几个小时。有了假服务器，工作就能并行推进。

向对方要一份描述速率限制和 schema 的文档，照着它精确地搭出假服务器。等你拿到真实权限时，故障测试套件已经就绪。

第一件事是向客户的运维团队问三个问题：API 返回 429 时带哪些头部？实际的超时是多少？上一次 schema 毫无预警地变更是什么时候？每个答案都会变成 FAULTS 变量里的一个场景。

简历上不要只写“用 MCP 构建了 Agent”。要写你搭建了一套包含多少个场景的故障注入测试，以及它在上线前抓到了哪些缺陷。阅读前线部署工程师（FDE）的岗位描述时，留意“reliability”“production readiness”“integration testing”这类字眼。

好的 FAULTS 集会随每一次真实事故不断扩充：客户系统每以一种新方式出故障，脚本就多一行。