feat: add pipeline template tools (#35)

* feat: add pipeline template tools

* feat: import PipelineTemplate model

* feat: add pipeline template formatting utility

* feat: import pipeline template tools

* feat: add pipeline template MCP tools

* test: add unit tests for pipeline template tools

* fix: linting and format

Author: mathislucka

src/deepset_mcp/main.py

@@ -16,6 +16,10 @@
     update_pipeline as update_pipeline_tool,
     validate_pipeline as validate_pipeline_tool,
 )
+from deepset_mcp.tools.pipeline_template import (
+    get_pipeline_template as get_pipeline_template_tool,
+    list_pipeline_templates as list_pipeline_templates_tool,
+)
 
 INITIALIZED_MODEL = StaticModel.from_pretrained("minishlab/potion-base-2M")
 
@@ -45,6 +49,35 @@ async def list_pipelines() -> str:
     return response
 
 
+@mcp.tool()
+async def list_pipeline_templates() -> str:
+    """Retrieves a list of all pipeline templates available within the currently configured deepset workspace.
+
+    Use this when you need to know the available pipeline templates and their capabilities.
+    """
+    workspace = get_workspace()
+    async with AsyncDeepsetClient() as client:
+        response = await list_pipeline_templates_tool(client, workspace)
+
+    return response
+
+
+@mcp.tool()
+async def get_pipeline_template(template_name: str) -> str:
+    """Fetches detailed configuration information for a specific pipeline template.
+
+    This includes its YAML configuration, metadata, and recommended use cases.
+    Use this when you need to inspect a specific template's structure or settings.
+
+    :param template_name: Name of the pipeline template to retrieve.
+    """
+    workspace = get_workspace()
+    async with AsyncDeepsetClient() as client:
+        response = await get_pipeline_template_tool(client, workspace, template_name)
+
+    return response
+
+
 @mcp.tool()
 async def get_pipeline(pipeline_name: str) -> str:
     """Fetches detailed configuration information for a specific pipeline, identified by its unique `pipeline_name`.

src/deepset_mcp/tools/formatting_utils.py

@@ -1,4 +1,40 @@
 from deepset_mcp.api.pipeline.models import DeepsetPipeline, PipelineValidationResult
+from deepset_mcp.api.pipeline_template.models import PipelineTemplate
+
+
+def pipeline_template_to_llm_readable_string(template: PipelineTemplate) -> str:
+    """Creates a string representation of a pipeline template that is readable by LLMs."""
+    template_parts = [
+        f'''<pipeline_template name="{template.template_name}" id="{template.pipeline_template_id}">
+
+### Basic Information
+
+**Name:** {template.template_name}
+**Author:** {template.author}
+**Description:** {template.description}
+'''
+    ]
+
+    if template.best_for:
+        template_parts.append("\n### Best For\n" + "\n".join(f"- {use}" for use in template.best_for))
+
+    if template.potential_applications:
+        template_parts.append(
+            "\n### Potential Applications\n" + "\n".join(f"- {app}" for app in template.potential_applications)
+        )
+
+    if template.tags:
+        template_parts.append("\n### Tags\n" + "\n".join(f"- {tag.name}" for tag in template.tags))
+
+    if template.yaml_config is not None:
+        template_parts.append("\n### Template Configuration")
+        template_parts.append(f"\n```yaml\n{template.yaml_config}\n```")
+
+    template_parts.append(
+        f'\n</pipeline_template name="{template.template_name}" id="{template.pipeline_template_id}">'
+    )
+
+    return "\n".join(template_parts)
 
 
 def pipeline_to_llm_readable_string(pipeline: DeepsetPipeline) -> str:

src/deepset_mcp/tools/pipeline_template.py

@@ -0,0 +1,26 @@
+from deepset_mcp.api.exceptions import ResourceNotFoundError, UnexpectedAPIError
+from deepset_mcp.api.protocols import AsyncClientProtocol
+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."""
+    try:
+        response = await client.pipeline_templates(workspace=workspace).list_templates()
+        formatted_templates = [pipeline_template_to_llm_readable_string(t) for t in response]
+        return "\n\n".join(formatted_templates)
+    except ResourceNotFoundError:
+        return f"There is no workspace named '{workspace}'. Did you mean to configure it?"
+    except UnexpectedAPIError as e:
+        return f"Failed to list pipeline templates: {e}"
+
+
+async def get_pipeline_template(client: AsyncClientProtocol, workspace: str, template_name: str) -> str:
+    """Fetches detailed information for a specific pipeline template, identified by its `template_name`."""
+    try:
+        response = await client.pipeline_templates(workspace=workspace).get_template(template_name)
+        return pipeline_template_to_llm_readable_string(response)
+    except ResourceNotFoundError:
+        return f"There is no pipeline template named '{template_name}' in workspace '{workspace}'."
+    except UnexpectedAPIError as e:
+        return f"Failed to fetch pipeline template '{template_name}': {e}"

test/unit/tools/test_pipeline_template.py

