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

> 一个小型 Python 服务器按脚本注入超时、429 和字段改名，向客户演示系统出故障时 Agent 如何应对。

原文: https://fdetimes.net/zh/guides/fake-mcp-server-fault-injection-testing/

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`：第一次调用被限流，第二次成功。这样，今天通过的测试明天照样通过。

```python
# 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`。

```python
# 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 遇到前者会等待后重试；遇到后者则停下来告知用户，而不是徒劳地重试。在同一个文件里加上这个测试：

```python
@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 会在有限时间内放弃。

```python
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 之外的两种临时错误。

**要点:** 可靠的 Agent 不是从不出错的 Agent，而是被你刻意拉去直面故障的 Agent。

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

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

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

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

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

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

**本周动手试试:**

- 挑出 Agent 调用最频繁的那个工具，单独为它写一个假服务器，包含五个场景：ok、429、429cap、hang、drift。
- 算一算一个 429 总共会产生多少个 HTTP 请求，包括 SDK 自己的重试。如果超过 3 个，就把所有重试逻辑收拢到同一层。
- 加一个 CI 测试，为挂起的工具调用设置 5 秒超时，并确保它通过。

## 信源

- [Tools - Model Context Protocol Specification (2025-06-18)](https://modelcontextprotocol.io/specification/2025-06-18/server/tools)

- [Cancellation - Model Context Protocol Specification (2025-06-18)](https://modelcontextprotocol.io/specification/2025-06-18/basic/utilities/cancellation)

- [Claude API errors](https://platform.claude.com/docs/en/api/errors)

- [Rate limits](https://platform.claude.com/docs/en/api/rate-limits)

- [Testing your FastMCP Server](https://gofastmcp.com/servers/testing)

- [modelcontextprotocol/python-sdk](https://github.com/modelcontextprotocol/python-sdk)