feat: add cursor-based pagination to pipeline template resource (#177)

* feat: add cursor-based pagination to pipeline template resource

* fix: do not expose after until fixed

* fix: format

Author: mathislucka

src/deepset_mcp/api/indexes/models.py

@@ -57,11 +57,3 @@ def __rich_repr__(self) -> Result:
         )
         yield "last_edited_at", self.last_edited_at.strftime("%m/%d/%Y %I:%M:%S %p") if self.last_edited_at else None
         yield "yaml_config", self.yaml_config if self.yaml_config is not None else "Get full index to see config."
-
-
-class IndexList(BaseModel):
-    """Response model for listing indexes."""
-
-    data: list[Index]
-    has_more: bool
-    total: int

src/deepset_mcp/api/pipeline_template/models.py

@@ -64,14 +64,6 @@ def populate_yaml_config(cls, values: Any) -> Any:
         return values
 
 
-class PipelineTemplateList(BaseModel):
-    """Response model for listing pipeline templates."""
-
-    data: list[PipelineTemplate]
-    has_more: bool
-    total: int
-
-
 class PipelineTemplateSearchResult(BaseModel):
     """Model representing a search result for pipeline templates."""
 

src/deepset_mcp/api/pipeline_template/protocols.py

@@ -4,7 +4,8 @@
 
 from typing import Protocol
 
-from deepset_mcp.api.pipeline_template.models import PipelineTemplate, PipelineTemplateList
+from deepset_mcp.api.pipeline_template.models import PipelineTemplate
+from deepset_mcp.api.shared_models import PaginatedResponse
 
 
 class PipelineTemplateResourceProtocol(Protocol):
@@ -14,8 +15,13 @@ async def get_template(self, template_name: str) -> PipelineTemplate:
         """Fetch a single pipeline template by its name."""
         ...
 
-    async def list_templates(
-        self, limit: int = 100, field: str = "created_at", order: str = "DESC", filter: str | None = None
-    ) -> PipelineTemplateList:
-        """List pipeline templates in the configured workspace."""
+    async def list(
+        self,
+        limit: int = 10,
+        after: str | None = None,
+        field: str = "created_at",
+        order: str = "DESC",
+        filter: str | None = None,
+    ) -> PaginatedResponse[PipelineTemplate]:
+        """Lists pipeline templates and returns the first page of results with pagination support."""
         ...

src/deepset_mcp/api/pipeline_template/resource.py

@@ -5,9 +5,10 @@
 from typing import Any
 
 from deepset_mcp.api.exceptions import UnexpectedAPIError
-from deepset_mcp.api.pipeline_template.models import PipelineTemplate, PipelineTemplateList
+from deepset_mcp.api.pipeline_template.models import PipelineTemplate
 from deepset_mcp.api.pipeline_template.protocols import PipelineTemplateResourceProtocol
 from deepset_mcp.api.protocols import AsyncClientProtocol
+from deepset_mcp.api.shared_models import PaginatedResponse
 from deepset_mcp.api.transport import raise_for_status
 
 
@@ -46,47 +47,58 @@ async def get_template(self, template_name: str) -> PipelineTemplate:
 
         return PipelineTemplate.model_validate(data)
 
