diff --git a/src/fastmcp/client/mixins/prompts.py b/src/fastmcp/client/mixins/prompts.py index e8907c194..2844b2c6c 100644 --- a/src/fastmcp/client/mixins/prompts.py +++ b/src/fastmcp/client/mixins/prompts.py @@ -20,6 +20,8 @@ from fastmcp.utilities.logging import get_logger logger = get_logger(__name__) +AUTO_PAGINATION_MAX_PAGES = 250 + # Type alias for task response union (SEP-1686 graceful degradation) PromptTaskResponseUnion = RootModel[ mcp.types.CreateTaskResult | mcp.types.GetPromptResult @@ -54,25 +56,31 @@ class ClientPromptsMixin: ) return result - async def list_prompts(self: Client) -> list[mcp.types.Prompt]: + async def list_prompts( + self: Client, + max_pages: int = AUTO_PAGINATION_MAX_PAGES, + ) -> list[mcp.types.Prompt]: """Retrieve all prompts available on the server. This method automatically fetches all pages if the server paginates results, returning the complete list. For manual pagination control (e.g., to handle large result sets incrementally), use list_prompts_mcp() with the cursor parameter. + Args: + max_pages: Maximum number of pages to fetch before raising. Defaults to 250. + Returns: list[mcp.types.Prompt]: A list of all Prompt objects. Raises: - RuntimeError: If called while the client is not connected. + RuntimeError: If the page limit is reached before pagination completes. McpError: If the request results in a TimeoutError | JSONRPCError """ all_prompts: list[mcp.types.Prompt] = [] cursor: str | None = None seen_cursors: set[str] = set() - while True: + for _ in range(max_pages): result = await self.list_prompts_mcp(cursor=cursor) all_prompts.extend(result.prompts) if not result.nextCursor: @@ -85,6 +93,13 @@ class ClientPromptsMixin: break seen_cursors.add(result.nextCursor) cursor = result.nextCursor + else: + raise RuntimeError( + f"[{self.name}] Reached auto-pagination limit" + f" ({max_pages} pages) for list_prompts." + " Use list_prompts_mcp() with cursor for manual pagination," + " or increase max_pages." + ) return all_prompts diff --git a/src/fastmcp/client/mixins/resources.py b/src/fastmcp/client/mixins/resources.py index f49f861f8..92889a2e6 100644 --- a/src/fastmcp/client/mixins/resources.py +++ b/src/fastmcp/client/mixins/resources.py @@ -19,6 +19,8 @@ from fastmcp.utilities.logging import get_logger logger = get_logger(__name__) +AUTO_PAGINATION_MAX_PAGES = 250 + # Type alias for task response union (SEP-1686 graceful degradation) ResourceTaskResponseUnion = RootModel[ mcp.types.CreateTaskResult | mcp.types.ReadResourceResult @@ -53,25 +55,31 @@ class ClientResourcesMixin: ) return result - async def list_resources(self: Client) -> list[mcp.types.Resource]: + async def list_resources( + self: Client, + max_pages: int = AUTO_PAGINATION_MAX_PAGES, + ) -> list[mcp.types.Resource]: """Retrieve all resources available on the server. This method automatically fetches all pages if the server paginates results, returning the complete list. For manual pagination control (e.g., to handle large result sets incrementally), use list_resources_mcp() with the cursor parameter. + Args: + max_pages: Maximum number of pages to fetch before raising. Defaults to 250. + Returns: list[mcp.types.Resource]: A list of all Resource objects. Raises: - RuntimeError: If called while the client is not connected. + RuntimeError: If the page limit is reached before pagination completes. McpError: If the request results in a TimeoutError | JSONRPCError """ all_resources: list[mcp.types.Resource] = [] cursor: str | None = None seen_cursors: set[str] = set() - while True: + for _ in range(max_pages): result = await self.list_resources_mcp(cursor=cursor) all_resources.extend(result.resources) if not result.nextCursor: @@ -84,6 +92,13 @@ class ClientResourcesMixin: break seen_cursors.add(result.nextCursor) cursor = result.nextCursor + else: + raise RuntimeError( + f"[{self.name}] Reached auto-pagination limit" + f" ({max_pages} pages) for list_resources." + " Use list_resources_mcp() with cursor for manual pagination," + " or increase max_pages." + ) return all_resources @@ -110,7 +125,10 @@ class ClientResourcesMixin: ) return result - async def list_resource_templates(self: Client) -> list[mcp.types.ResourceTemplate]: + async def list_resource_templates( + self: Client, + max_pages: int = AUTO_PAGINATION_MAX_PAGES, + ) -> list[mcp.types.ResourceTemplate]: """Retrieve all resource templates available on the server. This method automatically fetches all pages if the server paginates results, @@ -118,18 +136,21 @@ class ClientResourcesMixin: large result sets incrementally), use list_resource_templates_mcp() with the cursor parameter. + Args: + max_pages: Maximum number of pages to fetch before raising. Defaults to 250. + Returns: list[mcp.types.ResourceTemplate]: A list of all ResourceTemplate objects. Raises: - RuntimeError: If called while the client is not connected. + RuntimeError: If the page limit is reached before pagination completes. McpError: If the request results in a TimeoutError | JSONRPCError """ all_templates: list[mcp.types.ResourceTemplate] = [] cursor: str | None = None seen_cursors: set[str] = set() - while True: + for _ in range(max_pages): result = await self.list_resource_templates_mcp(cursor=cursor) all_templates.extend(result.resourceTemplates) if not result.nextCursor: @@ -143,6 +164,13 @@ class ClientResourcesMixin: break seen_cursors.add(result.nextCursor) cursor = result.nextCursor + else: + raise RuntimeError( + f"[{self.name}] Reached auto-pagination limit" + f" ({max_pages} pages) for list_resource_templates." + " Use list_resource_templates_mcp() with cursor for manual pagination," + " or increase max_pages." + ) return all_templates diff --git a/src/fastmcp/client/mixins/tools.py b/src/fastmcp/client/mixins/tools.py index f702e4938..3c41cc376 100644 --- a/src/fastmcp/client/mixins/tools.py +++ b/src/fastmcp/client/mixins/tools.py @@ -25,6 +25,8 @@ from fastmcp.utilities.types import get_cached_typeadapter logger = get_logger(__name__) +AUTO_PAGINATION_MAX_PAGES = 250 + # Type alias for task response union (SEP-1686 graceful degradation) ToolTaskResponseUnion = RootModel[mcp.types.CreateTaskResult | mcp.types.CallToolResult] @@ -57,25 +59,31 @@ class ClientToolsMixin: ) return result - async def list_tools(self: Client) -> list[mcp.types.Tool]: + async def list_tools( + self: Client, + max_pages: int = AUTO_PAGINATION_MAX_PAGES, + ) -> list[mcp.types.Tool]: """Retrieve all tools available on the server. This method automatically fetches all pages if the server paginates results, returning the complete list. For manual pagination control (e.g., to handle large result sets incrementally), use list_tools_mcp() with the cursor parameter. + Args: + max_pages: Maximum number of pages to fetch before raising. Defaults to 250. + Returns: list[mcp.types.Tool]: A list of all Tool objects. Raises: - RuntimeError: If called while the client is not connected. + RuntimeError: If the page limit is reached before pagination completes. McpError: If the request results in a TimeoutError | JSONRPCError """ all_tools: list[mcp.types.Tool] = [] cursor: str | None = None seen_cursors: set[str] = set() - while True: + for _ in range(max_pages): result = await self.list_tools_mcp(cursor=cursor) all_tools.extend(result.tools) if not result.nextCursor: @@ -88,6 +96,13 @@ class ClientToolsMixin: break seen_cursors.add(result.nextCursor) cursor = result.nextCursor + else: + raise RuntimeError( + f"[{self.name}] Reached auto-pagination limit" + f" ({max_pages} pages) for list_tools." + " Use list_tools_mcp() with cursor for manual pagination," + " or increase max_pages." + ) return all_tools diff --git a/tests/server/test_pagination.py b/tests/server/test_pagination.py index f90a763c0..f28fabe1b 100644 --- a/tests/server/test_pagination.py +++ b/tests/server/test_pagination.py @@ -420,6 +420,96 @@ class TestPaginationCycleDetection: assert len(tools) == 1 assert tools[0].name == "my_tool" + async def test_tools_raises_on_auto_pagination_limit(self) -> None: + """list_tools should raise RuntimeError after exceeding max_pages.""" + server = FastMCP() + + @server.tool + def my_tool() -> str: + return "ok" + + async with Client(server) as client: + original = client.list_tools_mcp + call_count = 0 + + async def returning_unique_cursor( + *, + cursor: str | None = None, + ) -> mcp.types.ListToolsResult: + nonlocal call_count + result = await original(cursor=cursor) + call_count += 1 + result.nextCursor = f"cursor-{call_count}" + return result + + with ( + patch.object( + client, "list_tools_mcp", side_effect=returning_unique_cursor + ), + pytest.raises(RuntimeError, match="auto-pagination limit"), + ): + await client.list_tools(max_pages=5) + + async def test_resources_raises_on_auto_pagination_limit(self) -> None: + """list_resources should raise RuntimeError after exceeding max_pages.""" + server = FastMCP() + + @server.resource("test://r") + def my_resource() -> str: + return "data" + + async with Client(server) as client: + original = client.list_resources_mcp + call_count = 0 + + async def returning_unique_cursor( + *, + cursor: str | None = None, + ) -> mcp.types.ListResourcesResult: + nonlocal call_count + result = await original(cursor=cursor) + call_count += 1 + result.nextCursor = f"cursor-{call_count}" + return result + + with ( + patch.object( + client, "list_resources_mcp", side_effect=returning_unique_cursor + ), + pytest.raises(RuntimeError, match="auto-pagination limit"), + ): + await client.list_resources(max_pages=5) + + async def test_prompts_raises_on_auto_pagination_limit(self) -> None: + """list_prompts should raise RuntimeError after exceeding max_pages.""" + server = FastMCP() + + @server.prompt + def my_prompt() -> str: + return "text" + + async with Client(server) as client: + original = client.list_prompts_mcp + call_count = 0 + + async def returning_unique_cursor( + *, + cursor: str | None = None, + ) -> mcp.types.ListPromptsResult: + nonlocal call_count + result = await original(cursor=cursor) + call_count += 1 + result.nextCursor = f"cursor-{call_count}" + return result + + with ( + patch.object( + client, "list_prompts_mcp", side_effect=returning_unique_cursor + ), + pytest.raises(RuntimeError, match="auto-pagination limit"), + ): + await client.list_prompts(max_pages=5) + async def test_normal_pagination_unaffected(self) -> None: """Cycle detection should not interfere with normal pagination.""" server = FastMCP(list_page_size=10)