Add list_all_* helpers that drain pagination on the client

Currently Client.list_tools / list_prompts / list_resources /
list_resource_templates return a single page and the caller has to loop
on next_cursor manually. Add list_all_tools / list_all_prompts /
list_all_resources / list_all_resource_templates that walk next_cursor
until exhausted, plus iter_all_* async iterators for streaming
consumers. The single-page methods get a docstring update pointing at
the new drains. ClientSessionGroup switches its tool/prompt/resource
aggregation to the drain helper so its consumers always see the full
collection across multi-page servers.

Implements the helper maxisbey endorsed in #2556.

Rebased onto the v2 rework: types import from mcp_types, the stream-spy
tests run in legacy mode, and the test Tool carries a valid input_schema.

Author: adityasingh2400

src/mcp/client/client.py

@@ -5,7 +5,7 @@
 import hashlib
 import logging
 import uuid
-from collections.abc import Awaitable, Callable, Mapping, Sequence
+from collections.abc import AsyncIterator, Awaitable, Callable, Mapping, Sequence
 from contextlib import AbstractAsyncContextManager, AsyncExitStack
 from dataclasses import KW_ONLY, dataclass, field
 from typing import Any, Literal, TypeVar, cast
@@ -32,12 +32,16 @@
     ListToolsResult,
     LoggingLevel,
     PaginatedRequestParams,
+    Prompt,
     PromptReference,
     ReadResourceResult,
     RequestParamsMeta,
+    Resource,
+    ResourceTemplate,
     ResourceTemplateReference,
     Result,
     ServerCapabilities,
+    Tool,
 )
 from mcp_types.version import HANDSHAKE_PROTOCOL_VERSIONS, MODERN_PROTOCOL_VERSIONS
 from typing_extensions import deprecated
@@ -580,7 +584,11 @@ async def list_resources(
         meta: RequestParamsMeta | None = None,
         cache_mode: CacheMode = "use",
     ) -> ListResourcesResult:
