From dada8893a9dc40219f032b6a27631f2c208c39be Mon Sep 17 00:00:00 2001
From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
Date: Sat, 17 Jan 2026 19:30:24 -0500
Subject: [PATCH 1/3] Add MCP-compliant pagination for list operations

Server authors opt-in by setting list_page_size on FastMCP.
Client convenience methods auto-fetch all pages transparently.
Use _mcp methods with cursor parameter for manual pagination.
---
 src/fastmcp/client/client.py                  | 127 +++++++---
 src/fastmcp/server/context.py                 |   8 +-
 src/fastmcp/server/server.py                  | 131 +++++++---
 src/fastmcp/utilities/pagination.py           |  73 ++++++
 tests/server/auth/test_authorization.py       |  27 ++-
 .../providers/proxy/test_proxy_server.py      |  23 +-
 tests/server/test_dependencies.py             |  17 +-
 tests/server/test_file_server.py              |   6 +-
 tests/server/test_pagination.py               | 225 ++++++++++++++++++
 tests/server/test_tool_annotations.py         |  13 +-
 10 files changed, 547 insertions(+), 103 deletions(-)
 create mode 100644 src/fastmcp/utilities/pagination.py
 create mode 100644 tests/server/test_pagination.py

diff --git a/src/fastmcp/client/client.py b/src/fastmcp/client/client.py
index e7a287112..5809e51bf 100644
--- a/src/fastmcp/client/client.py
+++ b/src/fastmcp/client/client.py
@@ -796,9 +796,14 @@ class Client(Generic[ClientTransportT]):
 
     # --- Resources ---
 
-    async def list_resources_mcp(self) -> mcp.types.ListResourcesResult:
+    async def list_resources_mcp(
+        self, *, cursor: str | None = None
+    ) -> mcp.types.ListResourcesResult:
         """Send a resources/list request and return the complete MCP protocol result.
 
+        Args:
+            cursor: Optional pagination cursor from a previous request's nextCursor.
+
         Returns:
             mcp.types.ListResourcesResult: The complete response object from the protocol,
                 containing the list of resources and any additional metadata.
@@ -810,28 +815,44 @@ class Client(Generic[ClientTransportT]):
         logger.debug(f"[{self.name}] called list_resources")
 
         result = await self._await_with_session_monitoring(
-            self.session.list_resources()
+            self.session.list_resources(cursor=cursor)
         )
         return result
 
     async def list_resources(self) -> list[mcp.types.Resource]:
-        """Retrieve a list of resources available on the server.
+        """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.
 
         Returns:
-            list[mcp.types.Resource]: A list of Resource objects.
+            list[mcp.types.Resource]: A list of all Resource objects.
 
         Raises:
             RuntimeError: If called while the client is not connected.
             McpError: If the request results in a TimeoutError | JSONRPCError
         """
-        result = await self.list_resources_mcp()
-        return result.resources
+        all_resources: list[mcp.types.Resource] = []
+        cursor: str | None = None
+
+        while True:
+            result = await self.list_resources_mcp(cursor=cursor)
+            all_resources.extend(result.resources)
+            if result.nextCursor is None:
+                break
+            cursor = result.nextCursor
+
+        return all_resources
 
     async def list_resource_templates_mcp(
-        self,
+        self, *, cursor: str | None = None
     ) -> mcp.types.ListResourceTemplatesResult:
         """Send a resources/listResourceTemplates request and return the complete MCP protocol result.
 
+        Args:
+            cursor: Optional pagination cursor from a previous request's nextCursor.
+
         Returns:
             mcp.types.ListResourceTemplatesResult: The complete response object from the protocol,
                 containing the list of resource templates and any additional metadata.
@@ -843,24 +864,36 @@ class Client(Generic[ClientTransportT]):
         logger.debug(f"[{self.name}] called list_resource_templates")
 
         result = await self._await_with_session_monitoring(
-            self.session.list_resource_templates()
+            self.session.list_resource_templates(cursor=cursor)
         )
         return result
 
-    async def list_resource_templates(
-        self,
-    ) -> list[mcp.types.ResourceTemplate]:
-        """Retrieve a list of resource templates available on the server.
+    async def list_resource_templates(self) -> list[mcp.types.ResourceTemplate]:
+        """Retrieve all resource templates 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_resource_templates_mcp() with the
+        cursor parameter.
 
         Returns:
-            list[mcp.types.ResourceTemplate]: A list of ResourceTemplate objects.
+            list[mcp.types.ResourceTemplate]: A list of all ResourceTemplate objects.
 
         Raises:
             RuntimeError: If called while the client is not connected.
             McpError: If the request results in a TimeoutError | JSONRPCError
         """
