modelcontextprotocol/python-sdk
Watch
1
0
Fork
You've already forked python-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix: send params as empty object for list methods without cursor (#1453)

Browse source
This commit is contained in:
Felix Weinberger 2025-10-14 11:11:38 +01:00 • committed by GitHub
commit de89457683
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 301 additions and 180 deletions
Download patch file Download diff file Expand all files Collapse all files

4
README.md
View file

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

4
examples/snippets/clients/pagination_client.py
View file

@ -6,7 +6,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:
@ -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")

165
src/mcp/client/session.py
View file

@ -1,11 +1,12 @@
import logging
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
@ -212,25 +213,79 @@ class ClientSession(
types.EmptyResult,
)
async def list_resources(self, cursor: str | None = None) -> types.ListResourcesResult:
"""Send a resources/list request."""
@overload
@deprecated("Use list_resources(params=PaginatedRequestParams(...)) instead")
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,
*,
params: types.PaginatedRequestParams | None = None,
) -> types.ListResourcesResult:
"""Send a resources/list request.
Args:
cursor: Simple cursor string for pagination (deprecated, use params instead)
params: Full pagination parameters including cursor and any future fields
"""
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=types.PaginatedRequestParams(cursor=cursor) if cursor is not None else None,
)
),
types.ClientRequest(types.ListResourcesRequest(params=request_params)),
types.ListResourcesResult,
)
async def list_resource_templates(self, cursor: str | None = None) -> types.ListResourceTemplatesResult:
"""Send a resources/templates/list request."""
@overload
@deprecated("Use list_resource_templates(params=PaginatedRequestParams(...)) instead")
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,
*,
params: types.PaginatedRequestParams | None = None,
) -> types.ListResourceTemplatesResult:
"""Send a resources/templates/list request.
Args:
cursor: Simple cursor string for pagination (deprecated, use params instead)
params: Full pagination parameters including cursor and any future fields
"""
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=types.PaginatedRequestParams(cursor=cursor) if cursor is not None else None,
)
),
types.ClientRequest(types.ListResourceTemplatesRequest(params=request_params)),
types.ListResourceTemplatesResult,
)
@ -317,14 +372,40 @@ class ClientSession(
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."""
@overload
@deprecated("Use list_prompts(params=PaginatedRequestParams(...)) instead")
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,
*,
params: types.PaginatedRequestParams | None = None,
) -> types.ListPromptsResult:
"""Send a prompts/list request.
Args:
cursor: Simple cursor string for pagination (deprecated, use params instead)
params: Full pagination parameters including cursor and any future fields
"""
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=types.PaginatedRequestParams(cursor=cursor) if cursor is not None else None,
)
),
types.ClientRequest(types.ListPromptsRequest(params=request_params)),
types.ListPromptsResult,
)
@ -363,14 +444,40 @@ class ClientSession(
types.CompleteResult,
)
async def list_tools(self, cursor: str | None = None) -> types.ListToolsResult:
"""Send a tools/list request."""
@overload
@deprecated("Use list_tools(params=PaginatedRequestParams(...)) instead")
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,
*,
params: types.PaginatedRequestParams | None = None,
) -> types.ListToolsResult:
"""Send a tools/list request.
Args:
cursor: Simple cursor string for pagination (deprecated, use params instead)
params: Full pagination parameters including cursor and any future fields
"""
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=types.PaginatedRequestParams(cursor=cursor) if cursor is not None else None,
)
),
types.ClientRequest(types.ListToolsRequest(params=request_params)),
types.ListToolsResult,
)

334
tests/client/test_list_methods_cursor.py
View file

@ -2,23 +2,22 @@ from collections.abc import Callable
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
from mcp.types import ListToolsRequest, ListToolsResult
from .conftest import StreamSpyCollection
pytestmark = pytest.mark.anyio
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"""
@ -29,186 +28,201 @@ async def test_list_tools_cursor_parameter(stream_spy: Callable[[], StreamSpyCol
"""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"] == ""
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"] == ""
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()
# 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"] == ""
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.
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:
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_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
"""
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"] == ""
_ = 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.
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.
"""
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 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 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 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 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",
[
"list_tools",
"list_resources",
"list_prompts",
"list_resource_templates",
],
)
async def test_list_methods_raises_error_when_both_cursor_and_params_provided(
full_featured_server: FastMCP,
method_name: str,
):
"""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, a ValueError should be raised
to prevent ambiguity.
"""
async with create_session(full_featured_server._mcp_server) as client_session:
method = getattr(client_session, method_name)
# 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():
"""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:
# Use params to explicitly send params: {} for strict server compatibility
result = await client_session.list_tools(params=types.PaginatedRequestParams())
assert result is not None