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

Add MCP-compliant pagination for list operations

Browse source 
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.
This commit is contained in:
Jeremiah Lowin 2026-01-17 19:30:24 -05:00
commit dada8893a9
10 changed files with 547 additions and 103 deletions
Download patch file Download diff file Expand all files Collapse all files

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

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

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

131
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,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]

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

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

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

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

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

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