switch pagination to single decorator with callback inspection

Author: maxisbey

README.md

@@ -1762,7 +1762,7 @@ server = Server("paginated-server")
 ITEMS = [f"Item {i}" for i in range(1, 101)]  # 100 items
 
 
-@server.list_resources_paginated()
+@server.list_resources()
 async def list_resources_paginated(cursor: types.Cursor | None) -> types.ListResourcesResult:
     """List resources with pagination support."""
     page_size = 10
@@ -1786,12 +1786,6 @@ async def list_resources_paginated(cursor: types.Cursor | None) -> types.ListRes
 _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 -->
@@ -1849,8 +1843,6 @@ _Full example: [examples/snippets/clients/pagination_client.py](https://github.c
 - **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

examples/servers/simple-pagination/mcp_simple_pagination/server.py

@@ -58,7 +58,7 @@ def main(port: int, transport: str) -> int:
     app = Server("mcp-simple-pagination")
 
     # Paginated list_tools - returns 5 tools per page
-    @app.list_tools_paginated()
+    @app.list_tools()
     async def list_tools_paginated(cursor: types.Cursor | None) -> types.ListToolsResult:
         page_size = 5
 
@@ -84,7 +84,7 @@ async def list_tools_paginated(cursor: types.Cursor | None) -> types.ListToolsRe
         return types.ListToolsResult(tools=page_tools, nextCursor=next_cursor)
 
     # Paginated list_resources - returns 10 resources per page
-    @app.list_resources_paginated()
+    @app.list_resources()
     async def list_resources_paginated(
         cursor: types.Cursor | None,
     ) -> types.ListResourcesResult:
@@ -112,7 +112,7 @@ async def list_resources_paginated(
         return types.ListResourcesResult(resources=page_resources, nextCursor=next_cursor)
 
     # Paginated list_prompts - returns 7 prompts per page
-    @app.list_prompts_paginated()
+    @app.list_prompts()
     async def list_prompts_paginated(
         cursor: types.Cursor | None,
     ) -> types.ListPromptsResult:

examples/snippets/servers/pagination_example.py

@@ -14,7 +14,7 @@
 ITEMS = [f"Item {i}" for i in range(1, 101)]  # 100 items
 
 
-@server.list_resources_paginated()
+@server.list_resources()
 async def list_resources_paginated(cursor: types.Cursor | None) -> types.ListResourcesResult:
     """List resources with pagination support."""
     page_size = 10

src/mcp/server/lowlevel/func_inspection.py

@@ -0,0 +1,54 @@
+import inspect
+from collections.abc import Callable
+from typing import Any
+
+
+def accepts_cursor(func: Callable[..., Any]) -> bool:
+    """
+    True if the function accepts a cursor parameter call, otherwise false.
+
+    `accepts_cursor` does not validate that the function will work. For
+    example, if `func` contains keyword-only arguments with no defaults,
+    then it will not work when used in the `lowlevel/server.py` code, but
+    this function will not raise an exception.
+    """
+    try:
+        sig = inspect.signature(func)
+    except (ValueError, TypeError):
+        return False
+
+    params = dict(sig.parameters.items())
+
+    method = inspect.ismethod(func)
+
+    if method:
+        params.pop("self", None)
+        params.pop("cls", None)
+
+    if len(params) == 0:
+        # No parameters at all - can't accept cursor
+        return False
+
+    # Check if ALL remaining parameters are keyword-only
+    all_keyword_only = all(param.kind == inspect.Parameter.KEYWORD_ONLY for param in params.values())
+
+    if all_keyword_only:
+        # If all params are keyword-only, check if they ALL have defaults
+        # If they do, the function can be called with no arguments -> no cursor
+        all_have_defaults = all(param.default is not inspect.Parameter.empty for param in params.values())
+        return not all_have_defaults  # False if all have defaults (no cursor), True otherwise
+
+    # Check if the ONLY parameter is **kwargs (VAR_KEYWORD)
+    # A function with only **kwargs can't accept a positional cursor argument
+    if len(params) == 1:
+        only_param = next(iter(params.values()))
+        if only_param.kind == inspect.Parameter.VAR_KEYWORD:
+            return False  # Can't pass positional cursor to **kwargs
+
+    # Has at least one positional or variadic parameter - can accept cursor
+    # Important note: this is designed to _not_ handle the situation where
+    # there are multiple keyword only arguments with no defaults. In those
+    # situations it's an invalid handler function, and will error. But it's
+    # not the responsibility of this function to check the validity of a
+    # callback.
+    return True

src/mcp/server/lowlevel/server.py

@@ -82,6 +82,7 @@ async def main():
 from typing_extensions import TypeVar
 
 import mcp.types as types
+from mcp.server.lowlevel.func_inspection import accepts_cursor
 from mcp.server.lowlevel.helper_types import ReadResourceContents
 from mcp.server.models import InitializationOptions
 from mcp.server.session import ServerSession
@@ -229,25 +230,29 @@ def request_context(
         return request_ctx.get()
 
     def list_prompts(self):
-        def decorator(func: Callable[[], Awaitable[list[types.Prompt]]]):
+        def decorator(
+            func: Callable[[], Awaitable[list[types.Prompt]]]
+            | Callable[[types.Cursor | None], Awaitable[types.ListPromptsResult]],
+        ):
             logger.debug("Registering handler for PromptListRequest")
+            pass_cursor = accepts_cursor(func)
 
-            async def handler(_: Any):
-                prompts = await func()
-                return types.ServerResult(types.ListPromptsResult(prompts=prompts))
+            if pass_cursor:
+                cursor_func = cast(Callable[[types.Cursor | None], Awaitable[types.ListPromptsResult]], func)
 
-            self.request_handlers[types.ListPromptsRequest] = handler
-            return func
+                async def cursor_handler(req: types.ListPromptsRequest):
+                    result = await cursor_func(req.params.cursor if req.params is not None else None)
+                    return types.ServerResult(result)
 
-        return decorator
+                handler = cursor_handler
+            else:
+                list_func = cast(Callable[[], Awaitable[list[types.Prompt]]], func)
 
-    def list_prompts_paginated(self):
-        def decorator(func: Callable[[types.Cursor | None], Awaitable[types.ListPromptsResult]]):
-            logger.debug("Registering handler for PromptListRequest with pagination")
+                async def list_handler(_: types.ListPromptsRequest):
+                    result = await list_func()
+                    return types.ServerResult(types.ListPromptsResult(prompts=result))
 
-            async def handler(req: types.ListPromptsRequest):
-                result = await func(req.params.cursor if req.params else None)
-                return types.ServerResult(result)
+                handler = list_handler
 
             self.request_handlers[types.ListPromptsRequest] = handler
             return func
@@ -270,25 +275,29 @@ async def handler(req: types.GetPromptRequest):
         return decorator
 
     def list_resources(self):
-        def decorator(func: Callable[[], Awaitable[list[types.Resource]]]):
+        def decorator(
+            func: Callable[[], Awaitable[list[types.Resource]]]
+            | Callable[[types.Cursor | None], Awaitable[types.ListResourcesResult]],
+        ):
             logger.debug("Registering handler for ListResourcesRequest")
+            pass_cursor = accepts_cursor(func)
 
-            async def handler(_: Any):
-                resources = await func()
-                return types.ServerResult(types.ListResourcesResult(resources=resources))
+            if pass_cursor:
+                cursor_func = cast(Callable[[types.Cursor | None], Awaitable[types.ListResourcesResult]], func)
 
-            self.request_handlers[types.ListResourcesRequest] = handler
-            return func
+                async def cursor_handler(req: types.ListResourcesRequest):
+                    result = await cursor_func(req.params.cursor if req.params is not None else None)
+                    return types.ServerResult(result)
 
-        return decorator
+                handler = cursor_handler
+            else:
+                list_func = cast(Callable[[], Awaitable[list[types.Resource]]], func)
 
-    def list_resources_paginated(self):
-        def decorator(func: Callable[[types.Cursor | None], Awaitable[types.ListResourcesResult]]):
-            logger.debug("Registering handler for ListResourcesRequest with pagination")
+                async def list_handler(_: types.ListResourcesRequest):
+                    result = await list_func()
+                    return types.ServerResult(types.ListResourcesResult(resources=result))
 
-            async def handler(req: types.ListResourcesRequest):
-                result = await func(req.params.cursor if req.params else None)
-                return types.ServerResult(result)
+                handler = list_handler
 
             self.request_handlers[types.ListResourcesRequest] = handler
             return func
@@ -406,33 +415,36 @@ async def handler(req: types.UnsubscribeRequest):
         return decorator
 
     def list_tools(self):
-        def decorator(func: Callable[[], Awaitable[list[types.Tool]]]):
+        def decorator(
+            func: Callable[[], Awaitable[list[types.Tool]]]
+            | Callable[[types.Cursor | None], Awaitable[types.ListToolsResult]],
+        ):
             logger.debug("Registering handler for ListToolsRequest")
-
-            async def handler(_: Any):
-                tools = await func()
-                # Refresh the tool cache
-                self._tool_cache.clear()
-                for tool in tools:
-                    self._tool_cache[tool.name] = tool
-                return types.ServerResult(types.ListToolsResult(tools=tools))
-
-            self.request_handlers[types.ListToolsRequest] = handler
-            return func
-
-        return decorator
-
-    def list_tools_paginated(self):
-        def decorator(func: Callable[[types.Cursor | None], Awaitable[types.ListToolsResult]]):
-            logger.debug("Registering paginated handler for ListToolsRequest")
-
-            async def handler(request: types.ListToolsRequest):
-                cursor = request.params.cursor if request.params else None
-                result = await func(cursor)
-                # Refresh the tool cache with returned tools
-                for tool in result.tools:
-                    self._tool_cache[tool.name] = tool
-                return types.ServerResult(result)
+            pass_cursor = accepts_cursor(func)
+
+            if pass_cursor:
+                cursor_func = cast(Callable[[types.Cursor | None], Awaitable[types.ListToolsResult]], func)
+
+                async def cursor_handler(req: types.ListToolsRequest):
+                    result = await cursor_func(req.params.cursor if req.params is not None else None)
+                    # Refresh the tool cache with returned tools
+                    for tool in result.tools:
+                        self._tool_cache[tool.name] = tool
+                    return types.ServerResult(result)
+
+                handler = cursor_handler
+            else:
+                list_func = cast(Callable[[], Awaitable[list[types.Tool]]], func)
+
+                async def list_handler(req: types.ListToolsRequest):
+                    result = await list_func()
+                    # Clear and refresh the entire tool cache
+                    self._tool_cache.clear()
+                    for tool in result:
+                        self._tool_cache[tool.name] = tool
+                    return types.ServerResult(types.ListToolsResult(tools=result))
+
+                handler = list_handler
 
             self.request_handlers[types.ListToolsRequest] = handler
             return func