Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 22 additions & 6 deletions src/mcp2cli/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -2729,14 +2729,14 @@ async def _mcp_session(
)

if list_mode:
result = await session.list_tools()
all_tools = await _list_all_tools(session)
tools = [
{
"name": t.name,
"description": t.description or "",
"inputSchema": t.inputSchema or {},
}
for t in result.tools
for t in all_tools
]
commands = extract_mcp_commands(tools)
if search_pattern:
Expand Down Expand Up @@ -3032,11 +3032,27 @@ def _extract_content_parts(content_list, *, attrs=("text", "data")) -> str:
return "\n".join(parts) if parts else ""


async def _list_all_tools(session):
"""Fetch every tool from an MCP session, following `nextCursor` until
exhausted. Per the MCP spec, tools/list is paginated and page size is
entirely up to the server, so a single call is not guaranteed to return
the full tool set: https://modelcontextprotocol.io/specification/2025-06-18/server/utilities/pagination
"""
tools = []
cursor = None
while True:
result = await session.list_tools(cursor=cursor)
tools.extend(result.tools)
cursor = result.nextCursor
if not cursor:
return tools


async def _dispatch_list_tools(session, params):
result = await session.list_tools()
tools = await _list_all_tools(session)
return [
{"name": t.name, "description": t.description or "", "inputSchema": t.inputSchema or {}}
for t in result.tools
for t in tools
]


Expand Down Expand Up @@ -3517,14 +3533,14 @@ def _fetch_mcp_tools(
tools_result: list[dict] = []

async def _extract_tools(session):
result = await session.list_tools()
all_tools = await _list_all_tools(session)
tools_result.extend(
{
"name": t.name,
"description": t.description or "",
"inputSchema": t.inputSchema or {},
}
for t in result.tools
for t in all_tools
)

async def _run():
Expand Down
52 changes: 52 additions & 0 deletions tests/test_helpers.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
"""Tests for helper functions and data structures."""

import argparse
import asyncio
import json
import shutil

Expand All @@ -10,6 +11,7 @@
_apply_head,
_collect_openapi_params,
_find_toon_cli,
_list_all_tools,
_split_at_subcommand,
_toon_encode,
cache_key_for,
Expand Down Expand Up @@ -609,3 +611,53 @@ def test_body_schemaless_json_object(self):
args = argparse.Namespace(args='{"key": "value"}', stdin=False)
_, _, _, body, _ = _collect_openapi_params(cmd, args)
assert body == {"args": {"key": "value"}}


class _FakePage:
def __init__(self, tools, next_cursor=None):
self.tools = tools
self.nextCursor = next_cursor


class _FakeSession:
"""Fake MCP session whose list_tools() serves canned pages, keyed by cursor."""

def __init__(self, pages):
# pages: list of (cursor_expected, _FakePage) in order
self._pages = pages
self.calls = []

async def list_tools(self, cursor=None):
self.calls.append(cursor)
expected_cursor, page = self._pages[len(self.calls) - 1]
assert cursor == expected_cursor, (
f"call {len(self.calls)}: expected cursor={expected_cursor!r}, got {cursor!r}"
)
return page


class TestListAllTools:
def test_single_page_no_cursor(self):
"""A server that never sets nextCursor is fetched in exactly one call."""
session = _FakeSession([(None, _FakePage(["a", "b"], next_cursor=None))])
tools = asyncio.run(_list_all_tools(session))
assert tools == ["a", "b"]
assert session.calls == [None]

def test_follows_next_cursor_across_pages(self):
"""A paginated server's tools are all collected across multiple calls."""
session = _FakeSession(
[
(None, _FakePage(["a", "b"], next_cursor="page2")),
("page2", _FakePage(["c", "d"], next_cursor="page3")),
("page3", _FakePage(["e"], next_cursor=None)),
]
)
tools = asyncio.run(_list_all_tools(session))
assert tools == ["a", "b", "c", "d", "e"]
assert session.calls == [None, "page2", "page3"]

def test_empty_result(self):
session = _FakeSession([(None, _FakePage([], next_cursor=None))])
tools = asyncio.run(_list_all_tools(session))
assert tools == []