feat: add get_logs tool for pipeline resource (#64)

* feat: add formatting utility for pipeline logs

* feat: add PipelineLogList import for formatting utils

* feat: add get_pipeline_logs tool function

* feat: import get_pipeline_logs_tool

* feat: add get_pipeline_logs MCP tool

* test: add import for get_pipeline_logs function

* test: fix imports for pipeline models

* test: add logs response and exception to FakePipelineResource

* test: implement get_logs method in FakePipelineResource

* test: add comprehensive tests for get_pipeline_logs function

* test: add comprehensive tests for formatting utils

* fix: format and lint

Author: mathislucka

src/deepset_mcp/main.py

@@ -20,6 +20,7 @@
 from deepset_mcp.tools.pipeline import (
     create_pipeline as create_pipeline_tool,
     get_pipeline as get_pipeline_tool,
+    get_pipeline_logs as get_pipeline_logs_tool,
     list_pipelines as list_pipelines_tool,
     update_pipeline as update_pipeline_tool,
     validate_pipeline as validate_pipeline_tool,
@@ -296,6 +297,29 @@ async def update_index(
     return response
 
 
+@mcp.tool()
+async def get_pipeline_logs(pipeline_name: str, limit: int = 30, level: str | None = None) -> str:
+    """Fetches logs for a specific pipeline in the deepset workspace.
+
+    Use this to debug pipeline issues, monitor pipeline execution, or understand what happened during pipeline runs.
+    The logs provide detailed information about pipeline operations, errors, and warnings.
+
+    :param pipeline_name: Name of the pipeline to fetch logs for.
+    :param limit: Maximum number of log entries to return (default: 30, max: 100).
+    :param level: Filter logs by level. Valid values: 'info', 'warning', 'error'. If not specified, returns all levels.
+    """
+    workspace = get_workspace()
+    async with AsyncDeepsetClient() as client:
+        response = await get_pipeline_logs_tool(
+            client=client,
+            workspace=workspace,
+            pipeline_name=pipeline_name,
+            limit=limit,
+            level=level,
+        )
+    return response
+
+
 #
 #
 # @mcp.tool()

src/deepset_mcp/tools/formatting_utils.py

@@ -1,4 +1,4 @@
-from deepset_mcp.api.pipeline.models import DeepsetPipeline, PipelineValidationResult
+from deepset_mcp.api.pipeline.models import DeepsetPipeline, PipelineLogList, PipelineValidationResult
 from deepset_mcp.api.pipeline_template.models import PipelineTemplate
 
 
@@ -85,3 +85,59 @@ def validation_result_to_llm_readable_string(validation_result: PipelineValidati
             result_parts.append(f"Error {i}\n- Code: {error.code}\n- Message: {error.message}\n")
 
     return "\n".join(result_parts)
+
+
+def pipeline_logs_to_llm_readable_string(logs: PipelineLogList, pipeline_name: str, level: str | None = None) -> str:
+    """Creates a string representation of pipeline logs that is readable by LLMs.
+
+    :param logs: The PipelineLogList containing log entries.
+    :param pipeline_name: The name of the pipeline the logs are from.
+    :param level: The log level filter that was applied, if any.
+
+    :returns: A formatted string representation of the logs.
+    """
+    if not logs.data:
+        filter_info = f" (filtered by level: {level})" if level else ""
+        return f"No logs found for pipeline '{pipeline_name}'{filter_info}."
+
+    log_parts = [f"### Logs for Pipeline '{pipeline_name}'"]
+
+    if level:
+        log_parts.append(f"**Filter Applied:** Level = {level}")
+
+    log_parts.extend(
+        [
+            f"**Total Logs:** {logs.total}",
+            f"**Showing:** {len(logs.data)} entries",
+            f"**Has More:** {'Yes' if logs.has_more else 'No'}",
+            "\n---\n",
+        ]
+    )
+
+    for i, log in enumerate(logs.data, 1):
+        log_entry = [
+            f"**Log Entry {i}**",
+            f"- **Timestamp:** {log.logged_at.strftime('%B %d, %Y %I:%M:%S %p')}",
+            f"- **Level:** {log.level}",
+            f"- **Origin:** {log.origin}",
+            f"- **Message:** {log.message}",
+        ]
+
+        if log.exceptions:
+            log_entry.append(f"- **Exceptions:** {log.exceptions}")
+
+        if log.extra_fields:
+            log_entry.append("- **Extra Fields:")
+            for key, value in log.extra_fields.items():
+                log_entry.append(f"  - {key}: {value}")
+
+        log_parts.append("\n".join(log_entry))
+
+        # Add separator between log entries (except for the last one)
+        if i < len(logs.data):
+            log_parts.append("")
+
+    if logs.has_more:
+        log_parts.append("\n*Note: There are more log entries available. Adjust the limit parameter to see more.*")
+
+    return "\n".join(log_parts)

src/deepset_mcp/tools/pipeline.py

@@ -109,3 +109,39 @@ async def update_pipeline(
         return f"Failed to update the pipeline '{pipeline_name}': {e}"
 
     return f"The pipeline '{pipeline_name}' was successfully updated."
+
+
+async def get_pipeline_logs(
+    client: AsyncClientProtocol,
+    workspace: str,
+    pipeline_name: str,
+    limit: int = 30,
+    level: str | None = None,
+) -> str:
+    """Fetches logs for a specific pipeline.
+
+    Retrieves log entries for the specified pipeline, with optional filtering by log level.
+    This is useful for debugging pipeline issues or monitoring pipeline execution.
+
+    :param client: The async client for API communication.
+    :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.
+
+    :returns: A formatted string containing the pipeline logs.
+    """
+    try:
+        logs = await client.pipelines(workspace=workspace).get_logs(
+            pipeline_name=pipeline_name, limit=limit, level=level
+        )
+    except ResourceNotFoundError:
+        return f"There is no pipeline named '{pipeline_name}' in workspace '{workspace}'."
+    except BadRequestError as e:
+        return f"Failed to fetch logs for pipeline '{pipeline_name}': {e}"
+    except UnexpectedAPIError as e:
+        return f"Failed to fetch logs for pipeline '{pipeline_name}': {e}"
+
+    from deepset_mcp.tools.formatting_utils import pipeline_logs_to_llm_readable_string
+
+    return pipeline_logs_to_llm_readable_string(logs, pipeline_name, level)

test/unit/tools/test_formatting_utils.py

@@ -0,0 +1,154 @@
+from datetime import datetime
+
+from deepset_mcp.api.pipeline.models import (
+    PipelineLog,
+    PipelineLogList,
+    PipelineValidationResult,
+    ValidationError,
+)
+from deepset_mcp.tools.formatting_utils import (
+    pipeline_logs_to_llm_readable_string,
+    validation_result_to_llm_readable_string,
+)
+
+
+def test_pipeline_logs_to_llm_readable_string_with_logs() -> None:
+    """Test formatting pipeline logs with data."""
+    log1 = PipelineLog(
+        log_id="log1",
+        message="Pipeline started successfully",
+        logged_at=datetime(2023, 1, 1, 12, 0, 0),
+        level="info",
+        origin="querypipeline",
+        exceptions=None,
+        extra_fields={},
+    )
+    log2 = PipelineLog(
+        log_id="log2",
+        message="Error processing document",
+        logged_at=datetime(2023, 1, 1, 12, 1, 30),
+        level="error",
+        origin="querypipeline",
+        exceptions="ValueError: Invalid document format",
+        extra_fields={"component": "document_reader", "file_name": "test.pdf"},
+    )
+
+    logs = PipelineLogList(data=[log1, log2], has_more=True, total=5)
+
+    result = pipeline_logs_to_llm_readable_string(logs, "test-pipeline")
+
+    # Check basic structure
+    assert "### Logs for Pipeline 'test-pipeline'" in result
+    assert "**Total Logs:** 5" in result
+    assert "**Showing:** 2 entries" in result
+    assert "**Has More:** Yes" in result
+
+    # Check log entry 1
+    assert "**Log Entry 1**" in result
+    assert "**Timestamp:** January 01, 2023 12:00:00 PM" in result
+    assert "**Level:** info" in result
+    assert "**Origin:** querypipeline" in result
+    assert "**Message:** Pipeline started successfully" in result
+
+    # Check log entry 2
+    assert "**Log Entry 2**" in result
+    assert "**Timestamp:** January 01, 2023 12:01:30 PM" in result
+    assert "**Level:** error" in result
+    assert "**Message:** Error processing document" in result
+    assert "**Exceptions:** ValueError: Invalid document format" in result
+    assert "component: document_reader" in result
+    assert "file_name: test.pdf" in result
+
+    # Check pagination note
+    assert "*Note: There are more log entries available. Adjust the limit parameter to see more.*" in result
+
+
+def test_pipeline_logs_to_llm_readable_string_empty() -> None:
+    """Test formatting empty pipeline logs."""
+    logs = PipelineLogList(data=[], has_more=False, total=0)
+
+    result = pipeline_logs_to_llm_readable_string(logs, "empty-pipeline")
+
+    assert result == "No logs found for pipeline 'empty-pipeline'."
+
+
+def test_pipeline_logs_to_llm_readable_string_empty_with_filter() -> None:
+    """Test formatting empty pipeline logs with level filter."""
+    logs = PipelineLogList(data=[], has_more=False, total=0)
+
+    result = pipeline_logs_to_llm_readable_string(logs, "test-pipeline", level="error")
+
+    assert result == "No logs found for pipeline 'test-pipeline' (filtered by level: error)."
+
+
+def test_pipeline_logs_to_llm_readable_string_with_level_filter() -> None:
+    """Test formatting pipeline logs with level filter applied."""
+    log = PipelineLog(
+        log_id="log1",
+        message="Critical error",
+        logged_at=datetime(2023, 1, 1, 12, 0, 0),
+        level="error",
+        origin="querypipeline",
+        exceptions=None,
+        extra_fields={},
+    )
+
+    logs = PipelineLogList(data=[log], has_more=False, total=1)
+
+    result = pipeline_logs_to_llm_readable_string(logs, "test-pipeline", level="error")
+
+    assert "### Logs for Pipeline 'test-pipeline'" in result
+    assert "**Filter Applied:** Level = error" in result
+    assert "**Total Logs:** 1" in result
+    assert "**Showing:** 1 entries" in result
+    assert "**Has More:** No" in result
+    assert "**Message:** Critical error" in result
+
+
+def test_pipeline_logs_to_llm_readable_string_no_pagination_note() -> None:
+    """Test that pagination note is not shown when has_more is False."""
+    log = PipelineLog(
+        log_id="log1",
+        message="Test message",
+        logged_at=datetime(2023, 1, 1, 12, 0, 0),
+        level="info",
+        origin="querypipeline",
+        exceptions=None,
+        extra_fields={},
+    )
+
+    logs = PipelineLogList(data=[log], has_more=False, total=1)
+
+    result = pipeline_logs_to_llm_readable_string(logs, "test-pipeline")
+
+    assert "*Note: There are more log entries available" not in result
+
+
+def test_validation_result_to_llm_readable_string_valid() -> None:
+    """Test formatting valid validation result."""
+    result = PipelineValidationResult(valid=True, errors=[])
+
+    formatted = validation_result_to_llm_readable_string(result)
+
+    assert "The provided pipeline configuration is valid." in formatted
+    assert "Validation Errors" not in formatted
+
+
+def test_validation_result_to_llm_readable_string_invalid() -> None:
+    """Test formatting invalid validation result."""
+    errors = [
+        ValidationError(code="YAML_ERROR", message="Syntax error in YAML"),
+        ValidationError(code="COMPONENT_ERROR", message="Unknown component type"),
+    ]
+    result = PipelineValidationResult(valid=False, errors=errors)
+
+    formatted = validation_result_to_llm_readable_string(result)
+
+    assert "The provided pipeline configuration is invalid." in formatted
+    assert "**Validation Errors**" in formatted
+    assert "Error 1" in formatted
+    assert "- Code: YAML_ERROR" in formatted
+    assert "- Message: Syntax error in YAML" in formatted
+    assert "Error 2" in formatted
+    assert "- Code: COMPONENT_ERROR" in formatted
+    assert "- Message: Unknown component type" in formatted