feat: add pipeline_type filter support to pipeline templates (#53)

* feat: add PipelineType enum and pipeline_type field to PipelineTemplate model

* feat: add filter parameter to list_templates method

* feat: update PipelineTemplateResourceProtocol to include filter parameter

* feat: add filter support to list_pipeline_templates tool

* test: add pipeline_type to sample template for testing

* test: add unit tests for filter and sorting parameters

* test: import PipelineType enum for updated tests

* test: update FakePipelineTemplateResource to support new parameters

* test: import Any for type hints

* test: add pipeline_type to test template instances

* test: add pipeline_type to second test template

* test: add pipeline_type to remaining test template

* test: add comprehensive tests for filter and sorting functionality

* feat: update MCP tool to support filter and sorting parameters

* test: add integration tests for filter and sorting functionality

* fix: tests

Author: mathislucka

src/deepset_mcp/api/pipeline_template/models.py

@@ -1,8 +1,16 @@
+from enum import StrEnum
 from uuid import UUID
 
 from pydantic import BaseModel, Field
 
 
+class PipelineType(StrEnum):
+    """Enum representing the type of a pipeline template."""
+
+    QUERY = "query"
+    INDEXING = "indexing"
+
+
 class PipelineTemplateTag(BaseModel):
     """Model representing a tag on a pipeline template."""
 
@@ -17,7 +25,9 @@ class PipelineTemplate(BaseModel):
     best_for: list[str]
     description: str
     template_name: str = Field(alias="pipeline_name")
+    display_name: str = Field(alias="name")
     pipeline_template_id: UUID = Field(alias="pipeline_template_id")
     potential_applications: list[str] = Field(alias="potential_applications")
     yaml_config: str | None = Field(None, alias="query_yaml")
     tags: list[PipelineTemplateTag]
+    pipeline_type: PipelineType

src/deepset_mcp/api/pipeline_template/resource.py

@@ -41,23 +41,36 @@ async def get_template(self, template_name: str) -> PipelineTemplate:
 
         return PipelineTemplate.model_validate(data)
 
-    async def list_templates(self, limit: int = 100) -> list[PipelineTemplate]:
+    async def list_templates(
+        self, limit: int = 100, field: str = "created_at", order: str = "DESC", filter: str | None = None
+    ) -> list[PipelineTemplate]:
         """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
         -------
         list[PipelineTemplate]
             List of pipeline templates
         """
+        params = {"limit": limit, "page_number": 1, "field": field, "order": order}
+
+        if filter is not None:
+            params["filter"] = filter
+
         response = await self._client.request(
             f"/v1/workspaces/{self._workspace}/pipeline_templates",
             method="GET",
-            params={"limit": limit, "page_number": 1, "field": "created_at", "order": "DESC"},
+            params=params,
         )
 
         raise_for_status(response)

src/deepset_mcp/api/protocols.py

@@ -134,7 +134,9 @@ 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) -> list[PipelineTemplate]:
+    async def list_templates(
+        self, limit: int = 100, field: str = "created_at", order: str = "DESC", filter: str | None = None
+    ) -> list[PipelineTemplate]:
         """List pipeline templates in the configured workspace."""
         ...
 

src/deepset_mcp/tools/pipeline_template.py

@@ -3,10 +3,29 @@
 from deepset_mcp.tools.formatting_utils import pipeline_template_to_llm_readable_string
 
 
-async def list_pipeline_templates(client: AsyncClientProtocol, workspace: str) -> str:
-    """Retrieves a list of all available pipeline templates."""
+async def list_pipeline_templates(
+    client: AsyncClientProtocol,
+    workspace: str,
+    limit: int = 100,
+    field: str = "created_at",
+    order: str = "DESC",
+    filter: str | None = None,
+) -> str:
+    """Retrieves a list of all available pipeline 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 filter: OData filter expression to filter templates by criteria.
+
+    :returns: Formatted string with template information.
+    """
     try:
-        response = await client.pipeline_templates(workspace=workspace).list_templates()
+        response = await client.pipeline_templates(workspace=workspace).list_templates(
+            limit=limit, field=field, order=order, filter=filter
+        )
         formatted_templates = [pipeline_template_to_llm_readable_string(t) for t in response]
         return "\n\n".join(formatted_templates)
     except ResourceNotFoundError:

test/integration/test_integration_pipeline_resource.py

