feat: add pipeline template resource (#25)

* feat: add pipeline template package

* feat: add pipeline template models

* feat: add pipeline template resource implementation

* feat: import PipelineTemplate model in protocols

* feat: add pipeline_templates method to AsyncClientProtocol

* feat: add PipelineTemplateResourceProtocol

* feat: import PipelineTemplateResource in client

* feat: add pipeline_templates method to AsyncDeepsetClient

* test: add pipeline template tests package

* test: add pipeline template resource tests

* test: create conftest with fake client implementation

* feat: add pipeline template tests directory

* refactor: remove duplicate test directory

* feat: add base test classes and helpers for pipeline template tests

* feat: add pipeline template resource method tests

* fix: minor issues; add integration tests

Author: mathislucka

src/deepset_mcp/api/client.py

@@ -4,6 +4,7 @@
 
 from deepset_mcp.api.haystack_service.resource import HaystackServiceResource
 from deepset_mcp.api.pipeline.resource import PipelineResource
+from deepset_mcp.api.pipeline_template.resource import PipelineTemplateResource
 from deepset_mcp.api.protocols import AsyncClientProtocol
 from deepset_mcp.api.transport import AsyncTransport, TransportProtocol, TransportResponse
 
@@ -98,3 +99,7 @@ def pipelines(self, workspace: str) -> PipelineResource:
     def haystack_service(self) -> HaystackServiceResource:
         """Resource to interact with the Haystack service API."""
         return HaystackServiceResource(client=self)
+
+    def pipeline_templates(self, workspace: str) -> PipelineTemplateResource:
+        """Resource to interact with pipeline templates in the specified workspace."""
+        return PipelineTemplateResource(client=self, workspace=workspace)

src/deepset_mcp/api/pipeline_template/__init__.py

@@ -0,0 +1,23 @@
+from uuid import UUID
+
+from pydantic import BaseModel, Field
+
+
+class PipelineTemplateTag(BaseModel):
+    """Model representing a tag on a pipeline template."""
+
+    name: str
+    tag_id: UUID
+
+
+class PipelineTemplate(BaseModel):
+    """Model representing a pipeline template."""
+
+    author: str
+    best_for: list[str]
+    description: str
+    template_name: str = Field(alias="pipeline_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]

src/deepset_mcp/api/pipeline_template/models.py

@@ -0,0 +1,69 @@
+from typing import Any
+
+from deepset_mcp.api.exceptions import UnexpectedAPIError
+from deepset_mcp.api.pipeline_template.models import PipelineTemplate
+from deepset_mcp.api.protocols import AsyncClientProtocol
+from deepset_mcp.api.transport import raise_for_status
+
+
+class PipelineTemplateResource:
+    """Resource for interacting with pipeline templates in a workspace."""
+
+    def __init__(self, client: AsyncClientProtocol, workspace: str) -> None:
+        """Initialize the pipeline template resource.
+
+        Parameters
+        ----------
+        client : AsyncClientProtocol
+            Client to use for making API requests
+        workspace : str
+            Workspace to operate in
+        """
+        self._client = client
+        self._workspace = workspace
+
+    async def get_template(self, template_name: str) -> PipelineTemplate:
+        """Fetch a single pipeline template by its name.
+
+        Parameters
+        ----------
+        template_name : str
+            Name of the template to fetch
+
+        Returns
+        -------
+        PipelineTemplate
+            The requested pipeline template
+        """
+        response = await self._client.request(f"/v1/workspaces/{self._workspace}/pipeline_templates/{template_name}")
+        raise_for_status(response)
+        data = response.json
+
+        return PipelineTemplate.model_validate(data)
+
+    async def list_templates(self, limit: int = 100) -> list[PipelineTemplate]:
+        """List pipeline templates in the configured workspace.
+
+        Parameters
+        ----------
+        limit : int, optional (default=100)
+            Maximum number of templates to return
+
+        Returns
+        -------
+        list[PipelineTemplate]
+            List of pipeline templates
+        """
+        response = await self._client.request(
+            f"/v1/workspaces/{self._workspace}/pipeline_templates?limit={limit}&page_number=1&field=created_at&order=DESC",
+            method="GET",
+        )
+
+        raise_for_status(response)
+
+        if response.json is None:
+            raise UnexpectedAPIError(message="Unexpected API response, no templates returned.")
+
+        response_data: dict[str, Any] = response.json
+
+        return [PipelineTemplate.model_validate(template) for template in response_data["data"]]

src/deepset_mcp/api/pipeline_template/resource.py

@@ -2,6 +2,7 @@
 from typing import Any, Protocol, Self
 
 from deepset_mcp.api.pipeline.models import DeepsetPipeline, NoContentResponse, PipelineValidationResult
+from deepset_mcp.api.pipeline_template.models import PipelineTemplate
 from deepset_mcp.api.transport import TransportResponse
 
 
@@ -48,6 +49,22 @@ def haystack_service(self) -> "HaystackServiceProtocol":
         """Access the Haystack service."""
         ...
 
+    def pipeline_templates(self, workspace: str) -> "PipelineTemplateResourceProtocol":
+        """Access pipeline templates in the specified workspace."""
+        ...
+
+
+class PipelineTemplateResourceProtocol(Protocol):
+    """Protocol defining the implementation for PipelineTemplateResource."""
+
+    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]:
+        """List pipeline templates in the configured workspace."""
+        ...
+
 
 class PipelineResourceProtocol(Protocol):
     """Protocol defining the implementation for PipelineResource."""

src/deepset_mcp/api/protocols.py

@@ -0,0 +1,95 @@
+import pytest
+
+from deepset_mcp.api.client import AsyncDeepsetClient
+from deepset_mcp.api.exceptions import ResourceNotFoundError
+from deepset_mcp.api.pipeline_template.models import PipelineTemplate
+from deepset_mcp.api.pipeline_template.resource import PipelineTemplateResource
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture
+async def template_resource(
+    client: AsyncDeepsetClient,
+    test_workspace: str,
+) -> PipelineTemplateResource:
+    """Create a PipelineTemplateResource instance for testing."""
+    return PipelineTemplateResource(client=client, workspace=test_workspace)
+
+
+@pytest.mark.asyncio
+async def test_get_template(
+    template_resource: PipelineTemplateResource,
+) -> None:
+    """Test getting a single pipeline template by name.
+
+    First lists all templates, then gets the first one by name.
+    """
+    # Get all templates to find an existing one
+    templates = await template_resource.list_templates()
+
+    # Skip if no templates are available
+    if not templates:
+        pytest.skip("No templates available in the test environment")
+
+    # Get the first template's name
+    template_name = templates[0].template_name
+
+    # Now get that specific template
+    template = await template_resource.get_template(template_name=template_name)
+
+    # Verify the template was retrieved correctly
+    assert template.template_name == template_name
+    assert template.pipeline_template_id is not None
+    assert isinstance(template.best_for, list)
+    assert isinstance(template.potential_applications, list)
+    assert isinstance(template.tags, list)
+
+
+@pytest.mark.asyncio
+async def test_get_nonexistent_template(
+    template_resource: PipelineTemplateResource,
+) -> None:
+    """Test error handling when getting a non-existent template."""
+    non_existent_name = "non-existent-template-xyz-123"
+
+    # Trying to get a non-existent template should raise an exception
+    with pytest.raises(ResourceNotFoundError):
+        await template_resource.get_template(template_name=non_existent_name)
+
+
+@pytest.mark.asyncio
+async def test_list_templates(
+    template_resource: PipelineTemplateResource,
+) -> None:
+    """Test listing templates."""
+    # Test listing templates with default limit
+    templates = await template_resource.list_templates()
+
+    # Verify that the templates are returned as a list
+    assert isinstance(templates, list)
+
+    # Skip further checks if no templates are available
+    if not templates:
+        pytest.skip("No templates available in the test environment")
+
+    # Verify the first template has the expected structure
+    template = templates[0]
+    assert isinstance(template, PipelineTemplate)
+    assert template.template_name is not None
+    assert template.author is not None
+    assert template.description is not None
+    assert template.pipeline_template_id is not None
+
+
+@pytest.mark.asyncio
+async def test_list_templates_with_limit(
+    template_resource: PipelineTemplateResource,
+) -> None:
+    """Test listing templates with a specific limit."""
+    # Test with a small limit
+    limit = 1
+    templates = await template_resource.list_templates(limit=limit)
+
+    # Verify that the number of templates is not more than the limit
+    assert len(templates) <= limit