-        result = await self.list_resource_templates_mcp()
-        return result.resourceTemplates
+        all_templates: list[mcp.types.ResourceTemplate] = []
+        cursor: str | None = None
+
+        while True:
+            result = await self.list_resource_templates_mcp(cursor=cursor)
+            all_templates.extend(result.resourceTemplates)
+            if result.nextCursor is None:
+                break
+            cursor = result.nextCursor
+
+        return all_templates
 
     async def read_resource_mcp(
         self, uri: AnyUrl | str, meta: dict[str, Any] | None = None
@@ -1071,9 +1104,14 @@ class Client(Generic[ClientTransportT]):
 
     # --- Prompts ---
 
-    async def list_prompts_mcp(self) -> mcp.types.ListPromptsResult:
+    async def list_prompts_mcp(
+        self, *, cursor: str | None = None
+    ) -> mcp.types.ListPromptsResult:
         """Send a prompts/list request and return the complete MCP protocol result.
 
+        Args:
+            cursor: Optional pagination cursor from a previous request's nextCursor.
+
         Returns:
             mcp.types.ListPromptsResult: The complete response object from the protocol,
                 containing the list of prompts and any additional metadata.
@@ -1084,21 +1122,36 @@ class Client(Generic[ClientTransportT]):
         """
         logger.debug(f"[{self.name}] called list_prompts")
 
-        result = await self._await_with_session_monitoring(self.session.list_prompts())
+        result = await self._await_with_session_monitoring(
+            self.session.list_prompts(cursor=cursor)
+        )
         return result
 
     async def list_prompts(self) -> list[mcp.types.Prompt]:
-        """Retrieve a list of prompts available on the server.
+        """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.
 
         Returns:
-            list[mcp.types.Prompt]: A list of Prompt objects.
+            list[mcp.types.Prompt]: A list of all Prompt objects.
 
         Raises:
             RuntimeError: If called while the client is not connected.
             McpError: If the request results in a TimeoutError | JSONRPCError
         """
-        result = await self.list_prompts_mcp()
-        return result.prompts
+        all_prompts: list[mcp.types.Prompt] = []
+        cursor: str | None = None
+
+        while True:
+            result = await self.list_prompts_mcp(cursor=cursor)
+            all_prompts.extend(result.prompts)
+            if result.nextCursor is None:
+                break
+            cursor = result.nextCursor
+
+        return all_prompts
 
     # --- Prompt ---
     async def get_prompt_mcp(
@@ -1375,9 +1428,14 @@ class Client(Generic[ClientTransportT]):
 
     # --- Tools ---
 
-    async def list_tools_mcp(self) -> mcp.types.ListToolsResult:
+    async def list_tools_mcp(
+        self, *, cursor: str | None = None
+    ) -> mcp.types.ListToolsResult:
         """Send a tools/list request and return the complete MCP protocol result.
 
+        Args:
+            cursor: Optional pagination cursor from a previous request's nextCursor.
+
         Returns:
             mcp.types.ListToolsResult: The complete response object from the protocol,
                 containing the list of tools and any additional metadata.
@@ -1388,21 +1446,36 @@ class Client(Generic[ClientTransportT]):
         """
         logger.debug(f"[{self.name}] called list_tools")
 
-        result = await self._await_with_session_monitoring(self.session.list_tools())
+        result = await self._await_with_session_monitoring(
+            self.session.list_tools(cursor=cursor)
+        )
         return result
 
     async def list_tools(self) -> list[mcp.types.Tool]:
-        """Retrieve a list of tools available on the server.
+        """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.
 
         Returns:
-            list[mcp.types.Tool]: A list of Tool objects.
+            list[mcp.types.Tool]: A list of all Tool objects.
 
         Raises:
             RuntimeError: If called while the client is not connected.
             McpError: If the request results in a TimeoutError | JSONRPCError
         """
-        result = await self.list_tools_mcp()
-        return result.tools
+        all_tools: list[mcp.types.Tool] = []
+        cursor: str | None = None
+
+        while True:
+            result = await self.list_tools_mcp(cursor=cursor)
+            all_tools.extend(result.tools)
+            if result.nextCursor is None:
+                break
+            cursor = result.nextCursor
+
+        return all_tools
 
     # --- Call Tool ---
 
diff --git a/src/fastmcp/server/context.py b/src/fastmcp/server/context.py
index 16e240ff5..aa5cfe658 100644
--- a/src/fastmcp/server/context.py
+++ b/src/fastmcp/server/context.py
@@ -346,7 +346,10 @@ class Context:
         Returns:
             List of Resource objects available on the server
         """
-        return await self.fastmcp._list_resources_mcp()
+        result = await self.fastmcp._list_resources_mcp(
+            mcp.types.ListResourcesRequest()
+        )
+        return result.resources
 
     async def list_prompts(self) -> list[SDKPrompt]:
         """List all available prompts from the server.
@@ -354,7 +357,8 @@ class Context:
         Returns:
             List of Prompt objects available on the server
         """
-        return await self.fastmcp._list_prompts_mcp()
+        result = await self.fastmcp._list_prompts_mcp(mcp.types.ListPromptsRequest())
+        return result.prompts
 
     async def get_prompt(
         self, name: str, arguments: dict[str, Any] | None = None
diff --git a/src/fastmcp/server/server.py b/src/fastmcp/server/server.py
index 1cde0cc4a..114dddece 100644
--- a/src/fastmcp/server/server.py
+++ b/src/fastmcp/server/server.py
@@ -43,10 +43,6 @@ from mcp.types import (
     ContentBlock,
     ToolAnnotations,
 )
-from mcp.types import Prompt as SDKPrompt
-from mcp.types import Resource as SDKResource
-from mcp.types import ResourceTemplate as SDKResourceTemplate
-from mcp.types import Tool as SDKTool
 from pydantic import AnyUrl
 from pydantic import ValidationError as PydanticValidationError
 from starlette.middleware import Middleware as ASGIMiddleware
@@ -101,6 +97,7 @@ from fastmcp.utilities.async_utils import gather
 from fastmcp.utilities.cli import log_server_banner
 from fastmcp.utilities.components import FastMCPComponent
 from fastmcp.utilities.logging import get_logger, temporary_log_level
+from fastmcp.utilities.pagination import paginate_sequence
 from fastmcp.utilities.types import FastMCPBaseModel, NotSet, NotSetT
 from fastmcp.utilities.versions import (
     VersionSpec,
@@ -275,6 +272,26 @@ def _lifespan_proxy(
     return wrap
 
 
+PaginateT = TypeVar("PaginateT")
+
+
+def _apply_pagination(
+    items: Sequence[PaginateT],
+    cursor: str | None,
+    page_size: int | None,
+) -> tuple[list[PaginateT], str | None]:
+    """Apply pagination to items, raising McpError for invalid cursors.
+
+    If page_size is None, returns all items without pagination.
+    """
+    if page_size is None:
+        return list(items), None
+    try:
+        return paginate_sequence(items, cursor, page_size)
+    except ValueError as e:
+        raise McpError(mcp.types.ErrorData(code=-32602, message=str(e))) from e
+
+
 class StateValue(FastMCPBaseModel):
     """Wrapper for stored context state values."""
 
@@ -301,6 +318,7 @@ class FastMCP(Provider, Generic[LifespanResultT]):
         exclude_tags: Collection[str] | None = None,
         on_duplicate: DuplicateBehavior | None = None,
         strict_input_validation: bool | None = None,
+        list_page_size: int | None = None,
         tasks: bool | None = None,
         session_state_store: AsyncKeyValue | None = None,
         # ---
@@ -368,6 +386,9 @@ class FastMCP(Provider, Generic[LifespanResultT]):
             else fastmcp.settings.mask_error_details
         )
 
+        # Store list_page_size for pagination of list operations
+        self._list_page_size: int | None = list_page_size
+
         if tool_serializer is not None and fastmcp.settings.deprecation_warnings:
             warnings.warn(
                 "The `tool_serializer` parameter is deprecated. "
@@ -738,16 +759,26 @@ class FastMCP(Provider, Generic[LifespanResultT]):
     def _setup_handlers(self) -> None:
         """Set up core MCP protocol handlers.
 
-        All handlers use decorator-based registration for consistency.
+        List handlers use SDK decorators that pass the request object to our handler
+        (needed for pagination cursor). The SDK also populates caches like _tool_cache.
+
+        Exception: list_resource_templates SDK decorator doesn't pass the request,
+        so we register that handler directly.
+
         The call_tool decorator is from the SDK (supports CreateTaskResult + validate_input).
         The read_resource and get_prompt decorators are from LowLevelServer to add
         CreateTaskResult support until the SDK provides it natively.
         """
         self._mcp_server.list_tools()(self._list_tools_mcp)
         self._mcp_server.list_resources()(self._list_resources_mcp)
-        self._mcp_server.list_resource_templates()(self._list_resource_templates_mcp)
         self._mcp_server.list_prompts()(self._list_prompts_mcp)
 
+        # list_resource_templates SDK decorator doesn't pass the request to handlers,
+        # so we register directly to get cursor access for pagination
+        self._mcp_server.request_handlers[mcp.types.ListResourceTemplatesRequest] = (
+            self._wrap_list_handler(self._list_resource_templates_mcp)
+        )
+
         self._mcp_server.call_tool(validate_input=self.strict_input_validation)(
             self._call_tool_mcp
         )
@@ -757,6 +788,17 @@ class FastMCP(Provider, Generic[LifespanResultT]):
         # Register SEP-1686 task protocol handlers
         self._setup_task_protocol_handlers()
 
+    def _wrap_list_handler(
+        self, handler: Callable[..., Awaitable[Any]]
+    ) -> Callable[..., Awaitable[mcp.types.ServerResult]]:
+        """Wrap a list handler to pass the request and return ServerResult."""
+
+        async def wrapper(request: Any) -> mcp.types.ServerResult:
+            result = await handler(request)
+            return mcp.types.ServerResult(result)
+
+        return wrapper
+
     def _setup_task_protocol_handlers(self) -> None:
         """Register SEP-1686 task protocol handlers with SDK.
 
@@ -1869,42 +1911,56 @@ class FastMCP(Provider, Generic[LifespanResultT]):
         """
         return list(self._additional_http_routes)
 
-    async def _list_tools_mcp(self) -> list[SDKTool]:
+    async def _list_tools_mcp(
+        self, request: mcp.types.ListToolsRequest
+    ) -> mcp.types.ListToolsResult:
         """
         List all available tools, in the format expected by the low-level MCP
-        server.
+        server. Supports pagination when list_page_size is configured.
         """
         logger.debug(f"[{self.name}] Handler called: list_tools")
 
         async with fastmcp.server.context.Context(fastmcp=self):
             tools = await self.get_tools(run_middleware=True)
-            return [
-                tool.to_mcp_tool(
-                    name=tool.name,
-                )
-                for tool in tools
-            ]
+            sdk_tools = [tool.to_mcp_tool(name=tool.name) for tool in tools]
+            # SDK may pass None for internal cache refresh despite type hint
+            cursor = (
+                request.params.cursor  # type: ignore[union-attr]
+                if request is not None and request.params
+                else None
+            )
+            page, next_cursor = _apply_pagination(
+                sdk_tools, cursor, self._list_page_size
+            )
+            return mcp.types.ListToolsResult(tools=page, nextCursor=next_cursor)
 
-    async def _list_resources_mcp(self) -> list[SDKResource]:
+    async def _list_resources_mcp(
+        self, request: mcp.types.ListResourcesRequest
+    ) -> mcp.types.ListResourcesResult:
         """
         List all available resources, in the format expected by the low-level MCP
-        server.
+        server. Supports pagination when list_page_size is configured.
         """
         logger.debug(f"[{self.name}] Handler called: list_resources")
 
         async with fastmcp.server.context.Context(fastmcp=self):
             resources = await self.get_resources(run_middleware=True)
-            return [
-                resource.to_mcp_resource(
-                    uri=str(resource.uri),
-                )
+            sdk_resources = [
+                resource.to_mcp_resource(uri=str(resource.uri))
                 for resource in resources
             ]
+            cursor = request.params.cursor if request.params else None
+            page, next_cursor = _apply_pagination(
+                sdk_resources, cursor, self._list_page_size
+            )
+            return mcp.types.ListResourcesResult(resources=page, nextCursor=next_cursor)
 
-    async def _list_resource_templates_mcp(self) -> list[SDKResourceTemplate]:
+    async def _list_resource_templates_mcp(
+        self, request: mcp.types.ListResourceTemplatesRequest
+    ) -> mcp.types.ListResourceTemplatesResult:
         """
         List all available resource templates, in the format expected by the low-level MCP
-        server.
+        server. Supports pagination when list_page_size is configured.
         """
         logger.debug(f"[{self.name}] Handler called: list_resource_templates")
 
@@ -1920,17 +1976,24 @@ class FastMCP(Provider, Generic[LifespanResultT]):
                 context=mw_context,
                 call_next=lambda context: self.get_resource_templates(),
             )
-            return [
-                template.to_mcp_template(
-                    uriTemplate=template.uri_template,
-                )
+            sdk_templates = [
+                template.to_mcp_template(uriTemplate=template.uri_template)
                 for template in templates
             ]
+            cursor = request.params.cursor if request.params else None
+            page, next_cursor = _apply_pagination(
+                sdk_templates, cursor, self._list_page_size
+            )
+            return mcp.types.ListResourceTemplatesResult(
+                resourceTemplates=page, nextCursor=next_cursor
+            )
 
-    async def _list_prompts_mcp(self) -> list[SDKPrompt]:
+    async def _list_prompts_mcp(
+        self, request: mcp.types.ListPromptsRequest
+    ) -> mcp.types.ListPromptsResult:
         """
         List all available prompts, in the format expected by the low-level MCP
-        server.
+        server. Supports pagination when list_page_size is configured.
         """
         logger.debug(f"[{self.name}] Handler called: list_prompts")
 
@@ -1946,12 +2009,12 @@ class FastMCP(Provider, Generic[LifespanResultT]):
                 context=mw_context,
                 call_next=lambda context: self.get_prompts(),
             )
-            return [
-                prompt.to_mcp_prompt(
-                    name=prompt.name,
-                )
-                for prompt in prompts
-            ]
+            sdk_prompts = [prompt.to_mcp_prompt(name=prompt.name) for prompt in prompts]
+            cursor = request.params.cursor if request.params else None
+            page, next_cursor = _apply_pagination(
+                sdk_prompts, cursor, self._list_page_size
+            )
+            return mcp.types.ListPromptsResult(prompts=page, nextCursor=next_cursor)
 
     async def _call_tool_mcp(
         self, key: str, arguments: dict[str, Any]
diff --git a/src/fastmcp/utilities/pagination.py b/src/fastmcp/utilities/pagination.py
new file mode 100644
index 000000000..b1df8d55f
--- /dev/null
+++ b/src/fastmcp/utilities/pagination.py
@@ -0,0 +1,73 @@
+"""Pagination utilities for MCP list operations."""
+
+from __future__ import annotations
+
+import base64
+import json
+from collections.abc import Sequence
+from dataclasses import dataclass
+from typing import TypeVar
+
+T = TypeVar("T")
+
+
+@dataclass
+class CursorState:
+    """Internal representation of pagination cursor state.
+
+    The cursor encodes the offset into the result set. This is opaque to clients
+    per the MCP spec - they should not parse or modify cursors.
+    """
+
+    offset: int
+
+    def encode(self) -> str:
+        """Encode cursor state to an opaque string."""
+        data = json.dumps({"o": self.offset})
+        return base64.urlsafe_b64encode(data.encode()).decode()
+
+    @classmethod
+    def decode(cls, cursor: str) -> CursorState:
+        """Decode cursor from an opaque string.
+
+        Raises:
+            ValueError: If the cursor is invalid or malformed.
+        """
+        try:
+            data = json.loads(base64.urlsafe_b64decode(cursor.encode()).decode())
+            return cls(offset=data["o"])
+        except (json.JSONDecodeError, KeyError, ValueError, TypeError) as e:
+            raise ValueError(f"Invalid cursor: {cursor}") from e
+
+
+def paginate_sequence(
+    items: Sequence[T],
+    cursor: str | None,
+    page_size: int,
+) -> tuple[list[T], str | None]:
+    """Paginate a sequence of items.
+
+    Args:
+        items: The full sequence to paginate.
+        cursor: Optional cursor from a previous request. None for first page.
+        page_size: Maximum number of items per page.
+
+    Returns:
+        Tuple of (page_items, next_cursor). next_cursor is None if no more pages.
+
+    Raises:
+        ValueError: If the cursor is invalid.
+    """
+    offset = 0
+    if cursor:
+        state = CursorState.decode(cursor)
+        offset = state.offset
+
+    end = offset + page_size
+    page = list(items[offset:end])
+
+    next_cursor = None
+    if end < len(items):
+        next_cursor = CursorState(offset=end).encode()
+
+    return page, next_cursor
diff --git a/tests/server/auth/test_authorization.py b/tests/server/auth/test_authorization.py
index e0de6aa15..e1ee8216e 100644
--- a/tests/server/auth/test_authorization.py
+++ b/tests/server/auth/test_authorization.py
@@ -2,6 +2,7 @@
 
 from unittest.mock import Mock
 
+import mcp.types as mcp_types
 import pytest
 from mcp.server.auth.middleware.auth_context import auth_context_var
 from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser
@@ -350,8 +351,8 @@ class TestAuthMiddleware:
             return "public"
 
         # No token - all tools filtered by middleware
-        tools = await mcp._list_tools_mcp()
-        assert len(tools) == 0
+        result = await mcp._list_tools_mcp(mcp_types.ListToolsRequest())
+        assert len(result.tools) == 0
 
     async def test_middleware_allows_tools_with_token(self):
         mcp = FastMCP(middleware=[AuthMiddleware(auth=require_auth)])
@@ -363,8 +364,8 @@ class TestAuthMiddleware:
         token = make_token()
         tok = set_token(token)
         try:
-            tools = await mcp._list_tools_mcp()
-            assert len(tools) == 1
+            result = await mcp._list_tools_mcp(mcp_types.ListToolsRequest())
+            assert len(result.tools) == 1
         finally:
             auth_context_var.reset(tok)
 
@@ -379,8 +380,8 @@ class TestAuthMiddleware:
         token = make_token(scopes=["read"])
         tok = set_token(token)
         try:
-            tools = await mcp._list_tools_mcp()
-            assert len(tools) == 0
+            result = await mcp._list_tools_mcp(mcp_types.ListToolsRequest())
+            assert len(result.tools) == 0
         finally:
             auth_context_var.reset(tok)
 
@@ -388,8 +389,8 @@ class TestAuthMiddleware:
         token = make_token(scopes=["api"])
         tok = set_token(token)
         try:
-            tools = await mcp._list_tools_mcp()
-            assert len(tools) == 1
+            result = await mcp._list_tools_mcp(mcp_types.ListToolsRequest())
+            assert len(result.tools) == 1
         finally:
             auth_context_var.reset(tok)
 
@@ -407,16 +408,16 @@ class TestAuthMiddleware:
             return "admin"
 
         # No token - public tool allowed, admin tool blocked
-        tools = await mcp._list_tools_mcp()
-        assert len(tools) == 1
-        assert tools[0].name == "public_tool"
+        result = await mcp._list_tools_mcp(mcp_types.ListToolsRequest())
+        assert len(result.tools) == 1
+        assert result.tools[0].name == "public_tool"
 
         # Token with admin scope - both allowed
         token = make_token(scopes=["admin"])
         tok = set_token(token)
         try:
-            tools = await mcp._list_tools_mcp()
-            assert len(tools) == 2
+            result = await mcp._list_tools_mcp(mcp_types.ListToolsRequest())
+            assert len(result.tools) == 2
         finally:
             auth_context_var.reset(tok)
 
diff --git a/tests/server/providers/proxy/test_proxy_server.py b/tests/server/providers/proxy/test_proxy_server.py
index ecad2062d..d047201dd 100644
--- a/tests/server/providers/proxy/test_proxy_server.py
+++ b/tests/server/providers/proxy/test_proxy_server.py
@@ -2,6 +2,7 @@ import inspect
 import json
 from typing import Any, cast
 
+import mcp.types as mcp_types
 import pytest
 from anyio import create_task_group
 from dirty_equals import Contains
@@ -285,10 +286,9 @@ class TestTools:
         assert tool.description is None
 
     async def test_list_tools_same_as_original(self, fastmcp_server, proxy_server):
-        assert (
-            await proxy_server._list_tools_mcp()
-            == await fastmcp_server._list_tools_mcp()
-        )
+        assert await proxy_server._list_tools_mcp(
+            mcp_types.ListToolsRequest()
+        ) == await fastmcp_server._list_tools_mcp(mcp_types.ListToolsRequest())
 
     async def test_call_tool_result_same_as_original(
         self, fastmcp_server: FastMCP, proxy_server: FastMCPProxy
@@ -371,10 +371,9 @@ class TestResources:
         assert wave_resource.icons == [Icon(src="https://example.com/wave-icon.png")]
 
     async def test_list_resources_same_as_original(self, fastmcp_server, proxy_server):
-        assert (
-            await proxy_server._list_resources_mcp()
-            == await fastmcp_server._list_resources_mcp()
-        )
+        assert await proxy_server._list_resources_mcp(
+            mcp_types.ListResourcesRequest()
+        ) == await fastmcp_server._list_resources_mcp(mcp_types.ListResourcesRequest())
 
     async def test_read_resource(self, proxy_server: FastMCPProxy):
         async with Client(proxy_server) as client:
@@ -489,8 +488,12 @@ class TestResourceTemplates:
     async def test_list_resource_templates_same_as_original(
         self, fastmcp_server, proxy_server
     ):
-        result = await fastmcp_server._list_resource_templates_mcp()
-        proxy_result = await proxy_server._list_resource_templates_mcp()
+        result = await fastmcp_server._list_resource_templates_mcp(
+            mcp_types.ListResourceTemplatesRequest()
+        )
+        proxy_result = await proxy_server._list_resource_templates_mcp(
+            mcp_types.ListResourceTemplatesRequest()
+        )
         assert proxy_result == result
 
     @pytest.mark.parametrize("id", [1, 2, 3])
diff --git a/tests/server/test_dependencies.py b/tests/server/test_dependencies.py
index 66b85dcac..2be1313f2 100644
--- a/tests/server/test_dependencies.py
+++ b/tests/server/test_dependencies.py
@@ -2,6 +2,7 @@
 
 from contextlib import asynccontextmanager, contextmanager
 
+import mcp.types as mcp_types
 import pytest
 from mcp.types import TextContent, TextResourceContents
 
@@ -135,8 +136,8 @@ async def test_dependencies_excluded_from_schema(mcp: FastMCP):
     ) -> str:
         return f"{name} is {age} years old"
 
-    tools = await mcp._list_tools_mcp()
-    tool = next(t for t in tools if t.name == "my_tool")
+    result = await mcp._list_tools_mcp(mcp_types.ListToolsRequest())
+    tool = next(t for t in result.tools if t.name == "my_tool")
 
     assert "name" in tool.inputSchema["properties"]
     assert "age" in tool.inputSchema["properties"]
@@ -466,8 +467,8 @@ async def test_connection_dependency_excluded_from_tool_schema(mcp: FastMCP):
     ) -> str:
         return name
 
-    tools = await mcp._list_tools_mcp()
-    tool = next(t for t in tools if t.name == "with_connection")
+    result = await mcp._list_tools_mcp(mcp_types.ListToolsRequest())
+    tool = next(t for t in result.tools if t.name == "with_connection")
 
     assert "name" in tool.inputSchema["properties"]
     assert "connection" not in tool.inputSchema["properties"]
@@ -588,8 +589,8 @@ async def test_external_user_cannot_override_dependency(mcp: FastMCP):
         return f"action={action},admin={admin}"
 
     # Verify dependency is NOT in the schema
-    tools = await mcp._list_tools_mcp()
-    tool = next(t for t in tools if t.name == "check_permission")
+    result = await mcp._list_tools_mcp(mcp_types.ListToolsRequest())
+    tool = next(t for t in result.tools if t.name == "check_permission")
     assert "admin" not in tool.inputSchema["properties"]
 
     # Normal call - dependency is resolved
@@ -859,8 +860,8 @@ class TestTransformContextAnnotations:
             return f"same={ctx1 is ctx2}"
 
         # Both ctx params should be excluded from schema
-        tools = await mcp._list_tools_mcp()
-        tool = next(t for t in tools if t.name == "tool_with_multiple_ctx")
+        result = await mcp._list_tools_mcp(mcp_types.ListToolsRequest())
+        tool = next(t for t in result.tools if t.name == "tool_with_multiple_ctx")
         assert "name" in tool.inputSchema["properties"]
         assert "ctx1" not in tool.inputSchema["properties"]
         assert "ctx2" not in tool.inputSchema["properties"]
diff --git a/tests/server/test_file_server.py b/tests/server/test_file_server.py
index 4b3eb9779..9bda34c10 100644
--- a/tests/server/test_file_server.py
+++ b/tests/server/test_file_server.py
@@ -76,10 +76,10 @@ def tools(mcp: FastMCP, test_dir: Path) -> FastMCP:
 
 
 async def test_list_resources(mcp: FastMCP):
-    resources = await mcp._list_resources_mcp()
-    assert len(resources) == 4
+    result = await mcp._list_resources_mcp(mcp_types.ListResourcesRequest())
+    assert len(result.resources) == 4
 
-    assert [str(r.uri) for r in resources] == [
+    assert [str(r.uri) for r in result.resources] == [
         "dir://test_dir",
         "file://test_dir/example.py",
         "file://test_dir/readme.md",
diff --git a/tests/server/test_pagination.py b/tests/server/test_pagination.py
new file mode 100644
index 000000000..fde5cccf3
--- /dev/null
+++ b/tests/server/test_pagination.py
@@ -0,0 +1,225 @@
+"""Tests for MCP pagination support."""
+
+from __future__ import annotations
+
+import pytest
+from mcp.shared.exceptions import McpError
+
+from fastmcp import Client, FastMCP
+from fastmcp.utilities.pagination import CursorState, paginate_sequence
+
+
+class TestCursorEncoding:
+    """Tests for cursor encoding/decoding."""
+
+    def test_encode_decode_roundtrip(self) -> None:
+        """Cursor should survive encode/decode roundtrip."""
+        state = CursorState(offset=100)
+        encoded = state.encode()
+        decoded = CursorState.decode(encoded)
+        assert decoded.offset == 100
+
+    def test_encode_produces_string(self) -> None:
+        """Encoded cursor should be a string."""
+        state = CursorState(offset=50)
+        encoded = state.encode()
+        assert isinstance(encoded, str)
+        assert len(encoded) > 0
+
+    def test_decode_invalid_base64_raises(self) -> None:
+        """Invalid base64 should raise ValueError."""
+        with pytest.raises(ValueError, match="Invalid cursor"):
+            CursorState.decode("not-valid-base64!!!")
+
+    def test_decode_invalid_json_raises(self) -> None:
+        """Valid base64 but invalid JSON should raise ValueError."""
+        import base64
+
+        invalid = base64.urlsafe_b64encode(b"not json").decode()
+        with pytest.raises(ValueError, match="Invalid cursor"):
+            CursorState.decode(invalid)
+
+    def test_decode_missing_offset_raises(self) -> None:
+        """JSON missing the offset key should raise ValueError."""
+        import base64
+        import json
+
+        invalid = base64.urlsafe_b64encode(json.dumps({"x": 1}).encode()).decode()
+        with pytest.raises(ValueError, match="Invalid cursor"):
+            CursorState.decode(invalid)
+
+
+class TestPaginateSequence:
+    """Tests for the paginate_sequence helper."""
+
+    def test_first_page_no_cursor(self) -> None:
+        """First page should start from beginning."""
+        items = list(range(25))
+        page, cursor = paginate_sequence(items, None, 10)
+        assert page == list(range(10))
+        assert cursor is not None
+
+    def test_second_page_with_cursor(self) -> None:
+        """Second page should continue from cursor."""
+        items = list(range(25))
+        _, cursor = paginate_sequence(items, None, 10)
+        page, next_cursor = paginate_sequence(items, cursor, 10)
+        assert page == list(range(10, 20))
+        assert next_cursor is not None
+
+    def test_last_page_returns_none_cursor(self) -> None:
+        """Last page should return None cursor."""
+        items = list(range(25))
+        _, c1 = paginate_sequence(items, None, 10)
+        _, c2 = paginate_sequence(items, c1, 10)
+        page, next_cursor = paginate_sequence(items, c2, 10)
+        assert page == list(range(20, 25))
+        assert next_cursor is None
+
+    def test_empty_list(self) -> None:
+        """Empty list should return empty page and no cursor."""
+        page, cursor = paginate_sequence([], None, 10)
+        assert page == []
+        assert cursor is None
+
+    def test_exact_page_size(self) -> None:
+        """List exactly matching page size should return no cursor."""
+        items = list(range(10))
+        page, cursor = paginate_sequence(items, None, 10)
+        assert page == items
+        assert cursor is None
+
+    def test_smaller_than_page_size(self) -> None:
+        """List smaller than page size should return all items."""
+        items = list(range(5))
+        page, cursor = paginate_sequence(items, None, 10)
+        assert page == items
+        assert cursor is None
+
+    def test_invalid_cursor_raises(self) -> None:
+        """Invalid cursor should raise ValueError."""
+        with pytest.raises(ValueError, match="Invalid cursor"):
+            paginate_sequence([1, 2, 3], "invalid!", 10)
+
+
+class TestServerPagination:
+    """Integration tests for server pagination."""
+
+    async def test_tools_pagination_returns_all_tools(self) -> None:
+        """Client should receive all tools across paginated requests."""
+        server = FastMCP(list_page_size=10)
+
+        for i in range(25):
+
+            @server.tool(name=f"tool_{i}")
+            def make_tool() -> str:
+                return "ok"
+
+        async with Client(server) as client:
+            tools = await client.list_tools()
+            assert len(tools) == 25
+            tool_names = {t.name for t in tools}
+            assert tool_names == {f"tool_{i}" for i in range(25)}
+
+    async def test_resources_pagination_returns_all_resources(self) -> None:
+        """Client should receive all resources across paginated requests."""
+        server = FastMCP(list_page_size=10)
+
+        for i in range(25):
+
+            @server.resource(f"test://resource_{i}")
+            def make_resource() -> str:
+                return "data"
+
+        async with Client(server) as client:
+            resources = await client.list_resources()
+            assert len(resources) == 25
+
+    async def test_prompts_pagination_returns_all_prompts(self) -> None:
+        """Client should receive all prompts across paginated requests."""
+        server = FastMCP(list_page_size=10)
+
+        for i in range(25):
+
+            @server.prompt(name=f"prompt_{i}")
+            def make_prompt() -> str:
+                return "text"
+
+        async with Client(server) as client:
+            prompts = await client.list_prompts()
+            assert len(prompts) == 25
+
+    async def test_manual_pagination(self) -> None:
+        """Client can manually paginate using cursor."""
+        server = FastMCP(list_page_size=10)
+
+        for i in range(25):
+
+            @server.tool(name=f"tool_{i}")
+            def make_tool() -> str:
+                return "ok"
+
+        async with Client(server) as client:
+            # First page
+            result = await client.list_tools_mcp()
+            assert len(result.tools) == 10
+            assert result.nextCursor is not None
+
+            # Second page
+            result2 = await client.list_tools_mcp(cursor=result.nextCursor)
+            assert len(result2.tools) == 10
+            assert result2.nextCursor is not None
+
+            # Third (last) page
+            result3 = await client.list_tools_mcp(cursor=result2.nextCursor)
+            assert len(result3.tools) == 5
+            assert result3.nextCursor is None
+
+    async def test_invalid_cursor_returns_error(self) -> None:
+        """Server should return MCP error for invalid cursor."""
+        server = FastMCP(list_page_size=10)
+
+        @server.tool
+        def my_tool() -> str:
+            return "ok"
+
+        async with Client(server) as client:
+            with pytest.raises(McpError) as exc:
+                await client.list_tools_mcp(cursor="invalid!")
+            assert exc.value.error.code == -32602
+
+    async def test_no_pagination_when_disabled(self) -> None:
+        """Without list_page_size, all items returned at once."""
+        server = FastMCP()  # No pagination
+
+        for i in range(25):
+
+            @server.tool(name=f"tool_{i}")
+            def make_tool() -> str:
+                return "ok"
+
+        async with Client(server) as client:
+            result = await client.list_tools_mcp()
+            assert len(result.tools) == 25
+            assert result.nextCursor is None
+
+    async def test_pagination_exact_page_boundary(self) -> None:
+        """Test pagination at exact page boundaries."""
+        server = FastMCP(list_page_size=10)
+
+        for i in range(20):  # Exactly 2 pages
+
+            @server.tool(name=f"tool_{i}")
+            def make_tool() -> str:
+                return "ok"
+
+        async with Client(server) as client:
+            # First page
+            result = await client.list_tools_mcp()
+            assert len(result.tools) == 10
+            assert result.nextCursor is not None
+
+            # Second (last) page
+            result2 = await client.list_tools_mcp(cursor=result.nextCursor)
+            assert len(result2.tools) == 10
+            assert result2.nextCursor is None
diff --git a/tests/server/test_tool_annotations.py b/tests/server/test_tool_annotations.py
index e90cde1ea..2649c282e 100644
--- a/tests/server/test_tool_annotations.py
+++ b/tests/server/test_tool_annotations.py
@@ -1,5 +1,6 @@
 from typing import Any
 
+import mcp.types as mcp_types
 from mcp.types import Tool as MCPTool
 from mcp.types import ToolAnnotations, ToolExecution
 
@@ -47,12 +48,12 @@ async def test_tool_annotations_in_mcp_protocol():
         return message
 
     # Check via MCP protocol
-    mcp_tools = await mcp._list_tools_mcp()
-    assert len(mcp_tools) == 1
-    assert mcp_tools[0].annotations is not None
-    assert mcp_tools[0].annotations.title == "Echo Tool"
-    assert mcp_tools[0].annotations.readOnlyHint is True
-    assert mcp_tools[0].annotations.openWorldHint is False
+    result = await mcp._list_tools_mcp(mcp_types.ListToolsRequest())
+    assert len(result.tools) == 1
+    assert result.tools[0].annotations is not None
+    assert result.tools[0].annotations.title == "Echo Tool"
+    assert result.tools[0].annotations.readOnlyHint is True
+    assert result.tools[0].annotations.openWorldHint is False
 
 
 async def test_tool_annotations_in_client_api():

From 96d4b1b606fca7c4495790434b852da151267c0b Mon Sep 17 00:00:00 2001
From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
Date: Sat, 17 Jan 2026 19:30:33 -0500
Subject: [PATCH 2/3] Add pagination documentation

---
 docs/clients/prompts.mdx    |  6 ++-
 docs/clients/resources.mdx  | 12 +++--
 docs/clients/tools.mdx      |  6 ++-
 docs/docs.json              |  1 +
 docs/servers/pagination.mdx | 87 +++++++++++++++++++++++++++++++++++++
 docs/servers/server.mdx     |  5 +++
 6 files changed, 109 insertions(+), 8 deletions(-)
 create mode 100644 docs/servers/pagination.mdx

diff --git a/docs/clients/prompts.mdx b/docs/clients/prompts.mdx
index 84dcdd9f5..53ce27738 100644
--- a/docs/clients/prompts.mdx
+++ b/docs/clients/prompts.mdx
@@ -13,13 +13,13 @@ Prompts are reusable message templates exposed by MCP servers. They can accept a
 
 ## Listing Prompts
 
-Use `list_prompts()` to retrieve all available prompt templates:
+Use `list_prompts()` to retrieve all available prompt templates. When the server paginates results, the client automatically fetches all pages and returns the complete list.
 
 ```python
 async with client:
     prompts = await client.list_prompts()
     # prompts -> list[mcp.types.Prompt]
-    
+
     for prompt in prompts:
         print(f"Prompt: {prompt.name}")
         print(f"Description: {prompt.description}")
@@ -31,6 +31,8 @@ async with client:
             print(f"Tags: {fastmcp_meta.get('tags', [])}")
 ```
 
+For manual pagination control, use `list_prompts_mcp()` with the `cursor` parameter. See [Pagination](/servers/pagination#manual-pagination) for details.
+
 ### Filtering by Tags
 
 <VersionBadge version="2.11.0" />
diff --git a/docs/clients/resources.mdx b/docs/clients/resources.mdx
index 865d669cc..9eb99a265 100644
--- a/docs/clients/resources.mdx
+++ b/docs/clients/resources.mdx
@@ -22,13 +22,13 @@ MCP servers expose two types of resources:
 
 ### Static Resources
 
-Use `list_resources()` to retrieve all static resources available on the server:
+Use `list_resources()` to retrieve all static resources available on the server. When the server paginates results, the client automatically fetches all pages and returns the complete list.
 
 ```python
 async with client:
     resources = await client.list_resources()
     # resources -> list[mcp.types.Resource]
-    
+
     for resource in resources:
         print(f"Resource URI: {resource.uri}")
         print(f"Name: {resource.name}")
@@ -40,15 +40,17 @@ async with client:
             print(f"Tags: {fastmcp_meta.get('tags', [])}")
 ```
 
+For manual pagination control, use `list_resources_mcp()` with the `cursor` parameter. See [Pagination](/servers/pagination#manual-pagination) for details.
+
 ### Resource Templates
 
-Use `list_resource_templates()` to retrieve available resource templates:
+Use `list_resource_templates()` to retrieve available resource templates. When the server paginates results, the client automatically fetches all pages and returns the complete list.
 
 ```python
 async with client:
     templates = await client.list_resource_templates()
     # templates -> list[mcp.types.ResourceTemplate]
-    
+
     for template in templates:
         print(f"Template URI: {template.uriTemplate}")
         print(f"Name: {template.name}")
@@ -59,6 +61,8 @@ async with client:
             print(f"Tags: {fastmcp_meta.get('tags', [])}")
 ```
 
+For manual pagination control, use `list_resource_templates_mcp()` with the `cursor` parameter. See [Pagination](/servers/pagination#manual-pagination) for details.
+
 ### Filtering by Tags
 
 <VersionBadge version="2.11.0" />
diff --git a/docs/clients/tools.mdx b/docs/clients/tools.mdx
index a01e2dd66..7fad067e3 100644
--- a/docs/clients/tools.mdx
+++ b/docs/clients/tools.mdx
@@ -13,13 +13,13 @@ Tools are executable functions exposed by MCP servers. The FastMCP client provid
 
 ## Discovering Tools
 
-Use `list_tools()` to retrieve all tools available on the server:
+Use `list_tools()` to retrieve all tools available on the server. When the server paginates results, the client automatically fetches all pages and returns the complete list.
 
 ```python
 async with client:
     tools = await client.list_tools()
     # tools -> list[mcp.types.Tool]
-    
+
     for tool in tools:
         print(f"Tool: {tool.name}")
         print(f"Description: {tool.description}")
@@ -31,6 +31,8 @@ async with client:
             print(f"Tags: {fastmcp_meta.get('tags', [])}")
 ```
 
+For manual pagination control, use `list_tools_mcp()` with the `cursor` parameter. See [Pagination](/servers/pagination#manual-pagination) for details.
+
 ### Filtering by Tags
 
 <VersionBadge version="2.11.0" />
diff --git a/docs/docs.json b/docs/docs.json
index f6b0649d1..ae934798c 100644
--- a/docs/docs.json
+++ b/docs/docs.json
@@ -127,6 +127,7 @@
                   "servers/lifespan",
                   "servers/logging",
                   "servers/middleware",
+                  "servers/pagination",
                   "servers/progress",
                   "servers/sampling",
                   "servers/storage-backends",
diff --git a/docs/servers/pagination.mdx b/docs/servers/pagination.mdx
new file mode 100644
index 000000000..ecdf64739
--- /dev/null
+++ b/docs/servers/pagination.mdx
@@ -0,0 +1,87 @@
+---
+title: Pagination
+sidebarTitle: Pagination
+description: Control how servers return large lists of components to clients.
+icon: page
+---
+
+import { VersionBadge } from '/snippets/version-badge.mdx'
+
+<VersionBadge version="3.0.0" />
+
+When a server exposes many tools, resources, or prompts, returning them all in a single response can be impractical. MCP supports pagination for list operations, allowing servers to return results in manageable chunks that clients can fetch incrementally.
+
+## Server Configuration
+
+By default, FastMCP servers return all components in a single response for backward compatibility. To enable pagination, set the `list_page_size` parameter when creating your server. This value determines the maximum number of items returned per page across all list operations.
+
+```python
+from fastmcp import FastMCP
+
+# Enable pagination with 50 items per page
+server = FastMCP("ComponentRegistry", list_page_size=50)
+
+# Register many tools, resources, prompts...
+for i in range(200):
+    @server.tool(name=f"tool_{i}")
+    def make_tool(x: int) -> int:
+        return x * i
+```
+
+When `list_page_size` is configured, the `tools/list`, `resources/list`, `resources/templates/list`, and `prompts/list` endpoints all paginate their responses. Each response includes a `nextCursor` field when more results exist, which clients use to fetch subsequent pages.
+
+### Cursor Format
+
+Cursors are opaque base64-encoded strings per the MCP specification. Clients should treat them as black boxes, passing them unchanged between requests. The cursor encodes the offset into the result set, but this is an implementation detail that may change.
+
+## Client Behavior
+
+The FastMCP Client handles pagination transparently. Convenience methods like `list_tools()`, `list_resources()`, `list_resource_templates()`, and `list_prompts()` automatically fetch all pages and return the complete list. Existing code continues to work without modification.
+
+```python
+from fastmcp import Client
+
+async with Client(server) as client:
+    # Returns all 200 tools, fetching pages automatically
+    tools = await client.list_tools()
+    print(f"Total tools: {len(tools)}")  # 200
+```
+
+### Manual Pagination
+
+For scenarios where you want to process results incrementally (memory-constrained environments, progress reporting, or early termination), use the `_mcp` variants with explicit cursor handling.
+
+```python
+from fastmcp import Client
+
+async with Client(server) as client:
+    # Fetch first page
+    result = await client.list_tools_mcp()
+    print(f"Page 1: {len(result.tools)} tools")
+
+    # Continue fetching while more pages exist
+    while result.nextCursor:
+        result = await client.list_tools_mcp(cursor=result.nextCursor)
+        print(f"Next page: {len(result.tools)} tools")
+```
+
+The `_mcp` methods return the raw MCP protocol objects, which include both the items and the `nextCursor` for the next page. When `nextCursor` is `None`, you've reached the end of the result set.
+
+All four list operations support manual pagination:
+
+| Operation | Convenience Method | Manual Method |
+|-----------|-------------------|---------------|
+| Tools | `list_tools()` | `list_tools_mcp(cursor=...)` |
+| Resources | `list_resources()` | `list_resources_mcp(cursor=...)` |
+| Resource Templates | `list_resource_templates()` | `list_resource_templates_mcp(cursor=...)` |
+| Prompts | `list_prompts()` | `list_prompts_mcp(cursor=...)` |
+
+## When to Use Pagination
+
+Pagination becomes valuable when your server exposes a large number of components. Consider enabling it when:
+
+- Your server dynamically generates many components (e.g., from a database or file system)
+- Memory usage is a concern for clients
+- You want to reduce initial response latency
+
+For servers with a fixed, modest number of components (fewer than 100), pagination adds complexity without meaningful benefit. The default behavior of returning everything in one response is simpler and efficient for typical use cases.
diff --git a/docs/servers/server.mdx b/docs/servers/server.mdx
index 777c7adf1..7f5c6fa41 100644
--- a/docs/servers/server.mdx
+++ b/docs/servers/server.mdx
@@ -95,6 +95,11 @@ The `FastMCP` constructor accepts several arguments:
   Controls how tool input parameters are validated. When `False` (default), FastMCP uses Pydantic's flexible validation that coerces compatible inputs (e.g., `"10"` → `10` for int parameters). When `True`, uses the MCP SDK's JSON Schema validation to validate inputs against the exact schema before passing them to your function, rejecting any type mismatches. The default mode improves compatibility with LLM clients while maintaining type safety. See [Input Validation Modes](/servers/tools#input-validation-modes) for details
 </ParamField>
 
+<ParamField body="list_page_size" type="int | None" default="None">
+  <VersionBadge version="3.0.0" />
+  Maximum number of items per page for list operations (`tools/list`, `resources/list`, etc.). When `None` (default), all results are returned in a single response. When set, responses are paginated and include a `nextCursor` for fetching additional pages. See [Pagination](/servers/pagination) for details
+</ParamField>
+
 </Card>
 ## Components
 

From 17d43d7991891dad357767efb0ce26d14f9f9099 Mon Sep 17 00:00:00 2001
From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
Date: Sat, 17 Jan 2026 19:50:33 -0500
Subject: [PATCH 3/3] Address review feedback: validation and edge cases

---
 docs/servers/pagination.mdx         | 15 ++++++++-----
 src/fastmcp/server/context.py       | 34 ++++++++++++++++++++++++-----
 src/fastmcp/server/server.py        |  2 ++
 src/fastmcp/utilities/pagination.py |  9 +++++++-
 tests/server/test_pagination.py     | 18 +++++++++++++++
 5 files changed, 66 insertions(+), 12 deletions(-)

diff --git a/docs/servers/pagination.mdx b/docs/servers/pagination.mdx
index ecdf64739..8c1002c7c 100644
--- a/docs/servers/pagination.mdx
+++ b/docs/servers/pagination.mdx
@@ -21,11 +21,16 @@ from fastmcp import FastMCP
 # Enable pagination with 50 items per page
 server = FastMCP("ComponentRegistry", list_page_size=50)
 
-# Register many tools, resources, prompts...
-for i in range(200):
-    @server.tool(name=f"tool_{i}")
-    def make_tool(x: int) -> int:
-        return x * i
+# Register tools (in practice, these might come from a database or config)
+@server.tool
+def search(query: str) -> str:
+    return f"Results for: {query}"
+
+@server.tool
+def analyze(data: str) -> dict:
+    return {"status": "analyzed", "data": data}
+
+# ... many more tools, resources, prompts
 ```
 
 When `list_page_size` is configured, the `tools/list`, `resources/list`, `resources/templates/list`, and `prompts/list` endpoints all paginate their responses. Each response includes a `nextCursor` field when more results exist, which clients use to fetch subsequent pages.
diff --git a/src/fastmcp/server/context.py b/src/fastmcp/server/context.py
index aa5cfe658..7b1f2f116 100644
--- a/src/fastmcp/server/context.py
+++ b/src/fastmcp/server/context.py
@@ -346,10 +346,20 @@ class Context:
         Returns:
             List of Resource objects available on the server
         """
-        result = await self.fastmcp._list_resources_mcp(
-            mcp.types.ListResourcesRequest()
-        )
-        return result.resources
+        all_resources: list[SDKResource] = []
+        cursor: str | None = None
+        while True:
+            request = mcp.types.ListResourcesRequest(
+                params=mcp.types.ListResourcesRequestParams(cursor=cursor)
+                if cursor
+                else None
+            )
+            result = await self.fastmcp._list_resources_mcp(request)
+            all_resources.extend(result.resources)
+            if result.nextCursor is None:
+                break
+            cursor = result.nextCursor
+        return all_resources
 
     async def list_prompts(self) -> list[SDKPrompt]:
         """List all available prompts from the server.
@@ -357,8 +367,20 @@ class Context:
         Returns:
             List of Prompt objects available on the server
         """
-        result = await self.fastmcp._list_prompts_mcp(mcp.types.ListPromptsRequest())
-        return result.prompts
+        all_prompts: list[SDKPrompt] = []
+        cursor: str | None = None
+        while True:
+            request = mcp.types.ListPromptsRequest(
+                params=mcp.types.ListPromptsRequestParams(cursor=cursor)
+                if cursor
+                else None
+            )
+            result = await self.fastmcp._list_prompts_mcp(request)
+            all_prompts.extend(result.prompts)
+            if result.nextCursor is None:
+                break
+            cursor = result.nextCursor
+        return all_prompts
 
     async def get_prompt(
         self, name: str, arguments: dict[str, Any] | None = None
diff --git a/src/fastmcp/server/server.py b/src/fastmcp/server/server.py
index 114dddece..c2f7ce297 100644
--- a/src/fastmcp/server/server.py
+++ b/src/fastmcp/server/server.py
@@ -387,6 +387,8 @@ class FastMCP(Provider, Generic[LifespanResultT]):
         )
 
         # Store list_page_size for pagination of list operations
+        if list_page_size is not None and list_page_size <= 0:
+            raise ValueError("list_page_size must be a positive integer")
         self._list_page_size: int | None = list_page_size
 
         if tool_serializer is not None and fastmcp.settings.deprecation_warnings:
diff --git a/src/fastmcp/utilities/pagination.py b/src/fastmcp/utilities/pagination.py
index b1df8d55f..48e2e45bb 100644
--- a/src/fastmcp/utilities/pagination.py
+++ b/src/fastmcp/utilities/pagination.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import base64
+import binascii
 import json
 from collections.abc import Sequence
 from dataclasses import dataclass
@@ -36,7 +37,13 @@ class CursorState:
         try:
             data = json.loads(base64.urlsafe_b64decode(cursor.encode()).decode())
             return cls(offset=data["o"])
-        except (json.JSONDecodeError, KeyError, ValueError, TypeError) as e:
+        except (
+            json.JSONDecodeError,
+            KeyError,
+            ValueError,
+            TypeError,
+            binascii.Error,
+        ) as e:
             raise ValueError(f"Invalid cursor: {cursor}") from e
 
 
diff --git a/tests/server/test_pagination.py b/tests/server/test_pagination.py
index fde5cccf3..c1387fb46 100644
--- a/tests/server/test_pagination.py
+++ b/tests/server/test_pagination.py
@@ -223,3 +223,21 @@ class TestServerPagination:
             result2 = await client.list_tools_mcp(cursor=result.nextCursor)
             assert len(result2.tools) == 10
             assert result2.nextCursor is None
+
+
+class TestPageSizeValidation:
+    """Tests for list_page_size validation."""
+
+    def test_zero_page_size_raises(self) -> None:
+        """Zero page size should raise ValueError."""
+        with pytest.raises(
+            ValueError, match="list_page_size must be a positive integer"
+        ):
+            FastMCP(list_page_size=0)
+
+    def test_negative_page_size_raises(self) -> None:
+        """Negative page size should raise ValueError."""
+        with pytest.raises(
+            ValueError, match="list_page_size must be a positive integer"
+        ):
+            FastMCP(list_page_size=-1)