逐页读完 MCP 工具目录：真实 stdio 分页、游标与失败边界

客户端启动后只显示 alpha，服务端明明还注册了 beta 和 gamma。连接建立成功，工具的输入 Schema 也能读取，问题却藏在发现阶段：代码把第一次 tools/list 的返回值当成了完整目录。目录较小时，这个问题容易被单页响应掩盖；当服务端按权限、后端批次或响应体大小拆分结果，遗漏才会暴露出来。

本文从这个故障出发，写一个真正经由标准输入输出与 MCP 服务端通信的 Python 客户端。服务端提供三个人工工具名，正常目录分成三页，条数依次是 1、0、2。中间的空页专门用于检验终止条件。示例中的数据与错误场景是合成的，初始化、tools/list、游标回传和协议错误由真实 SDK 处理。

读完后可以把 enumerate_tools 接到已有 ClientSession 上，获得一份经过边界检查的目录。这里不重复宿主安装和第一个工具的注册过程；没有调用模型、账户、外部 MCP 或真实业务系统，也没有验证任何宿主的工具选择界面。

代码固定使用 MCP Python SDK 1.29.0。本次 Python 3.12.14 环境中的真实 initialize 返回 protocolVersion=2025-11-25，因此可执行部分按该版本验收。2025-11-25 分页规范规定了不透明游标、服务端决定页大小以及用 nextCursor 继续查询的流程。客户端不应拆解、修改游标，旧版规范还明确要求不跨会话持久化它。

查阅日期为 2026-10-07。此时官方 latest 链接已经指向 2026-07-28 分页页面，页面增加了请求元数据说明，结果示例也出现了新字段，并明确空字符串属于有效游标。本实验没有实现新版全部消息契约，也没有用新版 SDK 复测；不要把下方的通过记录解释成 2026-07-28 协议兼容性证明。

我们的终止判断写成 next_cursor is None。空字符串会原样进入下一次请求，测试覆盖了这条路径。这样的写法也避免 Python 的真假值判断把空字符串误当成结束。旧版 SDK 的模型把缺省 nextCursor 映射为 None；本例以这个解析后的对象边界处理，不把 JSON null 单独宣称为规范推荐写法。

枚举函数只在完整走到末页时返回。中途遇到重复工具名、重复游标、页数用完或工具数量超限，会抛出 CatalogError。调用方不会误拿到一个看似成功的部分列表去覆盖旧目录。保留可用的旧目录、显示刷新失败以及决定何时重试，属于上层应用需要另行安排的行为。

seen_cursors 保存本轮响应已经给过的游标。A 后再返回 A 会被拦住，A、B、A 这样的长环也会被拦住。这是客户端的防循环策略，并非宣称协议为每一种重复情形规定了同一个错误码。即使服务端不断生成全新的游标，max_pages 仍能让本轮请求收敛。这里的 32 页只是样例默认预算，需要按自己的目录规模调整。

max_tools 限制成功累计的工具数量。本页数据在交给函数前已经由 SDK 接收和解析，因此这个限制无法防止一张超大响应先耗用内存。面对不可信远程服务，还需要传输层响应体限制、并发限制和更完整的资源配额。示例设置每次请求 3 秒超时，并给启动、初始化、枚举及退出包上 15 秒总预算；这些也只是本地小实验的选择。

客户端不能从游标里推导页码。为了让这一点能直接观察，本例由服务端生成随机字符串，把它映射到 generation 与页索引。页内容保存在内存中的固定列表，同一代次、同一页索引重复请求时返回同一个下一游标。测试比较两次响应，确认这一小段稳定性由代码实际兑现。

随机字符串不是完整的生产安全设计。这个服务端每次由客户端启动一个独立进程，进程里只有一份演示目录；token 映射不会持久化。生产多租户服务应把游标范围与已认证主体、授权视图、查询条件和目录版本绑定，还要处理过期、容量与权限撤销。即使采用签名游标，也必须检查签名之外的访问权限；知道 token 不应成为读取工具目录的额外授权。

目录变更采用“拒绝旧代次”的示例策略。change 模式在第二次查询前递增 generation，已发出的 token 因代次不匹配而得到 -32602。客户端不自行猜测一个替代位置，也不无限重开遍历。测试在捕获错误后显式启动一次新遍历，验证新一轮能够完成。这里没有保留旧快照供继续读取，也没有测试数据库事务快照。

这套快照和失效规则由本文样例定义。普通 tools/list 并没有为所有服务端提供同一种跨页事务保证。另一台服务端可能让目录在枚举中改变，而且不重复任何名称；仅靠本例的重名检查无法发现漏项。依赖严格一致目录的产品，需要先获得服务端具体的版本或快照契约。

需要 Python 3.12 和 MCP SDK 1.29.0。下面给出全部四个文件，保存到同一空目录即可。没有隐藏的数据文件、环境变量、密钥或额外服务。requirements.txt 只固定直接依赖；传递依赖由安装器解析，不能称为完全锁定的可重现环境。本文实际使用预装依赖，没有重新联网安装，下列安装步骤尚未在全新环境验收。

mkdir paginated-catalog
cd paginated-catalog
python3.12 -m venv .venv
# macOS / Linux
. .venv/bin/activate
# Windows PowerShell 改用：.venv\Scripts\Activate.ps1
# 保存下列四个文件后执行
python -m pip install -r requirements.txt
python client.py
python -m unittest -v

requirements.txt · 完整文件

mcp==1.29.0

服务端文件使用低层 Server，是为了接收 ListToolsRequest 本身并返回带 nextCursor 的 ListToolsResult。固定版本的 官方 server.py支持这种处理器签名，避免把 list[Tool] 的单页便利形式误当成分页接口。应用日志不要写入 stdout，那里用于 SDK 的协议帧。

server.py · 完整文件

"""Synthetic catalog served over real MCP stdio; Python MCP SDK 1.29.0."""
import argparse
import secrets
import anyio
from mcp import types
from mcp.server.lowlevel import Server
from mcp.server.stdio import stdio_server
from mcp.shared.exceptions import McpError

SCHEMA = {"type": "object", "properties": {}, "additionalProperties": False}

def tool(name):
    return types.Tool(name=name, description="Synthetic catalog entry: " + name,
                      inputSchema=SCHEMA)

class Catalog:
    def __init__(self, mode="normal"):
        self.mode = mode
        self.generation = 1
        self.calls = 0
        self.tokens = {}
        self.next_tokens = {}
        self.pages = [[tool("alpha")], [], [tool("beta"), tool("gamma")]]

    def advance(self, index):
        key = (self.generation, index)
        if key not in self.next_tokens:
            token = "" if self.mode == "empty-token" and index == 1 else secrets.token_urlsafe(18)
            self.next_tokens[key] = token
            self.tokens[token] = key
        return self.next_tokens[key]

    def page(self, cursor):
        self.calls += 1
        if self.mode == "change" and self.calls == 2:
            # Test-only mutation: old cursors must not silently read a new generation.
            self.generation += 1
        if cursor is None:
            index = 0
        else:
            key = self.tokens.get(cursor)
            if key is None or key[0] != self.generation:
                raise McpError(types.ErrorData(code=-32602, message="Invalid or expired catalog cursor"))
            index = key[1]
        if self.mode == "single":
            return types.ListToolsResult(tools=[tool("alpha")])
        if self.mode == "empty":
            return types.ListToolsResult(tools=[])
        if self.mode == "endless":
            return types.ListToolsResult(tools=[], nextCursor=self.advance(index + 1))
        if self.mode == "repeat" and cursor is not None:
            return types.ListToolsResult(tools=[], nextCursor=cursor)
        if self.mode == "cycle" and index == 2:
            return types.ListToolsResult(tools=[], nextCursor=self.advance(1))
        entries = self.pages[index]
        if self.mode == "duplicate" and index == 2:
            entries = [tool("alpha")]
        next_cursor = self.advance(index + 1) if index < len(self.pages) - 1 else None
        return types.ListToolsResult(tools=entries, nextCursor=next_cursor)

async def main(mode):
    catalog = Catalog(mode)
    server = Server("paginated-catalog-lab")

    @server.list_tools()
    async def list_tools(request: types.ListToolsRequest) -> types.ListToolsResult:
        cursor = request.params.cursor if request.params is not None else None
        return catalog.page(cursor)

    @server.call_tool()
    async def call_tool(name, arguments):
        # Every advertised entry is callable, but has no external side effect.
        if name not in {"alpha", "beta", "gamma"}:
            raise ValueError("Unknown synthetic tool")
        return [types.TextContent(type="text", text="synthetic:" + name)]

    async with stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--mode", default="normal", choices=["normal", "single", "empty", "empty-token", "repeat", "cycle", "endless", "duplicate", "change"])
    anyio.run(main, parser.parse_args().mode)

