别让 MCP 服务悄悄破坏智能体：录制回放与故障注入实战

MCP Server 不只是一个工具列表，它还是智能体运行时的一项外部依赖。本文用纯 Node.js 构建可运行的录制回放夹具，并通过契约断言和故障注入，在客户端或服务端升级前暴露兼容性问题。

MCP 的兼容性不止是“能连上”

智能体接入 MCP Server 后，通常会先执行 initialize，再通过 tools/list 发现工具，最后调用 tools/call。其中任何一层发生变化，都可能让上层表现异常：

变化位置	典型问题	应验证的边界
工具发现	工具被删除或重命名	工具名称及必要能力是否存在
参数 Schema	必填字段增加、类型改变	`inputSchema` 的关键约束
返回内容	`content` 结构变化	内容类型和业务字段
错误语义	把业务失败改成协议错误	JSON-RPC 错误与 `isError` 的区别
传输层	超时、断连、输出损坏	客户端能否停止等待并给出可诊断错误

因此，测试不应只模拟一个 JavaScript 函数。更稳妥的边界是 MCP 的 JSON-RPC 消息本身：记录客户端与服务端之间的请求和响应，在测试中原样回放，再有针对性地破坏传输。

录制内容可能包含提示词、文件路径或访问令牌。提交到仓库前应脱敏，并使用专门的测试账号生成样本。

先定义值得稳定的契约

契约测试不等于对整份响应做快照。服务版本、描述文字和非关键字段可能正常变化；真正值得固定的是智能体依赖的行为。

以一个 echo 工具为例，可以约定：

tools/list 中必须存在 echo；
参数是对象，text 为必填字符串；
成功调用返回 type: "text" 的内容；
参数不合法属于工具执行失败，以 isError: true 表示；
JSON-RPC error 保留给方法不存在、请求无效等协议级问题。

这种断言比完整快照更克制。新增可选工具或修改描述不会阻塞升级，但删除必要字段、改变错误分类会立即失败。

实现录制与回放夹具

MCP 的 stdio 传输使用按行分隔的 JSON-RPC 消息。下面的夹具在录制模式下启动真实 Server，双向转发并保存消息；在回放模式下，它按方法返回已录制的响应。代码只使用 Node.js 内置模块，建议使用 Node.js 20 或更高版本。

保存为 mcp-fixture.mjs：

import { spawn } from "node:child_process";
import { appendFileSync, readFileSync, writeFileSync } from "node:fs";
import { createInterface } from "node:readline";

const mode = process.env.MODE ?? "replay";
const cassette = process.env.CASSETTE ?? "./cassette.ndjson";
const emit = value => process.stdout.write(JSON.stringify(value) + "\n");

if (mode === "record") {
  const bin = process.env.UPSTREAM_BIN;
  if (!bin) throw new Error("UPSTREAM_BIN is required in record mode");

  writeFileSync(cassette, "");
  const args = JSON.parse(process.env.UPSTREAM_ARGS ?? "[]");
  const upstream = spawn(bin, args, { stdio: ["pipe", "pipe", "pipe"] });

  function bridge(source, target, direction) {
    createInterface({ input: source }).on("line", line => {
      appendFileSync(cassette, JSON.stringify({ direction, raw: line }) + "\n");
      target.write(line + "\n");
    });
  }

  bridge(process.stdin, upstream.stdin, "c2s");
  bridge(upstream.stdout, process.stdout, "s2c");
  upstream.stderr.pipe(process.stderr);
  upstream.on("exit", code => process.exit(code ?? 1));
} else {
  const rows = readFileSync(cassette, "utf8").trim().split("\n")
    .filter(Boolean).map(line => JSON.parse(line));
  const requests = new Map();
  const responses = new Map();

  for (const row of rows) {
    const message = JSON.parse(row.raw);
    if (row.direction === "c2s" && message.id !== undefined && message.method) {
      requests.set(String(message.id), message.method);
    }
    if (row.direction === "s2c" && message.id !== undefined) {
      const method = requests.get(String(message.id));
      if (method) {
        const queue = responses.get(method) ?? [];
        queue.push(message);
        responses.set(method, queue);
      }
    }
  }

  createInterface({ input: process.stdin }).on("line", line => {
    let request;
    try {
      request = JSON.parse(line);
    } catch {
      return;
    }

    // notification 没有 id，不需要响应。
    if (request.id === undefined) return;

    if (request.method === process.env.FAULT_METHOD) {
      if (process.env.FAULT === "timeout") return;
      if (process.env.FAULT === "disconnect") return process.exit(70);
      if (process.env.FAULT === "malformed") {
        return process.stdout.write("{not-json}\n");
      }
    }

    const recorded = responses.get(request.method)?.shift();
    if (!recorded) {
      return emit({
        jsonrpc: "2.0",
        id: request.id,
        error: { code: -32601, message: `No recording for ${request.method}` }
      });
    }

    emit({ ...recorded, id: request.id });
  });
}

