feat: add pagination to mcp_client list_tools_sync (strands-agents#436)

Author: dbschmigelski

src/strands/tools/mcp/mcp_client.py

@@ -16,13 +16,14 @@
 from concurrent import futures
 from datetime import timedelta
 from types import TracebackType
-from typing import Any, Callable, Coroutine, Dict, List, Optional, TypeVar, Union
+from typing import Any, Callable, Coroutine, Dict, Optional, TypeVar, Union
 
 from mcp import ClientSession, ListToolsResult
 from mcp.types import CallToolResult as MCPCallToolResult
 from mcp.types import ImageContent as MCPImageContent
 from mcp.types import TextContent as MCPTextContent
 
+from ...types import PaginatedList
 from ...types.exceptions import MCPClientInitializationError
 from ...types.media import ImageFormat
 from ...types.tools import ToolResult, ToolResultContent, ToolResultStatus
@@ -140,7 +141,7 @@ async def _set_close_event() -> None:
         self._background_thread = None
         self._session_id = uuid.uuid4()
 
-    def list_tools_sync(self) -> List[MCPAgentTool]:
+    def list_tools_sync(self, pagination_token: Optional[str] = None) -> PaginatedList[MCPAgentTool]:
         """Synchronously retrieves the list of available tools from the MCP server.
 
         This method calls the asynchronous list_tools method on the MCP session
@@ -154,14 +155,14 @@ def list_tools_sync(self) -> List[MCPAgentTool]:
             raise MCPClientInitializationError(CLIENT_SESSION_NOT_RUNNING_ERROR_MESSAGE)
 
         async def _list_tools_async() -> ListToolsResult:
-            return await self._background_thread_session.list_tools()
+            return await self._background_thread_session.list_tools(cursor=pagination_token)
 
         list_tools_response: ListToolsResult = self._invoke_on_background_thread(_list_tools_async()).result()
         self._log_debug_with_thread("received %d tools from MCP server", len(list_tools_response.tools))
 
         mcp_tools = [MCPAgentTool(tool, self) for tool in list_tools_response.tools]
         self._log_debug_with_thread("successfully adapted %d MCP tools", len(mcp_tools))
-        return mcp_tools
+        return PaginatedList[MCPAgentTool](mcp_tools, token=list_tools_response.nextCursor)
 
     def list_prompts_sync(self) -> ListPromptsResult:
         """Synchronously retrieves the list of available prompts from the MCP server.

src/strands/types/__init__.py

@@ -1 +1,5 @@
 """SDK type definitions."""
+
+from .collections import PaginatedList
+
+__all__ = ["PaginatedList"]

src/strands/types/collections.py

@@ -0,0 +1,23 @@
+"""Generic collection types for the Strands SDK."""
+
+from typing import Generic, List, Optional, TypeVar
+
+T = TypeVar("T")
+
+
+class PaginatedList(list, Generic[T]):
+    """A generic list-like object that includes a pagination token.
+
+    This maintains backwards compatibility by inheriting from list,
+    so existing code that expects List[T] will continue to work.
+    """
+
+    def __init__(self, data: List[T], token: Optional[str] = None):
+        """Initialize a PaginatedList with data and an optional pagination token.
+
+        Args:
+            data: The list of items to store.
+            token: Optional pagination token for retrieving additional items.
+        """
+        super().__init__(data)
+        self.pagination_token = token

tests/strands/tools/mcp/test_mcp_client.py

@@ -71,10 +71,11 @@ def test_list_tools_sync(mock_transport, mock_session):
     with MCPClient(mock_transport["transport_callable"]) as client:
         tools = client.list_tools_sync()
 
-        mock_session.list_tools.assert_called_once()
+        mock_session.list_tools.assert_called_once_with(cursor=None)
 
         assert len(tools) == 1
         assert tools[0].tool_name == "test_tool"
+        assert tools.pagination_token is None
 
 
 def test_list_tools_sync_session_not_active():
@@ -85,6 +86,34 @@ def test_list_tools_sync_session_not_active():
         client.list_tools_sync()
 
 
+def test_list_tools_sync_with_pagination_token(mock_transport, mock_session):
+    """Test that list_tools_sync correctly passes pagination token and returns next cursor."""
+    mock_tool = MCPTool(name="test_tool", description="A test tool", inputSchema={"type": "object", "properties": {}})
+    mock_session.list_tools.return_value = ListToolsResult(tools=[mock_tool], nextCursor="next_page_token")
+
+    with MCPClient(mock_transport["transport_callable"]) as client:
+        tools = client.list_tools_sync(pagination_token="current_page_token")
+
+        mock_session.list_tools.assert_called_once_with(cursor="current_page_token")
+        assert len(tools) == 1
+        assert tools[0].tool_name == "test_tool"
+        assert tools.pagination_token == "next_page_token"
+
+
+def test_list_tools_sync_without_pagination_token(mock_transport, mock_session):
+    """Test that list_tools_sync works without pagination token and handles missing next cursor."""
+    mock_tool = MCPTool(name="test_tool", description="A test tool", inputSchema={"type": "object", "properties": {}})
+    mock_session.list_tools.return_value = ListToolsResult(tools=[mock_tool])  # No nextCursor
+
+    with MCPClient(mock_transport["transport_callable"]) as client:
+        tools = client.list_tools_sync()
+
+        mock_session.list_tools.assert_called_once_with(cursor=None)
+        assert len(tools) == 1
+        assert tools[0].tool_name == "test_tool"
+        assert tools.pagination_token is None
+
+
 @pytest.mark.parametrize("is_error,expected_status", [(False, "success"), (True, "error")])
 def test_call_tool_sync_status(mock_transport, mock_session, is_error, expected_status):
     """Test that call_tool_sync correctly handles success and error results."""