每一轮都只调用一次 session.list_tools。首页使用 params=None；后续构造 PaginatedRequestParams(cursor=cursor)。SDK 的 ClientSession.list_tools 实现发送单个请求并返回单个 ListToolsResult，完整枚举需要应用自行循环。我们使用 params 形式，避免依赖该版本已经标记弃用的快捷 cursor 参数。

名称去重是有意选择的严格策略：同一工具在不同页重复出现就让本轮失败，避免不同 Schema 被静默覆盖。工具定义变化之后该保留哪一个版本，无法用一个字典赋值替业务决定。本函数按服务端顺序保留结果，没有额外排序，也没有执行这些工具。

client.py · 完整文件

"""Bounded enumeration. A partial catalog is never returned as success."""
import argparse
import json
import sys
from pathlib import Path
from importlib.metadata import version
from datetime import timedelta
import anyio
from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client

class CatalogError(RuntimeError):
    pass

async def enumerate_tools(session, *, max_pages=32, max_tools=10000):
    if type(max_pages) is not int or max_pages < 1:
        raise ValueError("max_pages must be a positive integer")
    if type(max_tools) is not int or max_tools < 0:
        raise ValueError("max_tools must be a non-negative integer")
    cursor = None
    seen_cursors = set()
    names = set()
    tools = []
    page_sizes = []
    for _ in range(max_pages):
        params = None if cursor is None else types.PaginatedRequestParams(cursor=cursor)
        result = await session.list_tools(params=params)
        page_sizes.append(len(result.tools))
        for entry in result.tools:
            if entry.name in names:
                raise CatalogError("duplicate tool name; discard incomplete catalog")
            names.add(entry.name)
            tools.append(entry)
            if len(tools) > max_tools:
                raise CatalogError("tool limit reached; discard incomplete catalog")
        next_cursor = result.nextCursor
        if next_cursor is None:
            return tools, page_sizes
        if next_cursor in seen_cursors:
            raise CatalogError("cursor repeated; discard incomplete catalog")
        seen_cursors.add(next_cursor)
        cursor = next_cursor
    raise CatalogError("page limit reached; discard incomplete catalog")

async def demo(mode, max_pages):
    server = Path(__file__).with_name("server.py")
    params = StdioServerParameters(command=sys.executable, args=[str(server), "--mode", mode])
    # Total budget includes startup, initialization, pagination and shutdown.
    with anyio.fail_after(15):
        async with stdio_client(params) as (read, write):
            async with ClientSession(read, write, read_timeout_seconds=timedelta(seconds=3)) as session:
                init = await session.initialize()
                entries, sizes = await enumerate_tools(session, max_pages=max_pages)
                print(json.dumps({"protocol": init.protocolVersion, "sdk": version("mcp"),
                    "mode": mode, "page_sizes": sizes, "names": [x.name for x in entries]}, ensure_ascii=False))

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--mode", default="normal")
    parser.add_argument("--max-pages", type=int, default=32)
    args = parser.parse_args()
    anyio.run(demo, args.mode, args.max_pages)

test_catalog.py 会用当前 Python 解释器启动 server.py，通过 stdio_client 和 ClientSession 初始化会话。除本地参数校验测试外，分页和协议错误都来自子进程，不是把 list_tools 替换成假的函数。每项测试拥有自己的服务进程，关闭由 SDK 的异步上下文管理器负责。测试进程正常退出可以证明本轮上下文走完；本文没有对任意失控子进程做强杀策略验证。

测试分别检查单页、空目录、空中间页、空字符串游标、同页重放、立即重复、隔页循环、无限新游标、页数边界、工具数量边界、名称重复、无效游标、目录换代和跨进程旧游标。正常模式还调用最后一页的 gamma，确认读到的名字对应本地合成工具；它只返回固定文本，没有业务副作用。

test_catalog.py · 完整文件

import json
import sys
import unittest
from contextlib import asynccontextmanager
from datetime import timedelta
from pathlib import Path
from importlib.metadata import version
import anyio
from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client
from mcp.shared.exceptions import McpError
from client import enumerate_tools, CatalogError

@asynccontextmanager
async def connect(mode):
    params = StdioServerParameters(command=sys.executable,
        args=[str(Path(__file__).with_name("server.py")), "--mode", mode])
    with anyio.fail_after(15):
        async with stdio_client(params) as (read, write):
            async with ClientSession(read, write, read_timeout_seconds=timedelta(seconds=3)) as session:
                init = await session.initialize()
                yield session, init

def real_test(fn):
    def run(self):
        anyio.run(fn, self)
    return run

class CatalogTests(unittest.TestCase):
    def test_sdk_version(self):
        self.assertEqual(version("mcp"), "1.29.0")

    @real_test
    async def test_normal_with_empty_middle_page(self):
        async with connect("normal") as (session, init):
            self.assertEqual(init.protocolVersion, "2025-11-25")
            self.assertIsNotNone(init.capabilities.tools)
            entries, sizes = await enumerate_tools(session)
            self.assertEqual([t.name for t in entries], ["alpha", "beta", "gamma"])
            self.assertEqual(sizes, [1, 0, 2])
            called = await session.call_tool("gamma", {})
            self.assertFalse(called.isError)
            self.assertEqual(called.content[0].text, "synthetic:gamma")

    @real_test
    async def test_one_page(self):
        async with connect("single") as (session, _):
            tools, sizes = await enumerate_tools(session, max_pages=1)
            self.assertEqual([t.name for t in tools], ["alpha"])
            self.assertEqual(sizes, [1])

    @real_test
    async def test_empty_final_page(self):
        async with connect("empty") as (session, _):
            self.assertEqual(await enumerate_tools(session), ([], [0]))

    @real_test
    async def test_empty_string_cursor_is_sent(self):
        async with connect("empty-token") as (session, _):
            first = await session.list_tools()
            self.assertEqual(first.nextCursor, "")
            tools, sizes = await enumerate_tools(session)
            self.assertEqual(sizes, [1, 0, 2])
            self.assertEqual(len(tools), 3)

    @real_test
    async def test_replay_is_stable_in_same_generation(self):
        async with connect("normal") as (session, _):
            first = await session.list_tools()
            again = await session.list_tools()
            self.assertEqual(first.model_dump(), again.model_dump())
            params = types.PaginatedRequestParams(cursor=first.nextCursor)
            second = await session.list_tools(params=params)
            replay = await session.list_tools(params=params)
            self.assertEqual(second.model_dump(), replay.model_dump())
            self.assertEqual(second.tools, [])
            self.assertIsNotNone(second.nextCursor)

    async def expect_catalog_error(self, mode, message, **kwargs):
        async with connect(mode) as (session, _):
            with self.assertRaisesRegex(CatalogError, message):
                await enumerate_tools(session, **kwargs)

    @real_test
    async def test_repeat_cursor(self):
        await self.expect_catalog_error("repeat", "cursor repeated")

    @real_test
    async def test_longer_cursor_cycle(self):
        await self.expect_catalog_error("cycle", "cursor repeated")

    @real_test
    async def test_unending_unique_cursors(self):
        await self.expect_catalog_error("endless", "page limit", max_pages=4)

    @real_test
    async def test_page_limit_before_completion(self):
        await self.expect_catalog_error("normal", "page limit", max_pages=2)

    @real_test
    async def test_exact_page_limit_succeeds(self):
        async with connect("normal") as (session, _):
            tools, sizes = await enumerate_tools(session, max_pages=3)
            self.assertEqual(len(tools), 3)
            self.assertEqual(len(sizes), 3)

    @real_test
    async def test_tool_limit(self):
        await self.expect_catalog_error("normal", "tool limit", max_tools=2)

    @real_test
    async def test_duplicate_name(self):
        await self.expect_catalog_error("duplicate", "duplicate tool")

    @real_test
    async def test_invalid_cursor(self):
        async with connect("normal") as (session, _):
            with self.assertRaises(McpError) as caught:
                await session.list_tools(params=types.PaginatedRequestParams(cursor="not-issued"))
            self.assertEqual(caught.exception.error.code, -32602)

    @real_test
    async def test_generation_change_rejects_old_cursor(self):
        async with connect("change") as (session, _):
            with self.assertRaises(McpError) as caught:
                await enumerate_tools(session)
            self.assertEqual(caught.exception.error.code, -32602)
            # Explicit new traversal; the client never retries automatically.
            tools, sizes = await enumerate_tools(session)
            self.assertEqual([t.name for t in tools], ["alpha", "beta", "gamma"])
            self.assertEqual(sizes, [1, 0, 2])

    @real_test
    async def test_cursor_not_valid_in_new_server(self):
        async with connect("normal") as (session, _):
            cursor = (await session.list_tools()).nextCursor
        async with connect("normal") as (session, _):
            with self.assertRaises(McpError) as caught:
                await session.list_tools(params=types.PaginatedRequestParams(cursor=cursor))
            self.assertEqual(caught.exception.error.code, -32602)

    @real_test
    async def test_invalid_local_limits_before_request(self):
        # None deliberately has no list_tools; validation must happen first.
        for limit in (0, -1, True, 1.5):
            with self.assertRaises(ValueError):
                await enumerate_tools(None, max_pages=limit)
        for limit in (-1, True, 1.5):
            with self.assertRaises(ValueError):
                await enumerate_tools(None, max_tools=limit)

