fix(client): list_all_tools prunes per-tool cache state on terminal page

Cubic P2 on PR #3083: a multi-page list_all_tools() drain called
list_tools() with a non-None cursor on every page, so
_absorb_tool_listing(complete=True) was never reached and the
per-tool cache (_x_mcp_header_maps, _tool_output_schemas) was
left in an additive-only state. Tools that had been present in
a prior list_tools() call but were absent from the drained
listing kept their cache entries indefinitely.

Run _absorb_tool_listing once more with complete=True on the
merged result after the loop ends. The merged listing is the
drained universe, so complete=True prunes cache entries for
tools no longer present, mirroring the single-page complete
listing semantic.

Test: seed the cache with stale-tool via a single-page complete
listing, then drain survivor across two pages, assert the
cache reflects only the drained universe. A second test asserts
the inverse: if every drained tool is in the union, all cache
entries stay.

Author: sanjibani

src/mcp/client/session.py

@@ -1195,9 +1195,13 @@ def _absorb_tool_listing(self, result: types.ListToolsResult, *, complete: bool)
     async def list_all_tools(self) -> list[types.Tool]:
         """Drain pagination across `tools/list` and return the full tool list.
 
-        Each page flows through the 2026 x-mcp-header per-page filter via
-        `list_tools` -> `_absorb_tool_listing`, so the returned list is exactly
-        what the cached state would expose.
+        Each intermediate page flows through `list_tools` -> `_absorb_tool_listing`
+        with `complete=False`, which populates per-tool cache state (`_x_mcp_header_maps`,
+        `_tool_output_schemas`) as a pure addition. After the loop ends, the helper
+        runs `_absorb_tool_listing` once more with `complete=True` on the merged
+        result so state for tools that disappeared across pages (or were filtered
+        by the per-page x-mcp-header MUST) is pruned to match the drained universe.
+        The returned list reflects the post-filter view that the cache exposes.
         """
         tools: list[types.Tool] = []
         cursor: str | None = None
@@ -1206,7 +1210,12 @@ async def list_all_tools(self) -> list[types.Tool]:
             tools.extend(result.tools)
             cursor = result.next_cursor
             if cursor is None:
-                return tools
+                break
+        # Final absorption pass with `complete=True` so the per-tool cache matches
+        # the drained listing (prunes entries for tools no longer present).
+        merged = types.ListToolsResult(tools=tools, next_cursor=None)
+        absorbed = self._absorb_tool_listing(merged, complete=True)
+        return absorbed.tools
 
     async def list_all_resources(self) -> list[types.Resource]:
         """Drain pagination across `resources/list` and return the full resource list."""

tests/client/test_client.py

@@ -1158,3 +1158,74 @@ async def on_list_tools(
             tools = await client.session.list_all_tools()
     assert captured == [None, "next-cursor"]
     assert [t.name for t in tools] == ["page1-item", "page2-item"]
+
+
+async def test_list_all_tools_prunes_per_tool_cache_state_for_tools_dropped_across_pages() -> None:
+    """Cubic P2 fix: after a multi-page drain, per-tool cache state (`_x_mcp_header_maps`,
+    `_tool_output_schemas`) must reflect the merged listing (the drained universe). Tools
+    that appeared in a previous `list_tools()` call but are not in the drained listing
+    should not leave stale cache entries (the merged-listing absorption pass with
+    `complete=True` prunes them just like a complete single-page listing would)."""
+
+    survivor = types.Tool(
+        name="survivor",
+        input_schema={"type": "object", "properties": {"region": {"type": "string", "x-mcp-header": "Region"}}},
+    )
+    stale_tool = types.Tool(name="stale-tool", input_schema={"type": "object"})
+    seen_pages: list[str | None] = []
+
+    async def on_list_tools(
+        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
+    ) -> types.ListToolsResult:
+        cursor = (params.cursor if params else None) or "p1"
+        seen_pages.append(cursor)
+        if seen_pages == ["p1"]:
+            # First call: complete=True listing that seeds the stale tool.
+            return types.ListToolsResult(tools=[stale_tool])
+        if cursor == "p1":
+            return types.ListToolsResult(tools=[survivor], next_cursor="p2")
+        return types.ListToolsResult(tools=[survivor])
+
+    server = Server("test", on_list_tools=on_list_tools)
+
+    with anyio.fail_after(5):
+        async with Client(server) as client:
+            # Seed: single-page complete listing with stale-tool.
+            seeded = await client.session.list_tools()
+            assert [t.name for t in seeded.tools] == ["stale-tool"]
+            assert set(client.session._x_mcp_header_maps) == {"stale-tool"}
+
+            # Drain: multi-page listing of survivor only.
+            drained = await client.session.list_all_tools()
+            assert [t.name for t in drained] == ["survivor", "survivor"]
+            # After drain, cache reflects only the drained universe.
+            assert set(client.session._x_mcp_header_maps) == {"survivor"}
+            assert set(client.session._tool_output_schemas) == {"survivor"}
+
+
+async def test_list_all_tools_does_not_prune_when_server_state_remains_intact() -> None:
+    """Sanity: if the drained listing is the only listing seen so far, every tool in
+    the union stays in the cache."""
+
+    a = types.Tool(name="a", input_schema={"type": "object"})
+    b = types.Tool(name="b", input_schema={"type": "object"})
+    c = types.Tool(name="c", input_schema={"type": "object"})
+
+    pages: list[types.ListToolsResult] = [
+        types.ListToolsResult(tools=[a, b], next_cursor="p2"),
+        types.ListToolsResult(tools=[c]),
+    ]
+
+    async def on_list_tools(
+        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
+    ) -> types.ListToolsResult:
+        cursor = (params.cursor if params else None) or "p1"
+        return pages[0] if cursor == "p1" else pages[1]
+
+    server = Server("test", on_list_tools=on_list_tools)
+
+    with anyio.fail_after(5):
+        async with Client(server) as client:
+            await client.session.list_all_tools()
+            assert set(client.session._x_mcp_header_maps) == {"a", "b", "c"}
+            assert set(client.session._tool_output_schemas) == {"a", "b", "c"}