@@ -222,9 +222,6 @@ async def test_validation_invalid_yaml(
     assert result.errors[0].code == "PIPELINE_SCHEMA_ERROR"
 
 
-@pytest.mark.skip(
-    reason="API not working correctly."
-)  # skip until deepset API correctly returns error for yaml syntax errors
 @pytest.mark.asyncio
 async def test_validation_syntax_error(
     pipeline_resource: PipelineResource,

test/integration/test_integration_pipeline_template_resource.py

@@ -93,3 +93,48 @@ async def test_list_templates_with_limit(
 
     # Verify that the number of templates is not more than the limit
     assert len(templates) <= limit
+
+
+@pytest.mark.asyncio
+async def test_list_templates_with_filter(
+    template_resource: PipelineTemplateResource,
+) -> None:
+    """Test listing templates with a pipeline type filter."""
+    # Test filtering by QUERY pipeline type
+    query_templates = await template_resource.list_templates(filter="pipeline_type eq 'QUERY'")
+
+    # Verify that all returned templates are QUERY type
+    assert isinstance(query_templates, list)
+
+    # If templates are available, verify they are all QUERY type
+    for template in query_templates:
+        assert isinstance(template, PipelineTemplate)
+        assert template.pipeline_type == "query"
+
+    # Test filtering by INDEXING pipeline type
+    indexing_templates = await template_resource.list_templates(filter="pipeline_type eq 'INDEXING'")
+
+    # Verify that all returned templates are INDEXING type
+    assert isinstance(indexing_templates, list)
+
+    # If templates are available, verify they are all INDEXING type
+    for template in indexing_templates:
+        assert isinstance(template, PipelineTemplate)
+        assert template.pipeline_type == "indexing"
+
+
+@pytest.mark.asyncio
+async def test_list_templates_with_custom_sorting(
+    template_resource: PipelineTemplateResource,
+) -> None:
+    """Test listing templates with custom sorting."""
+    # Test sorting by name in ascending order
+    templates = await template_resource.list_templates(field="name", order="ASC", limit=5)
+
+    # Verify that the templates are returned as a list
+    assert isinstance(templates, list)
+
+    # If we have multiple templates, verify they are sorted correctly
+    if len(templates) > 1:
+        for i in range(len(templates) - 1):
+            assert templates[i].display_name <= templates[i + 1].display_name

test/unit/api/pipeline_template/test_pipeline_template_resource.py

@@ -14,6 +14,7 @@ def create_sample_template(
     author: str = "deepset-ai",
     description: str = "A test template",
     template_id: str = "3fa85f64-5717-4562-b3fc-2c963f66afa6",
+    pipeline_type: str = "query",
 ) -> dict[str, Any]:
     """Create a sample pipeline template response dictionary for testing."""
     return {
@@ -29,6 +30,7 @@ def create_sample_template(
         "potential_applications": ["testing", "development"],
         "recommended_dataset": ["sample-data"],
         "tags": [{"name": "test", "tag_id": "d4a85f64-5717-4562-b3fc-2c963f66afa6"}],
+        "pipeline_type": pipeline_type,
     }
 
 
@@ -179,3 +181,75 @@ async def test_list_templates_empty_result(self) -> None:
 
         # Verify empty results
         assert len(result) == 0
+
+    @pytest.mark.asyncio
+    async def test_list_templates_with_filter(self) -> None:
+        """Test listing templates with a filter."""
+        # Create sample data
+        sample_templates = [
+            create_sample_template(
+                name="Query Template", template_id="1fa85f64-5717-4562-b3fc-2c963f66afa6", pipeline_type="query"
+            ),
+        ]
+
+        # Create client with predefined response
+        client = DummyClient(
+            responses={
+                "test-workspace/pipeline_templates": {
+                    "data": sample_templates,
+                    "has_more": False,
+                    "total": 1,
+                }
+            }
+        )
+
+        # Create resource and call list method with filter
+        resource = PipelineTemplateResource(client=client, workspace="test-workspace")
+        result = await resource.list_templates(filter="pipeline_type eq 'QUERY'")
+
+        # Verify results
+        assert len(result) == 1
+        assert isinstance(result[0], PipelineTemplate)
+        assert result[0].template_name == "Query Template"
+        assert result[0].pipeline_type == "query"
+
+        # Verify request includes filter
+        assert len(client.requests) == 1
+        assert client.requests[0]["endpoint"] == "/v1/workspaces/test-workspace/pipeline_templates"
+        assert "filter" in client.requests[0]["params"]
+        assert client.requests[0]["params"]["filter"] == "pipeline_type eq 'QUERY'"
+
+    @pytest.mark.asyncio
+    async def test_list_templates_with_custom_sorting(self) -> None:
+        """Test listing templates with custom field and order."""
+        # Create sample data
+        sample_templates = [
+            create_sample_template(name="Template A", template_id="1fa85f64-5717-4562-b3fc-2c963f66afa6"),
+            create_sample_template(name="Template B", template_id="2fa85f64-5717-4562-b3fc-2c963f66afa6"),
+        ]
+
+        # Create client with predefined response
+        client = DummyClient(
+            responses={
+                "test-workspace/pipeline_templates": {
+                    "data": sample_templates,
+                    "has_more": False,
+                    "total": 2,
+                }
+            }
+        )
+
+        # Create resource and call list method with custom sorting
+        resource = PipelineTemplateResource(client=client, workspace="test-workspace")
+        result = await resource.list_templates(field="name", order="ASC")
+
+        # Verify results
+        assert len(result) == 2
+        assert isinstance(result[0], PipelineTemplate)
+        assert result[0].template_name == "Template A"
+        assert result[1].template_name == "Template B"
+
+        # Verify request includes custom sorting
+        assert len(client.requests) == 1
+        assert client.requests[0]["params"]["field"] == "name"
+        assert client.requests[0]["params"]["order"] == "ASC"