if __name__ == "__main__":
    unittest.main(verbosity=2)

2026-10-07 的本地运行完成 17 项 unittest 测试，全部通过。正常命令 python client.py 的输出如下。随机游标不写入输出，避免把一个仅属于当前服务进程的 token 误作可长期复用的配置。

{"protocol": "2025-11-25", "sdk": "1.29.0", "mode": "normal", "page_sizes": [1, 0, 2], "names": ["alpha", "beta", "gamma"]}

然后执行 python client.py --mode repeat。程序应以失败退出，异常末尾包含 cursor repeated。执行 python client.py --mode endless --max-pages 4，则应包含 page limit。这些失败是教学目录故意生成的，不代表真实部署的服务端应这样响应。SDK 异步清理可能把错误包在 ExceptionGroup 中，排错时查看其中的 CatalogError，而不要只根据最外层标题判断错误类型。

想观察目录换代时，运行 python client.py --mode change，应得到 code=-32602 的 MCP 错误。测试中的第二次枚举是显式操作；命令行演示不自动恢复。应用接到类似错误后可以提示刷新，或在受限次数内重新枚举。重试预算应与同一用户操作绑定，否则服务端持续变化时，每次重启都会重新得到一个“有限”预算，整体仍可能没有终点。

目录获取成功仅说明这一轮消息及本地约束通过。工具出现于列表不等于允许执行，也不保证下一刻仍然可用。实际调用还要按当前授权、输入 Schema 和服务端错误处理。若不同用户看到不同工具，目录缓存必须按对应主体和权限上下文分开，不能共享这份演示进程中的内存缓存方式。

固定版本的 Tools 规范包含 tools.listChanged 能力与 notifications/tools/list_changed 通知。本文没有启用该通知能力，也未编写通知处理器；change 模式只是为了确定性触发旧游标失效。准备添加通知时，可以让正在进行的刷新作废，再发起有预算的新遍历，但通知与当前分页请求之间的竞态仍须依照服务端契约验收。

接入自己的目录时，先保留这组回归，把合成页替换为受控后端，再增加权限切换、游标过期、目录删除与并发变化的用例。若目标宿主声称自动完成分页，还应给它接入一份至少三页且包含空中间页的测试目录，核对最后一页工具是否真的可见。本次只证明自写客户端的行为，没有替任何宿主做这项验收。

前面的 enumerate_tools 会在重名时拒绝整轮结果，调用方保存的旧列表也确实没有被替换。把同一个 ClientSession 继续用于工具调用时，还有一份状态值得检查：SDK 为结果校验维护的 outputSchema 缓存。2026-10-13 的补充实验把 alpha 原来的 value:string 改成 value:integer，再用第二页的同名 alpha 触发拒绝。旧列表仍写着 string，随后返回字符串 old 的调用却被客户端拒绝了。

另一个方向更容易漏过：两张刷新页都省略 outputSchema。外层同样因重名失败，旧列表仍要求字符串；随后服务端返回数字 7，SDK 却接受了。把同一份结果交给保留的旧 Schema 单独校验，仍会失败。这两组观察把“没有替换列表”与“仍按旧合同校验调用结果”区分开来。

本节继续固定 Python MCP SDK 1.29.0、Python 3.12.14 与实际协商的 2025-11-25 协议。原有九节、三个 Python 文件、requirements.txt、元数据与日期均保留。下面增加一个上下文限定的候选会话入口、确定性输入和回归测试，所有服务与调用都只在本机新建的合成子进程中运行。没有账户、模型、远程业务调用或外部副作用。

核对 1.29.0 的官方 ClientSession 源码后，可以解释这个顺序：list_tools 先接收一页结果，再按名称保存这一页的 outputSchema，最后才把该页交给调用方。外层的重名判断发生在这之后。即使 enumerate_tools 抛错，已经完成的页级缓存赋值也没有被它撤回。下次成功返回的 call_tool 结果会走 SDK 的结构化结果校验，使用的是会话里的那份值。

客户端结果验证发生在 tools/call 响应到达之后。client_schema_error 不能说明服务端工具从未执行，更不表示它的副作用被回滚；请求轨迹会明确显示这次调用已经发出。本夹具只有固定数据返回，没有真实副作用。

省略 outputSchema 的工具被模型解析后，其属性为 None；这个版本仍会把 None 存入该名称的缓存。后续校验遇到这个值便跳过 Schema 验证。实验没有读取、修改或备份 SDK 的私有缓存，而是观察公开 list_tools / call_tool 行为，再用旧目录的 Schema 做独立核对。这样能看到实际影响，也避免把私有字段赋回去当作恢复接口。

夹具刻意在跨页目录中重复名称，并能返回与当前宣告 Schema 不一致的数据。服务端使用原始低层 CallToolRequest 处理器，返回 content=[]、structuredContent 和 isError=False，避免高层工具装饰器先替我们校验或额外查询目录。2025-11-25 Tools 规范的官方源文件要求宣告输出 Schema 的服务端给出符合它的结构化结果，并建议客户端验证。这里有意违反约定来检验客户端的防护；不要把这段夹具当作合规生产服务。

下面的 schema_server.py 按 fixture.pages 顺序响应，每一步核对收到的游标；调用只允许 alpha 和空参数。它在 finally 中写请求轨迹与 closed=true，标准输出仍完全留给 MCP 协议。

保存为 schema_server.py：

"""Deterministic, local-only MCP catalog fixture. No external tool actions."""
import json
from pathlib import Path
import sys
import anyio
from mcp import types
from mcp.server.lowlevel import Server
from mcp.server.stdio import stdio_server

async def main(fixture_path, audit_path):
    fixture = json.loads(Path(fixture_path).read_text())
    audit = {'requests': [], 'closed': False}
    server = Server('catalog-cache-fixture')
    position = 0

    @server.list_tools()
    async def list_tools(request: types.ListToolsRequest) -> types.ListToolsResult:
        nonlocal position
        cursor = request.params.cursor if request.params is not None else None
        step = fixture['pages'][position]
        if cursor != step['cursor']:
            raise ValueError('fixture cursor mismatch')
        position += 1
        audit['requests'].append({'method': 'tools/list', 'cursor': cursor})
        return types.ListToolsResult.model_validate(step['result'])

    # Deliberately raw low-level fixture: returns exactly the declared payload.
    # No high-level server output validation or implicit tool-list lookup.
    async def call_tool(request: types.CallToolRequest):
        if request.params.name != 'alpha' or request.params.arguments != {}:
            raise ValueError('unexpected synthetic call')
        audit['requests'].append({'method': 'tools/call', 'name': 'alpha'})
        result = types.CallToolResult(content=[],
            structuredContent=fixture['call_content'], isError=False)
        return types.ServerResult(result)

    server.request_handlers[types.CallToolRequest] = call_tool
    try:
        async with stdio_server() as (read, write):
            await server.run(read, write, server.create_initialization_options())
    finally:
        audit['closed'] = True
        Path(audit_path).write_text(json.dumps(audit, indent=2) + '\n')

if __name__ == '__main__':
    anyio.run(main, *sys.argv[1:])