-    async def list_templates(
-        self, limit: int = 100, field: str = "created_at", order: str = "DESC", filter: str | None = None
-    ) -> PipelineTemplateList:
-        """List pipeline templates in the configured workspace.
-
-        Parameters
-        ----------
-        limit : int, optional (default=100)
-            Maximum number of templates to return
-        field : str, optional (default="created_at")
-            Field to sort by
-        order : str, optional (default="DESC")
-            Sort order (ASC or DESC)
-        filter : str | None, optional (default=None)
-            OData filter expression for filtering templates
-
-        Returns
-        -------
-        PipelineTemplateList
-            List of pipeline templates with metadata
+    async def list(
+        self,
+        limit: int = 10,
+        after: str | None = None,
+        field: str = "created_at",
+        order: str = "DESC",
+        filter: str | None = None,
+    ) -> PaginatedResponse[PipelineTemplate]:
+        """Lists pipeline templates and returns the first page of results.
+
+        The returned object can be iterated over to fetch subsequent pages.
+
+        :param limit: The maximum number of pipeline templates to return per page.
+        :param after: The cursor to fetch the next page of results.
+        :param field: Field to sort by (default: "created_at").
+        :param order: Sort order, either "ASC" or "DESC" (default: "DESC").
+        :param filter: OData filter expression for filtering templates.
+        :returns: A `PaginatedResponse` object containing the first page of pipeline templates.
         """
-        params = {"limit": limit, "page_number": 1, "field": field, "order": order}
-
+        # TODO: Remove when fixed
+        if after is not None:
+            raise ValueError("Pagination using 'after' parameter is currently not supported by the deepset platform.")
+
+        # 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 templates older than the current cursor
+        # TODO: while 'before' signals templates 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, "field": field, "order": order}
         if filter is not None:
-            params["filter"] = filter
+            request_params["filter"] = filter
+        request_params = {k: v for k, v in request_params.items() if v is not None}
 
-        response = await self._client.request(
-            f"/v1/workspaces/{self._workspace}/pipeline_templates",
-            method="GET",
-            params=params,
-        )
-
-        raise_for_status(response)
+        # 2. Make the first API call using a private, stateless method
+        page = await self._list_api_call(**request_params)
 
-        if response.json is None:
-            raise UnexpectedAPIError(message="Unexpected API response, no templates returned.")
-
-        response_data: dict[str, Any] = response.json
+        # 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, "field": field, "order": order, "filter": filter},
+        )
+        return page
 
-        return PipelineTemplateList(
-            data=[PipelineTemplate.model_validate(template) for template in response_data["data"]],
-            has_more=response_data.get("has_more", False),
-            total=response_data.get("total", len(response_data["data"])),
+    async def _list_api_call(self, **kwargs: Any) -> PaginatedResponse[PipelineTemplate]:
+        """A private, stateless method that performs the raw API call."""
+        resp = await self._client.request(
+            endpoint=f"v1/workspaces/{self._workspace}/pipeline_templates", 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 PaginatedResponse[PipelineTemplate].create_with_cursor_field(resp.json, "name")

src/deepset_mcp/tools/pipeline_template.py

@@ -7,12 +7,12 @@
 from deepset_mcp.api.exceptions import ResourceNotFoundError, UnexpectedAPIError
 from deepset_mcp.api.pipeline_template.models import (
     PipelineTemplate,
-    PipelineTemplateList,
     PipelineTemplateSearchResult,
     PipelineTemplateSearchResults,
     PipelineType,
 )
 from deepset_mcp.api.protocols import AsyncClientProtocol
+from deepset_mcp.api.shared_models import PaginatedResponse
 from deepset_mcp.tools.model_protocol import ModelProtocol
 
 
@@ -21,27 +21,22 @@ async def list_templates(
     client: AsyncClientProtocol,
     workspace: str,
     limit: int = 100,
-    field: str = "created_at",
-    order: str = "DESC",
     pipeline_type: PipelineType | str | None = None,
-) -> PipelineTemplateList | str:
+    # after: str | None = None TODO
+) -> PaginatedResponse[PipelineTemplate] | str:
     """Retrieves a list of all available pipeline and indexing templates.
 
     :param client: The async client for API requests.
     :param workspace: The workspace to list templates from.
     :param limit: Maximum number of templates to return (default: 100).
-    :param field: Field to sort by (default: "created_at").
-    :param order: Sort order, either "ASC" or "DESC" (default: "DESC").
     :param pipeline_type: The type of pipeline to return.
 
     :returns: List of pipeline templates or error message.
     """
     try:
-        return await client.pipeline_templates(workspace=workspace).list_templates(
+        return await client.pipeline_templates(workspace=workspace).list(
             limit=limit,
-            field=field,
-            order=order,
-            filter=f"pipeline_type eq '{pipeline_type}'" if pipeline_type else None,
+            filter=f"pipeline_type eq '{pipeline_type}'" if pipeline_type else None,  # TODO: after=after
         )
     except ResourceNotFoundError:
         return f"There is no workspace named '{workspace}'. Did you mean to configure it?"
@@ -87,18 +82,21 @@ async def search_templates(
     :returns: Search results with similarity scores or error message.
     """
     try:
-        response = await client.pipeline_templates(workspace=workspace).list_templates(
-            filter=f"pipeline_type eq '{pipeline_type}'"
-        )
+        response = await client.pipeline_templates(workspace=workspace).list()
+        templates = response.data
+
+        # Filter by pipeline_type if specified
+        if pipeline_type:
+            templates = [t for t in templates if t.pipeline_type == pipeline_type]
     except UnexpectedAPIError as e:
         return f"Failed to retrieve pipeline templates: {e}"
 
-    if not response.data:
+    if not templates:
         return PipelineTemplateSearchResults(results=[], query=query, total_found=0)
 
     # Extract text for embedding from all templates
     template_texts: list[tuple[str, str]] = [
-        (template.template_name, f"{template.template_name} {template.description}") for template in response.data
+        (template.template_name, f"{template.template_name} {template.description}") for template in templates
     ]
     template_names: list[str] = [t[0] for t in template_texts]
 
@@ -122,7 +120,7 @@ async def search_templates(
     search_results = []
     for template_name, sim in top_templates:
         # Find the template object by name
-        template = next((t for t in response.data if t.template_name == template_name), None)
+        template = next((t for t in templates if t.template_name == template_name), None)
         if template:
             search_results.append(PipelineTemplateSearchResult(template=template, similarity_score=float(sim)))