mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-10-10 15:03:20 +02:00
d869ad0eb6CursorState.decode took the decoded offset verbatim. The value then reaches a slice and an addition, so a string or a float raised TypeError that _apply_pagination does not catch and the server reported as an internal error, while a negative offset sliced from the end and returned a valid-looking page plus a cursor that restarts mid-list. A cursor is opaque and server-produced, so anything but a whole non-negative count means it was tampered with. decode now raises inside its own guard, which the existing handler turns into the same "Invalid cursor" ValueError every other malformed cursor gets, and _apply_pagination maps to INVALID_PARAMS.
662 lines
23 KiB
Python
662 lines
23 KiB
Python
"""Tests for MCP pagination support."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import call, patch
|
|
|
|
import mcp_types
|
|
import pytest
|
|
from mcp.server.context import ServerRequestContext
|
|
from mcp.shared.exceptions import MCPError
|
|
from mcp.types import INVALID_PARAMS
|
|
|
|
from fastmcp import Client, FastMCP
|
|
from fastmcp.utilities.pagination import CursorState, paginate_sequence
|
|
|
|
|
|
class TestCursorEncoding:
|
|
"""Tests for cursor encoding/decoding."""
|
|
|
|
def test_encode_decode_roundtrip(self) -> None:
|
|
"""Cursor should survive encode/decode roundtrip."""
|
|
state = CursorState(offset=100)
|
|
encoded = state.encode()
|
|
decoded = CursorState.decode(encoded)
|
|
assert decoded.offset == 100
|
|
|
|
def test_encode_produces_string(self) -> None:
|
|
"""Encoded cursor should be a string."""
|
|
state = CursorState(offset=50)
|
|
encoded = state.encode()
|
|
assert isinstance(encoded, str)
|
|
assert len(encoded) > 0
|
|
|
|
def test_decode_invalid_base64_raises(self) -> None:
|
|
"""Invalid base64 should raise ValueError."""
|
|
with pytest.raises(ValueError, match="Invalid cursor"):
|
|
CursorState.decode("not-valid-base64!!!")
|
|
|
|
def test_decode_invalid_json_raises(self) -> None:
|
|
"""Valid base64 but invalid JSON should raise ValueError."""
|
|
import base64
|
|
|
|
invalid = base64.urlsafe_b64encode(b"not json").decode()
|
|
with pytest.raises(ValueError, match="Invalid cursor"):
|
|
CursorState.decode(invalid)
|
|
|
|
@pytest.mark.parametrize("offset", ["1", 1.5, -1, True, None, [0]])
|
|
def test_decode_rejects_a_malformed_offset(self, offset: object) -> None:
|
|
"""An offset that is not a whole non-negative count is an invalid cursor.
|
|
|
|
A string or a float reached the slice and raised TypeError, which the
|
|
server reports as an internal error rather than invalid parameters, and a
|
|
negative one sliced from the end and returned a valid-looking page.
|
|
"""
|
|
import base64
|
|
import json
|
|
|
|
invalid = base64.urlsafe_b64encode(json.dumps({"o": offset}).encode()).decode()
|
|
with pytest.raises(ValueError, match="Invalid cursor"):
|
|
CursorState.decode(invalid)
|
|
|
|
def test_decode_missing_offset_raises(self) -> None:
|
|
"""JSON missing the offset key should raise ValueError."""
|
|
import base64
|
|
import json
|
|
|
|
invalid = base64.urlsafe_b64encode(json.dumps({"x": 1}).encode()).decode()
|
|
with pytest.raises(ValueError, match="Invalid cursor"):
|
|
CursorState.decode(invalid)
|
|
|
|
|
|
class TestPaginateSequence:
|
|
"""Tests for the paginate_sequence helper."""
|
|
|
|
def test_first_page_no_cursor(self) -> None:
|
|
"""First page should start from beginning."""
|
|
items = list(range(25))
|
|
page, cursor = paginate_sequence(items, None, 10)
|
|
assert page == list(range(10))
|
|
assert cursor is not None
|
|
|
|
def test_second_page_with_cursor(self) -> None:
|
|
"""Second page should continue from cursor."""
|
|
items = list(range(25))
|
|
_, cursor = paginate_sequence(items, None, 10)
|
|
page, next_cursor = paginate_sequence(items, cursor, 10)
|
|
assert page == list(range(10, 20))
|
|
assert next_cursor is not None
|
|
|
|
def test_last_page_returns_none_cursor(self) -> None:
|
|
"""Last page should return None cursor."""
|
|
items = list(range(25))
|
|
_, c1 = paginate_sequence(items, None, 10)
|
|
_, c2 = paginate_sequence(items, c1, 10)
|
|
page, next_cursor = paginate_sequence(items, c2, 10)
|
|
assert page == list(range(20, 25))
|
|
assert next_cursor is None
|
|
|
|
def test_empty_list(self) -> None:
|
|
"""Empty list should return empty page and no cursor."""
|
|
page, cursor = paginate_sequence([], None, 10)
|
|
assert page == []
|
|
assert cursor is None
|
|
|
|
def test_exact_page_size(self) -> None:
|
|
"""List exactly matching page size should return no cursor."""
|
|
items = list(range(10))
|
|
page, cursor = paginate_sequence(items, None, 10)
|
|
assert page == items
|
|
assert cursor is None
|
|
|
|
def test_smaller_than_page_size(self) -> None:
|
|
"""List smaller than page size should return all items."""
|
|
items = list(range(5))
|
|
page, cursor = paginate_sequence(items, None, 10)
|
|
assert page == items
|
|
assert cursor is None
|
|
|
|
def test_invalid_cursor_raises(self) -> None:
|
|
"""Invalid cursor should raise ValueError."""
|
|
with pytest.raises(ValueError, match="Invalid cursor"):
|
|
paginate_sequence([1, 2, 3], "invalid!", 10)
|
|
|
|
|
|
class TestMalformedCursorParams:
|
|
"""A tampered cursor is invalid parameters, not an internal error."""
|
|
|
|
@staticmethod
|
|
def _cursor(offset: object) -> str:
|
|
import base64
|
|
import json
|
|
|
|
return base64.urlsafe_b64encode(json.dumps({"o": offset}).encode()).decode()
|
|
|
|
@pytest.mark.parametrize("offset", ["1", 1.5, -1])
|
|
def test_paginate_sequence_rejects_a_malformed_offset(self, offset: object) -> None:
|
|
with pytest.raises(ValueError, match="Invalid cursor"):
|
|
paginate_sequence([0, 1, 2], self._cursor(offset), page_size=2)
|
|
|
|
async def test_list_tools_reports_invalid_params(self) -> None:
|
|
"""The caller maps the ValueError to INVALID_PARAMS, so the client sees that."""
|
|
server = FastMCP(list_page_size=1)
|
|
|
|
@server.tool(name="one")
|
|
def one() -> int:
|
|
return 1
|
|
|
|
@server.tool(name="two")
|
|
def two() -> int:
|
|
return 2
|
|
|
|
async with Client(server) as client:
|
|
with pytest.raises(MCPError) as excinfo:
|
|
await client.list_tools_mcp(cursor=self._cursor(-1))
|
|
|
|
assert excinfo.value.error.code == INVALID_PARAMS
|
|
|
|
|
|
class TestServerPagination:
|
|
"""Integration tests for server pagination."""
|
|
|
|
@pytest.mark.parametrize(
|
|
"method,first_page,second_page",
|
|
[
|
|
(
|
|
"list_tools",
|
|
mcp_types.ListToolsResult(
|
|
tools=[
|
|
mcp_types.Tool(name="first", input_schema={"type": "object"})
|
|
],
|
|
next_cursor="",
|
|
),
|
|
mcp_types.ListToolsResult(
|
|
tools=[
|
|
mcp_types.Tool(name="second", input_schema={"type": "object"})
|
|
],
|
|
),
|
|
),
|
|
(
|
|
"list_resources",
|
|
mcp_types.ListResourcesResult(
|
|
resources=[mcp_types.Resource(name="first", uri="test://first")],
|
|
next_cursor="",
|
|
),
|
|
mcp_types.ListResourcesResult(
|
|
resources=[mcp_types.Resource(name="second", uri="test://second")],
|
|
),
|
|
),
|
|
(
|
|
"list_resource_templates",
|
|
mcp_types.ListResourceTemplatesResult(
|
|
resource_templates=[
|
|
mcp_types.ResourceTemplate(
|
|
name="first", uri_template="test://first/{id}"
|
|
)
|
|
],
|
|
next_cursor="",
|
|
),
|
|
mcp_types.ListResourceTemplatesResult(
|
|
resource_templates=[
|
|
mcp_types.ResourceTemplate(
|
|
name="second", uri_template="test://second/{id}"
|
|
)
|
|
],
|
|
),
|
|
),
|
|
(
|
|
"list_prompts",
|
|
mcp_types.ListPromptsResult(
|
|
prompts=[mcp_types.Prompt(name="first")],
|
|
next_cursor="",
|
|
),
|
|
mcp_types.ListPromptsResult(prompts=[mcp_types.Prompt(name="second")]),
|
|
),
|
|
],
|
|
)
|
|
async def test_empty_cursor_fetches_the_next_page(
|
|
self,
|
|
method: str,
|
|
first_page: mcp_types.PaginatedResult,
|
|
second_page: mcp_types.PaginatedResult,
|
|
) -> None:
|
|
async with Client(FastMCP()) as client:
|
|
with patch.object(
|
|
client, f"{method}_mcp", side_effect=[first_page, second_page]
|
|
) as list_page:
|
|
items = await getattr(client, method)(max_pages=2)
|
|
|
|
assert [item.name for item in items] == ["first", "second"]
|
|
extra_kwargs = {"cache_mode": "use"} if method == "list_tools" else {}
|
|
assert list_page.await_args_list == [
|
|
call(cursor=None, **extra_kwargs),
|
|
call(cursor="", **extra_kwargs),
|
|
]
|
|
|
|
async def test_tools_pagination_returns_all_tools(self) -> None:
|
|
"""Client should receive all tools across paginated requests."""
|
|
server = FastMCP(list_page_size=10)
|
|
|
|
for i in range(25):
|
|
|
|
@server.tool(name=f"tool_{i}")
|
|
def make_tool() -> str:
|
|
return "ok"
|
|
|
|
async with Client(server) as client:
|
|
tools = await client.list_tools()
|
|
assert len(tools) == 25
|
|
tool_names = {t.name for t in tools}
|
|
assert tool_names == {f"tool_{i}" for i in range(25)}
|
|
|
|
async def test_tools_follow_empty_cursor_over_mcp_session(self) -> None:
|
|
class EmptyCursorServer(FastMCP):
|
|
async def _on_list_tools(
|
|
self,
|
|
ctx: ServerRequestContext,
|
|
params: mcp_types.PaginatedRequestParams | None,
|
|
) -> mcp_types.ListToolsResult:
|
|
cursor = params.cursor if params is not None else None
|
|
assert cursor in (None, "")
|
|
return mcp_types.ListToolsResult(
|
|
tools=[
|
|
mcp_types.Tool(
|
|
name="first" if cursor is None else "second",
|
|
input_schema={"type": "object"},
|
|
)
|
|
],
|
|
next_cursor="" if cursor is None else None,
|
|
)
|
|
|
|
async with Client(EmptyCursorServer()) as client:
|
|
tools = await client.list_tools(max_pages=2)
|
|
|
|
assert [tool.name for tool in tools] == ["first", "second"]
|
|
|
|
async def test_resources_pagination_returns_all_resources(self) -> None:
|
|
"""Client should receive all resources across paginated requests."""
|
|
server = FastMCP(list_page_size=10)
|
|
|
|
for i in range(25):
|
|
|
|
@server.resource(f"test://resource_{i}")
|
|
def make_resource() -> str:
|
|
return "data"
|
|
|
|
async with Client(server) as client:
|
|
resources = await client.list_resources()
|
|
assert len(resources) == 25
|
|
|
|
async def test_prompts_pagination_returns_all_prompts(self) -> None:
|
|
"""Client should receive all prompts across paginated requests."""
|
|
server = FastMCP(list_page_size=10)
|
|
|
|
for i in range(25):
|
|
|
|
@server.prompt(name=f"prompt_{i}")
|
|
def make_prompt() -> str:
|
|
return "text"
|
|
|
|
async with Client(server) as client:
|
|
prompts = await client.list_prompts()
|
|
assert len(prompts) == 25
|
|
|
|
async def test_manual_pagination(self) -> None:
|
|
"""Client can manually paginate using cursor."""
|
|
server = FastMCP(list_page_size=10)
|
|
|
|
for i in range(25):
|
|
|
|
@server.tool(name=f"tool_{i}")
|
|
def make_tool() -> str:
|
|
return "ok"
|
|
|
|
async with Client(server) as client:
|
|
# First page
|
|
result = await client.list_tools_mcp()
|
|
assert len(result.tools) == 10
|
|
assert result.next_cursor is not None
|
|
|
|
# Second page
|
|
result2 = await client.list_tools_mcp(cursor=result.next_cursor)
|
|
assert len(result2.tools) == 10
|
|
assert result2.next_cursor is not None
|
|
|
|
# Third (last) page
|
|
result3 = await client.list_tools_mcp(cursor=result2.next_cursor)
|
|
assert len(result3.tools) == 5
|
|
assert result3.next_cursor is None
|
|
|
|
async def test_invalid_cursor_returns_error(self) -> None:
|
|
"""Server should return MCP error for invalid cursor."""
|
|
server = FastMCP(list_page_size=10)
|
|
|
|
@server.tool
|
|
def my_tool() -> str:
|
|
return "ok"
|
|
|
|
async with Client(server) as client:
|
|
with pytest.raises(MCPError) as exc:
|
|
await client.list_tools_mcp(cursor="invalid!")
|
|
assert exc.value.error.code == -32602
|
|
|
|
async def test_no_pagination_when_disabled(self) -> None:
|
|
"""Without list_page_size, all items returned at once."""
|
|
server = FastMCP() # No pagination
|
|
|
|
for i in range(25):
|
|
|
|
@server.tool(name=f"tool_{i}")
|
|
def make_tool() -> str:
|
|
return "ok"
|
|
|
|
async with Client(server) as client:
|
|
result = await client.list_tools_mcp()
|
|
assert len(result.tools) == 25
|
|
assert result.next_cursor is None
|
|
|
|
async def test_pagination_exact_page_boundary(self) -> None:
|
|
"""Test pagination at exact page boundaries."""
|
|
server = FastMCP(list_page_size=10)
|
|
|
|
for i in range(20): # Exactly 2 pages
|
|
|
|
@server.tool(name=f"tool_{i}")
|
|
def make_tool() -> str:
|
|
return "ok"
|
|
|
|
async with Client(server) as client:
|
|
# First page
|
|
result = await client.list_tools_mcp()
|
|
assert len(result.tools) == 10
|
|
assert result.next_cursor is not None
|
|
|
|
# Second (last) page
|
|
result2 = await client.list_tools_mcp(cursor=result.next_cursor)
|
|
assert len(result2.tools) == 10
|
|
assert result2.next_cursor is None
|
|
|
|
|
|
class TestPageSizeValidation:
|
|
"""Tests for list_page_size validation."""
|
|
|
|
def test_zero_page_size_raises(self) -> None:
|
|
"""Zero page size should raise ValueError."""
|
|
with pytest.raises(
|
|
ValueError, match="list_page_size must be a positive integer"
|
|
):
|
|
FastMCP(list_page_size=0)
|
|
|
|
def test_negative_page_size_raises(self) -> None:
|
|
"""Negative page size should raise ValueError."""
|
|
with pytest.raises(
|
|
ValueError, match="list_page_size must be a positive integer"
|
|
):
|
|
FastMCP(list_page_size=-1)
|
|
|
|
|
|
class TestPaginationCycleDetection:
|
|
"""Tests that auto-pagination terminates when the server returns cycling cursors."""
|
|
|
|
@pytest.mark.parametrize("next_cursor", ["stuck", ""])
|
|
async def test_tools_constant_cursor_terminates(self, next_cursor: str) -> None:
|
|
"""list_tools should stop if the server always returns the same cursor."""
|
|
server = FastMCP()
|
|
|
|
@server.tool
|
|
def my_tool() -> str:
|
|
return "ok"
|
|
|
|
async with Client(server) as client:
|
|
original = client.list_tools_mcp
|
|
|
|
async def returning_constant_cursor(
|
|
*,
|
|
cursor: str | None = None,
|
|
cache_mode: str = "use",
|
|
) -> mcp_types.ListToolsResult:
|
|
result = await original(cursor=cursor)
|
|
result.next_cursor = next_cursor
|
|
return result
|
|
|
|
with patch.object(
|
|
client, "list_tools_mcp", side_effect=returning_constant_cursor
|
|
):
|
|
tools = await client.list_tools()
|
|
|
|
# Should get tools from first page + one duplicate (the retry before
|
|
# detecting the cycle), then stop.
|
|
assert len(tools) == 2
|
|
assert all(t.name == "my_tool" for t in tools)
|
|
|
|
@pytest.mark.parametrize("next_cursor", ["stuck", ""])
|
|
async def test_prompts_constant_cursor_terminates(self, next_cursor: str) -> None:
|
|
"""list_prompts should stop if the server always returns the same cursor."""
|
|
server = FastMCP()
|
|
|
|
@server.prompt
|
|
def my_prompt() -> str:
|
|
return "text"
|
|
|
|
async with Client(server) as client:
|
|
original = client.list_prompts_mcp
|
|
|
|
async def returning_constant_cursor(
|
|
*,
|
|
cursor: str | None = None,
|
|
cache_mode: str = "use",
|
|
) -> mcp_types.ListPromptsResult:
|
|
result = await original(cursor=cursor)
|
|
result.next_cursor = next_cursor
|
|
return result
|
|
|
|
with patch.object(
|
|
client, "list_prompts_mcp", side_effect=returning_constant_cursor
|
|
):
|
|
prompts = await client.list_prompts()
|
|
|
|
assert len(prompts) == 2
|
|
assert all(p.name == "my_prompt" for p in prompts)
|
|
|
|
@pytest.mark.parametrize("next_cursor", ["stuck", ""])
|
|
async def test_resources_constant_cursor_terminates(self, next_cursor: str) -> None:
|
|
"""list_resources should stop if the server always returns the same cursor."""
|
|
server = FastMCP()
|
|
|
|
@server.resource("test://r")
|
|
def my_resource() -> str:
|
|
return "data"
|
|
|
|
async with Client(server) as client:
|
|
original = client.list_resources_mcp
|
|
|
|
async def returning_constant_cursor(
|
|
*,
|
|
cursor: str | None = None,
|
|
cache_mode: str = "use",
|
|
) -> mcp_types.ListResourcesResult:
|
|
result = await original(cursor=cursor)
|
|
result.next_cursor = next_cursor
|
|
return result
|
|
|
|
with patch.object(
|
|
client, "list_resources_mcp", side_effect=returning_constant_cursor
|
|
):
|
|
resources = await client.list_resources()
|
|
|
|
assert len(resources) == 2
|
|
assert all(r.name == "my_resource" for r in resources)
|
|
|
|
@pytest.mark.parametrize("next_cursor", ["stuck", ""])
|
|
async def test_resource_templates_constant_cursor_terminates(
|
|
self, next_cursor: str
|
|
) -> None:
|
|
"""list_resource_templates should stop if the server always returns the same cursor."""
|
|
server = FastMCP()
|
|
|
|
@server.resource("test://items/{item_id}")
|
|
def my_template(item_id: str) -> str:
|
|
return item_id
|
|
|
|
async with Client(server) as client:
|
|
original = client.list_resource_templates_mcp
|
|
|
|
async def returning_constant_cursor(
|
|
*,
|
|
cursor: str | None = None,
|
|
cache_mode: str = "use",
|
|
) -> mcp_types.ListResourceTemplatesResult:
|
|
result = await original(cursor=cursor)
|
|
result.next_cursor = next_cursor
|
|
return result
|
|
|
|
with patch.object(
|
|
client,
|
|
"list_resource_templates_mcp",
|
|
side_effect=returning_constant_cursor,
|
|
):
|
|
templates = await client.list_resource_templates()
|
|
|
|
assert len(templates) == 2
|
|
|
|
async def test_cycling_cursors_terminates(self) -> None:
|
|
"""list_tools should stop if the server cycles through a set of cursors."""
|
|
server = FastMCP()
|
|
|
|
@server.tool
|
|
def my_tool() -> str:
|
|
return "ok"
|
|
|
|
async with Client(server) as client:
|
|
call_count = 0
|
|
original = client.list_tools_mcp
|
|
|
|
async def returning_cycling_cursor(
|
|
*,
|
|
cursor: str | None = None,
|
|
cache_mode: str = "use",
|
|
) -> mcp_types.ListToolsResult:
|
|
nonlocal call_count
|
|
result = await original(cursor=cursor)
|
|
# Cycle through A -> B -> C -> A
|
|
cursors = ["A", "B", "C"]
|
|
result.next_cursor = cursors[call_count % 3]
|
|
call_count += 1
|
|
return result
|
|
|
|
with patch.object(
|
|
client, "list_tools_mcp", side_effect=returning_cycling_cursor
|
|
):
|
|
tools = await client.list_tools()
|
|
|
|
# A, B, C seen, then A is a duplicate → 4 calls total
|
|
assert call_count == 4
|
|
assert len(tools) == 4
|
|
|
|
async def test_tools_raises_on_auto_pagination_limit(self) -> None:
|
|
"""list_tools should raise RuntimeError after exceeding max_pages."""
|
|
server = FastMCP()
|
|
|
|
@server.tool
|
|
def my_tool() -> str:
|
|
return "ok"
|
|
|
|
async with Client(server) as client:
|
|
original = client.list_tools_mcp
|
|
call_count = 0
|
|
|
|
async def returning_unique_cursor(
|
|
*,
|
|
cursor: str | None = None,
|
|
cache_mode: str = "use",
|
|
) -> mcp_types.ListToolsResult:
|
|
nonlocal call_count
|
|
result = await original(cursor=cursor)
|
|
call_count += 1
|
|
result.next_cursor = f"cursor-{call_count}"
|
|
return result
|
|
|
|
with (
|
|
patch.object(
|
|
client, "list_tools_mcp", side_effect=returning_unique_cursor
|
|
),
|
|
pytest.raises(RuntimeError, match="auto-pagination limit"),
|
|
):
|
|
await client.list_tools(max_pages=5)
|
|
|
|
async def test_resources_raises_on_auto_pagination_limit(self) -> None:
|
|
"""list_resources should raise RuntimeError after exceeding max_pages."""
|
|
server = FastMCP()
|
|
|
|
@server.resource("test://r")
|
|
def my_resource() -> str:
|
|
return "data"
|
|
|
|
async with Client(server) as client:
|
|
original = client.list_resources_mcp
|
|
call_count = 0
|
|
|
|
async def returning_unique_cursor(
|
|
*,
|
|
cursor: str | None = None,
|
|
cache_mode: str = "use",
|
|
) -> mcp_types.ListResourcesResult:
|
|
nonlocal call_count
|
|
result = await original(cursor=cursor)
|
|
call_count += 1
|
|
result.next_cursor = f"cursor-{call_count}"
|
|
return result
|
|
|
|
with (
|
|
patch.object(
|
|
client, "list_resources_mcp", side_effect=returning_unique_cursor
|
|
),
|
|
pytest.raises(RuntimeError, match="auto-pagination limit"),
|
|
):
|
|
await client.list_resources(max_pages=5)
|
|
|
|
async def test_prompts_raises_on_auto_pagination_limit(self) -> None:
|
|
"""list_prompts should raise RuntimeError after exceeding max_pages."""
|
|
server = FastMCP()
|
|
|
|
@server.prompt
|
|
def my_prompt() -> str:
|
|
return "text"
|
|
|
|
async with Client(server) as client:
|
|
original = client.list_prompts_mcp
|
|
call_count = 0
|
|
|
|
async def returning_unique_cursor(
|
|
*,
|
|
cursor: str | None = None,
|
|
cache_mode: str = "use",
|
|
) -> mcp_types.ListPromptsResult:
|
|
nonlocal call_count
|
|
result = await original(cursor=cursor)
|
|
call_count += 1
|
|
result.next_cursor = f"cursor-{call_count}"
|
|
return result
|
|
|
|
with (
|
|
patch.object(
|
|
client, "list_prompts_mcp", side_effect=returning_unique_cursor
|
|
),
|
|
pytest.raises(RuntimeError, match="auto-pagination limit"),
|
|
):
|
|
await client.list_prompts(max_pages=5)
|
|
|
|
async def test_normal_pagination_unaffected(self) -> None:
|
|
"""Cycle detection should not interfere with normal pagination."""
|
|
server = FastMCP(list_page_size=10)
|
|
|
|
for i in range(25):
|
|
|
|
@server.tool(name=f"tool_{i}")
|
|
def make_tool() -> str:
|
|
return "ok"
|
|
|
|
async with Client(server) as client:
|
|
tools = await client.list_tools()
|
|
assert len(tools) == 25
|
|
assert len({t.name for t in tools}) == 25
|