候选路径会先建立一个新的 stdio 进程与 ClientSession，再把原样保留的 enumerate_tools 用在这段新连接上。只有读到末页并通过原有名称、游标和容量检查，candidate_catalog 才把 session 与完整 entries 一起 yield 给调用方。失败发生在 yield 之前，调用方没有拿到候选对象；退出候选上下文后，再使用原来的活动会话。

这个入口接收 StdioServerParameters，可以接到另一个受控本地服务；其中的命令和参数仍须由可信宿主提供。示例固定 3 秒请求预算和 15 秒整段上下文预算，后者包括调用方在 async with 内做的工作，不能直接套到长期运行的宿主。入口严格检查 SDK 和协商版本，升级后需要明确修改约束并重跑。

保存为 candidate_catalog.py：

"""Fixed-version, context-bound stdio candidate; not a session pool or swap manager."""
from contextlib import asynccontextmanager
from dataclasses import dataclass
from datetime import timedelta
from importlib.metadata import version
import sys
from typing import TextIO
import anyio
from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client
from client import enumerate_tools

@dataclass(frozen=True)
class ReadyCatalog:
    session: ClientSession
    entries: tuple[types.Tool, ...]
    page_sizes: tuple[int, ...]

@asynccontextmanager
async def initialized_session(params: StdioServerParameters, *, errlog: TextIO = sys.stderr):
    if version("mcp") != "1.29.0":
        raise RuntimeError("This example requires mcp==1.29.0")
    # A small-demo budget for the entire context, including caller's work.
    with anyio.fail_after(15):
        async with stdio_client(params, errlog=errlog) as (read, write):
            async with ClientSession(read, write,
                    read_timeout_seconds=timedelta(seconds=3)) as session:
                init = await session.initialize()
                if init.protocolVersion != "2025-11-25":
                    raise RuntimeError("Unexpected negotiated protocol")
                yield session

@asynccontextmanager
async def candidate_catalog(params: StdioServerParameters, *, errlog: TextIO = sys.stderr,
                            max_pages=32, max_tools=10000):
    """Yield only a fully enumerated candidate, usable inside this context only.

    This opens a separate session. It never receives or changes an active session.
    Exit in the same task and normal nested context order; do not transfer ownership.
    """
    async with initialized_session(params, errlog=errlog) as session:
        entries, sizes = await enumerate_tools(session, max_pages=max_pages, max_tools=max_tools)
        yield ReadyCatalog(session, tuple(entries), tuple(sizes))

ReadyCatalog 只在它所属的 async with 内有效。tuple 与 frozen dataclass 固定了外层容器，内部 Tool 和 Schema 对象仍可变，调用方应只读使用。不要把 ready.session 存到上下文外继续调用，也不要在另一个任务关闭它。本例按正常嵌套顺序进入、退出 AnyIO/SDK 上下文，没有实现可转移的会话所有者、并发指针替换或旧会话的排空退役。

在隔离场景中，活动会话 A 只收到一次最初的 tools/list，以及刷新前后的两次 tools/call；失败的候选 B 收到两次 tools/list，没有 tools/call。另一组输入让 A 两次都返回数字 7，客户端前后都拒绝它。这条反向控制很重要：只检查 old 仍能调用成功，无法发现校验已被清空。成功候选也分别检查接受 7 和拒绝 old，确保两页读完后确实采用 integer 合同。

本页当前提供完整代码块，没有可点击下载的 ZIP 入口。先新建一个只放本文示例的 source 目录：保存原文的 client.py、server.py、test_catalog.py 与 requirements.txt，再把本节新增文件放入同一目录；八份输入保存在 fixtures 子目录。合计 21 个文件，包括 sample-output.json 这份历史输出参考；不能把它当作本次运行结果。无需隐藏环境变量、密钥或已运行的服务。

# 先按代码块的文件名保存全部 21 个文件，再进入 source
cd source
python3.12 -m venv .venv
# macOS / Linux
. .venv/bin/activate
# Windows PowerShell 使用 .venv\Scripts\Activate.ps1；本次未验 Windows
python -m pip install -r requirements-lock.txt
python -B client.py
python -B -m unittest -v test_catalog.py
python -B run_cache_probe.py
python -B -m unittest -v test_session_cache.py
python -B check_mutants.py
python -B -O run_cache_probe.py
python -B -O -m unittest -v test_session_cache.py
python -B -O check_mutants.py

2026-10-20 复跑：先区分启动预算和目录断言

这次从实际页面重新提取的 21 个文件与内容记录逐字一致。原样运行 python -B -m unittest -v test_catalog.py test_session_cache.py，21 项中 11 项通过、10 项报错；错误发生在初始化阶段，包括明确的 3 秒响应超时，以及 SDK 清理期间的 BrokenResourceError。另选三项、不带 -B 重跑，仍有一项报错。因此不能把历史通过记录当作当前机器上原预算必然通过的保证，也不能只看到最外层 ExceptionGroup 就判断是分页逻辑出错。

在独立副本里，仅把三个文件中的请求预算 3 秒改为 10 秒、整段上下文预算 15 秒改为 60 秒，普通模式的 21 项测试全部通过，独立六场景探针的八份请求审计匹配，四个错误变异也被指定检查检出。但同一副本用 -O 运行 21 项测试时，仍有 20 项通过、1 项在候选会话进入阶段出现 BrokenResourceError；这一轮没有继续运行 -O 变异检查。分页算法、候选隔离、夹具、固定预期和错误处理均未改动。这个受控对照说明本次结果对预算敏感，不足以证明每个 BrokenResourceError 都由超时造成；不要捕获并忽略它，也不要无限提高超时来换取绿色结果。

如需复现这个诊断，先在 source 内运行下面的 POSIX shell 命令（本次实测 Linux；未验 Windows）。它只创建旁边的 source-budget-check，不改原目录；目标目录已存在或预算锚点不匹配时停止。10 秒与 60 秒仍是本次实验预算，并非生产推荐值；60 秒同样包含调用方在 async with 内的工作。

python - <<'PY'
from pathlib import Path
import shutil

source = Path.cwd()  # 当前目录必须是上面新建的 source
candidate = source.with_name(source.name + "-budget-check")
updates = {}
for name in ("client.py", "test_catalog.py", "candidate_catalog.py"):
    text = (source / name).read_text(encoding="utf-8")
    for old, new in (("anyio.fail_after(15)", "anyio.fail_after(60)"),
                     ("timedelta(seconds=3)", "timedelta(seconds=10)")):
        if text.count(old) != 1:
            raise SystemExit(f"unexpected budget anchor in {name}: {old}")
        text = text.replace(old, new, 1)
    updates[name] = text
# 目标已存在时直接失败，不覆盖已有目录；不复制虚拟环境或旧运行结果。
shutil.copytree(source, candidate,
                ignore=shutil.ignore_patterns(".venv", "__pycache__", "runs"))
for name, text in updates.items():
    (candidate / name).write_text(text, encoding="utf-8")
print(candidate)
PY
cd ../source-budget-check
python -B -m unittest -v test_catalog.py test_session_cache.py
python -B run_cache_probe.py
python -B check_mutants.py
python -B -O -m unittest -v test_catalog.py test_session_cache.py

本轮使用预装 Python 3.12.14、MCP 1.29.0、AnyIO 4.14.2、jsonschema 4.26.0 与 Pydantic 2.13.4；并非重新安装 requirements-lock.txt 的结果。29 项锁定版本中有两项不同：cryptography 为 50.0.0（清单为 50.0.2），pycparser 为 3.0（清单为 3.11）。不要把该对照写成完整锁定环境、全新安装、Windows、其他 SDK 或生产并发切换的验收。

requirements.txt 继续保持原来的单行内容。新增 requirements-lock.txt 固定这次 Linux / Python 3.12.14 环境中完整解析出的依赖版本，包含传递依赖及其额外依赖；它没有包文件摘要，也没有内置 wheel。新建虚拟环境安装后已用该组版本复跑；其他系统、镜像源和未来下载可用性仍需另验。

保存为 requirements-lock.txt：

# Linux / CPython 3.12.14 resolved dependencies, including requested extras.
# Exact versions only; this file does not pin wheel hashes or include packages.
annotated-types==0.8.0
anyio==4.14.2
attrs==26.1.0
certifi==2026.7.22
cffi==2.1.1
click==8.4.2
cryptography==50.0.2
h11==0.16.0
httpcore==1.0.9
httpx-sse==0.4.3
httpx==0.28.1
idna==3.18
jsonschema-specifications==2025.9.1
jsonschema==4.26.0
mcp==1.29.0
pycparser==3.11
pydantic-core==2.46.4
pydantic-settings==2.15.0
pydantic==2.13.4
pyjwt==2.13.0
python-dotenv==1.2.2
python-multipart==0.0.32
referencing==0.37.0
rpds-py==2026.6.3
sse-starlette==3.4.8
starlette==1.6.0
typing-extensions==4.16.0
typing-inspection==0.4.3
uvicorn==0.52.1

