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

refactor: flatten the methods in Client (#1914)

Browse source
This commit is contained in:
Marcelo Trylesinski 2026-01-20 13:12:07 +01:00 • committed by GitHub
commit bcb07c2b25
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 213 additions and 315 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.github/workflows/shared.yml vendored
View file

@ -58,7 +58,7 @@ jobs:
- name: Run pytest with coverage
shell: bash
run: |
uv run --frozen --no-sync coverage run -m pytest
uv run --frozen --no-sync coverage run -m pytest -n auto
uv run --frozen --no-sync coverage combine
uv run --frozen --no-sync coverage report

3
CLAUDE.md
View file

@ -17,6 +17,7 @@ This document contains critical information about working with this codebase. Fo
- Functions must be focused and small
- Follow existing patterns exactly
- Line length: 120 chars maximum
- FORBIDDEN: imports inside functions
3. Testing Requirements
- Framework: `uv run --frozen pytest`
@ -25,6 +26,8 @@ This document contains critical information about working with this codebase. Fo
- Coverage: test edge cases and errors
- New features require tests
- Bug fixes require regression tests
- IMPORTANT: The `tests/client/test_client.py` is the most well designed test file. Follow its patterns.
- IMPORTANT: Be minimal, and focus on E2E tests: Use the `mcp.client.Client` whenever possible.
- For commits fixing bugs or adding features based on user reports add:

1
pyproject.toml
View file

@ -171,7 +171,6 @@ xfail_strict = true
addopts = """
--color=yes
--capture=fd
--numprocesses auto
"""
filterwarnings = [
"error",

7
src/mcp/client/_memory.py
View file

@ -36,12 +36,7 @@ class InMemoryTransport:
result = await client.call_tool("my_tool", {...})
"""
def __init__(
self,
server: Server[Any] | FastMCP,
*,
raise_exceptions: bool = False,
) -> None:
def __init__(self, server: Server[Any] | FastMCP, *, raise_exceptions: bool = False) -> None:
"""Initialize the in-memory transport.
Args:

61
src/mcp/client/client.py
View file

@ -58,6 +58,7 @@ class Client:
self,
server: Server[Any] | FastMCP,
*,
# TODO(Marcelo): When do `raise_exceptions=True` actually raises?
raise_exceptions: bool = False,
read_timeout_seconds: float | None = None,
sampling_callback: SamplingFnT | None = None,
@ -125,14 +126,9 @@ class Client:
self._exit_stack = exit_stack.pop_all()
return self
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: Any,
) -> None:
async def __aexit__(self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: Any) -> None:
"""Exit the async context manager."""
if self._exit_stack:
if self._exit_stack: # pragma: no branch
await self._exit_stack.__aexit__(exc_type, exc_val, exc_tb)
self._session = None
@ -177,28 +173,22 @@ class Client:
"""Set the logging level on the server."""
return await self.session.set_logging_level(level)
async def list_resources(
self,
params: types.PaginatedRequestParams | None = None,
) -> types.ListResourcesResult:
async def list_resources(self, *, cursor: str | None = None) -> types.ListResourcesResult:
"""List available resources from the server."""
return await self.session.list_resources(params=params)
return await self.session.list_resources(params=types.PaginatedRequestParams(cursor=cursor))
async def list_resource_templates(
self,
params: types.PaginatedRequestParams | None = None,
) -> types.ListResourceTemplatesResult:
async def list_resource_templates(self, *, cursor: str | None = None) -> types.ListResourceTemplatesResult:
"""List available resource templates from the server."""
return await self.session.list_resource_templates(params=params)
return await self.session.list_resource_templates(params=types.PaginatedRequestParams(cursor=cursor))
async def read_resource(self, uri: str | AnyUrl) -> types.ReadResourceResult:
"""Read a resource from the server.
Args:
uri: The URI of the resource to read
uri: The URI of the resource to read.
Returns:
The resource content
The resource content.
"""
return await self.session.read_resource(uri)
@ -239,18 +229,11 @@ class Client:
meta=meta,
)
async def list_prompts(
self,
params: types.PaginatedRequestParams | None = None,
) -> types.ListPromptsResult:
async def list_prompts(self, *, cursor: str | None = None) -> types.ListPromptsResult:
"""List available prompts from the server."""
return await self.session.list_prompts(params=params)
return await self.session.list_prompts(params=types.PaginatedRequestParams(cursor=cursor))
async def get_prompt(
self,
name: str,
arguments: dict[str, str] | None = None,
) -> types.GetPromptResult:
async def get_prompt(self, name: str, arguments: dict[str, str] | None = None) -> types.GetPromptResult:
"""Get a prompt from the server.
Args:
@ -258,7 +241,7 @@ class Client:
arguments: Arguments to pass to the prompt
Returns:
The prompt content
The prompt content.
"""
return await self.session.get_prompt(name=name, arguments=arguments)
@ -276,21 +259,15 @@ class Client:
context_arguments: Additional context arguments
Returns:
Completion suggestions
Completion suggestions.
"""
return await self.session.complete(
ref=ref,
argument=argument,
context_arguments=context_arguments,
)
return await self.session.complete(ref=ref, argument=argument, context_arguments=context_arguments)
async def list_tools(
self,
params: types.PaginatedRequestParams | None = None,
) -> types.ListToolsResult:
async def list_tools(self, *, cursor: str | None = None) -> types.ListToolsResult:
"""List available tools from the server."""
return await self.session.list_tools(params=params)
return await self.session.list_tools(params=types.PaginatedRequestParams(cursor=cursor))
async def send_roots_list_changed(self) -> None:
"""Send a notification that the roots list has changed."""
await self.session.send_roots_list_changed()
# TODO(Marcelo): Currently, there is no way for the server to handle this. We should add support.
await self.session.send_roots_list_changed() # pragma: no cover

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

@ -349,10 +349,7 @@ class ClientSession(
Args:
params: Full pagination parameters including cursor and any future fields
"""
return await self.send_request(
types.ListPromptsRequest(params=params),
types.ListPromptsResult,
)
return await self.send_request(types.ListPromptsRequest(params=params), types.ListPromptsResult)
async def get_prompt(self, name: str, arguments: dict[str, str] | None = None) -> types.GetPromptResult:
"""Send a prompts/get request."""