录制时，让 MCP Client 把这个进程当作 Server 启动，同时由夹具拉起真实 Server：

MODE=record \
CASSETTE=./cassette.ndjson \
UPSTREAM_BIN=node \
UPSTREAM_ARGS='["./real-server.mjs"]' \
node mcp-fixture.mjs

这里没有直接匹配录制时的请求 ID，而是按方法保存响应队列，并在回放时替换为当前请求 ID。这样可以避免客户端升级后仅因 ID 生成策略改变而误报。它适合顺序调用测试；如果同一方法存在并发请求，还应增加参数摘要或场景名称作为匹配键。

编写契约测试与最小录制样本

为了让示例无需真实 Server 也能运行，先创建一个最小 cassette.ndjson。每一行都是夹具保存的方向和原始消息：

{"direction":"c2s","raw":"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\"}"}
{"direction":"s2c","raw":"{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"protocolVersion\":\"2024-11-05\",\"capabilities\":{\"tools\":{}},\"serverInfo\":{\"name\":\"echo-test\",\"version\":\"1.0.0\"}}}"}
{"direction":"c2s","raw":"{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"tools/list\"}"}
{"direction":"s2c","raw":"{\"jsonrpc\":\"2.0\",\"id\":2,\"result\":{\"tools\":[{\"name\":\"echo\",\"description\":\"Echo text\",\"inputSchema\":{\"type\":\"object\",\"properties\":{\"text\":{\"type\":\"string\"}},\"required\":[\"text\"],\"additionalProperties\":false}}]}}"}
{"direction":"c2s","raw":"{\"jsonrpc\":\"2.0\",\"id\":3,\"method\":\"tools/call\"}"}
{"direction":"s2c","raw":"{\"jsonrpc\":\"2.0\",\"id\":3,\"result\":{\"content\":[{\"type\":\"text\",\"text\":\"hello\"}]}}"}
{"direction":"c2s","raw":"{\"jsonrpc\":\"2.0\",\"id\":4,\"method\":\"tools/call\"}"}
{"direction":"s2c","raw":"{\"jsonrpc\":\"2.0\",\"id\":4,\"result\":{\"content\":[{\"type\":\"text\",\"text\":\"text is required\"}],\"isError\":true}}"}

再保存以下测试为 contract.test.mjs：

import test from "node:test";
import assert from "node:assert/strict";
import { spawn } from "node:child_process";
import { createInterface } from "node:readline";

class Client {
  constructor(fault) {
    this.id = 0;
    this.pending = new Map();
    this.child = spawn(process.execPath, ["mcp-fixture.mjs"], {
      env: {
        ...process.env,
        MODE: "replay",
        CASSETTE: "./cassette.ndjson",
        FAULT_METHOD: fault ? "tools/call" : "",
        FAULT: fault ?? ""
      },
      stdio: ["pipe", "pipe", "inherit"]
    });

    createInterface({ input: this.child.stdout }).on("line", line => {
      try {
        const message = JSON.parse(line);
        const pending = this.pending.get(message.id);
        if (pending) {
          clearTimeout(pending.timer);
          this.pending.delete(message.id);
          pending.resolve(message);
        }
      } catch {
        this.rejectAll(new Error("malformed response"));
      }
    });
    this.child.on("exit", () => this.rejectAll(new Error("server exited")));
  }

  rejectAll(error) {
    for (const item of this.pending.values()) {
      clearTimeout(item.timer);
      item.reject(error);
    }
    this.pending.clear();
  }

