feat: use StrEnum for pipeline log level filtering (#65)

* feat: add LogLevel StrEnum for pipeline log filtering

* feat: import LogLevel enum

* feat: update get_logs method to use LogLevel enum

* feat: import LogLevel enum in pipeline tools

* feat: update get_pipeline_logs to use LogLevel enum

* feat: import LogLevel enum in main module

* feat: update MCP tool to convert string level to LogLevel enum

* feat: import LogLevel enum in pipeline tests

* feat: update fake resource method signature to use LogLevel enum

* feat: update test to use LogLevel enum

* feat: import LogLevel enum in pipeline resource tests

* feat: update unit test to use LogLevel enum

* feat: update unit test for warning level to use LogLevel enum

* feat: import LogLevel enum in integration tests

* fix: format and lint and types

Author: mathislucka

src/deepset_mcp/api/pipeline/log_level.py

@@ -0,0 +1,9 @@
+from enum import StrEnum
+
+
+class LogLevel(StrEnum):
+    """Log level filter options for pipeline logs."""
+
+    INFO = "info"
+    WARNING = "warning"
+    ERROR = "error"

src/deepset_mcp/api/pipeline/resource.py

@@ -1,6 +1,7 @@
 from typing import TYPE_CHECKING, Any
 
 from deepset_mcp.api.exceptions import UnexpectedAPIError
+from deepset_mcp.api.pipeline.log_level import LogLevel
 from deepset_mcp.api.pipeline.models import (
     DeepsetPipeline,
     NoContentResponse,
@@ -173,13 +174,13 @@ async def get_logs(
         self,
         pipeline_name: str,
         limit: int = 30,
-        level: str | None = None,
+        level: LogLevel | None = None,
     ) -> PipelineLogList:
         """Fetch logs for a specific pipeline.
 
         :param pipeline_name: Name of the pipeline to fetch logs for.
         :param limit: Maximum number of log entries to return.
-        :param level: Filter logs by level (info, warning, error). If None, returns all levels.
+        :param level: Filter logs by level. If None, returns all levels.
 
         :returns: A PipelineLogList containing the log entries.
         """

src/deepset_mcp/api/protocols.py

@@ -2,6 +2,7 @@
 from typing import Any, Protocol, Self, TypeVar, overload
 
 from deepset_mcp.api.indexes.models import Index, IndexList
+from deepset_mcp.api.pipeline.log_level import LogLevel
 from deepset_mcp.api.pipeline.models import (
     DeepsetPipeline,
     NoContentResponse,
@@ -182,7 +183,7 @@ async def get_logs(
         self,
         pipeline_name: str,
         limit: int = 30,
-        level: str | None = None,
+        level: LogLevel | None = None,
     ) -> PipelineLogList:
         """Fetch logs for a specific pipeline."""
         ...

src/deepset_mcp/main.py

@@ -6,6 +6,7 @@
 from model2vec import StaticModel
 
 from deepset_mcp.api.client import AsyncDeepsetClient
+from deepset_mcp.api.pipeline.log_level import LogLevel
 from deepset_mcp.tools.haystack_service import (
     get_component_definition as get_component_definition_tool,
     list_component_families as list_component_families_tool,
@@ -309,13 +310,22 @@ async def get_pipeline_logs(pipeline_name: str, limit: int = 30, level: str | No
     :param level: Filter logs by level. Valid values: 'info', 'warning', 'error'. If not specified, returns all levels.
     """
     workspace = get_workspace()
+
+    # Convert string level to LogLevel enum if provided
+    log_level: LogLevel | None = None
+    if level is not None:
+        try:
+            log_level = LogLevel(level)
+        except ValueError:
+            return f"Invalid log level '{level}'. Valid values are: 'info', 'warning', 'error'."
+
     async with AsyncDeepsetClient() as client:
         response = await get_pipeline_logs_tool(
             client=client,
             workspace=workspace,
             pipeline_name=pipeline_name,
             limit=limit,
-            level=level,
+            level=log_level,
         )
     return response
 

src/deepset_mcp/tools/pipeline.py

@@ -1,6 +1,7 @@
 import yaml
 
 from deepset_mcp.api.exceptions import BadRequestError, ResourceNotFoundError, UnexpectedAPIError
+from deepset_mcp.api.pipeline.log_level import LogLevel
 from deepset_mcp.api.protocols import AsyncClientProtocol
 from deepset_mcp.tools.formatting_utils import pipeline_to_llm_readable_string, validation_result_to_llm_readable_string
 
@@ -116,7 +117,7 @@ async def get_pipeline_logs(
     workspace: str,
     pipeline_name: str,
     limit: int = 30,
-    level: str | None = None,
+    level: LogLevel | None = None,
 ) -> str:
     """Fetches logs for a specific pipeline.
 
@@ -127,7 +128,7 @@ async def get_pipeline_logs(
     :param workspace: The workspace name.
     :param pipeline_name: Name of the pipeline to fetch logs for.
     :param limit: Maximum number of log entries to return (default: 30).
-    :param level: Filter logs by level (info, warning, error). If None, returns all levels.
+    :param level: Filter logs by level. If None, returns all levels.
 
     :returns: A formatted string containing the pipeline logs.
     """

test/unit/api/pipeline/test_pipeline_resource.py

@@ -3,6 +3,7 @@
 import pytest
 
 from deepset_mcp.api.exceptions import UnexpectedAPIError
+from deepset_mcp.api.pipeline.log_level import LogLevel
 from deepset_mcp.api.pipeline.models import (
     DeepsetPipeline,
     PipelineLog,
@@ -717,7 +718,7 @@ async def test_get_logs_with_level_filter(self) -> None:
 
         # Create resource and call get_logs method with level filter
         resource = PipelineResource(client=client, workspace="test-workspace")
-        result = await resource.get_logs(pipeline_name="test-pipeline", level="error")
+        result = await resource.get_logs(pipeline_name="test-pipeline", level=LogLevel.ERROR)
 
         # Verify results
         assert len(result.data) == 2
@@ -747,7 +748,7 @@ async def test_get_logs_with_warning_level(self) -> None:
 
         # Create resource and call get_logs method with warning level
         resource = PipelineResource(client=client, workspace="test-workspace")
-        result = await resource.get_logs(pipeline_name="test-pipeline", level="warning")
+        result = await resource.get_logs(pipeline_name="test-pipeline", level=LogLevel.WARNING)
 
         # Verify results
         assert len(result.data) == 1

test/unit/tools/test_pipeline.py

@@ -7,6 +7,7 @@
     ResourceNotFoundError,
     UnexpectedAPIError,
 )
+from deepset_mcp.api.pipeline.log_level import LogLevel
 from deepset_mcp.api.pipeline.models import (
     DeepsetPipeline,
     NoContentResponse,
@@ -96,7 +97,7 @@ async def get_logs(
         self,
         pipeline_name: str,
         limit: int = 30,
-        level: str | None = None,
+        level: LogLevel | None = None,
     ) -> PipelineLogList:
         if self._logs_exception:
             raise self._logs_exception
@@ -471,7 +472,7 @@ async def test_get_pipeline_logs_with_level_filter() -> None:
     resource = FakePipelineResource(logs_response=logs)
     client = FakeClient(resource)
 
-    result = await get_pipeline_logs(client, workspace="ws", pipeline_name="test-pipeline", level="error")
+    result = await get_pipeline_logs(client, workspace="ws", pipeline_name="test-pipeline", level=LogLevel.ERROR)
 
     assert "No logs found for pipeline 'test-pipeline' (filtered by level: error)" in result