feat(client): add list_all_* helpers that drain pagination

Closes #2556.

The single-page list_tools / list_resources / list_resource_templates /
list_prompts methods return the server's first page plus a next_cursor;
callers that don't loop on the cursor silently see only page zero.
Four list_all_* helpers on ClientSession drain the cursor until the
server stops sending one and return the flattened list:

  list_all_tools()             -> list[Tool]
  list_all_resources()         -> list[Resource]
  list_all_resource_templates() -> list[ResourceTemplate]
  list_all_prompts()           -> list[Prompt]

Each helper reuses the existing public list_* method, so per-page
post-processing (e.g. the 2026 x-mcp-header filter in _absorb_tool_listing)
keeps running across the drain. Single-page and empty-list cases work
without a loop.

Tests cover: multi-page drain, single-page short-circuit, empty result,
helpers-dont-share-cursor-state, terminal-empty-page-stops-loop, and
cursor-roundtrip (the server's next_cursor comes back unchanged as
params.cursor on the next request). 9 new tests pass alongside the
existing 112 in tests/client.

Author: sanjibani

src/mcp/client/session.py

@@ -1183,6 +1183,64 @@ def _absorb_tool_listing(self, result: types.ListToolsResult, *, complete: bool)
 
         return result
 
+    # ---------- Pagination-draining helpers ----------
+    #
+    # The single-page `list_*` methods return the server's first page plus a
+    # `next_cursor`; callers who don't loop on the cursor silently get only
+    # page zero. The four `list_all_*` helpers drain all pages and return a
+    # flat list, so the common "give me everything" use case stops hiding
+    # pagination behind what looks like a single call. The single-page
+    # methods are unchanged.
+
+    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.
+        """
+        tools: list[types.Tool] = []
+        cursor: str | None = None
+        while True:
+            result = await self.list_tools(params=types.PaginatedRequestParams(cursor=cursor))
+            tools.extend(result.tools)
+            cursor = result.next_cursor
+            if cursor is None:
+                return tools
+
+    async def list_all_resources(self) -> list[types.Resource]:
+        """Drain pagination across `resources/list` and return the full resource list."""
+        items: list[types.Resource] = []
+        cursor: str | None = None
+        while True:
+            result = await self.list_resources(params=types.PaginatedRequestParams(cursor=cursor))
+            items.extend(result.resources)
+            cursor = result.next_cursor
+            if cursor is None:
+                return items
+
+    async def list_all_resource_templates(self) -> list[types.ResourceTemplate]:
+        """Drain pagination across `resources/templates/list` and return the full template list."""
+        templates: list[types.ResourceTemplate] = []
+        cursor: str | None = None
+        while True:
+            result = await self.list_resource_templates(params=types.PaginatedRequestParams(cursor=cursor))
+            templates.extend(result.resource_templates)
+            cursor = result.next_cursor
+            if cursor is None:
+                return templates
+
+    async def list_all_prompts(self) -> list[types.Prompt]:
+        """Drain pagination across `prompts/list` and return the full prompt list."""
+        items: list[types.Prompt] = []
+        cursor: str | None = None
+        while True:
+            result = await self.list_prompts(params=types.PaginatedRequestParams(cursor=cursor))
+            items.extend(result.prompts)
+            cursor = result.next_cursor
+            if cursor is None:
+                return items
+
     @deprecated("The roots capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
     async def send_roots_list_changed(self) -> None:
         """Send a roots/list_changed notification."""

tests/client/test_client.py

@@ -946,3 +946,215 @@ async def elicitation_callback(
     assert result == snapshot(
         ReadResourceResult(contents=[TextResourceContents(uri="memory://gated", text="unlocked")])
     )
+
+
+# ---------- list_all_* pagination-draining helpers ----------
+#
+# Each helper drives the existing single-page list_* method through a
+# cursor loop and returns the flattened list. The fixtures return one
+# page per call in order; the helper under test must dispatch until the
+# server stops sending a next_cursor.
+
+
+def _make_tools(names: list[str]) -> list[types.Tool]:
+    return [types.Tool(name=n, input_schema={"type": "object"}) for n in names]
+
+
+def _make_resources(uris: list[str]) -> list[types.Resource]:
+    return [types.Resource(uri=u, name=u.rsplit("/", 1)[-1]) for u in uris]
+
+
+def _make_resource_templates(uris: list[str]) -> list[types.ResourceTemplate]:
+    return [types.ResourceTemplate(uri_template=u, name=u.rsplit("/", 1)[-1]) for u in uris]
+
+
+def _make_prompts(names: list[str]) -> list[types.Prompt]:
+    return [types.Prompt(name=n) for n in names]
+
+
+async def test_list_all_tools_drains_pagination() -> None:
+    """`list_all_tools` keeps calling `tools/list` with the previous next_cursor until the
+    server returns a page with `next_cursor=None`, and returns the flattened list."""
+
+    pages: list[types.ListToolsResult] = [
+        types.ListToolsResult(tools=_make_tools(["a", "b"]), next_cursor="p2"),
+        types.ListToolsResult(tools=_make_tools(["c"]), next_cursor="p3"),
+        types.ListToolsResult(tools=_make_tools(["d", "e"])),
+    ]
+
+    async def on_list_tools(
+        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
+    ) -> types.ListToolsResult:
+        cursor = (params.cursor if params else None) or "p1"
+        idx = {"p1": 0, "p2": 1, "p3": 2}[cursor]
+        return pages[idx]
+
+    server = Server("test", on_list_tools=on_list_tools)
+
+    with anyio.fail_after(5):
+        async with Client(server) as client:
+            tools = await client.session.list_all_tools()
+    assert [t.name for t in tools] == ["a", "b", "c", "d", "e"]
+
+
+async def test_list_all_tools_single_page_works() -> None:
+    """`list_all_tools` returns the single page's items when no cursor is involved."""
+
+    async def on_list_tools(
+        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
+    ) -> types.ListToolsResult:
+        return types.ListToolsResult(tools=_make_tools(["only"]))
+
+    server = Server("test", on_list_tools=on_list_tools)
+
+    with anyio.fail_after(5):
+        async with Client(server) as client:
+            tools = await client.session.list_all_tools()
+    assert [t.name for t in tools] == ["only"]
+
+
+async def test_list_all_tools_empty_returns_empty_list() -> None:
+    async def on_list_tools(
+        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
+    ) -> types.ListToolsResult:
+        return types.ListToolsResult(tools=[])
+
+    server = Server("test", on_list_tools=on_list_tools)
+
+    with anyio.fail_after(5):
+        async with Client(server) as client:
+            tools = await client.session.list_all_tools()
+    assert tools == []
+
+
+async def test_list_all_resources_drains_pagination() -> None:
+    pages: list[types.ListResourcesResult] = [
+        types.ListResourcesResult(resources=_make_resources(["memory://r1", "memory://r2"]), next_cursor="p2"),
+        types.ListResourcesResult(resources=_make_resources(["memory://r3"])),
+    ]
+
+    async def on_list_resources(
+        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
+    ) -> types.ListResourcesResult:
+        cursor = (params.cursor if params else None) or "p1"
+        return pages[0] if cursor == "p1" else pages[1]
+
+    server = Server("test", on_list_resources=on_list_resources)
+
+    with anyio.fail_after(5):
+        async with Client(server) as client:
+            resources = await client.session.list_all_resources()
+    assert [r.uri for r in resources] == ["memory://r1", "memory://r2", "memory://r3"]
+
+
+async def test_list_all_resource_templates_drains_pagination() -> None:
+    pages: list[types.ListResourceTemplatesResult] = [
+        types.ListResourceTemplatesResult(
+            resource_templates=_make_resource_templates(["tmpl://{a}"]),
+            next_cursor="p2",
+        ),
+        types.ListResourceTemplatesResult(resource_templates=_make_resource_templates(["tmpl://{b}"])),
+    ]
+
+    async def on_list_resource_templates(
+        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
+    ) -> types.ListResourceTemplatesResult:
+        cursor = (params.cursor if params else None) or "p1"
+        return pages[0] if cursor == "p1" else pages[1]
+
+    server = Server("test", on_list_resource_templates=on_list_resource_templates)
+
+    with anyio.fail_after(5):
+        async with Client(server) as client:
+            templates = await client.session.list_all_resource_templates()
+    assert [t.uri_template for t in templates] == ["tmpl://{a}", "tmpl://{b}"]
+
+
+async def test_list_all_prompts_drains_pagination() -> None:
+    pages: list[types.ListPromptsResult] = [
+        types.ListPromptsResult(prompts=_make_prompts(["p1"]), next_cursor="p2"),
+        types.ListPromptsResult(prompts=_make_prompts(["p2", "p3"])),
+    ]
+
+    async def on_list_prompts(
+        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
+    ) -> types.ListPromptsResult:
+        cursor = (params.cursor if params else None) or "p1"
+        return pages[0] if cursor == "p1" else pages[1]
+
+    server = Server("test", on_list_prompts=on_list_prompts)
+
+    with anyio.fail_after(5):
+        async with Client(server) as client:
+            prompts = await client.session.list_all_prompts()
+    assert [p.name for p in prompts] == ["p1", "p2", "p3"]
+
+
+async def test_list_all_helpers_do_not_share_cursor_state() -> None:
+    """Each `list_all_*` helper starts at cursor=None; calling one does not affect another's
+    pagination state (the helpers read no shared cursor, only the server's per-page next_cursor)."""
+
+    async def on_list_tools(
+        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
+    ) -> types.ListToolsResult:
+        return types.ListToolsResult(tools=_make_tools(["t1"]))
+
+    async def on_list_prompts(
+        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
+    ) -> types.ListPromptsResult:
+        return types.ListPromptsResult(prompts=_make_prompts(["only-prompt"]))
+
+    server = Server("test", on_list_tools=on_list_tools, on_list_prompts=on_list_prompts)
+
+    with anyio.fail_after(5):
+        async with Client(server) as client:
+            tools = await client.session.list_all_tools()
+            prompts = await client.session.list_all_prompts()
+    assert [t.name for t in tools] == ["t1"]
+    assert [p.name for p in prompts] == ["only-prompt"]
+
+
+async def test_list_all_tools_stops_when_server_sends_terminal_empty_page() -> None:
+    """A terminal page with cursor=None (even if zero items) ends the drain loop."""
+
+    pages: list[types.ListToolsResult] = [
+        types.ListToolsResult(tools=_make_tools(["first"]), next_cursor="p2"),
+        types.ListToolsResult(tools=[], next_cursor=None),
+        types.ListToolsResult(tools=_make_tools(["never-returned"])),
+    ]
+
+    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:
+            tools = await client.session.list_all_tools()
+    assert [t.name for t in tools] == ["first"]
+
+
+async def test_list_all_helpers_pass_cursor_through_paginated_request_params() -> None:
+    """The helpers must hand the server's `next_cursor` back as `params.cursor` on the next call
+    (not as a positional arg or otherwise mangled). Server-side capture proves it."""
+
+    captured: list[str | None] = []
+
+    async def on_list_tools(
+        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
+    ) -> types.ListToolsResult:
+        captured.append(params.cursor if params else None)
+        if len(captured) == 1:
+            return types.ListToolsResult(tools=_make_tools(["page1-item"]), next_cursor="next-cursor")
+        return types.ListToolsResult(tools=_make_tools(["page2-item"]), next_cursor=None)
+
+    server = Server("test", on_list_tools=on_list_tools)
+
+    with anyio.fail_after(5):
+        async with Client(server) as client:
+            tools = await client.session.list_all_tools()
+    assert captured == [None, "next-cursor"]
+    assert [t.name for t in tools] == ["page1-item", "page2-item"]