diff --git a/docs/clients/prompts.mdx b/docs/clients/prompts.mdx
index 84dcdd9f5d..53ce27738b 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
diff --git a/docs/clients/resources.mdx b/docs/clients/resources.mdx
index 865d669cc9..9eb99a265b 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
diff --git a/docs/clients/tools.mdx b/docs/clients/tools.mdx
index a01e2dd662..7fad067e34 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
diff --git a/docs/docs.json b/docs/docs.json
index f6b0649d14..ae934798cb 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 0000000000..8c1002c7c1
--- /dev/null
+++ b/docs/servers/pagination.mdx
@@ -0,0 +1,92 @@
+---
+title: Pagination
+sidebarTitle: Pagination
+description: Control how servers return large lists of components to clients.
+icon: page
+---
+
+import { VersionBadge } from '/snippets/version-badge.mdx'
+
+
+
+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 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.
+
+### 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 777c7adf16..7f5c6fa419 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
+
+
+ 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
+
+
## Components
diff --git a/src/fastmcp/client/client.py b/src/fastmcp/client/client.py
index e7a287112e..5809e51bff 100644
--- a/src/fastmcp/client/client.py
+++ b/src/fastmcp/client/client.py
@@ -796,9 +796,14 @@ async def send_roots_list_changed(self) -> None:
# --- 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 @@ async def list_resources_mcp(self) -> mcp.types.ListResourcesResult:
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 @@ async def list_resource_templates_mcp(
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 @@ async def _read_resource_as_task(
# --- 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 @@ async def list_prompts_mcp(self) -> mcp.types.ListPromptsResult:
"""
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 @@ async def complete(
# --- 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 @@ async def list_tools_mcp(self) -> mcp.types.ListToolsResult:
"""
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 16e240ff5a..7b1f2f1160 100644
--- a/src/fastmcp/server/context.py
+++ b/src/fastmcp/server/context.py
@@ -346,7 +346,20 @@ async def list_resources(self) -> list[SDKResource]:
Returns:
List of Resource objects available on the server
"""
- return await self.fastmcp._list_resources_mcp()
+ 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.
@@ -354,7 +367,20 @@ async def list_prompts(self) -> list[SDKPrompt]:
Returns:
List of Prompt objects available on the server
"""
- return await self.fastmcp._list_prompts_mcp()
+ 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 1cde0cc4a9..c2f7ce2974 100644
--- a/src/fastmcp/server/server.py
+++ b/src/fastmcp/server/server.py
@@ -43,10 +43,6 @@
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.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 @@ async def wrap(
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 @@ def __init__(
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,11 @@ def __init__(
else fastmcp.settings.mask_error_details
)
+ # 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:
warnings.warn(
"The `tool_serializer` parameter is deprecated. "
@@ -738,16 +761,26 @@ def run(
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 +790,17 @@ def _setup_handlers(self) -> None:
# 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 +1913,56 @@ def _get_additional_http_routes(self) -> list[BaseRoute]:
"""
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 +1978,24 @@ async def _list_resource_templates_mcp(self) -> list[SDKResourceTemplate]:
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 +2011,12 @@ async def _list_prompts_mcp(self) -> list[SDKPrompt]:
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 0000000000..48e2e45bb5
--- /dev/null
+++ b/src/fastmcp/utilities/pagination.py
@@ -0,0 +1,80 @@
+"""Pagination utilities for MCP list operations."""
+
+from __future__ import annotations
+
+import base64
+import binascii
+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,
+ binascii.Error,
+ ) 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 e0de6aa150..e1ee8216e6 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 @@ def public_tool() -> str:
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 @@ def public_tool() -> str:
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 @@ def api_tool() -> str:
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 @@ def api_tool() -> str:
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 @@ def admin_tool() -> str:
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 ecad2062d3..d047201dd9 100644
--- a/tests/server/providers/proxy/test_proxy_server.py
+++ b/tests/server/providers/proxy/test_proxy_server.py
@@ -2,6 +2,7 @@
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 @@ async def test_tool_without_description(self, proxy_server):
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 @@ async def test_get_resources_meta(self, proxy_server):
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 @@ async def test_get_resource_templates_meta(self, proxy_server):
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 66b85dcac1..2be1313f27 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 my_tool(
) -> 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 with_connection(
) -> 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 check_permission(
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 @@ async def tool_with_multiple_ctx(
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 4b3eb97791..9bda34c106 100644
--- a/tests/server/test_file_server.py
+++ b/tests/server/test_file_server.py
@@ -76,10 +76,10 @@ def delete_file(path: str) -> bool:
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 0000000000..c1387fb464
--- /dev/null
+++ b/tests/server/test_pagination.py
@@ -0,0 +1,243 @@
+"""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
+
+
+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)
diff --git a/tests/server/test_tool_annotations.py b/tests/server/test_tool_annotations.py
index e90cde1eab..2649c282e2 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 @@ def echo(message: str) -> str:
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():