@@ -0,0 +1,139 @@
+from uuid import UUID
+
+import pytest
+
+from deepset_mcp.api.exceptions import ResourceNotFoundError, UnexpectedAPIError
+from deepset_mcp.api.pipeline_template.models import PipelineTemplate, PipelineTemplateTag
+from deepset_mcp.tools.pipeline_template import get_pipeline_template, list_pipeline_templates
+from test.unit.conftest import BaseFakeClient
+
+
+class FakePipelineTemplateResource:
+    def __init__(
+        self,
+        list_response: list[PipelineTemplate] | None = None,
+        get_response: PipelineTemplate | None = None,
+        list_exception: Exception | None = None,
+        get_exception: Exception | None = None,
+    ) -> None:
+        self._list_response = list_response
+        self._get_response = get_response
+        self._list_exception = list_exception
+        self._get_exception = get_exception
+
+    async def list_templates(self, limit: int = 100) -> list[PipelineTemplate]:
+        if self._list_exception:
+            raise self._list_exception
+        if self._list_response is not None:
+            return self._list_response
+        raise NotImplementedError
+
+    async def get_template(self, template_name: str) -> PipelineTemplate:
+        if self._get_exception:
+            raise self._get_exception
+        if self._get_response is not None:
+            return self._get_response
+        raise NotImplementedError
+
+
+class FakeClient(BaseFakeClient):
+    def __init__(self, resource: FakePipelineTemplateResource) -> None:
+        self._resource = resource
+        super().__init__()
+
+    def pipeline_templates(self, workspace: str) -> FakePipelineTemplateResource:
+        return self._resource
+
+
+@pytest.mark.asyncio
+async def test_list_pipeline_templates_returns_formatted_string() -> None:
+    template1 = PipelineTemplate(
+        pipeline_name="template1",
+        pipeline_template_id=UUID("00000000-0000-0000-0000-000000000001"),
+        author="Alice Smith",
+        description="First template",
+        best_for=["use case 1", "use case 2"],
+        potential_applications=["app 1", "app 2"],
+        query_yaml="config1: value1",
+        tags=[PipelineTemplateTag(name="tag1", tag_id=UUID("10000000-0000-0000-0000-000000000001"))],
+    )
+    template2 = PipelineTemplate(
+        pipeline_name="template2",
+        pipeline_template_id=UUID("00000000-0000-0000-0000-000000000002"),
+        author="Bob Jones",
+        description="Second template",
+        best_for=["use case 3"],
+        potential_applications=["app 3"],
+        query_yaml="config2: value2",
+        tags=[PipelineTemplateTag(name="tag2", tag_id=UUID("20000000-0000-0000-0000-000000000002"))],
+    )
+    resource = FakePipelineTemplateResource(list_response=[template1, template2])
+    client = FakeClient(resource)
+    result = await list_pipeline_templates(client, workspace="ws1")
+
+    assert result.count("<pipeline_template name=") == 2
+    assert "template1" in result
+    assert "template2" in result
+    assert "Alice Smith" in result
+    assert "Bob Jones" in result
+
+
+@pytest.mark.asyncio
+async def test_list_pipeline_templates_handles_resource_not_found() -> None:
+    resource = FakePipelineTemplateResource(list_exception=ResourceNotFoundError())
+    client = FakeClient(resource)
+    result = await list_pipeline_templates(client, workspace="invalid_ws")
+
+    assert "no workspace named 'invalid_ws'" in result.lower()
+
+
+@pytest.mark.asyncio
+async def test_list_pipeline_templates_handles_unexpected_error() -> None:
+    resource = FakePipelineTemplateResource(list_exception=UnexpectedAPIError(status_code=500, message="Server error"))
+    client = FakeClient(resource)
+    result = await list_pipeline_templates(client, workspace="ws1")
+
+    assert "Failed to list pipeline templates" in result
+    assert "Server error" in result
+
+
+@pytest.mark.asyncio
+async def test_get_pipeline_template_returns_formatted_string() -> None:
+    template = PipelineTemplate(
+        pipeline_name="test_template",
+        pipeline_template_id=UUID("00000000-0000-0000-0000-000000000001"),
+        author="Eve Brown",
+        description="Test template",
+        best_for=["use case 1"],
+        potential_applications=["app 1"],
+        query_yaml="config: value",
+        tags=[PipelineTemplateTag(name="tag1", tag_id=UUID("10000000-0000-0000-0000-000000000001"))],
+    )
+    resource = FakePipelineTemplateResource(get_response=template)
+    client = FakeClient(resource)
+    result = await get_pipeline_template(client, workspace="ws1", template_name="test_template")
+
+    assert "test_template" in result
+    assert "Eve Brown" in result
+    assert "Test template" in result
+    assert "config: value" in result
+    assert "tag1" in result
+
+
+@pytest.mark.asyncio
+async def test_get_pipeline_template_handles_resource_not_found() -> None:
+    resource = FakePipelineTemplateResource(get_exception=ResourceNotFoundError())
+    client = FakeClient(resource)
+    result = await get_pipeline_template(client, workspace="ws1", template_name="invalid_template")
+
+    assert "no pipeline template named 'invalid_template'" in result.lower()
+
+
+@pytest.mark.asyncio
+async def test_get_pipeline_template_handles_unexpected_error() -> None:
+    resource = FakePipelineTemplateResource(get_exception=UnexpectedAPIError(status_code=500, message="Server error"))
+    client = FakeClient(resource)
+    result = await get_pipeline_template(client, workspace="ws1", template_name="test_template")
+
+    assert "Failed to fetch pipeline template 'test_template'" in result
+    assert "Server error" in result