in_place 两份输入先返回一页完整 string 目录，再依次返回两页重名目录。active 两份输入始终只有原来的 string 目录；candidate 两份输入只给重名刷新页。success 两份输入把第二页改为 beta，因而可完整交付。active / active_invalid 与 candidate_success / candidate_success_invalid 各自保持页内容一致，只改变 call_content，以分别检验接受与拒绝。candidate / candidate_invalid 使用相同重名页，但启动成两个独立进程；它们的调用内容不会被请求。字符串 next 只是本地固定游标，客户端原样回传，未赋予它业务含义。

{
  "pages": [
    {
      "cursor": null,
      "result": {
        "tools": [
          {
            "name": "alpha",
            "description": "Synthetic entry",
            "inputSchema": {
              "type": "object",
              "properties": {},
              "additionalProperties": false
            },
            "outputSchema": {
              "type": "object",
              "properties": {
                "value": {
                  "type": "string"
                }
              },
              "required": [
                "value"
              ],
              "additionalProperties": false
            }
          }
        ]
      }
    }
  ],
  "call_content": {
    "value": "old"
  }
}

{
  "pages": [
    {
      "cursor": null,
      "result": {
        "tools": [
          {
            "name": "alpha",
            "description": "Synthetic entry",
            "inputSchema": {
              "type": "object",
              "properties": {},
              "additionalProperties": false
            },
            "outputSchema": {
              "type": "object",
              "properties": {
                "value": {
                  "type": "string"
                }
              },
              "required": [
                "value"
              ],
              "additionalProperties": false
            }
          }
        ]
      }
    }
  ],
  "call_content": {
    "value": 7
  }
}

{
  "pages": [
    {
      "cursor": null,
      "result": {
        "tools": [
          {
            "name": "alpha",
            "description": "Synthetic entry",
            "inputSchema": {
              "type": "object",
              "properties": {},
              "additionalProperties": false
            },
            "outputSchema": {
              "type": "object",
              "properties": {
                "value": {
                  "type": "integer"
                }
              },
              "required": [
                "value"
              ],
              "additionalProperties": false
            }
          }
        ],
        "nextCursor": "next"
      }
    },
    {
      "cursor": "next",
      "result": {
        "tools": [
          {
            "name": "alpha",
            "description": "Synthetic entry",
            "inputSchema": {
              "type": "object",
              "properties": {},
              "additionalProperties": false
            },
            "outputSchema": {
              "type": "object",
              "properties": {
                "value": {
                  "type": "integer"
                }
              },
              "required": [
                "value"
              ],
              "additionalProperties": false
            }
          }
        ]
      }
    }
  ],
  "call_content": {
    "value": "unused"
  }
}

{
  "pages": [
    {
      "cursor": null,
      "result": {
        "tools": [
          {
            "name": "alpha",
            "description": "Synthetic entry",
            "inputSchema": {
              "type": "object",
              "properties": {},
              "additionalProperties": false
            },
            "outputSchema": {
              "type": "object",
              "properties": {
                "value": {
                  "type": "integer"
                }
              },
              "required": [
                "value"
              ],
              "additionalProperties": false
            }
          }
        ],
        "nextCursor": "next"
      }
    },
    {
      "cursor": "next",
      "result": {
        "tools": [
          {
            "name": "alpha",
            "description": "Synthetic entry",
            "inputSchema": {
              "type": "object",
              "properties": {},
              "additionalProperties": false
            },
            "outputSchema": {
              "type": "object",
              "properties": {
                "value": {
                  "type": "integer"
                }
              },
              "required": [
                "value"
              ],
              "additionalProperties": false
            }
          }
        ]
      }
    }
  ],
  "call_content": {
    "value": "unused"
  }
}

{
  "pages": [
    {
      "cursor": null,
      "result": {
        "tools": [
          {
            "name": "alpha",
            "description": "Synthetic entry",
            "inputSchema": {
              "type": "object",
              "properties": {},
              "additionalProperties": false
            },
            "outputSchema": {
              "type": "object",
              "properties": {
                "value": {
                  "type": "integer"
                }
              },
              "required": [
                "value"
              ],
              "additionalProperties": false
            }
          }
        ],
        "nextCursor": "next"
      }
    },
    {
      "cursor": "next",
      "result": {
        "tools": [
          {
            "name": "beta",
            "description": "Synthetic entry",
            "inputSchema": {
              "type": "object",
              "properties": {},
              "additionalProperties": false
            },
            "outputSchema": {
              "type": "object",
              "properties": {
                "value": {
                  "type": "integer"
                }
              },
              "required": [
                "value"
              ],
              "additionalProperties": false
            }
          }
        ]
      }
    }
  ],
  "call_content": {
    "value": 7
  }
}

{
  "pages": [
    {
      "cursor": null,
      "result": {
        "tools": [
          {
            "name": "alpha",
            "description": "Synthetic entry",
            "inputSchema": {
              "type": "object",
              "properties": {},
              "additionalProperties": false
            },
            "outputSchema": {
              "type": "object",
              "properties": {
                "value": {
                  "type": "integer"
                }
              },
              "required": [
                "value"
              ],
              "additionalProperties": false
            }
          }
        ],
        "nextCursor": "next"
      }
    },
    {
      "cursor": "next",
      "result": {
        "tools": [
          {
            "name": "beta",
            "description": "Synthetic entry",
            "inputSchema": {
              "type": "object",
              "properties": {},
              "additionalProperties": false
            },
            "outputSchema": {
              "type": "object",
              "properties": {
                "value": {
                  "type": "integer"
                }
              },
              "required": [
                "value"
              ],
              "additionalProperties": false
            }
          }
        ]
      }
    }
  ],
  "call_content": {
    "value": "old"
  }
}

{
  "pages": [
    {
      "cursor": null,
      "result": {
        "tools": [
          {
            "name": "alpha",
            "description": "Synthetic entry",
            "inputSchema": {
              "type": "object",
              "properties": {},
              "additionalProperties": false
            },
            "outputSchema": {
              "type": "object",
              "properties": {
                "value": {
                  "type": "string"
                }
              },
              "required": [
                "value"
              ],
              "additionalProperties": false
            }
          }
        ]
      }
    },
    {
      "cursor": null,
      "result": {
        "tools": [
          {
            "name": "alpha",
            "description": "Synthetic entry",
            "inputSchema": {
              "type": "object",
              "properties": {},
              "additionalProperties": false
            },
            "outputSchema": {
              "type": "object",
              "properties": {
                "value": {
                  "type": "integer"
                }
              },
              "required": [
                "value"
              ],
              "additionalProperties": false
            }
          }
        ],
        "nextCursor": "next"
      }
    },
    {
      "cursor": "next",
      "result": {
        "tools": [
          {
            "name": "alpha",
            "description": "Synthetic entry",
            "inputSchema": {
              "type": "object",
              "properties": {},
              "additionalProperties": false
            },
            "outputSchema": {
              "type": "object",
              "properties": {
                "value": {
                  "type": "integer"
                }
              },
              "required": [
                "value"
              ],
              "additionalProperties": false
            }
          }
        ]
      }
    }
  ],
  "call_content": {
    "value": "old"
  }
}

{
  "pages": [
    {
      "cursor": null,
      "result": {
        "tools": [
          {
            "name": "alpha",
            "description": "Synthetic entry",
            "inputSchema": {
              "type": "object",
              "properties": {},
              "additionalProperties": false
            },
            "outputSchema": {
              "type": "object",
              "properties": {
                "value": {
                  "type": "string"
                }
              },
              "required": [
                "value"
              ],
              "additionalProperties": false
            }
          }
        ]
      }
    },
    {
      "cursor": null,
      "result": {
        "tools": [
          {
            "name": "alpha",
            "description": "Synthetic entry",
            "inputSchema": {
              "type": "object",
              "properties": {},
              "additionalProperties": false
            }
          }
        ],
        "nextCursor": "next"
      }
    },
    {
      "cursor": "next",
      "result": {
        "tools": [
          {
            "name": "alpha",
            "description": "Synthetic entry",
            "inputSchema": {
              "type": "object",
              "properties": {},
              "additionalProperties": false
            }
          }
        ]
      }
    }
  ],
  "call_content": {
    "value": 7
  }
}

