feat: add cursor-based pagination to index resource (#176)

Author: mathislucka

src/deepset_mcp/api/indexes/protocols.py

@@ -4,25 +4,26 @@
 
 from typing import Protocol
 
-from deepset_mcp.api.indexes.models import Index, IndexList
+from deepset_mcp.api.indexes.models import Index
 from deepset_mcp.api.pipeline.models import PipelineValidationResult
+from deepset_mcp.api.shared_models import PaginatedResponse
 
 
 class IndexResourceProtocol(Protocol):
     """Protocol defining the implementation for IndexResource."""
 
-    async def list(self, limit: int = 10, page_number: int = 1) -> IndexList:
+    async def list(self, limit: int = 10, after: str | None = None) -> PaginatedResponse[Index]:
         """List indexes in the configured workspace."""
         ...
 
     async def get(self, index_name: str) -> Index:
         """Fetch a single index by its name."""
         ...
 
-    async def create(self, name: str, yaml_config: str, description: str | None = None) -> Index:
+    async def create(self, index_name: str, yaml_config: str, description: str | None = None) -> Index:
         """Create a new index with the given name and configuration.
 
-        :param name: Name of the index
+        :param index_name: Name of the index
         :param yaml_config: YAML configuration for the index
         :param description: Optional description for the index
         :returns: Created index details

src/deepset_mcp/api/indexes/resource.py

@@ -2,13 +2,15 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
+from typing import Any
 from urllib.parse import quote
 
 from deepset_mcp.api.exceptions import UnexpectedAPIError
-from deepset_mcp.api.indexes.models import Index, IndexList
+from deepset_mcp.api.indexes.models import Index
 from deepset_mcp.api.indexes.protocols import IndexResourceProtocol
 from deepset_mcp.api.pipeline.models import PipelineValidationResult, ValidationError
 from deepset_mcp.api.protocols import AsyncClientProtocol
+from deepset_mcp.api.shared_models import PaginatedResponse
 from deepset_mcp.api.transport import raise_for_status
 
 
@@ -24,26 +26,45 @@ def __init__(self, client: AsyncClientProtocol, workspace: str) -> None:
         self._client = client
         self._workspace = workspace
 
-    async def list(self, limit: int = 10, page_number: int = 1) -> IndexList:
-        """List all indexes.
+    async def list(self, limit: int = 10, after: str | None = None) -> PaginatedResponse[Index]:
+        """Lists indexes and returns the first page of results.
 
-        :param limit: Maximum number of indexes to return.
-        :param page_number: Page number for pagination.
+        The returned object can be iterated over to fetch subsequent pages.
 
-        :returns: List of indexes.
+        :param limit: The maximum number of indexes to return per page.
+        :param after: The cursor to fetch the next page of results.
+        :returns: A `PaginatedResponse` object containing the first page of indexes.
         """
-        params = {
-            "limit": limit,
-            "page_number": page_number,
-        }
-
-        response = await self._client.request(
-            f"/v1/workspaces/{quote(self._workspace, safe='')}/indexes", params=params
+        # 1. Prepare arguments for the initial API call
+        # TODO: Pagination in the deepset API is currently implemented in an unintuitive way.
+        # TODO: The cursor is always time based (created_at) and after signifies indexes older than the current cursor
+        # TODO: while 'before' signals indexes younger than the current cursor.
+        # TODO: This is applied irrespective of any sort (e.g. name) that would conflict with this approach.
+        # TODO: Change this to 'after' once the behaviour is fixed on the deepset API
+        request_params = {"limit": limit, "before": after}
+        request_params = {k: v for k, v in request_params.items() if v is not None}
+
+        # 2. Make the first API call using a private, stateless method
+        page = await self._list_api_call(**request_params)
+
+        # 3. Inject the logic needed for subsequent fetches into the response object
+        page._inject_paginator(
+            fetch_func=self._list_api_call,
+            # Base args for the *next* fetch don't include initial cursors
+            base_args={"limit": limit},
         )
+        return page
 
-        raise_for_status(response)
+    async def _list_api_call(self, **kwargs: Any) -> PaginatedResponse[Index]:
+        """A private, stateless method that performs the raw API call."""
+        resp = await self._client.request(
+            endpoint=f"v1/workspaces/{quote(self._workspace, safe='')}/indexes", method="GET", params=kwargs
+        )
+        raise_for_status(resp)
+        if resp.json is None:
+            raise UnexpectedAPIError(status_code=resp.status_code, message="Empty response", detail=None)
 
-        return IndexList.model_validate(response.json)
+        return PaginatedResponse[Index].create_with_cursor_field(resp.json, "pipeline_index_id")
 
     async def get(self, index_name: str) -> Index:
         """Get a specific index.

src/deepset_mcp/tools/indexes.py

@@ -3,23 +3,29 @@
 # SPDX-License-Identifier: Apache-2.0
 
 from deepset_mcp.api.exceptions import BadRequestError, ResourceNotFoundError, UnexpectedAPIError
-from deepset_mcp.api.indexes.models import Index, IndexList
+from deepset_mcp.api.indexes.models import Index
 from deepset_mcp.api.pipeline import PipelineValidationResult
 from deepset_mcp.api.protocols import AsyncClientProtocol
+from deepset_mcp.api.shared_models import PaginatedResponse
 
 
-async def list_indexes(*, client: AsyncClientProtocol, workspace: str) -> IndexList | str:
-    """Use this to list available indexes on the deepset platform in your workspace.
+async def list_indexes(
+    *, client: AsyncClientProtocol, workspace: str, after: str | None = None
+) -> PaginatedResponse[Index] | str:
+    """Retrieves a list of all indexes available within the currently configured deepset workspace.
 
-    :param client: Deepset API client to use for requesting indexes.
-    :param workspace: Workspace of which to list indexes.
+    :param client: The async client for API communication.
+    :param workspace: The workspace name.
+    :param after: The cursor to fetch the next page of results.
+        If there are more results to fetch, the cursor will appear as `next_cursor` on the response.
+    :returns: List of indexes or error message.
     """
     try:
-        result = await client.indexes(workspace=workspace).list()
-    except ResourceNotFoundError as e:
-        return f"Error listing indexes. Error: {e.message} ({e.status_code})"
-
-    return result
+        return await client.indexes(workspace=workspace).list(after=after)
+    except ResourceNotFoundError:
+        return f"There is no workspace named '{workspace}'. Did you mean to configure it?"
+    except (BadRequestError, UnexpectedAPIError) as e:
+        return f"Failed to list indexes: {e}"
 
 
 async def get_index(*, client: AsyncClientProtocol, workspace: str, index_name: str) -> Index | str:
@@ -55,7 +61,7 @@ async def create_index(
     """
     try:
         result = await client.indexes(workspace=workspace).create(
-            name=index_name, yaml_config=yaml_configuration, description=description
+            index_name=index_name, yaml_config=yaml_configuration, description=description
         )
     except ResourceNotFoundError:
         return f"There is no workspace named '{workspace}'. Did you mean to configure it?"

test/integration/test_integration_index_resource.py

@@ -11,6 +11,7 @@
 from deepset_mcp.api.indexes.models import Index
 from deepset_mcp.api.indexes.resource import IndexResource
 from deepset_mcp.api.pipeline.models import PipelineValidationResult
+from deepset_mcp.api.shared_models import PaginatedResponse
 
 pytestmark = pytest.mark.integration
 
@@ -238,25 +239,48 @@ async def test_list_indexes(
 
     # Test listing without pagination
     indexes = await index_resource.list(limit=10)
+    assert isinstance(indexes, PaginatedResponse)
     assert len(indexes.data) == 3
 
     # Verify our created indexes are in the list
     retrieved_names = [p.name for p in indexes.data]
     for name in index_names:
         assert name in retrieved_names
 
-    # Test pagination
-    if len(indexes.data) > 1:
-        # Get the first page with 1 item
-        first_page = await index_resource.list(limit=1)
-        assert len(first_page.data) == 1
 
-        # Get the second page
-        second_page = await index_resource.list(page_number=2, limit=1)
-        assert len(second_page.data) == 1
+@pytest.mark.asyncio
+async def test_pagination_iteration(
+    index_resource: IndexResource,
+    valid_index_config: str,
+) -> None:
+    """Test iterating over multiple pages of indexes using the async iterator."""
+    # Create several test indexes
+    config = json.loads(valid_index_config)
+    index_names = []
+    for i in range(5):
+        index_name = f"test-pagination-index-{i}"
+        index_names.append(index_name)
+        await index_resource.create(index_name=index_name, yaml_config=config["yaml_config"])
+
+    # Get the first page with a small limit to ensure pagination
+    paginator = await index_resource.list(limit=2)
+
+    # Collect all indexes by iterating through pages
+    all_indexes = []
+    async for index in paginator:
+        all_indexes.append(index)
+
+    # Verify we got all our created indexes (at least 5)
+    assert len(all_indexes) >= 5
 
-        # Verify they're different indexes
-        assert first_page.data[0].pipeline_index_id != second_page.data[0].pipeline_index_id
+    # Verify all indexes are Index instances
+    for index in all_indexes:
+        assert isinstance(index, Index)
+
+    # Verify our created indexes are in the results
+    retrieved_names = [p.name for p in all_indexes]
+    for name in index_names:
+        assert name in retrieved_names
 
 
 @pytest.mark.asyncio