  request(method, params = {}) {
    const id = ++this.id;
    return new Promise((resolve, reject) => {
      const timer = setTimeout(() => {
        this.pending.delete(id);
        reject(new Error("request timeout"));
      }, 300);
      this.pending.set(id, { resolve, reject, timer });
      this.child.stdin.write(JSON.stringify({ jsonrpc: "2.0", id, method, params }) + "\n");
    });
  }

  notify(method) {
    this.child.stdin.write(JSON.stringify({ jsonrpc: "2.0", method }) + "\n");
  }

  close() {
    this.child.kill();
  }
}

async function initialized(fault) {
  const client = new Client(fault);
  await client.request("initialize", {
    protocolVersion: "2024-11-05",
    capabilities: {},
    clientInfo: { name: "contract-test", version: "1.0.0" }
  });
  client.notify("notifications/initialized");
  const list = await client.request("tools/list");
  return { client, list };
}

test("工具发现、Schema 与错误语义保持兼容", async () => {
  const { client, list } = await initialized();
  const echo = list.result.tools.find(tool => tool.name === "echo");
  assert.ok(echo);
  assert.equal(echo.inputSchema.properties.text.type, "string");
  assert.ok(echo.inputSchema.required.includes("text"));

  const ok = await client.request("tools/call", {
    name: "echo", arguments: { text: "hello" }
  });
  assert.deepEqual(ok.result.content[0], { type: "text", text: "hello" });

  const failed = await client.request("tools/call", {
    name: "echo", arguments: {}
  });
  assert.equal(failed.result.isError, true);
  client.close();
});

for (const [fault, expected] of [
  ["timeout", /timeout/],
  ["disconnect", /exited/],
  ["malformed", /malformed/]
]) {
  test(`客户端处理 ${fault}`, async () => {
    const { client } = await initialized(fault);
    await assert.rejects(
      client.request("tools/call", { name: "echo", arguments: { text: "x" } }),
      expected
    );
    client.close();
  });
}

运行命令如下：

node --test contract.test.mjs

超时时间只是测试夹具的等待上限，不代表生产环境的推荐配置。生产客户端应根据工具成本设置超时，并在超时后清理挂起请求；断连时要拒绝所有等待中的 Promise；畸形响应则应记录原始诊断信息，但避免把可能含敏感数据的完整内容写入普通日志。

把验证放到升级流程里

实践中可以保留两类测试。第一类使用脱敏录制文件，运行快速、结果稳定，适合每次提交执行；第二类连接真实 MCP Server，验证认证、网络和部署配置，适合定时任务或发布前检查。

升级客户端时，用同一份录制文件执行契约与故障测试，可以发现协议解析和错误处理回归。升级 Server 时，重新录制到临时文件，对比工具名称、关键 Schema 和错误分类，确认后再更新仓库中的基线。

录制文件本身也要版本化。建议在目录名中包含 Server 名称和场景，例如 cassettes/filesystem/read-success.ndjson，而不要只保存一份不断覆盖的全局文件。涉及分页、同名工具多次调用或并发调用时，应扩展匹配规则，不能假设所有交互都能仅按方法顺序回放。

总结

把 MCP Server 当作第三方依赖后，测试重点会从“调用是否成功”转向“依赖边界是否仍然兼容”：

在 JSON-RPC 层录制和回放，覆盖初始化、工具发现与调用；
只固定智能体真正依赖的工具名称、Schema、内容结构和错误语义；
区分工具业务失败、协议错误、超时、断连与畸形响应；
在客户端和 Server 升级前运行同一组契约测试；
对录制数据脱敏，并为并发和多场景设计明确的匹配规则。

录制回放不能替代真实环境测试，但它能提供一个可重复、可审查的兼容性边界，让 MCP 的变化在进入智能体运行时之前就暴露出来。

别让 MCP 服务悄悄破坏智能体：录制回放与故障注入实战

MCP 的兼容性不止是“能连上”

先定义值得稳定的契约

实现录制与回放夹具

编写契约测试与最小录制样本

把验证放到升级流程里

总结

相关文章

模型升级不靠猜：用契约测试守住 LLM 应用行为边界

别让 AI 值班变成盲目执行：用 Node.js 构建证据驱动的止损闭环

AI 应用也需要供应链清单：统一追踪模型、提示词与工具版本

评论