expected.json 和 expected-audits.json 是事先按这六个场景写定的预期，运行时只读，不能用刚得到的观察结果覆盖它们。第一份检查旧目录快照、旧 Schema 的独立校验、候选关闭顺序和实际调用结果；第二份检查每个服务恰好收到哪些列表与调用请求。多一次 SDK 隐式列表查询也会使轨迹不符。

保存为 expected.json：

{
  "in_place_integer": {
    "saved_schema_type": "string",
    "saved_snapshot_unchanged": true,
    "refresh": "duplicate_tool_name",
    "call": {
      "status": "client_schema_error"
    },
    "saved_schema_accepts_payload": true
  },
  "in_place_no_schema": {
    "saved_schema_type": "string",
    "saved_snapshot_unchanged": true,
    "refresh": "duplicate_tool_name",
    "call": {
      "status": "accepted",
      "content": {
        "value": 7
      }
    },
    "saved_schema_accepts_payload": false
  },
  "isolated_failed_candidate": {
    "before_call": {
      "status": "accepted",
      "content": {
        "value": "old"
      }
    },
    "refresh": "duplicate_tool_name",
    "candidate_closed_before_active_call": true,
    "saved_snapshot_unchanged": true,
    "after_call": {
      "status": "accepted",
      "content": {
        "value": "old"
      }
    }
  },
  "isolated_retains_validation": {
    "before_call": {
      "status": "client_schema_error"
    },
    "refresh": "duplicate_tool_name",
    "candidate_closed_before_active_call": true,
    "saved_snapshot_unchanged": true,
    "after_call": {
      "status": "client_schema_error"
    }
  },
  "accepted_candidate": {
    "names": [
      "alpha",
      "beta"
    ],
    "page_sizes": [
      1,
      1
    ],
    "call": {
      "status": "accepted",
      "content": {
        "value": 7
      }
    },
    "closed_after_context": true
  },
  "accepted_candidate_validates": {
    "names": [
      "alpha",
      "beta"
    ],
    "page_sizes": [
      1,
      1
    ],
    "call": {
      "status": "client_schema_error"
    },
    "closed_after_context": true
  }
}

{
  "in_place_integer-audit": {
    "requests": [
      {
        "method": "tools/list",
        "cursor": null
      },
      {
        "method": "tools/list",
        "cursor": null
      },
      {
        "method": "tools/list",
        "cursor": "next"
      },
      {
        "method": "tools/call",
        "name": "alpha"
      }
    ],
    "closed": true
  },
  "in_place_no_schema-audit": {
    "requests": [
      {
        "method": "tools/list",
        "cursor": null
      },
      {
        "method": "tools/list",
        "cursor": null
      },
      {
        "method": "tools/list",
        "cursor": "next"
      },
      {
        "method": "tools/call",
        "name": "alpha"
      }
    ],
    "closed": true
  },
  "active-audit": {
    "requests": [
      {
        "method": "tools/list",
        "cursor": null
      },
      {
        "method": "tools/call",
        "name": "alpha"
      },
      {
        "method": "tools/call",
        "name": "alpha"
      }
    ],
    "closed": true
  },
  "candidate-audit": {
    "requests": [
      {
        "method": "tools/list",
        "cursor": null
      },
      {
        "method": "tools/list",
        "cursor": "next"
      }
    ],
    "closed": true
  },
  "active_invalid-audit": {
    "requests": [
      {
        "method": "tools/list",
        "cursor": null
      },
      {
        "method": "tools/call",
        "name": "alpha"
      },
      {
        "method": "tools/call",
        "name": "alpha"
      }
    ],
    "closed": true
  },
  "candidate_invalid-audit": {
    "requests": [
      {
        "method": "tools/list",
        "cursor": null
      },
      {
        "method": "tools/list",
        "cursor": "next"
      }
    ],
    "closed": true
  },
  "candidate_success-audit": {
    "requests": [
      {
        "method": "tools/list",
        "cursor": null
      },
      {
        "method": "tools/list",
        "cursor": "next"
      },
      {
        "method": "tools/call",
        "name": "alpha"
      }
    ],
    "closed": true
  },
  "candidate_success_invalid-audit": {
    "requests": [
      {
        "method": "tools/list",
        "cursor": null
      },
      {
        "method": "tools/list",
        "cursor": "next"
      },
      {
        "method": "tools/call",
        "name": "alpha"
      }
    ],
    "closed": true
  }
}

run_cache_probe.py 每次创建全新的 runs/run-* 目录，旧审计文件不会参与本次判断。仅期望的 CatalogError 被归为 duplicate_tool_name；ExceptionGroup 中只要混入其他异常就继续抛出。客户端 Schema 异常检查稳定前缀和关键类型诊断，完整错误文本另外写入本轮目录，不绑定整段 jsonschema 格式。

保存为 run_cache_probe.py：

"""True stdio, fixed independent oracles, and checks that remain active under -O."""
from contextlib import asynccontextmanager
from importlib.metadata import version
import json
from pathlib import Path
import platform
import sys
import tempfile
import anyio
from jsonschema import validate, ValidationError
from mcp import StdioServerParameters
from client import enumerate_tools, CatalogError
from candidate_catalog import initialized_session, candidate_catalog

ROOT = Path(__file__).resolve().parent

def require(condition, message):
    if not condition:
        raise AssertionError(message)

def leaves(error):
    if isinstance(error, BaseExceptionGroup):
        return [leaf for child in error.exceptions for leaf in leaves(child)]
    return [error]

def is_duplicate_error(error):
    errors = leaves(error)
    return bool(errors) and all(isinstance(e, CatalogError) and
        str(e) == 'duplicate tool name; discard incomplete catalog' for e in errors)

def parameters(name, out):
    flags = ['-B'] + (['-O'] if sys.flags.optimize else [])
    return StdioServerParameters(command=sys.executable,
        args=[*flags, str(ROOT/'schema_server.py'), str(ROOT/'fixtures'/f'{name}.json'),
              str(out/(name+'-audit.json'))])

@asynccontextmanager
async def connect(name, out, *, candidate=False):
    factory = candidate_catalog if candidate else initialized_session
    with (out/(name+'-stderr.txt')).open('w') as errors:
        async with factory(parameters(name, out), errlog=errors) as result:
            yield result


def closed(name, out):
    audit = json.loads((out/(name+'-audit.json')).read_text())
    require(audit['closed'] is True, 'server finalizer did not finish: '+name)
    return True

def snapshot(entries):
    return [entry.model_dump(mode='json', exclude_none=True) for entry in entries]

async def reject_refresh(session):
    try:
        await enumerate_tools(session)
    except CatalogError as error:
        require(is_duplicate_error(error), 'wrong catalog rejection')
        return 'duplicate_tool_name'
    raise AssertionError('bad catalog accepted')

async def call_outcome(session, out, label, *, invalid_text=None):
    try:
        result = await session.call_tool('alpha', {})
    except RuntimeError as error:
        (out/(label+'-client-error.txt')).write_text(str(error)+'\n')
        require(str(error).startswith('Invalid structured content returned by tool alpha:'),
                'unexpected client error')
        require(invalid_text is not None and invalid_text in str(error), 'wrong schema failure')
        return {'status': 'client_schema_error'}
    require(not result.isError, 'fixture returned an unexpected tool error')
    return {'status': 'accepted', 'content': result.structuredContent}

async def in_place(name, out):
    async with connect(name, out) as session:
        saved, _ = await enumerate_tools(session)
        before = snapshot(saved)
        failure = await reject_refresh(session)
        after = await call_outcome(session, out, name,
            invalid_text="'old' is not of type 'integer'")
        saved_schema_accepts = True
        payload = json.loads((ROOT/'fixtures'/f'{name}.json').read_text())['call_content']
        try:
            validate(payload, saved[0].outputSchema)
        except ValidationError:
            saved_schema_accepts = False
        observed = {'saved_schema_type': saved[0].outputSchema['properties']['value']['type'],
            'saved_snapshot_unchanged': snapshot(saved) == before,
            'refresh': failure, 'call': after,
            'saved_schema_accepts_payload': saved_schema_accepts}
    closed(name, out)
    return observed