-        """List available resources from the server."""
+        """List a single page of available resources from the server.
+
+        Returns one page only. The result may include a `next_cursor` if more
+        pages are available. Use `list_all_resources` to drain every page.
+        """
         return await self._cached_fetch(
             "resources/list",
             cursor=cursor,
@@ -596,7 +604,12 @@ async def list_resource_templates(
         meta: RequestParamsMeta | None = None,
         cache_mode: CacheMode = "use",
     ) -> ListResourceTemplatesResult:
-        """List available resource templates from the server."""
+        """List a single page of available resource templates from the server.
+
+        Returns one page only. The result may include a `next_cursor` if more
+        pages are available. Use `list_all_resource_templates` to drain every
+        page.
+        """
         return await self._cached_fetch(
             "resources/templates/list",
             cursor=cursor,
@@ -814,7 +827,11 @@ async def list_prompts(
         meta: RequestParamsMeta | None = None,
         cache_mode: CacheMode = "use",
     ) -> ListPromptsResult:
-        """List available prompts from the server."""
+        """List a single page of available prompts from the server.
+
+        Returns one page only. The result may include a `next_cursor` if more
+        pages are available. Use `list_all_prompts` to drain every page.
+        """
         return await self._cached_fetch(
             "prompts/list",
             cursor=cursor,
@@ -912,7 +929,11 @@ async def list_tools(
         meta: RequestParamsMeta | None = None,
         cache_mode: CacheMode = "use",
     ) -> ListToolsResult:
-        """List available tools from the server."""
+        """List a single page of available tools from the server.
+
+        Returns one page only. The result may include a `next_cursor` if more
+        pages are available. Use `list_all_tools` to drain every page.
+        """
         return await self._cached_fetch(
             "tools/list",
             cursor=cursor,
@@ -927,6 +948,77 @@ async def list_tools(
             ),
         )
 
+    async def iter_all_tools(self, *, meta: RequestParamsMeta | None = None) -> AsyncIterator[Tool]:
+        """Yield every tool from the server, paging through `next_cursor`.
+
+        Useful for streaming consumers that want to process tools without
+        materializing the full list in memory.
+        """
+        cursor: str | None = None
+        while True:
+            result = await self.list_tools(cursor=cursor, meta=meta)
+            for tool in result.tools:
+                yield tool
+            if result.next_cursor is None:
+                return
+            cursor = result.next_cursor
+
+    async def list_all_tools(self, *, meta: RequestParamsMeta | None = None) -> list[Tool]:
+        """List every tool from the server, draining `next_cursor` across pages.
+
+        Unlike `list_tools`, which returns one page, this walks pagination
+        until the server reports no further pages and returns the combined
+        list.
+        """
+        return [tool async for tool in self.iter_all_tools(meta=meta)]
+
+    async def iter_all_prompts(self, *, meta: RequestParamsMeta | None = None) -> AsyncIterator[Prompt]:
+        """Yield every prompt from the server, paging through `next_cursor`."""
+        cursor: str | None = None
+        while True:
+            result = await self.list_prompts(cursor=cursor, meta=meta)
+            for prompt in result.prompts:
+                yield prompt
+            if result.next_cursor is None:
+                return
+            cursor = result.next_cursor
+
+    async def list_all_prompts(self, *, meta: RequestParamsMeta | None = None) -> list[Prompt]:
+        """List every prompt from the server, draining `next_cursor` across pages."""
+        return [prompt async for prompt in self.iter_all_prompts(meta=meta)]
+
+    async def iter_all_resources(self, *, meta: RequestParamsMeta | None = None) -> AsyncIterator[Resource]:
+        """Yield every resource from the server, paging through `next_cursor`."""
+        cursor: str | None = None
+        while True:
+            result = await self.list_resources(cursor=cursor, meta=meta)
+            for resource in result.resources:
+                yield resource
+            if result.next_cursor is None:
+                return
+            cursor = result.next_cursor
+
+    async def list_all_resources(self, *, meta: RequestParamsMeta | None = None) -> list[Resource]:
+        """List every resource from the server, draining `next_cursor` across pages."""
+        return [resource async for resource in self.iter_all_resources(meta=meta)]
+
+    async def iter_all_resource_templates(
+        self, *, meta: RequestParamsMeta | None = None
+    ) -> AsyncIterator[ResourceTemplate]:
+        """Yield every resource template from the server, paging through `next_cursor`."""
+        cursor: str | None = None
+        while True:
+            result = await self.list_resource_templates(cursor=cursor, meta=meta)
+            for template in result.resource_templates:
+                yield template
+            if result.next_cursor is None:
+                return
+            cursor = result.next_cursor
+
+    async def list_all_resource_templates(self, *, meta: RequestParamsMeta | None = None) -> list[ResourceTemplate]:
+        """List every resource template from the server, draining `next_cursor` across pages."""
+        return [template async for template in self.iter_all_resource_templates(meta=meta)]
+
     @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 notification that the roots list has changed."""

src/mcp/client/session_group.py

@@ -8,7 +8,7 @@
 
 import contextlib
 import logging
-from collections.abc import Callable
+from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
 from types import TracebackType
 from typing import Any, Literal, TypeAlias, overload
@@ -67,6 +67,28 @@ class StreamableHttpParameters(BaseModel):
 ServerParameters: TypeAlias = StdioServerParameters | SseServerParameters | StreamableHttpParameters
 
 
+async def _drain_paginated(
+    fetch_page: Callable[..., Awaitable[Any]],
+    attribute: str,
+) -> list[Any]:
+    """Drain a paginated `session.list_*` call across `next_cursor` pages.
+
+    `fetch_page` is one of the ClientSession `list_*` methods that takes a
+    `params=PaginatedRequestParams(...)` keyword. `attribute` is the name of
+    the list attribute on the result (e.g. `"tools"`, `"prompts"`).
+    """
+    items: list[Any] = []
+    cursor: str | None = None
+    while True:
+        params = types.PaginatedRequestParams(cursor=cursor) if cursor is not None else None
+        result = await fetch_page(params=params)
+        items.extend(getattr(result, attribute))
+        next_cursor = getattr(result, "next_cursor", None)
+        if next_cursor is None:
+            return items
+        cursor = next_cursor
+
+
 # Use dataclass instead of Pydantic BaseModel
 # because Pydantic BaseModel cannot handle Protocol fields.
 @dataclass
@@ -383,9 +405,11 @@ async def _aggregate_components(self, server_info: types.Implementation, session
         tools_temp: dict[str, types.Tool] = {}
         tool_to_session_temp: dict[str, mcp.ClientSession] = {}
 
-        # Query the server for its prompts and aggregate to list.
+        # Query the server for its prompts and aggregate to list. Drain
+        # pagination so we don't drop later pages on servers that split
+        # results across multiple `next_cursor` responses.
         try:
-            prompts = (await session.list_prompts()).prompts
+            prompts = await _drain_paginated(session.list_prompts, "prompts")
             for prompt in prompts:
                 name = self._component_name(prompt.name, server_info)
                 prompts_temp[name] = prompt
@@ -395,7 +419,7 @@ async def _aggregate_components(self, server_info: types.Implementation, session
 
         # Query the server for its resources and aggregate to list.
         try:
-            resources = (await session.list_resources()).resources
+            resources = await _drain_paginated(session.list_resources, "resources")
             for resource in resources:
                 name = self._component_name(resource.name, server_info)
                 resources_temp[name] = resource
@@ -405,7 +429,7 @@ async def _aggregate_components(self, server_info: types.Implementation, session
 
         # Query the server for its tools and aggregate to list.
         try:
-            tools = (await session.list_tools()).tools
+            tools = await _drain_paginated(session.list_tools, "tools")
             for tool in tools:
                 name = self._component_name(tool.name, server_info)
                 tools_temp[name] = tool