ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-22 21:44:18 +02:00
Code Issues Projects Releases Packages Wiki Activity

Merge pull request #2903 from jlowin/feature/pagination-support

Browse source
This commit is contained in:
Jeremiah Lowin 2026-01-17 20:03:03 -05:00 • committed by GitHub
commit 2eac7ab1d1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
16 changed files with 710 additions and 111 deletions
Download patch file Download diff file Expand all files Collapse all files

6
docs/clients/prompts.mdx
View file

@ -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" />

12
docs/clients/resources.mdx
View file

@ -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" />

6
docs/clients/tools.mdx
View file

@ -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" />

1
docs/docs.json
View file

@ -127,6 +127,7 @@
"servers/lifespan",
"servers/logging",
"servers/middleware",
"servers/pagination",
"servers/progress",
"servers/sampling",
"servers/storage-backends",

92
docs/servers/pagination.mdx Normal file
View file

@ -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'
<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 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.

5
docs/servers/server.mdx
View file

@ -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

127
src/fastmcp/client/client.py
View file

@ -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 ---

30
src/fastmcp/server/context.py
View file

@ -346,7 +346,20 @@ class Context:
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 @@ class Context:
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

133
src/fastmcp/server/server.py
View file

@ -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,11 @@ class FastMCP(Provider, Generic[LifespanResultT]):
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 @@ 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 +790,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 +1913,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 +1978,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 +2011,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]

80
src/fastmcp/utilities/pagination.py Normal file
View file

@ -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

27
tests/server/auth/test_authorization.py
View file

@ -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)

23
tests/server/providers/proxy/test_proxy_server.py
View file

@ -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])

17
tests/server/test_dependencies.py
View file

@ -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"]

6
tests/server/test_file_server.py
View file

@ -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",

243
tests/server/test_pagination.py Normal file
View file

@ -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)

13
tests/server/test_tool_annotations.py
View file

@ -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():