async def isolated(out, *, negative=False):
    name = 'active_invalid' if negative else 'active'
    candidate_name = 'candidate_invalid' if negative else 'candidate'
    async with connect(name, out) as active:
        saved, _ = await enumerate_tools(active)
        before = snapshot(saved)
        first = await call_outcome(active, out, name+'-before',
            invalid_text="7 is not of type 'string'")
        failure = None
        try:
            async with connect(candidate_name, out, candidate=True):
                raise AssertionError('candidate unexpectedly yielded')
        except Exception as error:
            if not is_duplicate_error(error):
                raise
            failure = 'duplicate_tool_name'
        candidate_closed = closed(candidate_name, out)
        last = await call_outcome(active, out, name+'-after',
            invalid_text="7 is not of type 'string'")
        observed = {'before_call': first, 'refresh': failure,
            'candidate_closed_before_active_call': candidate_closed,
            'saved_snapshot_unchanged': snapshot(saved) == before,
            'after_call': last}
    closed(name, out)
    return observed

async def accepted_candidate(out, *, negative=False):
    name = 'candidate_success_invalid' if negative else 'candidate_success'
    async with connect(name, out, candidate=True) as ready:
        observed = {'names': [entry.name for entry in ready.entries],
            'page_sizes': list(ready.page_sizes),
            'call': await call_outcome(ready.session, out, name,
                invalid_text="'old' is not of type 'integer'")}
    observed['closed_after_context'] = closed(name, out)
    return observed

async def run_all(out):
    observed = {'in_place_integer': await in_place('in_place_integer', out),
        'in_place_no_schema': await in_place('in_place_no_schema', out),
        'isolated_failed_candidate': await isolated(out),
        'isolated_retains_validation': await isolated(out, negative=True),
        'accepted_candidate': await accepted_candidate(out),
        'accepted_candidate_validates': await accepted_candidate(out, negative=True)}
    expected = json.loads((ROOT/'expected.json').read_text())
    require(observed == expected, 'case oracle mismatch:\n'+json.dumps(observed, indent=2))
    audits = {p.stem: json.loads(p.read_text()) for p in sorted(out.glob('*-audit.json'))}
    expected_audits = json.loads((ROOT/'expected-audits.json').read_text())
    require(audits == expected_audits, 'request audit mismatch:\n'+json.dumps(audits, indent=2))
    return {'environment': {'python': platform.python_version(),
        'mcp': version('mcp'), 'jsonschema': version('jsonschema')},
        'cases': observed, 'audits': audits}

async def main():
    (ROOT/'runs').mkdir(exist_ok=True)
    out = Path(tempfile.mkdtemp(prefix='run-', dir=ROOT/'runs'))
    result = await run_all(out)
    (out/'observed.json').write_text(json.dumps(result, indent=2)+'\n')
    print(json.dumps(result, indent=2))
    print('PASS: 6 fixed scenarios; 8 process finalizers; exact request traces')
    print('Evidence: '+str(out))

if __name__ == '__main__':
    anyio.run(main)

{
  "environment": {
    "python": "3.12.14",
    "mcp": "1.29.0",
    "jsonschema": "4.26.0"
  },
  "cases": {
    "in_place_integer": {
      "saved_schema_type": "string",
      "saved_snapshot_unchanged": true,
      "refresh": "duplicate_tool_name",
      "call": {
        "status": "client_schema_error"
      },
      "saved_schema_accepts_payload": true
    },
    "in_place_no_schema": {
      "saved_schema_type": "string",
      "saved_snapshot_unchanged": true,
      "refresh": "duplicate_tool_name",
      "call": {
        "status": "accepted",
        "content": {
          "value": 7
        }
      },
      "saved_schema_accepts_payload": false
    },
    "isolated_failed_candidate": {
      "before_call": {
        "status": "accepted",
        "content": {
          "value": "old"
        }
      },
      "refresh": "duplicate_tool_name",
      "candidate_closed_before_active_call": true,
      "saved_snapshot_unchanged": true,
      "after_call": {
        "status": "accepted",
        "content": {
          "value": "old"
        }
      }
    },
    "isolated_retains_validation": {
      "before_call": {
        "status": "client_schema_error"
      },
      "refresh": "duplicate_tool_name",
      "candidate_closed_before_active_call": true,
      "saved_snapshot_unchanged": true,
      "after_call": {
        "status": "client_schema_error"
      }
    },
    "accepted_candidate": {
      "names": [
        "alpha",
        "beta"
      ],
      "page_sizes": [
        1,
        1
      ],
      "call": {
        "status": "accepted",
        "content": {
          "value": 7
        }
      },
      "closed_after_context": true
    },
    "accepted_candidate_validates": {
      "names": [
        "alpha",
        "beta"
      ],
      "page_sizes": [
        1,
        1
      ],
      "call": {
        "status": "client_schema_error"
      },
      "closed_after_context": true
    }
  },
  "audits": {
    "active-audit": {
      "requests": [
        {
          "method": "tools/list",
          "cursor": null
        },
        {
          "method": "tools/call",
          "name": "alpha"
        },
        {
          "method": "tools/call",
          "name": "alpha"
        }
      ],
      "closed": true
    },
    "active_invalid-audit": {
      "requests": [
        {
          "method": "tools/list",
          "cursor": null
        },
        {
          "method": "tools/call",
          "name": "alpha"
        },
        {
          "method": "tools/call",
          "name": "alpha"
        }
      ],
      "closed": true
    },
    "candidate-audit": {
      "requests": [
        {
          "method": "tools/list",
          "cursor": null
        },
        {
          "method": "tools/list",
          "cursor": "next"
        }
      ],
      "closed": true
    },
    "candidate_invalid-audit": {
      "requests": [
        {
          "method": "tools/list",
          "cursor": null
        },
        {
          "method": "tools/list",
          "cursor": "next"
        }
      ],
      "closed": true
    },
    "candidate_success-audit": {
      "requests": [
        {
          "method": "tools/list",
          "cursor": null
        },
        {
          "method": "tools/list",
          "cursor": "next"
        },
        {
          "method": "tools/call",
          "name": "alpha"
        }
      ],
      "closed": true
    },
    "candidate_success_invalid-audit": {
      "requests": [
        {
          "method": "tools/list",
          "cursor": null
        },
        {
          "method": "tools/list",
          "cursor": "next"
        },
        {
          "method": "tools/call",
          "name": "alpha"
        }
      ],
      "closed": true
    },
    "in_place_integer-audit": {
      "requests": [
        {
          "method": "tools/list",
          "cursor": null
        },
        {
          "method": "tools/list",
          "cursor": null
        },
        {
          "method": "tools/list",
          "cursor": "next"
        },
        {
          "method": "tools/call",
          "name": "alpha"
        }
      ],
      "closed": true
    },
    "in_place_no_schema-audit": {
      "requests": [
        {
          "method": "tools/list",
          "cursor": null
        },
        {
          "method": "tools/list",
          "cursor": null
        },
        {
          "method": "tools/list",
          "cursor": "next"
        },
        {
          "method": "tools/call",
          "name": "alpha"
        }
      ],
      "closed": true
    }
  }
}

正常退出时，控制台先输出 environment、六项 cases 与八份 audits，随后是以下结尾；Evidence 后的路径每次变化。cases 必须逐字段等于上面的 expected.json，audits 必须逐字段等于 expected-audits.json。

PASS: 6 fixed scenarios; 8 process finalizers; exact request traces
Evidence: <本次新建的 runs/run-* 目录>

in_place_integer 的完整错误文本位于 in_place_integer-client-error.txt，关键诊断为 old 不是 integer；活动会话的数字反向控制则为 7 不是 string。closed=true 说明夹具 finalizer 已运行，检查它之前也已退出 SDK 的 stdio 上下文。固定版本的 stdio 实现包含关闭输入、等待及终止后备流程；本实验没有读取子进程 returncode，因此不把这一证据写成“八个进程均以 0 正常退出”。

新增测试有四个 unittest 方法：完整核对六个固定场景；重名候选在 yield 前拒绝并关闭；调用方在成功候选上下文内抛错时保留异常并清理；只允许一页时因预算不足拒绝、关闭，而且没有工具调用。每个测试都使用独立临时目录和真实子进程。

保存为 test_session_cache.py：

import json
from pathlib import Path
import tempfile
import unittest
import anyio
from client import CatalogError
from candidate_catalog import candidate_catalog
from run_cache_probe import run_all, parameters, connect, closed, leaves, is_duplicate_error

