From 03b89e9444d5b3302b8c6a9d0ee65dcb0312d0e4 Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Thu, 9 Oct 2025 15:06:01 +0100 Subject: [PATCH 1/8] fix: send params as empty object for list methods without cursor Some external MCP servers implement strict JSON-RPC validation requiring the params field to always be present. Previously, the Python SDK omitted params entirely when cursor=None, causing validation errors. Now sends params: {} when no cursor is provided, matching TypeScript SDK behavior and fixing compatibility with strict servers. Reported-by: justin-yi-wang --- src/mcp/client/session.py | 8 ++-- tests/client/test_list_methods_cursor.py | 50 ++++++++++++++++++++---- 2 files changed, 46 insertions(+), 12 deletions(-) diff --git a/src/mcp/client/session.py b/src/mcp/client/session.py index bcf80d62a..2471999d5 100644 --- a/src/mcp/client/session.py +++ b/src/mcp/client/session.py @@ -217,7 +217,7 @@ async def list_resources(self, cursor: str | None = None) -> types.ListResources return await self.send_request( types.ClientRequest( types.ListResourcesRequest( - params=types.PaginatedRequestParams(cursor=cursor) if cursor is not None else None, + params=types.PaginatedRequestParams(cursor=cursor), ) ), types.ListResourcesResult, @@ -228,7 +228,7 @@ async def list_resource_templates(self, cursor: str | None = None) -> types.List return await self.send_request( types.ClientRequest( types.ListResourceTemplatesRequest( - params=types.PaginatedRequestParams(cursor=cursor) if cursor is not None else None, + params=types.PaginatedRequestParams(cursor=cursor), ) ), types.ListResourceTemplatesResult, @@ -322,7 +322,7 @@ async def list_prompts(self, cursor: str | None = None) -> types.ListPromptsResu return await self.send_request( types.ClientRequest( types.ListPromptsRequest( - params=types.PaginatedRequestParams(cursor=cursor) if cursor is not None else None, + params=types.PaginatedRequestParams(cursor=cursor), ) ), types.ListPromptsResult, @@ -368,7 +368,7 @@ async def list_tools(self, cursor: str | None = None) -> types.ListToolsResult: result = await self.send_request( types.ClientRequest( types.ListToolsRequest( - params=types.PaginatedRequestParams(cursor=cursor) if cursor is not None else None, + params=types.PaginatedRequestParams(cursor=cursor), ) ), types.ListToolsResult, diff --git a/tests/client/test_list_methods_cursor.py b/tests/client/test_list_methods_cursor.py index b31b704a4..9267e6839 100644 --- a/tests/client/test_list_methods_cursor.py +++ b/tests/client/test_list_methods_cursor.py @@ -2,8 +2,10 @@ import pytest +from mcp.server import Server from mcp.server.fastmcp import FastMCP from mcp.shared.memory import create_connected_server_and_client_session as create_session +from mcp.types import ListToolsRequest, ListToolsResult from .conftest import StreamSpyCollection @@ -36,7 +38,7 @@ async def test_tool_2() -> str: _ = await client_session.list_tools() list_tools_requests = spies.get_client_requests(method="tools/list") assert len(list_tools_requests) == 1 - assert list_tools_requests[0].params is None + assert list_tools_requests[0].params == {} spies.clear() @@ -44,7 +46,7 @@ async def test_tool_2() -> str: _ = await client_session.list_tools(cursor=None) list_tools_requests = spies.get_client_requests(method="tools/list") assert len(list_tools_requests) == 1 - assert list_tools_requests[0].params is None + assert list_tools_requests[0].params == {} spies.clear() @@ -86,7 +88,7 @@ async def test_resource() -> str: _ = await client_session.list_resources() list_resources_requests = spies.get_client_requests(method="resources/list") assert len(list_resources_requests) == 1 - assert list_resources_requests[0].params is None + assert list_resources_requests[0].params == {} spies.clear() @@ -94,7 +96,7 @@ async def test_resource() -> str: _ = await client_session.list_resources(cursor=None) list_resources_requests = spies.get_client_requests(method="resources/list") assert len(list_resources_requests) == 1 - assert list_resources_requests[0].params is None + assert list_resources_requests[0].params == {} spies.clear() @@ -135,7 +137,7 @@ async def test_prompt(name: str) -> str: _ = await client_session.list_prompts() list_prompts_requests = spies.get_client_requests(method="prompts/list") assert len(list_prompts_requests) == 1 - assert list_prompts_requests[0].params is None + assert list_prompts_requests[0].params == {} spies.clear() @@ -143,7 +145,7 @@ async def test_prompt(name: str) -> str: _ = await client_session.list_prompts(cursor=None) list_prompts_requests = spies.get_client_requests(method="prompts/list") assert len(list_prompts_requests) == 1 - assert list_prompts_requests[0].params is None + assert list_prompts_requests[0].params == {} spies.clear() @@ -185,7 +187,7 @@ async def test_template(name: str) -> str: _ = await client_session.list_resource_templates() list_templates_requests = spies.get_client_requests(method="resources/templates/list") assert len(list_templates_requests) == 1 - assert list_templates_requests[0].params is None + assert list_templates_requests[0].params == {} spies.clear() @@ -193,7 +195,7 @@ async def test_template(name: str) -> str: _ = await client_session.list_resource_templates(cursor=None) list_templates_requests = spies.get_client_requests(method="resources/templates/list") assert len(list_templates_requests) == 1 - assert list_templates_requests[0].params is None + assert list_templates_requests[0].params == {} spies.clear() @@ -212,3 +214,35 @@ async def test_template(name: str) -> str: assert len(list_templates_requests) == 1 assert list_templates_requests[0].params is not None assert list_templates_requests[0].params["cursor"] == "" + + +async def test_list_tools_with_strict_server_validation(): + """Test that list_tools works with strict servers require a params field, + even if it is empty. + + Some MCP servers may implement strict JSON-RPC validation that requires + the params field to always be present in requests, even if empty {}. + + This test ensures such servers are supported by the client SDK for list_resources + requests without a cursor. + """ + + server = Server("strict_server") + + @server.list_tools() + async def handle_list_tools(request: ListToolsRequest) -> ListToolsResult: + """Strict handler that validates params field exists""" + + # Simulate strict server validation + if request.params is None: + raise ValueError( + "Strict server validation failed: params field must be present. " + "Expected params: {} for requests without cursor." + ) + + # Return empty tools list + return ListToolsResult(tools=[]) + + async with create_session(server) as client_session: + result = await client_session.list_tools() + assert result is not None From 7db0e258740bc2765254969aac119cbe0eeadc76 Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Thu, 9 Oct 2025 17:57:01 +0100 Subject: [PATCH 2/8] feat: deprecate cursor parameter in favor of params for list methods Deprecate the cursor parameter in list_tools, list_prompts, list_resources, and list_resource_templates methods, replacing it with a new params parameter that accepts PaginatedRequestParams. This change maintains complete backwards compatibility while providing opt-in support for strict MCP servers that require the params field to always be present in JSON-RPC requests (even if empty). When params is not provided or is None, the SDK continues to omit the params field entirely, matching the previous behavior. The cursor parameter now issues a DeprecationWarning directing developers to use the new params parameter. This matches the TypeScript SDK's API design. Reported-by: justin-yi-wang --- src/mcp/client/session.py | 117 +++++++++++++++++------ tests/client/test_list_methods_cursor.py | 24 +++-- 2 files changed, 104 insertions(+), 37 deletions(-) diff --git a/src/mcp/client/session.py b/src/mcp/client/session.py index 2471999d5..07167f59f 100644 --- a/src/mcp/client/session.py +++ b/src/mcp/client/session.py @@ -1,4 +1,5 @@ import logging +import warnings from datetime import timedelta from typing import Any, Protocol @@ -212,25 +213,55 @@ async def set_logging_level(self, level: types.LoggingLevel) -> types.EmptyResul types.EmptyResult, ) - async def list_resources(self, cursor: str | None = None) -> types.ListResourcesResult: - """Send a resources/list request.""" + async def list_resources( + self, + cursor: str | None = None, + *, + params: types.PaginatedRequestParams | None = None, + ) -> types.ListResourcesResult: + """Send a resources/list request. + + Args: + cursor: (Deprecated) Pagination cursor. Use params parameter instead. + params: Pagination parameters. Defaults to None (omits params field). + """ + if cursor is not None: + warnings.warn( + "cursor parameter is deprecated, use params=PaginatedRequestParams(cursor=...) instead", + DeprecationWarning, + stacklevel=2, + ) + if params is None: + params = types.PaginatedRequestParams(cursor=cursor) + return await self.send_request( - types.ClientRequest( - types.ListResourcesRequest( - params=types.PaginatedRequestParams(cursor=cursor), - ) - ), + types.ClientRequest(types.ListResourcesRequest(params=params)), types.ListResourcesResult, ) - async def list_resource_templates(self, cursor: str | None = None) -> types.ListResourceTemplatesResult: - """Send a resources/templates/list request.""" + async def list_resource_templates( + self, + cursor: str | None = None, + *, + params: types.PaginatedRequestParams | None = None, + ) -> types.ListResourceTemplatesResult: + """Send a resources/templates/list request. + + Args: + cursor: (Deprecated) Pagination cursor. Use params parameter instead. + params: Pagination parameters. Defaults to None (omits params field). + """ + if cursor is not None: + warnings.warn( + "cursor parameter is deprecated, use params=PaginatedRequestParams(cursor=...) instead", + DeprecationWarning, + stacklevel=2, + ) + if params is None: + params = types.PaginatedRequestParams(cursor=cursor) + return await self.send_request( - types.ClientRequest( - types.ListResourceTemplatesRequest( - params=types.PaginatedRequestParams(cursor=cursor), - ) - ), + types.ClientRequest(types.ListResourceTemplatesRequest(params=params)), types.ListResourceTemplatesResult, ) @@ -317,14 +348,29 @@ async def _validate_tool_result(self, name: str, result: types.CallToolResult) - except SchemaError as e: raise RuntimeError(f"Invalid schema for tool {name}: {e}") - async def list_prompts(self, cursor: str | None = None) -> types.ListPromptsResult: - """Send a prompts/list request.""" + async def list_prompts( + self, + cursor: str | None = None, + *, + params: types.PaginatedRequestParams | None = None, + ) -> types.ListPromptsResult: + """Send a prompts/list request. + + Args: + cursor: (Deprecated) Pagination cursor. Use params parameter instead. + params: Pagination parameters. Defaults to None (omits params field). + """ + if cursor is not None: + warnings.warn( + "cursor parameter is deprecated, use params=PaginatedRequestParams(cursor=...) instead", + DeprecationWarning, + stacklevel=2, + ) + if params is None: + params = types.PaginatedRequestParams(cursor=cursor) + return await self.send_request( - types.ClientRequest( - types.ListPromptsRequest( - params=types.PaginatedRequestParams(cursor=cursor), - ) - ), + types.ClientRequest(types.ListPromptsRequest(params=params)), types.ListPromptsResult, ) @@ -363,14 +409,29 @@ async def complete( types.CompleteResult, ) - async def list_tools(self, cursor: str | None = None) -> types.ListToolsResult: - """Send a tools/list request.""" + async def list_tools( + self, + cursor: str | None = None, + *, + params: types.PaginatedRequestParams | None = None, + ) -> types.ListToolsResult: + """Send a tools/list request. + + Args: + cursor: (Deprecated) Pagination cursor. Use params parameter instead. + params: Pagination parameters. Defaults to None (omits params field). + """ + if cursor is not None: + warnings.warn( + "cursor parameter is deprecated, use params=PaginatedRequestParams(cursor=...) instead", + DeprecationWarning, + stacklevel=2, + ) + if params is None: + params = types.PaginatedRequestParams(cursor=cursor) + result = await self.send_request( - types.ClientRequest( - types.ListToolsRequest( - params=types.PaginatedRequestParams(cursor=cursor), - ) - ), + types.ClientRequest(types.ListToolsRequest(params=params)), types.ListToolsResult, ) diff --git a/tests/client/test_list_methods_cursor.py b/tests/client/test_list_methods_cursor.py index 9267e6839..f3ab36cb6 100644 --- a/tests/client/test_list_methods_cursor.py +++ b/tests/client/test_list_methods_cursor.py @@ -2,6 +2,7 @@ import pytest +import mcp.types as types from mcp.server import Server from mcp.server.fastmcp import FastMCP from mcp.shared.memory import create_connected_server_and_client_session as create_session @@ -12,6 +13,7 @@ pytestmark = pytest.mark.anyio +@pytest.mark.filterwarnings("ignore::DeprecationWarning") async def test_list_tools_cursor_parameter(stream_spy: Callable[[], StreamSpyCollection]): """Test that the cursor parameter is accepted for list_tools and that it is correctly passed to the server. @@ -38,7 +40,7 @@ async def test_tool_2() -> str: _ = await client_session.list_tools() list_tools_requests = spies.get_client_requests(method="tools/list") assert len(list_tools_requests) == 1 - assert list_tools_requests[0].params == {} + assert list_tools_requests[0].params is None spies.clear() @@ -46,7 +48,7 @@ async def test_tool_2() -> str: _ = await client_session.list_tools(cursor=None) list_tools_requests = spies.get_client_requests(method="tools/list") assert len(list_tools_requests) == 1 - assert list_tools_requests[0].params == {} + assert list_tools_requests[0].params is None spies.clear() @@ -67,6 +69,7 @@ async def test_tool_2() -> str: assert list_tools_requests[0].params["cursor"] == "" +@pytest.mark.filterwarnings("ignore::DeprecationWarning") async def test_list_resources_cursor_parameter(stream_spy: Callable[[], StreamSpyCollection]): """Test that the cursor parameter is accepted for list_resources and that it is correctly passed to the server. @@ -88,7 +91,7 @@ async def test_resource() -> str: _ = await client_session.list_resources() list_resources_requests = spies.get_client_requests(method="resources/list") assert len(list_resources_requests) == 1 - assert list_resources_requests[0].params == {} + assert list_resources_requests[0].params is None spies.clear() @@ -96,7 +99,7 @@ async def test_resource() -> str: _ = await client_session.list_resources(cursor=None) list_resources_requests = spies.get_client_requests(method="resources/list") assert len(list_resources_requests) == 1 - assert list_resources_requests[0].params == {} + assert list_resources_requests[0].params is None spies.clear() @@ -117,6 +120,7 @@ async def test_resource() -> str: assert list_resources_requests[0].params["cursor"] == "" +@pytest.mark.filterwarnings("ignore::DeprecationWarning") async def test_list_prompts_cursor_parameter(stream_spy: Callable[[], StreamSpyCollection]): """Test that the cursor parameter is accepted for list_prompts and that it is correctly passed to the server. @@ -137,7 +141,7 @@ async def test_prompt(name: str) -> str: _ = await client_session.list_prompts() list_prompts_requests = spies.get_client_requests(method="prompts/list") assert len(list_prompts_requests) == 1 - assert list_prompts_requests[0].params == {} + assert list_prompts_requests[0].params is None spies.clear() @@ -145,7 +149,7 @@ async def test_prompt(name: str) -> str: _ = await client_session.list_prompts(cursor=None) list_prompts_requests = spies.get_client_requests(method="prompts/list") assert len(list_prompts_requests) == 1 - assert list_prompts_requests[0].params == {} + assert list_prompts_requests[0].params is None spies.clear() @@ -166,6 +170,7 @@ async def test_prompt(name: str) -> str: assert list_prompts_requests[0].params["cursor"] == "" +@pytest.mark.filterwarnings("ignore::DeprecationWarning") async def test_list_resource_templates_cursor_parameter(stream_spy: Callable[[], StreamSpyCollection]): """Test that the cursor parameter is accepted for list_resource_templates and that it is correctly passed to the server. @@ -187,7 +192,7 @@ async def test_template(name: str) -> str: _ = await client_session.list_resource_templates() list_templates_requests = spies.get_client_requests(method="resources/templates/list") assert len(list_templates_requests) == 1 - assert list_templates_requests[0].params == {} + assert list_templates_requests[0].params is None spies.clear() @@ -195,7 +200,7 @@ async def test_template(name: str) -> str: _ = await client_session.list_resource_templates(cursor=None) list_templates_requests = spies.get_client_requests(method="resources/templates/list") assert len(list_templates_requests) == 1 - assert list_templates_requests[0].params == {} + assert list_templates_requests[0].params is None spies.clear() @@ -244,5 +249,6 @@ async def handle_list_tools(request: ListToolsRequest) -> ListToolsResult: return ListToolsResult(tools=[]) async with create_session(server) as client_session: - result = await client_session.list_tools() + # Use params to explicitly send params: {} for strict server compatibility + result = await client_session.list_tools(params=types.PaginatedRequestParams()) assert result is not None From b779c12200353f9f05da6fd2d37ba58b2a7bf151 Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Thu, 9 Oct 2025 18:03:14 +0100 Subject: [PATCH 3/8] test: add params parameter tests and precedence verification Add comprehensive tests for the new params parameter across all four list methods (list_tools, list_resources, list_prompts, list_resource_templates). Tests verify: 1. params parameter works correctly when omitted, set to None, or set to empty 2. params parameter correctly passes cursor values 3. params parameter takes precedence over deprecated cursor parameter when both are provided, ensuring a safe migration path for existing code These tests ensure the backwards compatibility guarantees are preserved and prevent regressions in precedence behavior. --- tests/client/test_list_methods_cursor.py | 331 +++++++++++++++++++++++ 1 file changed, 331 insertions(+) diff --git a/tests/client/test_list_methods_cursor.py b/tests/client/test_list_methods_cursor.py index f3ab36cb6..5777d1a99 100644 --- a/tests/client/test_list_methods_cursor.py +++ b/tests/client/test_list_methods_cursor.py @@ -221,6 +221,337 @@ async def test_template(name: str) -> str: assert list_templates_requests[0].params["cursor"] == "" +async def test_list_tools_params_parameter(stream_spy: Callable[[], StreamSpyCollection]): + """Test that the params parameter works correctly for list_tools. + + This tests the new params parameter API (non-deprecated) to ensure + it correctly handles all parameter combinations. + """ + server = FastMCP("test") + + # Create a couple of test tools + @server.tool(name="test_tool_1") + async def test_tool_1() -> str: + """First test tool""" + return "Result 1" + + @server.tool(name="test_tool_2") + async def test_tool_2() -> str: + """Second test tool""" + return "Result 2" + + async with create_session(server._mcp_server) as client_session: + spies = stream_spy() + + # Test without params parameter (omitted) + _ = await client_session.list_tools() + list_tools_requests = spies.get_client_requests(method="tools/list") + assert len(list_tools_requests) == 1 + assert list_tools_requests[0].params is None + + spies.clear() + + # Test with params=None + _ = await client_session.list_tools(params=None) + list_tools_requests = spies.get_client_requests(method="tools/list") + assert len(list_tools_requests) == 1 + assert list_tools_requests[0].params is None + + spies.clear() + + # Test with empty params (for strict servers) + _ = await client_session.list_tools(params=types.PaginatedRequestParams()) + list_tools_requests = spies.get_client_requests(method="tools/list") + assert len(list_tools_requests) == 1 + assert list_tools_requests[0].params is not None + assert list_tools_requests[0].params.get("cursor") is None + + spies.clear() + + # Test with params containing cursor + _ = await client_session.list_tools(params=types.PaginatedRequestParams(cursor="some_cursor_value")) + list_tools_requests = spies.get_client_requests(method="tools/list") + assert len(list_tools_requests) == 1 + assert list_tools_requests[0].params is not None + assert list_tools_requests[0].params["cursor"] == "some_cursor_value" + + +async def test_list_resources_params_parameter(stream_spy: Callable[[], StreamSpyCollection]): + """Test that the params parameter works correctly for list_resources. + + This tests the new params parameter API (non-deprecated) to ensure + it correctly handles all parameter combinations. + """ + server = FastMCP("test") + + # Create a test resource + @server.resource("resource://test/data") + async def test_resource() -> str: + """Test resource""" + return "Test data" + + async with create_session(server._mcp_server) as client_session: + spies = stream_spy() + + # Test without params parameter (omitted) + _ = await client_session.list_resources() + list_resources_requests = spies.get_client_requests(method="resources/list") + assert len(list_resources_requests) == 1 + assert list_resources_requests[0].params is None + + spies.clear() + + # Test with params=None + _ = await client_session.list_resources(params=None) + list_resources_requests = spies.get_client_requests(method="resources/list") + assert len(list_resources_requests) == 1 + assert list_resources_requests[0].params is None + + spies.clear() + + # Test with empty params (for strict servers) + _ = await client_session.list_resources(params=types.PaginatedRequestParams()) + list_resources_requests = spies.get_client_requests(method="resources/list") + assert len(list_resources_requests) == 1 + assert list_resources_requests[0].params is not None + assert list_resources_requests[0].params.get("cursor") is None + + spies.clear() + + # Test with params containing cursor + _ = await client_session.list_resources(params=types.PaginatedRequestParams(cursor="some_cursor")) + list_resources_requests = spies.get_client_requests(method="resources/list") + assert len(list_resources_requests) == 1 + assert list_resources_requests[0].params is not None + assert list_resources_requests[0].params["cursor"] == "some_cursor" + + +async def test_list_prompts_params_parameter(stream_spy: Callable[[], StreamSpyCollection]): + """Test that the params parameter works correctly for list_prompts. + + This tests the new params parameter API (non-deprecated) to ensure + it correctly handles all parameter combinations. + """ + server = FastMCP("test") + + # Create a test prompt + @server.prompt() + async def test_prompt(name: str) -> str: + """Test prompt""" + return f"Hello, {name}!" + + async with create_session(server._mcp_server) as client_session: + spies = stream_spy() + + # Test without params parameter (omitted) + _ = await client_session.list_prompts() + list_prompts_requests = spies.get_client_requests(method="prompts/list") + assert len(list_prompts_requests) == 1 + assert list_prompts_requests[0].params is None + + spies.clear() + + # Test with params=None + _ = await client_session.list_prompts(params=None) + list_prompts_requests = spies.get_client_requests(method="prompts/list") + assert len(list_prompts_requests) == 1 + assert list_prompts_requests[0].params is None + + spies.clear() + + # Test with empty params (for strict servers) + _ = await client_session.list_prompts(params=types.PaginatedRequestParams()) + list_prompts_requests = spies.get_client_requests(method="prompts/list") + assert len(list_prompts_requests) == 1 + assert list_prompts_requests[0].params is not None + assert list_prompts_requests[0].params.get("cursor") is None + + spies.clear() + + # Test with params containing cursor + _ = await client_session.list_prompts(params=types.PaginatedRequestParams(cursor="some_cursor")) + list_prompts_requests = spies.get_client_requests(method="prompts/list") + assert len(list_prompts_requests) == 1 + assert list_prompts_requests[0].params is not None + assert list_prompts_requests[0].params["cursor"] == "some_cursor" + + +async def test_list_resource_templates_params_parameter(stream_spy: Callable[[], StreamSpyCollection]): + """Test that the params parameter works correctly for list_resource_templates. + + This tests the new params parameter API (non-deprecated) to ensure + it correctly handles all parameter combinations. + """ + server = FastMCP("test") + + # Create a test resource template + @server.resource("resource://test/{name}") + async def test_template(name: str) -> str: + """Test resource template""" + return f"Data for {name}" + + async with create_session(server._mcp_server) as client_session: + spies = stream_spy() + + # Test without params parameter (omitted) + _ = await client_session.list_resource_templates() + list_templates_requests = spies.get_client_requests(method="resources/templates/list") + assert len(list_templates_requests) == 1 + assert list_templates_requests[0].params is None + + spies.clear() + + # Test with params=None + _ = await client_session.list_resource_templates(params=None) + list_templates_requests = spies.get_client_requests(method="resources/templates/list") + assert len(list_templates_requests) == 1 + assert list_templates_requests[0].params is None + + spies.clear() + + # Test with empty params (for strict servers) + _ = await client_session.list_resource_templates(params=types.PaginatedRequestParams()) + list_templates_requests = spies.get_client_requests(method="resources/templates/list") + assert len(list_templates_requests) == 1 + assert list_templates_requests[0].params is not None + assert list_templates_requests[0].params.get("cursor") is None + + spies.clear() + + # Test with params containing cursor + _ = await client_session.list_resource_templates(params=types.PaginatedRequestParams(cursor="some_cursor")) + list_templates_requests = spies.get_client_requests(method="resources/templates/list") + assert len(list_templates_requests) == 1 + assert list_templates_requests[0].params is not None + assert list_templates_requests[0].params["cursor"] == "some_cursor" + + +@pytest.mark.filterwarnings("ignore::DeprecationWarning") +async def test_list_tools_params_takes_precedence_over_cursor( + stream_spy: Callable[[], StreamSpyCollection], +): + """Test that params parameter takes precedence over cursor parameter. + + When both cursor and params are provided, params should be used and + cursor should be ignored, ensuring safe migration path. + """ + server = FastMCP("test") + + @server.tool(name="test_tool") + async def test_tool() -> str: + """Test tool""" + return "Result" + + async with create_session(server._mcp_server) as client_session: + spies = stream_spy() + + # Call with both cursor and params - params should take precedence + _ = await client_session.list_tools( + cursor="old_cursor", + params=types.PaginatedRequestParams(cursor="new_cursor"), + ) + list_tools_requests = spies.get_client_requests(method="tools/list") + assert len(list_tools_requests) == 1 + # Verify params takes precedence (new_cursor should be used, not old_cursor) + assert list_tools_requests[0].params is not None + assert list_tools_requests[0].params["cursor"] == "new_cursor" + + +@pytest.mark.filterwarnings("ignore::DeprecationWarning") +async def test_list_resources_params_takes_precedence_over_cursor( + stream_spy: Callable[[], StreamSpyCollection], +): + """Test that params parameter takes precedence over cursor parameter. + + When both cursor and params are provided, params should be used and + cursor should be ignored, ensuring safe migration path. + """ + server = FastMCP("test") + + @server.resource("resource://test/data") + async def test_resource() -> str: + """Test resource""" + return "Test data" + + async with create_session(server._mcp_server) as client_session: + spies = stream_spy() + + # Call with both cursor and params - params should take precedence + _ = await client_session.list_resources( + cursor="old_cursor", + params=types.PaginatedRequestParams(cursor="new_cursor"), + ) + list_resources_requests = spies.get_client_requests(method="resources/list") + assert len(list_resources_requests) == 1 + # Verify params takes precedence (new_cursor should be used, not old_cursor) + assert list_resources_requests[0].params is not None + assert list_resources_requests[0].params["cursor"] == "new_cursor" + + +@pytest.mark.filterwarnings("ignore::DeprecationWarning") +async def test_list_prompts_params_takes_precedence_over_cursor( + stream_spy: Callable[[], StreamSpyCollection], +): + """Test that params parameter takes precedence over cursor parameter. + + When both cursor and params are provided, params should be used and + cursor should be ignored, ensuring safe migration path. + """ + server = FastMCP("test") + + @server.prompt() + async def test_prompt(name: str) -> str: + """Test prompt""" + return f"Hello, {name}!" + + async with create_session(server._mcp_server) as client_session: + spies = stream_spy() + + # Call with both cursor and params - params should take precedence + _ = await client_session.list_prompts( + cursor="old_cursor", + params=types.PaginatedRequestParams(cursor="new_cursor"), + ) + list_prompts_requests = spies.get_client_requests(method="prompts/list") + assert len(list_prompts_requests) == 1 + # Verify params takes precedence (new_cursor should be used, not old_cursor) + assert list_prompts_requests[0].params is not None + assert list_prompts_requests[0].params["cursor"] == "new_cursor" + + +@pytest.mark.filterwarnings("ignore::DeprecationWarning") +async def test_list_resource_templates_params_takes_precedence_over_cursor( + stream_spy: Callable[[], StreamSpyCollection], +): + """Test that params parameter takes precedence over cursor parameter. + + When both cursor and params are provided, params should be used and + cursor should be ignored, ensuring safe migration path. + """ + server = FastMCP("test") + + @server.resource("resource://test/{name}") + async def test_template(name: str) -> str: + """Test resource template""" + return f"Data for {name}" + + async with create_session(server._mcp_server) as client_session: + spies = stream_spy() + + # Call with both cursor and params - params should take precedence + _ = await client_session.list_resource_templates( + cursor="old_cursor", + params=types.PaginatedRequestParams(cursor="new_cursor"), + ) + list_templates_requests = spies.get_client_requests( + method="resources/templates/list" + ) + assert len(list_templates_requests) == 1 + # Verify params takes precedence (new_cursor should be used, not old_cursor) + assert list_templates_requests[0].params is not None + assert list_templates_requests[0].params["cursor"] == "new_cursor" + + async def test_list_tools_with_strict_server_validation(): """Test that list_tools works with strict servers require a params field, even if it is empty. From a98c4f79056f79570bb5a8f75d85b64103fb68ea Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Thu, 9 Oct 2025 18:07:34 +0100 Subject: [PATCH 4/8] refactor: use parametrized tests to reduce test duplication Replace 13 individual test functions with 3 parametrized test functions that cover all four list methods (list_tools, list_resources, list_prompts, list_resource_templates). Changes: - Created full_featured_server fixture with tools, resources, prompts, and templates - test_list_methods_cursor_parameter: parametrized version of 4 cursor tests - test_list_methods_params_parameter: parametrized version of 4 params tests - test_list_methods_params_takes_precedence_over_cursor: parametrized precedence tests - Kept test_list_tools_with_strict_server_validation separate (different scenario) Result: Reduced from 600+ lines to ~240 lines while maintaining identical coverage (13 test cases, all passing). --- tests/client/test_list_methods_cursor.py | 565 +++++------------------ 1 file changed, 108 insertions(+), 457 deletions(-) diff --git a/tests/client/test_list_methods_cursor.py b/tests/client/test_list_methods_cursor.py index 5777d1a99..d39366c96 100644 --- a/tests/client/test_list_methods_cursor.py +++ b/tests/client/test_list_methods_cursor.py @@ -13,16 +13,11 @@ pytestmark = pytest.mark.anyio -@pytest.mark.filterwarnings("ignore::DeprecationWarning") -async def test_list_tools_cursor_parameter(stream_spy: Callable[[], StreamSpyCollection]): - """Test that the cursor parameter is accepted for list_tools - and that it is correctly passed to the server. - - See: https://modelcontextprotocol.io/specification/2025-03-26/server/utilities/pagination#request-format - """ +@pytest.fixture +async def full_featured_server(): + """Create a server with tools, resources, prompts, and templates.""" server = FastMCP("test") - # Create a couple of test tools @server.tool(name="test_tool_1") async def test_tool_1() -> str: """First test tool""" @@ -33,523 +28,179 @@ async def test_tool_2() -> str: """Second test tool""" return "Result 2" - async with create_session(server._mcp_server) as client_session: - spies = stream_spy() - - # Test without cursor parameter (omitted) - _ = await client_session.list_tools() - list_tools_requests = spies.get_client_requests(method="tools/list") - assert len(list_tools_requests) == 1 - assert list_tools_requests[0].params is None - - spies.clear() - - # Test with cursor=None - _ = await client_session.list_tools(cursor=None) - list_tools_requests = spies.get_client_requests(method="tools/list") - assert len(list_tools_requests) == 1 - assert list_tools_requests[0].params is None - - spies.clear() - - # Test with cursor as string - _ = await client_session.list_tools(cursor="some_cursor_value") - list_tools_requests = spies.get_client_requests(method="tools/list") - assert len(list_tools_requests) == 1 - assert list_tools_requests[0].params is not None - assert list_tools_requests[0].params["cursor"] == "some_cursor_value" - - spies.clear() - - # Test with empty string cursor - _ = await client_session.list_tools(cursor="") - list_tools_requests = spies.get_client_requests(method="tools/list") - assert len(list_tools_requests) == 1 - assert list_tools_requests[0].params is not None - assert list_tools_requests[0].params["cursor"] == "" - - -@pytest.mark.filterwarnings("ignore::DeprecationWarning") -async def test_list_resources_cursor_parameter(stream_spy: Callable[[], StreamSpyCollection]): - """Test that the cursor parameter is accepted for list_resources - and that it is correctly passed to the server. - - See: https://modelcontextprotocol.io/specification/2025-03-26/server/utilities/pagination#request-format - """ - server = FastMCP("test") - - # Create a test resource @server.resource("resource://test/data") async def test_resource() -> str: """Test resource""" return "Test data" - async with create_session(server._mcp_server) as client_session: - spies = stream_spy() - - # Test without cursor parameter (omitted) - _ = await client_session.list_resources() - list_resources_requests = spies.get_client_requests(method="resources/list") - assert len(list_resources_requests) == 1 - assert list_resources_requests[0].params is None - - spies.clear() - - # Test with cursor=None - _ = await client_session.list_resources(cursor=None) - list_resources_requests = spies.get_client_requests(method="resources/list") - assert len(list_resources_requests) == 1 - assert list_resources_requests[0].params is None - - spies.clear() - - # Test with cursor as string - _ = await client_session.list_resources(cursor="some_cursor") - list_resources_requests = spies.get_client_requests(method="resources/list") - assert len(list_resources_requests) == 1 - assert list_resources_requests[0].params is not None - assert list_resources_requests[0].params["cursor"] == "some_cursor" - - spies.clear() - - # Test with empty string cursor - _ = await client_session.list_resources(cursor="") - list_resources_requests = spies.get_client_requests(method="resources/list") - assert len(list_resources_requests) == 1 - assert list_resources_requests[0].params is not None - assert list_resources_requests[0].params["cursor"] == "" - - -@pytest.mark.filterwarnings("ignore::DeprecationWarning") -async def test_list_prompts_cursor_parameter(stream_spy: Callable[[], StreamSpyCollection]): - """Test that the cursor parameter is accepted for list_prompts - and that it is correctly passed to the server. - See: https://modelcontextprotocol.io/specification/2025-03-26/server/utilities/pagination#request-format - """ - server = FastMCP("test") - - # Create a test prompt @server.prompt() async def test_prompt(name: str) -> str: """Test prompt""" return f"Hello, {name}!" - async with create_session(server._mcp_server) as client_session: - spies = stream_spy() - - # Test without cursor parameter (omitted) - _ = await client_session.list_prompts() - list_prompts_requests = spies.get_client_requests(method="prompts/list") - assert len(list_prompts_requests) == 1 - assert list_prompts_requests[0].params is None - - spies.clear() - - # Test with cursor=None - _ = await client_session.list_prompts(cursor=None) - list_prompts_requests = spies.get_client_requests(method="prompts/list") - assert len(list_prompts_requests) == 1 - assert list_prompts_requests[0].params is None - - spies.clear() - - # Test with cursor as string - _ = await client_session.list_prompts(cursor="some_cursor") - list_prompts_requests = spies.get_client_requests(method="prompts/list") - assert len(list_prompts_requests) == 1 - assert list_prompts_requests[0].params is not None - assert list_prompts_requests[0].params["cursor"] == "some_cursor" - - spies.clear() + @server.resource("resource://test/{name}") + async def test_template(name: str) -> str: + """Test resource template""" + return f"Data for {name}" - # Test with empty string cursor - _ = await client_session.list_prompts(cursor="") - list_prompts_requests = spies.get_client_requests(method="prompts/list") - assert len(list_prompts_requests) == 1 - assert list_prompts_requests[0].params is not None - assert list_prompts_requests[0].params["cursor"] == "" + return server +@pytest.mark.parametrize( + "method_name,request_method", + [ + ("list_tools", "tools/list"), + ("list_resources", "resources/list"), + ("list_prompts", "prompts/list"), + ("list_resource_templates", "resources/templates/list"), + ], +) @pytest.mark.filterwarnings("ignore::DeprecationWarning") -async def test_list_resource_templates_cursor_parameter(stream_spy: Callable[[], StreamSpyCollection]): - """Test that the cursor parameter is accepted for list_resource_templates - and that it is correctly passed to the server. +async def test_list_methods_cursor_parameter( + stream_spy: Callable[[], StreamSpyCollection], + full_featured_server: FastMCP, + method_name: str, + request_method: str, +): + """Test that the cursor parameter is accepted and correctly passed to the server. + + Covers: list_tools, list_resources, list_prompts, list_resource_templates See: https://modelcontextprotocol.io/specification/2025-03-26/server/utilities/pagination#request-format """ - server = FastMCP("test") - - # Create a test resource template - @server.resource("resource://test/{name}") - async def test_template(name: str) -> str: - """Test resource template""" - return f"Data for {name}" - - async with create_session(server._mcp_server) as client_session: + async with create_session(full_featured_server._mcp_server) as client_session: spies = stream_spy() # Test without cursor parameter (omitted) - _ = await client_session.list_resource_templates() - list_templates_requests = spies.get_client_requests(method="resources/templates/list") - assert len(list_templates_requests) == 1 - assert list_templates_requests[0].params is None + method = getattr(client_session, method_name) + _ = await method() + requests = spies.get_client_requests(method=request_method) + assert len(requests) == 1 + assert requests[0].params is None spies.clear() # Test with cursor=None - _ = await client_session.list_resource_templates(cursor=None) - list_templates_requests = spies.get_client_requests(method="resources/templates/list") - assert len(list_templates_requests) == 1 - assert list_templates_requests[0].params is None + _ = await method(cursor=None) + requests = spies.get_client_requests(method=request_method) + assert len(requests) == 1 + assert requests[0].params is None spies.clear() # Test with cursor as string - _ = await client_session.list_resource_templates(cursor="some_cursor") - list_templates_requests = spies.get_client_requests(method="resources/templates/list") - assert len(list_templates_requests) == 1 - assert list_templates_requests[0].params is not None - assert list_templates_requests[0].params["cursor"] == "some_cursor" + _ = await method(cursor="some_cursor_value") + requests = spies.get_client_requests(method=request_method) + assert len(requests) == 1 + assert requests[0].params is not None + assert requests[0].params["cursor"] == "some_cursor_value" spies.clear() # Test with empty string cursor - _ = await client_session.list_resource_templates(cursor="") - list_templates_requests = spies.get_client_requests(method="resources/templates/list") - assert len(list_templates_requests) == 1 - assert list_templates_requests[0].params is not None - assert list_templates_requests[0].params["cursor"] == "" - - -async def test_list_tools_params_parameter(stream_spy: Callable[[], StreamSpyCollection]): - """Test that the params parameter works correctly for list_tools. - - This tests the new params parameter API (non-deprecated) to ensure - it correctly handles all parameter combinations. - """ - server = FastMCP("test") - - # Create a couple of test tools - @server.tool(name="test_tool_1") - async def test_tool_1() -> str: - """First test tool""" - return "Result 1" - - @server.tool(name="test_tool_2") - async def test_tool_2() -> str: - """Second test tool""" - return "Result 2" - - async with create_session(server._mcp_server) as client_session: - spies = stream_spy() - - # Test without params parameter (omitted) - _ = await client_session.list_tools() - list_tools_requests = spies.get_client_requests(method="tools/list") - assert len(list_tools_requests) == 1 - assert list_tools_requests[0].params is None - - spies.clear() - - # Test with params=None - _ = await client_session.list_tools(params=None) - list_tools_requests = spies.get_client_requests(method="tools/list") - assert len(list_tools_requests) == 1 - assert list_tools_requests[0].params is None - - spies.clear() - - # Test with empty params (for strict servers) - _ = await client_session.list_tools(params=types.PaginatedRequestParams()) - list_tools_requests = spies.get_client_requests(method="tools/list") - assert len(list_tools_requests) == 1 - assert list_tools_requests[0].params is not None - assert list_tools_requests[0].params.get("cursor") is None - - spies.clear() - - # Test with params containing cursor - _ = await client_session.list_tools(params=types.PaginatedRequestParams(cursor="some_cursor_value")) - list_tools_requests = spies.get_client_requests(method="tools/list") - assert len(list_tools_requests) == 1 - assert list_tools_requests[0].params is not None - assert list_tools_requests[0].params["cursor"] == "some_cursor_value" - - -async def test_list_resources_params_parameter(stream_spy: Callable[[], StreamSpyCollection]): - """Test that the params parameter works correctly for list_resources. - - This tests the new params parameter API (non-deprecated) to ensure - it correctly handles all parameter combinations. - """ - server = FastMCP("test") - - # Create a test resource - @server.resource("resource://test/data") - async def test_resource() -> str: - """Test resource""" - return "Test data" - - async with create_session(server._mcp_server) as client_session: - spies = stream_spy() - - # Test without params parameter (omitted) - _ = await client_session.list_resources() - list_resources_requests = spies.get_client_requests(method="resources/list") - assert len(list_resources_requests) == 1 - assert list_resources_requests[0].params is None - - spies.clear() - - # Test with params=None - _ = await client_session.list_resources(params=None) - list_resources_requests = spies.get_client_requests(method="resources/list") - assert len(list_resources_requests) == 1 - assert list_resources_requests[0].params is None - - spies.clear() - - # Test with empty params (for strict servers) - _ = await client_session.list_resources(params=types.PaginatedRequestParams()) - list_resources_requests = spies.get_client_requests(method="resources/list") - assert len(list_resources_requests) == 1 - assert list_resources_requests[0].params is not None - assert list_resources_requests[0].params.get("cursor") is None - - spies.clear() - - # Test with params containing cursor - _ = await client_session.list_resources(params=types.PaginatedRequestParams(cursor="some_cursor")) - list_resources_requests = spies.get_client_requests(method="resources/list") - assert len(list_resources_requests) == 1 - assert list_resources_requests[0].params is not None - assert list_resources_requests[0].params["cursor"] == "some_cursor" - - -async def test_list_prompts_params_parameter(stream_spy: Callable[[], StreamSpyCollection]): - """Test that the params parameter works correctly for list_prompts. - - This tests the new params parameter API (non-deprecated) to ensure - it correctly handles all parameter combinations. - """ - server = FastMCP("test") - - # Create a test prompt - @server.prompt() - async def test_prompt(name: str) -> str: - """Test prompt""" - return f"Hello, {name}!" - - async with create_session(server._mcp_server) as client_session: - spies = stream_spy() - - # Test without params parameter (omitted) - _ = await client_session.list_prompts() - list_prompts_requests = spies.get_client_requests(method="prompts/list") - assert len(list_prompts_requests) == 1 - assert list_prompts_requests[0].params is None - - spies.clear() - - # Test with params=None - _ = await client_session.list_prompts(params=None) - list_prompts_requests = spies.get_client_requests(method="prompts/list") - assert len(list_prompts_requests) == 1 - assert list_prompts_requests[0].params is None - - spies.clear() - - # Test with empty params (for strict servers) - _ = await client_session.list_prompts(params=types.PaginatedRequestParams()) - list_prompts_requests = spies.get_client_requests(method="prompts/list") - assert len(list_prompts_requests) == 1 - assert list_prompts_requests[0].params is not None - assert list_prompts_requests[0].params.get("cursor") is None - - spies.clear() - - # Test with params containing cursor - _ = await client_session.list_prompts(params=types.PaginatedRequestParams(cursor="some_cursor")) - list_prompts_requests = spies.get_client_requests(method="prompts/list") - assert len(list_prompts_requests) == 1 - assert list_prompts_requests[0].params is not None - assert list_prompts_requests[0].params["cursor"] == "some_cursor" - + _ = await method(cursor="") + requests = spies.get_client_requests(method=request_method) + assert len(requests) == 1 + assert requests[0].params is not None + assert requests[0].params["cursor"] == "" + + +@pytest.mark.parametrize( + "method_name,request_method", + [ + ("list_tools", "tools/list"), + ("list_resources", "resources/list"), + ("list_prompts", "prompts/list"), + ("list_resource_templates", "resources/templates/list"), + ], +) +async def test_list_methods_params_parameter( + stream_spy: Callable[[], StreamSpyCollection], + full_featured_server: FastMCP, + method_name: str, + request_method: str, +): + """Test that the params parameter works correctly for list methods. -async def test_list_resource_templates_params_parameter(stream_spy: Callable[[], StreamSpyCollection]): - """Test that the params parameter works correctly for list_resource_templates. + Covers: list_tools, list_resources, list_prompts, list_resource_templates This tests the new params parameter API (non-deprecated) to ensure it correctly handles all parameter combinations. """ - server = FastMCP("test") - - # Create a test resource template - @server.resource("resource://test/{name}") - async def test_template(name: str) -> str: - """Test resource template""" - return f"Data for {name}" - - async with create_session(server._mcp_server) as client_session: + async with create_session(full_featured_server._mcp_server) as client_session: spies = stream_spy() + method = getattr(client_session, method_name) # Test without params parameter (omitted) - _ = await client_session.list_resource_templates() - list_templates_requests = spies.get_client_requests(method="resources/templates/list") - assert len(list_templates_requests) == 1 - assert list_templates_requests[0].params is None + _ = await method() + requests = spies.get_client_requests(method=request_method) + assert len(requests) == 1 + assert requests[0].params is None spies.clear() # Test with params=None - _ = await client_session.list_resource_templates(params=None) - list_templates_requests = spies.get_client_requests(method="resources/templates/list") - assert len(list_templates_requests) == 1 - assert list_templates_requests[0].params is None + _ = await method(params=None) + requests = spies.get_client_requests(method=request_method) + assert len(requests) == 1 + assert requests[0].params is None spies.clear() # Test with empty params (for strict servers) - _ = await client_session.list_resource_templates(params=types.PaginatedRequestParams()) - list_templates_requests = spies.get_client_requests(method="resources/templates/list") - assert len(list_templates_requests) == 1 - assert list_templates_requests[0].params is not None - assert list_templates_requests[0].params.get("cursor") is None + _ = await method(params=types.PaginatedRequestParams()) + requests = spies.get_client_requests(method=request_method) + assert len(requests) == 1 + assert requests[0].params is not None + assert requests[0].params.get("cursor") is None spies.clear() # Test with params containing cursor - _ = await client_session.list_resource_templates(params=types.PaginatedRequestParams(cursor="some_cursor")) - list_templates_requests = spies.get_client_requests(method="resources/templates/list") - assert len(list_templates_requests) == 1 - assert list_templates_requests[0].params is not None - assert list_templates_requests[0].params["cursor"] == "some_cursor" - - -@pytest.mark.filterwarnings("ignore::DeprecationWarning") -async def test_list_tools_params_takes_precedence_over_cursor( - stream_spy: Callable[[], StreamSpyCollection], -): - """Test that params parameter takes precedence over cursor parameter. - - When both cursor and params are provided, params should be used and - cursor should be ignored, ensuring safe migration path. - """ - server = FastMCP("test") - - @server.tool(name="test_tool") - async def test_tool() -> str: - """Test tool""" - return "Result" - - async with create_session(server._mcp_server) as client_session: - spies = stream_spy() - - # Call with both cursor and params - params should take precedence - _ = await client_session.list_tools( - cursor="old_cursor", - params=types.PaginatedRequestParams(cursor="new_cursor"), - ) - list_tools_requests = spies.get_client_requests(method="tools/list") - assert len(list_tools_requests) == 1 - # Verify params takes precedence (new_cursor should be used, not old_cursor) - assert list_tools_requests[0].params is not None - assert list_tools_requests[0].params["cursor"] == "new_cursor" - - + _ = await method(params=types.PaginatedRequestParams(cursor="some_cursor_value")) + requests = spies.get_client_requests(method=request_method) + assert len(requests) == 1 + assert requests[0].params is not None + assert requests[0].params["cursor"] == "some_cursor_value" + + +@pytest.mark.parametrize( + "method_name,request_method", + [ + ("list_tools", "tools/list"), + ("list_resources", "resources/list"), + ("list_prompts", "prompts/list"), + ("list_resource_templates", "resources/templates/list"), + ], +) @pytest.mark.filterwarnings("ignore::DeprecationWarning") -async def test_list_resources_params_takes_precedence_over_cursor( +async def test_list_methods_params_takes_precedence_over_cursor( stream_spy: Callable[[], StreamSpyCollection], + full_featured_server: FastMCP, + method_name: str, + request_method: str, ): """Test that params parameter takes precedence over cursor parameter. - When both cursor and params are provided, params should be used and - cursor should be ignored, ensuring safe migration path. - """ - server = FastMCP("test") - - @server.resource("resource://test/data") - async def test_resource() -> str: - """Test resource""" - return "Test data" - - async with create_session(server._mcp_server) as client_session: - spies = stream_spy() - - # Call with both cursor and params - params should take precedence - _ = await client_session.list_resources( - cursor="old_cursor", - params=types.PaginatedRequestParams(cursor="new_cursor"), - ) - list_resources_requests = spies.get_client_requests(method="resources/list") - assert len(list_resources_requests) == 1 - # Verify params takes precedence (new_cursor should be used, not old_cursor) - assert list_resources_requests[0].params is not None - assert list_resources_requests[0].params["cursor"] == "new_cursor" - - -@pytest.mark.filterwarnings("ignore::DeprecationWarning") -async def test_list_prompts_params_takes_precedence_over_cursor( - stream_spy: Callable[[], StreamSpyCollection], -): - """Test that params parameter takes precedence over cursor parameter. + Covers: list_tools, list_resources, list_prompts, list_resource_templates When both cursor and params are provided, params should be used and cursor should be ignored, ensuring safe migration path. """ - server = FastMCP("test") - - @server.prompt() - async def test_prompt(name: str) -> str: - """Test prompt""" - return f"Hello, {name}!" - - async with create_session(server._mcp_server) as client_session: + async with create_session(full_featured_server._mcp_server) as client_session: spies = stream_spy() + method = getattr(client_session, method_name) # Call with both cursor and params - params should take precedence - _ = await client_session.list_prompts( + _ = await method( cursor="old_cursor", params=types.PaginatedRequestParams(cursor="new_cursor"), ) - list_prompts_requests = spies.get_client_requests(method="prompts/list") - assert len(list_prompts_requests) == 1 - # Verify params takes precedence (new_cursor should be used, not old_cursor) - assert list_prompts_requests[0].params is not None - assert list_prompts_requests[0].params["cursor"] == "new_cursor" - - -@pytest.mark.filterwarnings("ignore::DeprecationWarning") -async def test_list_resource_templates_params_takes_precedence_over_cursor( - stream_spy: Callable[[], StreamSpyCollection], -): - """Test that params parameter takes precedence over cursor parameter. - - When both cursor and params are provided, params should be used and - cursor should be ignored, ensuring safe migration path. - """ - server = FastMCP("test") - - @server.resource("resource://test/{name}") - async def test_template(name: str) -> str: - """Test resource template""" - return f"Data for {name}" - - async with create_session(server._mcp_server) as client_session: - spies = stream_spy() - - # Call with both cursor and params - params should take precedence - _ = await client_session.list_resource_templates( - cursor="old_cursor", - params=types.PaginatedRequestParams(cursor="new_cursor"), - ) - list_templates_requests = spies.get_client_requests( - method="resources/templates/list" - ) - assert len(list_templates_requests) == 1 + requests = spies.get_client_requests(method=request_method) + assert len(requests) == 1 # Verify params takes precedence (new_cursor should be used, not old_cursor) - assert list_templates_requests[0].params is not None - assert list_templates_requests[0].params["cursor"] == "new_cursor" + assert requests[0].params is not None + assert requests[0].params["cursor"] == "new_cursor" async def test_list_tools_with_strict_server_validation(): From 3217282179be31325b77f374d67d3d4684133128 Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Mon, 13 Oct 2025 17:19:16 +0100 Subject: [PATCH 5/8] refactor: add @overload decorators and @deprecated for list methods Implements feedback from maxisbey on PR #1453. This change improves type safety and deprecation handling for list methods (list_tools, list_resources, list_prompts, list_resource_templates): - Add @overload decorators with three signatures to provide clear IDE typing: - cursor: str (positional argument) - params: PaginatedRequestParams (keyword-only argument) - no arguments - Replace warnings.warn() with @deprecated decorator from typing_extensions for cleaner deprecation signaling - Add validation to raise ValueError when both cursor and params are provided, preventing ambiguous parameter combinations - Update tests to expect ValueError when both parameters are specified This ensures the Python SDK has type checking equivalent to the TypeScript SDK and provides better developer experience through IDE assistance and clear deprecation messaging. --- README.md | 4 +- .../snippets/clients/pagination_client.py | 4 +- src/mcp/client/session.py | 140 ++++++++++++------ tests/client/test_list_methods_cursor.py | 38 ++--- 4 files changed, 112 insertions(+), 74 deletions(-) diff --git a/README.md b/README.md index 48d1c1742..a0daf5c6d 100644 --- a/README.md +++ b/README.md @@ -1840,7 +1840,7 @@ import asyncio from mcp.client.session import ClientSession from mcp.client.stdio import StdioServerParameters, stdio_client -from mcp.types import Resource +from mcp.types import PaginatedRequestParams, Resource async def list_all_resources() -> None: @@ -1857,7 +1857,7 @@ async def list_all_resources() -> None: while True: # Fetch a page of resources - result = await session.list_resources(cursor=cursor) + result = await session.list_resources(params=PaginatedRequestParams(cursor=cursor)) all_resources.extend(result.resources) print(f"Fetched {len(result.resources)} resources") diff --git a/examples/snippets/clients/pagination_client.py b/examples/snippets/clients/pagination_client.py index 4df1aec60..1805d2d31 100644 --- a/examples/snippets/clients/pagination_client.py +++ b/examples/snippets/clients/pagination_client.py @@ -6,7 +6,7 @@ from mcp.client.session import ClientSession from mcp.client.stdio import StdioServerParameters, stdio_client -from mcp.types import Resource +from mcp.types import PaginatedRequestParams, Resource async def list_all_resources() -> None: @@ -23,7 +23,7 @@ async def list_all_resources() -> None: while True: # Fetch a page of resources - result = await session.list_resources(cursor=cursor) + result = await session.list_resources(params=PaginatedRequestParams(cursor=cursor)) all_resources.extend(result.resources) print(f"Fetched {len(result.resources)} resources") diff --git a/src/mcp/client/session.py b/src/mcp/client/session.py index 07167f59f..0acf88715 100644 --- a/src/mcp/client/session.py +++ b/src/mcp/client/session.py @@ -1,12 +1,12 @@ import logging -import warnings from datetime import timedelta -from typing import Any, Protocol +from typing import Any, Protocol, overload import anyio.lowlevel from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream from jsonschema import SchemaError, ValidationError, validate from pydantic import AnyUrl, TypeAdapter +from typing_extensions import deprecated import mcp.types as types from mcp.shared.context import RequestContext @@ -213,6 +213,16 @@ async def set_logging_level(self, level: types.LoggingLevel) -> types.EmptyResul types.EmptyResult, ) + @deprecated("Use params=PaginatedRequestParams(...) instead") + @overload + async def list_resources(self, cursor: str | None) -> types.ListResourcesResult: ... + + @overload + async def list_resources(self, *, params: types.PaginatedRequestParams | None) -> types.ListResourcesResult: ... + + @overload + async def list_resources(self) -> types.ListResourcesResult: ... + async def list_resources( self, cursor: str | None = None, @@ -222,23 +232,36 @@ async def list_resources( """Send a resources/list request. Args: - cursor: (Deprecated) Pagination cursor. Use params parameter instead. - params: Pagination parameters. Defaults to None (omits params field). + cursor: Simple cursor string for pagination (deprecated, use params instead) + params: Full pagination parameters including cursor and any future fields """ - if cursor is not None: - warnings.warn( - "cursor parameter is deprecated, use params=PaginatedRequestParams(cursor=...) instead", - DeprecationWarning, - stacklevel=2, - ) - if params is None: - params = types.PaginatedRequestParams(cursor=cursor) + if params is not None and cursor is not None: + raise ValueError("Cannot specify both cursor and params") + + if params is not None: + request_params = params + elif cursor is not None: + request_params = types.PaginatedRequestParams(cursor=cursor) + else: + request_params = None return await self.send_request( - types.ClientRequest(types.ListResourcesRequest(params=params)), + types.ClientRequest(types.ListResourcesRequest(params=request_params)), types.ListResourcesResult, ) + @deprecated("Use params=PaginatedRequestParams(...) instead") + @overload + async def list_resource_templates(self, cursor: str | None) -> types.ListResourceTemplatesResult: ... + + @overload + async def list_resource_templates( + self, *, params: types.PaginatedRequestParams | None + ) -> types.ListResourceTemplatesResult: ... + + @overload + async def list_resource_templates(self) -> types.ListResourceTemplatesResult: ... + async def list_resource_templates( self, cursor: str | None = None, @@ -248,20 +271,21 @@ async def list_resource_templates( """Send a resources/templates/list request. Args: - cursor: (Deprecated) Pagination cursor. Use params parameter instead. - params: Pagination parameters. Defaults to None (omits params field). + cursor: Simple cursor string for pagination (deprecated, use params instead) + params: Full pagination parameters including cursor and any future fields """ - if cursor is not None: - warnings.warn( - "cursor parameter is deprecated, use params=PaginatedRequestParams(cursor=...) instead", - DeprecationWarning, - stacklevel=2, - ) - if params is None: - params = types.PaginatedRequestParams(cursor=cursor) + if params is not None and cursor is not None: + raise ValueError("Cannot specify both cursor and params") + + if params is not None: + request_params = params + elif cursor is not None: + request_params = types.PaginatedRequestParams(cursor=cursor) + else: + request_params = None return await self.send_request( - types.ClientRequest(types.ListResourceTemplatesRequest(params=params)), + types.ClientRequest(types.ListResourceTemplatesRequest(params=request_params)), types.ListResourceTemplatesResult, ) @@ -330,7 +354,7 @@ async def _validate_tool_result(self, name: str, result: types.CallToolResult) - """Validate the structured content of a tool result against its output schema.""" if name not in self._tool_output_schemas: # refresh output schema cache - await self.list_tools() + await self.list_tools() # type: ignore[reportDeprecated] output_schema = None if name in self._tool_output_schemas: @@ -348,6 +372,16 @@ async def _validate_tool_result(self, name: str, result: types.CallToolResult) - except SchemaError as e: raise RuntimeError(f"Invalid schema for tool {name}: {e}") + @deprecated("Use params=PaginatedRequestParams(...) instead") + @overload + async def list_prompts(self, cursor: str | None) -> types.ListPromptsResult: ... + + @overload + async def list_prompts(self, *, params: types.PaginatedRequestParams | None) -> types.ListPromptsResult: ... + + @overload + async def list_prompts(self) -> types.ListPromptsResult: ... + async def list_prompts( self, cursor: str | None = None, @@ -357,20 +391,21 @@ async def list_prompts( """Send a prompts/list request. Args: - cursor: (Deprecated) Pagination cursor. Use params parameter instead. - params: Pagination parameters. Defaults to None (omits params field). + cursor: Simple cursor string for pagination (deprecated, use params instead) + params: Full pagination parameters including cursor and any future fields """ - if cursor is not None: - warnings.warn( - "cursor parameter is deprecated, use params=PaginatedRequestParams(cursor=...) instead", - DeprecationWarning, - stacklevel=2, - ) - if params is None: - params = types.PaginatedRequestParams(cursor=cursor) + if params is not None and cursor is not None: + raise ValueError("Cannot specify both cursor and params") + + if params is not None: + request_params = params + elif cursor is not None: + request_params = types.PaginatedRequestParams(cursor=cursor) + else: + request_params = None return await self.send_request( - types.ClientRequest(types.ListPromptsRequest(params=params)), + types.ClientRequest(types.ListPromptsRequest(params=request_params)), types.ListPromptsResult, ) @@ -409,6 +444,16 @@ async def complete( types.CompleteResult, ) + @deprecated("Use params=PaginatedRequestParams(...) instead") + @overload + async def list_tools(self, cursor: str | None) -> types.ListToolsResult: ... + + @overload + async def list_tools(self, *, params: types.PaginatedRequestParams | None) -> types.ListToolsResult: ... + + @overload + async def list_tools(self) -> types.ListToolsResult: ... + async def list_tools( self, cursor: str | None = None, @@ -418,20 +463,21 @@ async def list_tools( """Send a tools/list request. Args: - cursor: (Deprecated) Pagination cursor. Use params parameter instead. - params: Pagination parameters. Defaults to None (omits params field). + cursor: Simple cursor string for pagination (deprecated, use params instead) + params: Full pagination parameters including cursor and any future fields """ - if cursor is not None: - warnings.warn( - "cursor parameter is deprecated, use params=PaginatedRequestParams(cursor=...) instead", - DeprecationWarning, - stacklevel=2, - ) - if params is None: - params = types.PaginatedRequestParams(cursor=cursor) + if params is not None and cursor is not None: + raise ValueError("Cannot specify both cursor and params") + + if params is not None: + request_params = params + elif cursor is not None: + request_params = types.PaginatedRequestParams(cursor=cursor) + else: + request_params = None result = await self.send_request( - types.ClientRequest(types.ListToolsRequest(params=params)), + types.ClientRequest(types.ListToolsRequest(params=request_params)), types.ListToolsResult, ) diff --git a/tests/client/test_list_methods_cursor.py b/tests/client/test_list_methods_cursor.py index d39366c96..e99f622f4 100644 --- a/tests/client/test_list_methods_cursor.py +++ b/tests/client/test_list_methods_cursor.py @@ -165,42 +165,34 @@ async def test_list_methods_params_parameter( @pytest.mark.parametrize( - "method_name,request_method", + "method_name", [ - ("list_tools", "tools/list"), - ("list_resources", "resources/list"), - ("list_prompts", "prompts/list"), - ("list_resource_templates", "resources/templates/list"), + "list_tools", + "list_resources", + "list_prompts", + "list_resource_templates", ], ) -@pytest.mark.filterwarnings("ignore::DeprecationWarning") -async def test_list_methods_params_takes_precedence_over_cursor( - stream_spy: Callable[[], StreamSpyCollection], +async def test_list_methods_raises_error_when_both_cursor_and_params_provided( full_featured_server: FastMCP, method_name: str, - request_method: str, ): - """Test that params parameter takes precedence over cursor parameter. + """Test that providing both cursor and params raises ValueError. Covers: list_tools, list_resources, list_prompts, list_resource_templates - When both cursor and params are provided, params should be used and - cursor should be ignored, ensuring safe migration path. + When both cursor and params are provided, a ValueError should be raised + to prevent ambiguity. """ async with create_session(full_featured_server._mcp_server) as client_session: - spies = stream_spy() method = getattr(client_session, method_name) - # Call with both cursor and params - params should take precedence - _ = await method( - cursor="old_cursor", - params=types.PaginatedRequestParams(cursor="new_cursor"), - ) - requests = spies.get_client_requests(method=request_method) - assert len(requests) == 1 - # Verify params takes precedence (new_cursor should be used, not old_cursor) - assert requests[0].params is not None - assert requests[0].params["cursor"] == "new_cursor" + # Call with both cursor and params - should raise ValueError + with pytest.raises(ValueError, match="Cannot specify both cursor and params"): + await method( + cursor="old_cursor", + params=types.PaginatedRequestParams(cursor="new_cursor"), + ) async def test_list_tools_with_strict_server_validation(): From d4b0608e7322bf71e39f42f7d61bcd681468ea05 Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Mon, 13 Oct 2025 17:57:21 +0100 Subject: [PATCH 6/8] fix: flip order of @overload & @deprecated to match docs See: https://docs.python.org/3.13/library/warnings.html#warnings.deprecated @deprecated combined with @overload is specifically called out to be placed _after_ the @overload. --- src/mcp/client/session.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/mcp/client/session.py b/src/mcp/client/session.py index 0acf88715..e9911721b 100644 --- a/src/mcp/client/session.py +++ b/src/mcp/client/session.py @@ -213,8 +213,8 @@ async def set_logging_level(self, level: types.LoggingLevel) -> types.EmptyResul types.EmptyResult, ) - @deprecated("Use params=PaginatedRequestParams(...) instead") @overload + @deprecated("Use params=PaginatedRequestParams(...) instead") async def list_resources(self, cursor: str | None) -> types.ListResourcesResult: ... @overload @@ -250,8 +250,8 @@ async def list_resources( types.ListResourcesResult, ) - @deprecated("Use params=PaginatedRequestParams(...) instead") @overload + @deprecated("Use params=PaginatedRequestParams(...) instead") async def list_resource_templates(self, cursor: str | None) -> types.ListResourceTemplatesResult: ... @overload @@ -372,8 +372,8 @@ async def _validate_tool_result(self, name: str, result: types.CallToolResult) - except SchemaError as e: raise RuntimeError(f"Invalid schema for tool {name}: {e}") - @deprecated("Use params=PaginatedRequestParams(...) instead") @overload + @deprecated("Use params=PaginatedRequestParams(...) instead") async def list_prompts(self, cursor: str | None) -> types.ListPromptsResult: ... @overload @@ -444,8 +444,8 @@ async def complete( types.CompleteResult, ) - @deprecated("Use params=PaginatedRequestParams(...) instead") @overload + @deprecated("Use params=PaginatedRequestParams(...) instead") async def list_tools(self, cursor: str | None) -> types.ListToolsResult: ... @overload From 28c6f0d64e39f5087bd82e47b08d902b051a4304 Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Mon, 13 Oct 2025 18:03:31 +0100 Subject: [PATCH 7/8] fix: improve deprecation warning string --- src/mcp/client/session.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/mcp/client/session.py b/src/mcp/client/session.py index e9911721b..3a4a349ec 100644 --- a/src/mcp/client/session.py +++ b/src/mcp/client/session.py @@ -214,7 +214,7 @@ async def set_logging_level(self, level: types.LoggingLevel) -> types.EmptyResul ) @overload - @deprecated("Use params=PaginatedRequestParams(...) instead") + @deprecated("Use list_resources(params=PaginatedRequestParams(...)) instead") async def list_resources(self, cursor: str | None) -> types.ListResourcesResult: ... @overload @@ -251,7 +251,7 @@ async def list_resources( ) @overload - @deprecated("Use params=PaginatedRequestParams(...) instead") + @deprecated("Use list_resource_templates(params=PaginatedRequestParams(...)) instead") async def list_resource_templates(self, cursor: str | None) -> types.ListResourceTemplatesResult: ... @overload @@ -373,7 +373,7 @@ async def _validate_tool_result(self, name: str, result: types.CallToolResult) - raise RuntimeError(f"Invalid schema for tool {name}: {e}") @overload - @deprecated("Use params=PaginatedRequestParams(...) instead") + @deprecated("Use list_prompts(params=PaginatedRequestParams(...)) instead") async def list_prompts(self, cursor: str | None) -> types.ListPromptsResult: ... @overload @@ -445,7 +445,7 @@ async def complete( ) @overload - @deprecated("Use params=PaginatedRequestParams(...) instead") + @deprecated("Use list_tools(params=PaginatedRequestParams(...)) instead") async def list_tools(self, cursor: str | None) -> types.ListToolsResult: ... @overload From ee989cf2bf1476277faddfee077d62e39e759b30 Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Tue, 14 Oct 2025 11:04:24 +0100 Subject: [PATCH 8/8] fix: remove unnecessary lint ignore --- src/mcp/client/session.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/mcp/client/session.py b/src/mcp/client/session.py index 3a4a349ec..45c57089c 100644 --- a/src/mcp/client/session.py +++ b/src/mcp/client/session.py @@ -354,7 +354,7 @@ async def _validate_tool_result(self, name: str, result: types.CallToolResult) - """Validate the structured content of a tool result against its output schema.""" if name not in self._tool_output_schemas: # refresh output schema cache - await self.list_tools() # type: ignore[reportDeprecated] + await self.list_tools() output_schema = None if name in self._tool_output_schemas: