feat: add get_logs method to PipelineHandle and PipelineResource (#56)

* feat: add log models for pipeline logs

* feat: add PipelineLogList import to handle

* feat: add get_logs method to PipelineHandle

* feat: add PipelineLogList import to resource

* feat: add get_logs method to PipelineResource

* feat: add PipelineLogList import to protocols

* feat: add get_logs method to PipelineResourceProtocol

* feat: add comprehensive unit tests for pipeline logs functionality

* feat: add default implementation for pipelines method in BaseFakeClient

* feat: add integration tests for pipeline logs functionality

* feat: export log models in pipeline package init

* refactor: remove PipelineHandle class in favor of simpler approach

* refactor: remove PipelineHandle import after deleting handle

* refactor: change list method to return DeepsetPipeline instead of handles

* refactor: change get method to return DeepsetPipeline instead of handle

* refactor: remove PipelineHandle import and add DeepsetPipeline import

* refactor: update protocol signatures to return DeepsetPipeline instead of handles

* refactor: remove handle tests since we're removing the handle class

* refactor: remove PipelineHandle import from logs tests

* feat: create clean pipeline logs tests without handle references

* refactor: remove old logs test file with handle references

* feat: recreate pipeline logs tests without handle references

* refactor: remove duplicate test file

* refactor: update integration tests to use resource directly instead of handle

* refactor: update integration test to use resource directly for error logs

* refactor: remove PipelineHandle import from resource tests

* refactor: update list test to expect DeepsetPipeline instead of handle

* refactor: update get test to expect DeepsetPipeline instead of handle

* refactor: update get without yaml test to expect DeepsetPipeline instead of handle

* refactor: update special characters test to expect DeepsetPipeline instead of handle

* refactor: remove PipelineHandle from pipeline package __init__.py

* fix: tests

Author: mathislucka

src/deepset_mcp/api/pipeline/__init__.py

@@ -1,12 +1,19 @@
-from .handle import PipelineHandle
-from .models import DeepsetPipeline, NoContentResponse, PipelineValidationResult, ValidationError
+from .models import (
+    DeepsetPipeline,
+    NoContentResponse,
+    PipelineLog,
+    PipelineLogList,
+    PipelineValidationResult,
+    ValidationError,
+)
 from .resource import PipelineResource
 
 __all__ = [
-    "PipelineHandle",
     "DeepsetPipeline",
     "NoContentResponse",
     "PipelineValidationResult",
     "ValidationError",
     "PipelineResource",
+    "PipelineLog",
+    "PipelineLogList",
 ]

src/deepset_mcp/api/pipeline/handle.py

@@ -1,5 +1,6 @@
 from datetime import datetime
 from enum import StrEnum
+from typing import Any
 
 from pydantic import BaseModel, Field
 
@@ -59,3 +60,23 @@ class NoContentResponse(BaseModel):
 
     success: bool = True
     message: str = "No content"
+
+
+class PipelineLog(BaseModel):
+    """Model representing a single log entry from a pipeline."""
+
+    log_id: str
+    message: str
+    logged_at: datetime
+    level: str
+    origin: str
+    exceptions: str | None = None
+    extra_fields: dict[str, Any] = Field(default_factory=dict)
+
+
+class PipelineLogList(BaseModel):
+    """Model representing a paginated list of pipeline logs."""
+
+    data: list[PipelineLog]
+    has_more: bool
+    total: int

src/deepset_mcp/api/pipeline/models.py

@@ -4,13 +4,13 @@
 from deepset_mcp.api.pipeline.models import (
     DeepsetPipeline,
     NoContentResponse,
+    PipelineLogList,
     PipelineValidationResult,
     ValidationError,
 )
 from deepset_mcp.api.transport import raise_for_status
 
 if TYPE_CHECKING:
-    from deepset_mcp.api.pipeline.handle import PipelineHandle
     from deepset_mcp.api.protocols import AsyncClientProtocol
 
 
@@ -67,16 +67,14 @@ async def list(
         self,
         page_number: int = 1,
         limit: int = 10,
-    ) -> list["PipelineHandle"]:
+    ) -> list[DeepsetPipeline]:
         """
         Retrieve pipeline in the configured workspace with optional pagination.
 
         :param page_number: Page number for paging.
         :param limit: Max number of items to return.
-        :return: List of PipelineHandle instances.
+        :return: List of DeepsetPipeline instances.
         """
-        from deepset_mcp.api.pipeline.handle import PipelineHandle
-
         params: dict[str, Any] = {
             "page_number": page_number,
             "limit": limit,
@@ -97,12 +95,10 @@ async def list(
         else:
             pipelines = []
 
-        return [PipelineHandle(pipeline=pipeline, resource=self) for pipeline in pipelines]
+        return pipelines
 
-    async def get(self, pipeline_name: str, include_yaml: bool = True) -> "PipelineHandle":
+    async def get(self, pipeline_name: str, include_yaml: bool = True) -> DeepsetPipeline:
         """Fetch a single pipeline by its name."""
-        from deepset_mcp.api.pipeline.handle import PipelineHandle
-
         resp = await self._client.request(endpoint=f"v1/workspaces/{self._workspace}/pipelines/{pipeline_name}")
         raise_for_status(resp)
 
@@ -118,7 +114,7 @@ async def get(self, pipeline_name: str, include_yaml: bool = True) -> "PipelineH
             if yaml_response.json is not None:
                 pipeline.yaml_config = yaml_response.json["query_yaml"]
 
-        return PipelineHandle(pipeline=pipeline, resource=self)
+        return pipeline
 
     async def create(self, name: str, yaml_config: str) -> NoContentResponse:
         """Create a new pipeline with a name and YAML config."""
@@ -172,3 +168,40 @@ async def update(
             return response
 
         raise ValueError("Either `updated_pipeline_name` or `yaml_config` must be provided.")
+
+    async def get_logs(
+        self,
+        pipeline_name: str,
+        limit: int = 30,
+        level: str | 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.
+
+        :returns: A PipelineLogList containing the log entries.
+        """
+        params: dict[str, Any] = {
+            "limit": limit,
+            "filter": "origin eq 'querypipeline'",
+        }
+
+        # Add level filter if specified
+        if level is not None:
+            params["filter"] = f"level eq '{level}' and origin eq 'querypipeline'"
+
+        resp = await self._client.request(
+            endpoint=f"v1/workspaces/{self._workspace}/pipelines/{pipeline_name}/logs",
+            method="GET",
+            params=params,
+        )
+
+        raise_for_status(resp)
+
+        if resp.json is not None:
+            return PipelineLogList.model_validate(resp.json)
+        else:
+            # Return empty log list if no response
+            return PipelineLogList(data=[], has_more=False, total=0)

src/deepset_mcp/api/pipeline/resource.py

@@ -1,14 +1,16 @@
 from types import TracebackType
-from typing import TYPE_CHECKING, Any, Protocol, Self, TypeVar, overload
+from typing import Any, Protocol, Self, TypeVar, overload
 
 from deepset_mcp.api.indexes.models import Index, IndexList
-from deepset_mcp.api.pipeline.models import NoContentResponse, PipelineValidationResult
+from deepset_mcp.api.pipeline.models import (
+    DeepsetPipeline,
+    NoContentResponse,
+    PipelineLogList,
+    PipelineValidationResult,
+)
 from deepset_mcp.api.pipeline_template.models import PipelineTemplate
 from deepset_mcp.api.transport import TransportResponse
 
-if TYPE_CHECKING:
-    from deepset_mcp.api.pipeline.handle import PipelineHandle
-
 
 class HaystackServiceProtocol(Protocol):
     """Protocol defining the implementation for HaystackService."""
@@ -151,15 +153,15 @@ async def validate(self, yaml_config: str) -> PipelineValidationResult:
         """Validate a pipeline's YAML configuration against the API."""
         ...
 
-    async def get(self, pipeline_name: str, include_yaml: bool = True) -> "PipelineHandle":
+    async def get(self, pipeline_name: str, include_yaml: bool = True) -> DeepsetPipeline:
         """Fetch a single pipeline by its name."""
         ...
 
     async def list(
         self,
         page_number: int = 1,
         limit: int = 10,
-    ) -> list["PipelineHandle"]:
+    ) -> list[DeepsetPipeline]:
         """List pipelines in the configured workspace with optional pagination."""
         ...
 
@@ -175,3 +177,12 @@ async def update(
     ) -> NoContentResponse:
         """Update name and/or YAML config of an existing pipeline."""
         ...
+
+    async def get_logs(
+        self,
+        pipeline_name: str,
+        limit: int = 30,
+        level: str | None = None,
+    ) -> PipelineLogList:
+        """Fetch logs for a specific pipeline."""
+        ...

src/deepset_mcp/api/protocols.py

@@ -8,15 +8,15 @@
 async def list_pipelines(client: AsyncClientProtocol, workspace: str) -> str:
     """Retrieves a list of all pipeline available within the currently configured deepset workspace."""
     response = await client.pipelines(workspace=workspace).list()
-    formatted_pipelines = [pipeline_to_llm_readable_string(h.pipeline) for h in response]
+    formatted_pipelines = [pipeline_to_llm_readable_string(p) for p in response]
 
     return "\n\n".join(formatted_pipelines)
 
 
 async def get_pipeline(client: AsyncClientProtocol, workspace: str, pipeline_name: str) -> str:
     """Fetches detailed configuration information for a specific pipeline, identified by its unique `pipeline_name`."""
     response = await client.pipelines(workspace=workspace).get(pipeline_name)
-    return pipeline_to_llm_readable_string(response.pipeline)
+    return pipeline_to_llm_readable_string(response)
 
 
 async def validate_pipeline(client: AsyncClientProtocol, workspace: str, yaml_configuration: str) -> str:
@@ -70,14 +70,14 @@ async def update_pipeline(
     replacement snippet is used to update the pipeline's configuration in the target workspace.
     """
     try:
-        original_handle = await client.pipelines(workspace=workspace).get(pipeline_name=pipeline_name)
+        original_pipeline = await client.pipelines(workspace=workspace).get(pipeline_name=pipeline_name)
     except ResourceNotFoundError:
         return f"There is no pipeline named '{pipeline_name}'. Did you mean to create it?"
 
-    if original_handle.yaml_config is None:
+    if original_pipeline.yaml_config is None:
         raise ValueError("The pipeline does not have a YAML configuration.")
 
-    occurrences = original_handle.yaml_config.count(original_config_snippet)
+    occurrences = original_pipeline.yaml_config.count(original_config_snippet)
 
     if occurrences == 0:
         return f"No occurrences of the provided configuration snippet were found in the pipeline '{pipeline_name}'."
@@ -88,7 +88,7 @@ async def update_pipeline(
             f"'{pipeline_name}'. Specify a more precise snippet to proceed with the update."
         )
 
-    updated_yaml_configuration = original_handle.yaml_config.replace(
+    updated_yaml_configuration = original_pipeline.yaml_config.replace(
         original_config_snippet, replacement_config_snippet, 1
     )