40
src/mcp/shared/session.py
View file

@ -234,12 +234,12 @@ class BaseSession(
metadata: MessageMetadata = None,
progress_callback: ProgressFnT | None = None,
) -> ReceiveResultT:
"""Sends a request and wait for a response. Raises an McpError if the
response contains an error. If a request read timeout is provided, it
will take precedence over the session read timeout.
"""Sends a request and wait for a response.
Do not use this method to emit notifications! Use send_notification()
instead.
Raises an McpError if the response contains an error. If a request read timeout is provided, it will take
precedence over the session read timeout.
Do not use this method to emit notifications! Use send_notification() instead.
"""
request_id = self._request_id
self._request_id = request_id + 1
@ -261,15 +261,10 @@ class BaseSession(
try:
jsonrpc_request = JSONRPCRequest(jsonrpc="2.0", id=request_id, **request_data)
await self._write_stream.send(SessionMessage(message=jsonrpc_request, metadata=metadata))
# request read timeout takes precedence over session read timeout
timeout = None
if request_read_timeout_seconds is not None: # pragma: no cover
timeout = request_read_timeout_seconds
elif self._session_read_timeout_seconds is not None: # pragma: no cover
timeout = self._session_read_timeout_seconds
timeout = request_read_timeout_seconds or self._session_read_timeout_seconds
try:
with anyio.fail_after(timeout):
@ -279,9 +274,8 @@ class BaseSession(
ErrorData(
code=httpx.codes.REQUEST_TIMEOUT,
message=(
f"Timed out while waiting for response to "
f"{request.__class__.__name__}. Waited "
f"{timeout} seconds."
f"Timed out while waiting for response to {request.__class__.__name__}. "
f"Waited {timeout} seconds."
),
)
)
@ -302,9 +296,7 @@ class BaseSession(
notification: SendNotificationT,
related_request_id: RequestId | None = None,
) -> None:
"""Emits a notification, which is a one-way message that does not expect
a response.
"""
"""Emits a notification, which is a one-way message that does not expect a response."""
# Some transport implementations may need to set the related_request_id
# to attribute to the notifications to the request that triggered them.
jsonrpc_notification = JSONRPCNotification(
@ -373,11 +365,7 @@ class BaseSession(
error_response = JSONRPCError(
jsonrpc="2.0",
id=message.message.id,
error=ErrorData(
code=INVALID_PARAMS,
message="Invalid request parameters",
data="",
),
error=ErrorData(code=INVALID_PARAMS, message="Invalid request parameters", data=""),
)
session_message = SessionMessage(message=error_response)
await self._write_stream.send(session_message)
@ -518,13 +506,9 @@ class BaseSession(
total: float | None = None,
message: str | None = None,
) -> None:
"""Sends a progress notification for a request that is currently being
processed.
"""
"""Sends a progress notification for a request that is currently being processed."""
async def _handle_incoming(
self,
req: RequestResponder[ReceiveRequestT, SendResultT] | ReceiveNotificationT | Exception,
self, req: RequestResponder[ReceiveRequestT, SendResultT] | ReceiveNotificationT | Exception
) -> None:
"""A generic handler for incoming messages. Overwritten by subclasses."""
pass # pragma: no cover

409
tests/client/test_client.py
View file

@ -1,14 +1,36 @@
"""Tests for the unified Client class."""
from unittest.mock import AsyncMock, patch
from __future__ import annotations
import anyio
import pytest
from inline_snapshot import snapshot
import mcp.types as types
from mcp.client.client import Client
from mcp.server import Server
from mcp.server.fastmcp import FastMCP
from mcp.types import EmptyResult, Resource
from mcp.types import (
CallToolResult,
EmptyResult,
GetPromptResult,
ListPromptsResult,
ListResourcesResult,
ListResourceTemplatesResult,
ListToolsResult,
Prompt,
PromptArgument,
PromptMessage,
PromptsCapability,
ReadResourceResult,
Resource,
ResourcesCapability,
ServerCapabilities,
TextContent,
TextResourceContents,
Tool,
ToolsCapability,
)
pytestmark = pytest.mark.anyio
@ -20,13 +42,27 @@ def simple_server() -> Server:
@server.list_resources()
async def handle_list_resources():
return [
Resource(
uri="memory://test",
name="Test Resource",
description="A test resource",
)
]
return [Resource(uri="memory://test", name="Test Resource", description="A test resource")]
@server.subscribe_resource()
async def handle_subscribe_resource(uri: str):
pass
@server.unsubscribe_resource()
async def handle_unsubscribe_resource(uri: str):
pass
@server.set_logging_level()
async def handle_set_logging_level(level: str):
pass
@server.completion()
async def handle_completion(
ref: types.PromptReference | types.ResourceTemplateReference,
argument: types.CompletionArgument,
context: types.CompletionContext | None,
) -> types.Completion | None:
return types.Completion(values=[])
return server
@ -41,11 +77,6 @@ def app() -> FastMCP:
"""Greet someone by name."""
return f"Hello, {name}!"
@server.tool()
def add(a: int, b: int) -> int:
"""Add two numbers."""
return a + b
@server.resource("test://resource")
def test_resource() -> str:
"""A test resource."""
@ -59,278 +90,198 @@ def app() -> FastMCP:
return server
async def test_creates_client(app: FastMCP):
"""Test that from_server creates a connected client."""
async with Client(app) as client:
assert client is not None
async def test_client_is_initialized(app: FastMCP):
"""Test that the client is initialized after entering context."""
async with Client(app) as client:
caps = client.server_capabilities
assert caps is not None
assert caps.tools is not None
assert client.server_capabilities == snapshot(
ServerCapabilities(
experimental={},
prompts=PromptsCapability(list_changed=False),
resources=ResourcesCapability(subscribe=False, list_changed=False),
tools=ToolsCapability(list_changed=False),
)
)
async def test_with_simple_server(simple_server: Server):
async def test_client_with_simple_server(simple_server: Server):
"""Test that from_server works with a basic Server instance."""
async with Client(simple_server) as client:
assert client is not None
caps = client.server_capabilities
assert caps is not None
# Verify list_resources works and returns expected resource
resources = await client.list_resources()
assert len(resources.resources) == 1
assert resources.resources[0].uri == "memory://test"
assert resources == snapshot(
ListResourcesResult(
resources=[Resource(name="Test Resource", uri="memory://test", description="A test resource")]
)
)
async def test_ping_returns_empty_result(app: FastMCP):
"""Test that ping returns an EmptyResult."""
async def test_client_send_ping(app: FastMCP):
async with Client(app) as client:
result = await client.send_ping()
assert isinstance(result, EmptyResult)
assert result == snapshot(EmptyResult())
async def test_list_tools(app: FastMCP):
"""Test listing tools."""
async def test_client_list_tools(app: FastMCP):
async with Client(app) as client:
result = await client.list_tools()
assert result.tools is not None
tool_names = [t.name for t in result.tools]
assert "greet" in tool_names
assert "add" in tool_names
assert result == snapshot(
ListToolsResult(
tools=[
Tool(
name="greet",
description="Greet someone by name.",
input_schema={
"properties": {"name": {"title": "Name", "type": "string"}},
"required": ["name"],
"title": "greetArguments",
"type": "object",
},
output_schema={
"properties": {"result": {"title": "Result", "type": "string"}},
"required": ["result"],
"title": "greetOutput",
"type": "object",
},
)
]
)
)
async def test_list_tools_with_pagination(app: FastMCP):
"""Test listing tools with pagination params."""
from mcp.types import PaginatedRequestParams
async with Client(app) as client:
result = await client.list_tools(params=PaginatedRequestParams())
assert result.tools is not None
async def test_call_tool(app: FastMCP):
"""Test calling a tool."""
async def test_client_call_tool(app: FastMCP):
async with Client(app) as client:
result = await client.call_tool("greet", {"name": "World"})
assert result.content is not None
assert len(result.content) > 0
content_str = str(result.content[0])
assert "Hello, World!" in content_str
async def test_call_tool_with_multiple_args(app: FastMCP):
"""Test calling a tool with multiple arguments."""
async with Client(app) as client:
result = await client.call_tool("add", {"a": 5, "b": 3})
assert result.content is not None
content_str = str(result.content[0])
assert "8" in content_str
async def test_list_resources(app: FastMCP):
"""Test listing resources."""
async with Client(app) as client:
result = await client.list_resources()
# FastMCP may have different resource listing behavior
assert result is not None
assert result == snapshot(
CallToolResult(
content=[TextContent(text="Hello, World!")],
structured_content={"result": "Hello, World!"},
)
)
async def test_read_resource(app: FastMCP):
"""Test reading a resource."""
async with Client(app) as client:
result = await client.read_resource("test://resource")
assert result.contents is not None
assert len(result.contents) > 0
async def test_list_prompts(app: FastMCP):
"""Test listing prompts."""
async with Client(app) as client:
result = await client.list_prompts()
prompt_names = [p.name for p in result.prompts]
assert "greeting_prompt" in prompt_names
assert result == snapshot(
ReadResourceResult(
contents=[TextResourceContents(uri="test://resource", mime_type="text/plain", text="Test content")]
)
)
async def test_get_prompt(app: FastMCP):
"""Test getting a prompt."""
async with Client(app) as client:
result = await client.get_prompt("greeting_prompt", {"name": "Alice"})
assert result.messages is not None
assert len(result.messages) > 0
assert result == snapshot(
GetPromptResult(
description="A greeting prompt.",
messages=[PromptMessage(role="user", content=TextContent(text="Please greet Alice warmly."))],
)
)
async def test_session_property(app: FastMCP):
"""Test that the session property returns the ClientSession."""
from mcp.client.session import ClientSession
async with Client(app) as client:
session = client.session
assert isinstance(session, ClientSession)
async def test_session_is_same_as_internal(app: FastMCP):
"""Test that session property returns consistent instance."""
async with Client(app) as client:
session1 = client.session
session2 = client.session
assert session1 is session2
async def test_enters_and_exits_cleanly(app: FastMCP):
"""Test that the client enters and exits cleanly."""
async with Client(app) as client:
# Should be able to use client
await client.send_ping()
# After exiting, resources should be cleaned up
async def test_exception_during_use(app: FastMCP):
"""Test that exceptions during use don't prevent cleanup."""
with pytest.raises(Exception): # May be wrapped in ExceptionGroup by anyio
async with Client(app) as client:
await client.send_ping()
raise ValueError("Test exception")
# Should exit cleanly despite exception
async def test_aexit_without_aenter(app: FastMCP):
"""Test that calling __aexit__ without __aenter__ doesn't raise."""
client = Client(app)
# This should not raise even though __aenter__ was never called
await client.__aexit__(None, None, None)
assert client._session is None
async def test_server_capabilities_after_init(app: FastMCP):
"""Test server_capabilities property after initialization."""
async with Client(app) as client:
caps = client.server_capabilities
assert caps is not None
# FastMCP should advertise tools capability
assert caps.tools is not None
def test_session_property_before_enter(app: FastMCP):
def test_client_session_property_before_enter(app: FastMCP):
"""Test that accessing session before context manager raises RuntimeError."""
client = Client(app)
with pytest.raises(RuntimeError, match="Client must be used within an async context manager"):
_ = client.session
client.session
async def test_reentry_raises_runtime_error(app: FastMCP):
async def test_client_reentry_raises_runtime_error(app: FastMCP):
"""Test that reentering a client raises RuntimeError."""
async with Client(app) as client:
with pytest.raises(RuntimeError, match="Client is already entered"):
await client.__aenter__()
async def test_cleanup_on_init_failure(app: FastMCP):
"""Test that resources are cleaned up if initialization fails."""
with patch("mcp.client.client.ClientSession") as mock_session_class:
# Create a mock context manager that fails on __aenter__
mock_session = AsyncMock()
mock_session.__aenter__.side_effect = RuntimeError("Session init failed")
mock_session.__aexit__ = AsyncMock(return_value=None)
mock_session_class.return_value = mock_session
client = Client(app)
with pytest.raises(BaseException) as exc_info:
await client.__aenter__()
# The error should contain our message (may be wrapped in ExceptionGroup)
# Use repr() to see nested exceptions in ExceptionGroup
assert "Session init failed" in repr(exc_info.value)
# Verify the client is in a clean state (session should be None)
assert client._session is None
async def test_send_progress_notification(app: FastMCP):
async def test_client_send_progress_notification():
"""Test sending progress notification."""
received_from_client = None
event = anyio.Event()
server = Server(name="test_server")
@server.progress_notification()
async def handle_progress_notification(
progress_token: str | int,
progress: float = 0.0,
total: float | None = None,
message: str | None = None,
) -> None:
nonlocal received_from_client
received_from_client = {"progress_token": progress_token, "progress": progress}
event.set()
async with Client(server) as client:
await client.send_progress_notification(progress_token="token123", progress=50.0)
await event.wait()
assert received_from_client == snapshot({"progress_token": "token123", "progress": 50.0})
async def test_client_subscribe_resource(simple_server: Server):
async with Client(simple_server) as client:
result = await client.subscribe_resource("memory://test")
assert result == snapshot(EmptyResult())
async def test_client_unsubscribe_resource(simple_server: Server):
async with Client(simple_server) as client:
result = await client.unsubscribe_resource("memory://test")
assert result == snapshot(EmptyResult())
async def test_client_set_logging_level(simple_server: Server):
"""Test setting logging level."""
async with Client(simple_server) as client:
result = await client.set_logging_level("debug")
assert result == snapshot(EmptyResult())
async def test_client_list_resources_with_params(app: FastMCP):
"""Test listing resources with params parameter."""
async with Client(app) as client:
# Send a progress notification - this should not raise
await client.send_progress_notification(
progress_token="test-token",
progress=50.0,
total=100.0,
message="Half done",
result = await client.list_resources()
assert result == snapshot(
ListResourcesResult(
resources=[
Resource(
name="test_resource",
uri="test://resource",
description="A test resource.",
mime_type="text/plain",
)
]
)
)
async def test_subscribe_resource(app: FastMCP):
"""Test subscribing to a resource."""
async with Client(app) as client:
# Mock the session's subscribe_resource since FastMCP doesn't support it
with patch.object(client.session, "subscribe_resource", return_value=EmptyResult()):
result = await client.subscribe_resource("test://resource")
assert isinstance(result, EmptyResult)
async def test_unsubscribe_resource(app: FastMCP):
"""Test unsubscribing from a resource."""
async with Client(app) as client:
# Mock the session's unsubscribe_resource since FastMCP doesn't support it
with patch.object(client.session, "unsubscribe_resource", return_value=EmptyResult()):
result = await client.unsubscribe_resource("test://resource")
assert isinstance(result, EmptyResult)
async def test_send_roots_list_changed(app: FastMCP):
"""Test sending roots list changed notification."""
async with Client(app) as client:
# Send roots list changed notification - should not raise
await client.send_roots_list_changed()
async def test_set_logging_level(app: FastMCP):
"""Test setting logging level."""
async with Client(app) as client:
# Mock the session's set_logging_level since FastMCP doesn't support it
with patch.object(client.session, "set_logging_level", return_value=EmptyResult()):
result = await client.set_logging_level("debug")
assert isinstance(result, EmptyResult)
async def test_list_resources_with_params(app: FastMCP):
"""Test listing resources with params parameter."""
async with Client(app) as client:
result = await client.list_resources(params=types.PaginatedRequestParams())
assert result is not None
async def test_list_resource_templates_with_params(app: FastMCP):
async def test_client_list_resource_templates(app: FastMCP):
"""Test listing resource templates with params parameter."""
async with Client(app) as client:
result = await client.list_resource_templates(params=types.PaginatedRequestParams())
assert result is not None
async def test_list_resource_templates_default(app: FastMCP):
"""Test listing resource templates with no params or cursor."""
async with Client(app) as client:
result = await client.list_resource_templates()
assert result is not None
assert result == snapshot(ListResourceTemplatesResult(resource_templates=[]))
async def test_list_prompts_with_params(app: FastMCP):
async def test_list_prompts(app: FastMCP):
"""Test listing prompts with params parameter."""
async with Client(app) as client:
result = await client.list_prompts(params=types.PaginatedRequestParams())
assert result is not None
result = await client.list_prompts()
assert result == snapshot(
ListPromptsResult(
prompts=[
Prompt(
name="greeting_prompt",
description="A greeting prompt.",
arguments=[PromptArgument(name="name", required=True)],
)
]
)
)
async def test_complete_with_prompt_reference(app: FastMCP):
async def test_complete_with_prompt_reference(simple_server: Server):
"""Test getting completions for a prompt argument."""
async with Client(app) as client:
ref = types.PromptReference(type="ref/prompt", name="greeting_prompt")
# Mock the session's complete method since FastMCP may not support it
with patch.object(
client.session,
"complete",
return_value=types.CompleteResult(completion=types.Completion(values=[])),
):
result = await client.complete(ref=ref, argument={"name": "test"})
assert result is not None
async with Client(simple_server) as client:
ref = types.PromptReference(type="ref/prompt", name="test_prompt")
result = await client.complete(ref=ref, argument={"name": "arg", "value": "test"})
assert result == snapshot(types.CompleteResult(completion=types.Completion(values=[])))

22
tests/client/test_list_methods_cursor.py
View file

@ -73,12 +73,12 @@ async def test_list_methods_params_parameter(
_ = await method()
requests = spies.get_client_requests(method=request_method)
assert len(requests) == 1
assert requests[0].params is None
assert requests[0].params is None or "cursor" not in requests[0].params
spies.clear()
# Test with params containing cursor
_ = await method(params=types.PaginatedRequestParams(cursor="from_params"))
_ = await method(cursor="from_params")
requests = spies.get_client_requests(method=request_method)
assert len(requests) == 1
assert requests[0].params is not None
@ -87,7 +87,7 @@ async def test_list_methods_params_parameter(
spies.clear()
# Test with empty params
_ = await method(params=types.PaginatedRequestParams())
_ = await method()
requests = spies.get_client_requests(method=request_method)
assert len(requests) == 1
# Empty params means no cursor
@ -99,7 +99,7 @@ async def test_list_tools_with_strict_server_validation(
):
"""Test pagination with a server that validates request format strictly."""
async with Client(full_featured_server) as client:
result = await client.list_tools(params=types.PaginatedRequestParams())
result = await client.list_tools()
assert isinstance(result, ListToolsResult)
assert len(result.tools) > 0
@ -112,19 +112,11 @@ async def test_list_tools_with_lowlevel_server():
async def handle_list_tools(request: ListToolsRequest) -> ListToolsResult:
# Echo back what cursor we received in the tool description
cursor = request.params.cursor if request.params else None
return ListToolsResult(
tools=[
types.Tool(
name="test_tool",
description=f"cursor={cursor}",
input_schema={},
)
]
)
return ListToolsResult(tools=[types.Tool(name="test_tool", description=f"cursor={cursor}", input_schema={})])
async with Client(server) as client:
result = await client.list_tools(params=types.PaginatedRequestParams())
result = await client.list_tools()
assert result.tools[0].description == "cursor=None"
result = await client.list_tools(params=types.PaginatedRequestParams(cursor="page2"))
result = await client.list_tools(cursor="page2")
assert result.tools[0].description == "cursor=page2"