class SessionCacheTests(unittest.TestCase):
    def test_fixed_scenarios(self):
        with tempfile.TemporaryDirectory() as directory:
            anyio.run(run_all, Path(directory))

    def test_rejected_candidate_never_yields(self):
        async def check(out):
            yielded = False
            error = None
            try:
                async with connect('candidate', out, candidate=True):
                    yielded = True
            except Exception as caught:
                error = caught
            self.assertFalse(yielded, 'candidate unexpectedly yielded')
            self.assertIsNotNone(error)
            self.assertTrue(is_duplicate_error(error))
            self.assertTrue(closed('candidate', out))
            audit = json.loads((out/'candidate-audit.json').read_text())
            self.assertEqual(audit['requests'], [
                {'method':'tools/list', 'cursor':None},
                {'method':'tools/list', 'cursor':'next'}])
        with tempfile.TemporaryDirectory() as directory:
            anyio.run(check, Path(directory))

    def test_caller_error_propagates_after_cleanup(self):
        async def check(out):
            error = None
            try:
                async with connect('candidate_success', out, candidate=True) as ready:
                    self.assertEqual([entry.name for entry in ready.entries], ['alpha', 'beta'])
                    raise RuntimeError('synthetic caller failure')
            except Exception as caught:
                error = caught
            self.assertIsNotNone(error, 'caller failure was swallowed')
            self.assertEqual([(type(e), str(e)) for e in leaves(error)],
                             [(RuntimeError, 'synthetic caller failure')])
            self.assertTrue(closed('candidate_success', out))
            audit = json.loads((out/'candidate_success-audit.json').read_text())
            self.assertEqual([r['method'] for r in audit['requests']], ['tools/list', 'tools/list'])
        with tempfile.TemporaryDirectory() as directory:
            anyio.run(check, Path(directory))

    def test_page_budget_rejects_before_yield_and_closes(self):
        async def check(out):
            yielded = False
            error = None
            with (out/'stderr.txt').open('w') as errors:
                try:
                    async with candidate_catalog(parameters('candidate_success', out),
                                                 max_pages=1, errlog=errors):
                        yielded = True
                except Exception as caught:
                    error = caught
            self.assertFalse(yielded)
            self.assertIsNotNone(error)
            self.assertEqual([(type(e), str(e)) for e in leaves(error)],
                [(CatalogError, 'page limit reached; discard incomplete catalog')])
            self.assertTrue(closed('candidate_success', out))
            audit = json.loads((out/'candidate_success-audit.json').read_text())
            self.assertEqual(audit['requests'], [{'method':'tools/list', 'cursor':None}])
        with tempfile.TemporaryDirectory() as directory:
            anyio.run(check, Path(directory))

if __name__ == '__main__':
    unittest.main(verbosity=2)

变异脚本在临时副本中每次只改一处：取消重名检查、只读首页就交付候选、吞掉目录拒绝后交付空候选、吞掉调用方异常。四个错误版本必须分别触发指定测试与指定诊断才算被检出；编译失败、导入失败或其他意外异常不能冒充成功检出。

保存为 check_mutants.py：

"""Execute four wrong versions in fresh copies; never edit the real source tree."""
import json
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile

ROOT = Path(__file__).resolve().parent
ENUMERATE = '        entries, sizes = await enumerate_tools(session, max_pages=max_pages, max_tools=max_tools)'
YIELD = '        yield ReadyCatalog(session, tuple(entries), tuple(sizes))'
MUTANTS = [
    ('ignore_duplicate', 'client.py', '            if entry.name in names:',
     '            if False and entry.name in names:', 'test_fixed_scenarios', 'bad catalog accepted'),
    ('publish_first_page', 'candidate_catalog.py', ENUMERATE,
     '        page = await session.list_tools()\n        entries, sizes = page.tools, [len(page.tools)]',
     'test_rejected_candidate_never_yields', 'candidate unexpectedly yielded'),
    ('swallow_catalog_rejection', 'candidate_catalog.py', ENUMERATE,
     '        try:\n    '+ENUMERATE+'\n        except RuntimeError:\n            entries, sizes = [], []',
     'test_rejected_candidate_never_yields', 'candidate unexpectedly yielded'),
    ('swallow_caller_failure', 'candidate_catalog.py', YIELD,
     '        try:\n    '+YIELD+'\n        except RuntimeError:\n            pass',
     'test_caller_error_propagates_after_cleanup', 'caller failure was swallowed'),
]

def main():
    results = []
    for name, file, old, new, method, marker in MUTANTS:
        with tempfile.TemporaryDirectory(prefix='catalog-mutant-') as directory:
            target = Path(directory)/'source'
            target.mkdir()
            files = ['client.py', 'candidate_catalog.py', 'run_cache_probe.py',
                     'schema_server.py', 'test_session_cache.py',
                     'expected.json', 'expected-audits.json']
            for filename in files:
                shutil.copyfile(ROOT/filename, target/filename)
            shutil.copytree(ROOT/'fixtures', target/'fixtures')
            path = target/file
            original = path.read_text()
            if original.count(old) != 1:
                raise RuntimeError('mutation anchor not unique: '+name)
            mutated = original.replace(old, new, 1)
            compile(mutated, file, 'exec')
            path.write_text(mutated)
            flags = ['-B'] + (['-O'] if sys.flags.optimize else [])
            command = [sys.executable, *flags, '-m', 'unittest',
                       'test_session_cache.SessionCacheTests.'+method]
            completed = subprocess.run(command, cwd=target, text=True, capture_output=True, timeout=60)
            diagnostic = completed.stdout+completed.stderr
            # A syntax/import/unrelated failure cannot count as killing a mutant.
            if completed.returncode == 0 or marker not in diagnostic:
                raise RuntimeError('mutant survived or failed unexpectedly: '+name+'\n'+diagnostic)
            results.append({'mutant':name, 'detected_by':method, 'required_failure':marker,
                            'returncode':completed.returncode})
    print(json.dumps(results, indent=2))
    print('PASS: 4 mutants rejected by their designated checks')

if __name__ == '__main__':
    main()

2026-10-13 的记录中，source 文件、从文章代码块重新提取的文件、当时的公开 ZIP 解压后的文件都分别运行了原 17 项测试、新增 4 项测试、六场景探针和四项变异检查，并覆盖普通模式与 -O。新增探针使用显式 require，unittest 使用断言方法，-O 不会移除这些核对；新增夹具子进程也传入 -O。原文启动子进程的代码完全保留，其子进程不会因父进程的 -O 自动继承优化标志。

这些通过记录覆盖本文明确写出的输入与生命周期路径。它们没有验证任意卡死子进程的强制清理、真实宿主 UI、生产并发切换、远端权限变化或所有 SDK 版本。图 4 已按原始比例渲染并检查；HTML 做结构与可复制代码验收，未启动浏览器做像素级验收。

• 候选枚举因重名、游标循环或预算耗尽失败：关闭该候选，保留失败原因；本轮没有完整新目录。修正服务端目录或重新确认预算后，再显式建立新候选，从首页开始，不复用失败游标。
• 已经在活动会话上做过失败刷新：不要仅凭旧列表未变就继续相信旧校验行为。停止把这段会话当作旧合同的依据，按宿主生命周期关闭并重新建立可信会话，再重新完整枚举；不通过改写 SDK 私有字段来假装修复。
• 新候选完整通过：本例只允许在候选上下文内使用它。产品级接管还需要处理在途调用、所有者任务、退出顺序和旧会话退役，应单独设计并测试。
• 服务端拒绝调用、权限撤销或业务合同变化：按当前授权和服务端错误处理。旧目录和旧会话都不能替代持续授权，也不保证工具仍可用。

如果环境不能同时建立两段连接，就不能直接采用这个隔离方式，需要另行定义可接受的刷新停顿或 SDK 支持的生命周期。新连接也可能看到不同主体、权限或后端视图，接入时必须核对。所用版本的分页规范描述了不透明游标与继续读取流程；本节没有从中推导跨会话目录快照或事务保证。

这里处理的是一个固定 SDK 内部页级状态与应用整轮验收不一致的边界。它与“旧刷新晚完成，覆盖新一代目录”的并发代次检查分别需要验收：即使拒绝了过期刷新提交，只要它在共享活动会话里读过列表页，也不能据此推断会话缓存保持原样。当前结论只适用于本文验证的 Python MCP SDK 1.29.0；升级版本、其他语言实现和生产会话管理器都需要自己的证据。