From 28361c0e67a04c12c37228ad1e8b34fc716da816 Mon Sep 17 00:00:00 2001 From: ZXT-zjbiliy <3240102335@zju.edu.cn> Date: Sun, 7 Jun 2026 15:16:13 +0800 Subject: [PATCH] feat(client): add list_all helpers for paginated MCP operations Add list_all_tools, list_all_resources, list_all_resource_templates, and list_all_prompts methods to Client that automatically follow next_cursor and return all items across all pages. Closes #2556 --- src/mcp/client/client.py | 77 ++++++++++++ tests/client/test_list_all_helpers.py | 170 ++++++++++++++++++++++++++ 2 files changed, 247 insertions(+) create mode 100644 tests/client/test_list_all_helpers.py diff --git a/src/mcp/client/client.py b/src/mcp/client/client.py index b33fea405..762972761 100644 --- a/src/mcp/client/client.py +++ b/src/mcp/client/client.py @@ -26,10 +26,14 @@ ListToolsResult, LoggingLevel, PaginatedRequestParams, + Prompt, PromptReference, ReadResourceResult, RequestParamsMeta, + Resource, + ResourceTemplate, ResourceTemplateReference, + Tool, ) @@ -302,6 +306,79 @@ async def list_tools(self, *, cursor: str | None = None, meta: RequestParamsMeta """List available tools from the server.""" return await self.session.list_tools(params=PaginatedRequestParams(cursor=cursor, _meta=meta)) + async def list_all_tools(self, *, meta: RequestParamsMeta | None = None) -> ListToolsResult: + """List all available tools from the server, draining pagination automatically. + + Follows ``next_cursor`` until the server returns no more pages and + returns a single :class:`ListToolsResult` whose ``tools`` list contains + every tool across all pages. + + The tool output-schema cache is populated as a side effect (same as + :meth:`list_tools`). + """ + all_tools: list[Tool] = [] + cursor: str | None = None + while True: + result = await self.list_tools(cursor=cursor, meta=meta) + all_tools.extend(result.tools) + if result.next_cursor is None: + break + cursor = result.next_cursor + return ListToolsResult(tools=all_tools) + + async def list_all_resources(self, *, meta: RequestParamsMeta | None = None) -> ListResourcesResult: + """List all available resources from the server, draining pagination automatically. + + Follows ``next_cursor`` until the server returns no more pages and + returns a single :class:`ListResourcesResult` whose ``resources`` list + contains every resource across all pages. + """ + all_resources: list[Resource] = [] + cursor: str | None = None + while True: + result = await self.list_resources(cursor=cursor, meta=meta) + all_resources.extend(result.resources) + if result.next_cursor is None: + break + cursor = result.next_cursor + return ListResourcesResult(resources=all_resources) + + async def list_all_resource_templates( + self, *, meta: RequestParamsMeta | None = None + ) -> ListResourceTemplatesResult: + """List all available resource templates from the server, draining pagination automatically. + + Follows ``next_cursor`` until the server returns no more pages and + returns a single :class:`ListResourceTemplatesResult` whose + ``resource_templates`` list contains every template across all pages. + """ + all_templates: list[ResourceTemplate] = [] + cursor: str | None = None + while True: + result = await self.list_resource_templates(cursor=cursor, meta=meta) + all_templates.extend(result.resource_templates) + if result.next_cursor is None: + break + cursor = result.next_cursor + return ListResourceTemplatesResult(resource_templates=all_templates) + + async def list_all_prompts(self, *, meta: RequestParamsMeta | None = None) -> ListPromptsResult: + """List all available prompts from the server, draining pagination automatically. + + Follows ``next_cursor`` until the server returns no more pages and + returns a single :class:`ListPromptsResult` whose ``prompts`` list + contains every prompt across all pages. + """ + all_prompts: list[Prompt] = [] + cursor: str | None = None + while True: + result = await self.list_prompts(cursor=cursor, meta=meta) + all_prompts.extend(result.prompts) + if result.next_cursor is None: + break + cursor = result.next_cursor + return ListPromptsResult(prompts=all_prompts) + async def send_roots_list_changed(self) -> None: """Send a notification that the roots list has changed.""" # TODO(Marcelo): Currently, there is no way for the server to handle this. We should add support. diff --git a/tests/client/test_list_all_helpers.py b/tests/client/test_list_all_helpers.py new file mode 100644 index 000000000..ad0db5d26 --- /dev/null +++ b/tests/client/test_list_all_helpers.py @@ -0,0 +1,170 @@ +"""Tests for the list_all_* helpers on Client that drain pagination automatically.""" + +import pytest + +from mcp import types +from mcp.client.client import Client +from mcp.server import Server, ServerRequestContext +from mcp.types import ( + ListPromptsResult, + ListResourcesResult, + ListResourceTemplatesResult, + ListToolsResult, + Prompt, + Resource, + ResourceTemplate, + Tool, +) + +pytestmark = pytest.mark.anyio + + +async def test_list_all_tools_drains_pagination() -> None: + """list_all_tools follows next_cursor and returns all tools across pages.""" + pages: dict[str | None, tuple[list[str], str | None]] = { + None: (["alpha", "beta"], "page-2"), + "page-2": (["gamma"], None), + } + + async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestParams | None) -> ListToolsResult: + assert params is not None + names, next_cursor = pages[params.cursor] + return ListToolsResult( + tools=[Tool(name=n, input_schema={"type": "object"}) for n in names], + next_cursor=next_cursor, + ) + + server = Server("paginated", on_list_tools=list_tools) + + async with Client(server) as client: + result = await client.list_all_tools() + + assert [t.name for t in result.tools] == ["alpha", "beta", "gamma"] + assert result.next_cursor is None + + +async def test_list_all_tools_single_page() -> None: + """list_all_tools works when the server returns all tools in a single page.""" + + async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestParams | None) -> ListToolsResult: + return ListToolsResult( + tools=[ + Tool(name="only", input_schema={"type": "object"}), + ] + ) + + server = Server("single", on_list_tools=list_tools) + + async with Client(server) as client: + result = await client.list_all_tools() + + assert [t.name for t in result.tools] == ["only"] + + +async def test_list_all_tools_empty() -> None: + """list_all_tools returns an empty list when the server has no tools.""" + + async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestParams | None) -> ListToolsResult: + return ListToolsResult(tools=[]) + + server = Server("empty", on_list_tools=list_tools) + + async with Client(server) as client: + result = await client.list_all_tools() + + assert result.tools == [] + + +async def test_list_all_resources_drains_pagination() -> None: + """list_all_resources follows next_cursor and returns all resources across pages.""" + pages: dict[str | None, tuple[list[str], str | None]] = { + None: (["res-a"], "page-2"), + "page-2": (["res-b", "res-c"], None), + } + + async def list_resources( + ctx: ServerRequestContext, params: types.PaginatedRequestParams | None + ) -> ListResourcesResult: + assert params is not None + names, next_cursor = pages[params.cursor] + return ListResourcesResult( + resources=[Resource(uri=f"test://{n}", name=n) for n in names], + next_cursor=next_cursor, + ) + + server = Server("paginated", on_list_resources=list_resources) + + async with Client(server) as client: + result = await client.list_all_resources() + + assert [r.name for r in result.resources] == ["res-a", "res-b", "res-c"] + + +async def test_list_all_resource_templates_drains_pagination() -> None: + """list_all_resource_templates follows next_cursor and returns all templates across pages.""" + pages: dict[str | None, tuple[list[str], str | None]] = { + None: (["tmpl-a"], "page-2"), + "page-2": (["tmpl-b"], None), + } + + async def list_resource_templates( + ctx: ServerRequestContext, params: types.PaginatedRequestParams | None + ) -> ListResourceTemplatesResult: + assert params is not None + names, next_cursor = pages[params.cursor] + return ListResourceTemplatesResult( + resource_templates=[ResourceTemplate(name=n, uri_template=f"{n}://{{id}}") for n in names], + next_cursor=next_cursor, + ) + + server = Server("paginated", on_list_resource_templates=list_resource_templates) + + async with Client(server) as client: + result = await client.list_all_resource_templates() + + assert [t.name for t in result.resource_templates] == ["tmpl-a", "tmpl-b"] + + +async def test_list_all_prompts_drains_pagination() -> None: + """list_all_prompts follows next_cursor and returns all prompts across pages.""" + pages: dict[str | None, tuple[list[str], str | None]] = { + None: (["greet", "farewell"], "page-2"), + "page-2": (["summarize"], None), + } + + async def list_prompts(ctx: ServerRequestContext, params: types.PaginatedRequestParams | None) -> ListPromptsResult: + assert params is not None + names, next_cursor = pages[params.cursor] + return ListPromptsResult( + prompts=[Prompt(name=n) for n in names], + next_cursor=next_cursor, + ) + + server = Server("paginated", on_list_prompts=list_prompts) + + async with Client(server) as client: + result = await client.list_all_prompts() + + assert [p.name for p in result.prompts] == ["greet", "farewell", "summarize"] + + +async def test_list_all_tools_populates_output_schema_cache() -> None: + """list_all_tools populates the tool output-schema cache (same as list_tools).""" + + async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestParams | None) -> ListToolsResult: + return ListToolsResult( + tools=[ + Tool( + name="cached_tool", + input_schema={"type": "object"}, + output_schema={"type": "object", "properties": {"x": {"type": "integer"}}}, + ), + ] + ) + + server = Server("schema-cache", on_list_tools=list_tools) + + async with Client(server) as client: + await client.list_all_tools() + # The cache should be populated + assert "cached_tool" in client.session._tool_output_schemas