Skip to content

feat: add paginated list decorators for prompts, resources, and tools #1286

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 4 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions .pre-commit-config.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -25,22 +25,22 @@ repos:
hooks:
- id: ruff-format
name: Ruff Format
entry: uv run ruff
entry: uv run --frozen ruff
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same for below

Suggested change
entry: uv run --frozen ruff
entry: ruff

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

good idea, updated

args: [format]
language: system
types: [python]
pass_filenames: false
- id: ruff
name: Ruff
entry: uv run ruff
entry: uv run --frozen ruff
args: ["check", "--fix", "--exit-non-zero-on-fix"]
types: [python]
language: system
pass_filenames: false
exclude: ^README\.md$
- id: pyright
name: pyright
entry: uv run pyright
entry: uv run --frozen pyright
language: system
types: [python]
pass_filenames: false
Expand All @@ -52,7 +52,7 @@ repos:
pass_filenames: false
- id: readme-snippets
name: Check README snippets are up to date
entry: uv run scripts/update_readme_snippets.py --check
entry: uv run --frozen scripts/update_readme_snippets.py --check
language: system
files: ^(README\.md|examples/.*\.py|scripts/update_readme_snippets\.py)$
pass_filenames: false
116 changes: 116 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@
- [Mounting to an Existing ASGI Server](#mounting-to-an-existing-asgi-server)
- [Advanced Usage](#advanced-usage)
- [Low-Level Server](#low-level-server)
- [Pagination (Advanced)](#pagination-advanced)
- [Writing MCP Clients](#writing-mcp-clients)
- [Client Display Utilities](#client-display-utilities)
- [OAuth Authentication for Clients](#oauth-authentication-for-clients)
Expand Down Expand Up @@ -1727,6 +1728,121 @@ Tools can return data in three ways:

When an `outputSchema` is defined, the server automatically validates the structured output against the schema. This ensures type safety and helps catch errors early.

### Pagination (Advanced)

For servers that need to handle large datasets, the low-level server provides paginated versions of list operations. This is an optional optimization - most servers won't need pagination unless they're dealing with hundreds or thousands of items.

#### Server-side Implementation

<!-- snippet-source examples/snippets/servers/pagination_example.py -->
```python
"""
Example of implementing pagination with MCP server decorators.
"""

from pydantic import AnyUrl

import mcp.types as types
from mcp.server.lowlevel import Server

# Initialize the server
server = Server("paginated-server")

# Sample data to paginate
ITEMS = [f"Item {i}" for i in range(1, 101)] # 100 items


@server.list_resources_paginated()
async def list_resources_paginated(cursor: types.Cursor | None) -> types.ListResourcesResult:
"""List resources with pagination support."""
page_size = 10

# Parse cursor to get offset
start = 0 if cursor is None else int(cursor)
end = start + page_size

# Get page of resources
page_items = [
types.Resource(uri=AnyUrl(f"resource://items/{item}"), name=item, description=f"Description for {item}")
for item in ITEMS[start:end]
]

# Determine next cursor
next_cursor = str(end) if end < len(ITEMS) else None

return types.ListResourcesResult(resources=page_items, nextCursor=next_cursor)
```

_Full example: [examples/snippets/servers/pagination_example.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/pagination_example.py)_
<!-- /snippet-source -->

Similar decorators are available for all list operations:

- `@server.list_tools_paginated()` - for paginating tools
- `@server.list_resources_paginated()` - for paginating resources
- `@server.list_prompts_paginated()` - for paginating prompts

#### Client-side Consumption

<!-- snippet-source examples/snippets/clients/pagination_client.py -->
```python
"""
Example of consuming paginated MCP endpoints from a client.
"""

import asyncio

from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client
from mcp.types import Resource


async def list_all_resources() -> None:
"""Fetch all resources using pagination."""
async with stdio_client(StdioServerParameters(command="uv", args=["run", "mcp-simple-pagination"])) as (
read,
write,
):
async with ClientSession(read, write) as session:
await session.initialize()

all_resources: list[Resource] = []
cursor = None

while True:
# Fetch a page of resources
result = await session.list_resources(cursor=cursor)
all_resources.extend(result.resources)

print(f"Fetched {len(result.resources)} resources")

# Check if there are more pages
if result.nextCursor:
cursor = result.nextCursor
else:
break

print(f"Total resources: {len(all_resources)}")


if __name__ == "__main__":
asyncio.run(list_all_resources())
```

_Full example: [examples/snippets/clients/pagination_client.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/clients/pagination_client.py)_
<!-- /snippet-source -->

#### Key Points

- **Cursors are opaque strings** - the server defines the format (numeric offsets, timestamps, etc.)
- **Return `nextCursor=None`** when there are no more pages
- **Backward compatible** - clients that don't support pagination will still work (they'll just get the first page)
- **Flexible page sizes** - Each endpoint can define its own page size based on data characteristics

> **NOTE**: The paginated decorators (`list_tools_paginated()`, `list_resources_paginated()`, `list_prompts_paginated()`) are mutually exclusive with their non-paginated counterparts and cannot be used together on the same server instance.

See the [simple-pagination example](examples/servers/simple-pagination) for a complete implementation.

### Writing MCP Clients

The SDK provides a high-level client interface for connecting to MCP servers using various [transports](https://modelcontextprotocol.io/specification/2025-03-26/basic/transports):
Expand Down
77 changes: 77 additions & 0 deletions examples/servers/simple-pagination/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,77 @@
# MCP Simple Pagination

A simple MCP server demonstrating pagination for tools, resources, and prompts using cursor-based pagination.

## Usage

Start the server using either stdio (default) or SSE transport:

```bash
# Using stdio transport (default)
uv run mcp-simple-pagination

# Using SSE transport on custom port
uv run mcp-simple-pagination --transport sse --port 8000
```

The server exposes:

- 25 tools (paginated, 5 per page)
- 30 resources (paginated, 10 per page)
- 20 prompts (paginated, 7 per page)

Each paginated list returns a `nextCursor` when more pages are available. Use this cursor in subsequent requests to retrieve the next page.

## Example

Using the MCP client, you can retrieve paginated items like this using the STDIO transport:

```python
import asyncio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client


async def main():
async with stdio_client(
StdioServerParameters(command="uv", args=["run", "mcp-simple-pagination"])
) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()

# Get first page of tools
tools_page1 = await session.list_tools()
print(f"First page: {len(tools_page1.tools)} tools")
print(f"Next cursor: {tools_page1.nextCursor}")

# Get second page using cursor
if tools_page1.nextCursor:
tools_page2 = await session.list_tools(cursor=tools_page1.nextCursor)
print(f"Second page: {len(tools_page2.tools)} tools")

# Similarly for resources
resources_page1 = await session.list_resources()
print(f"First page: {len(resources_page1.resources)} resources")

# And for prompts
prompts_page1 = await session.list_prompts()
print(f"First page: {len(prompts_page1.prompts)} prompts")


asyncio.run(main())
```

## Pagination Details

The server uses simple numeric indices as cursors for demonstration purposes. In production scenarios, you might use:

- Database offsets or row IDs
- Timestamps for time-based pagination
- Opaque tokens encoding pagination state

The pagination implementation demonstrates:

- Handling `None` cursor for the first page
- Returning `nextCursor` when more data exists
- Gracefully handling invalid cursors
- Different page sizes for different resource types
Empty file.
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
import sys

from .server import main

sys.exit(main()) # type: ignore[call-arg]
Loading
Loading