feat: introduce PipelineHandle for future pipeline operations (#54)

* feat: add PipelineHandle class for future pipeline operations

* fix: remove existing __init__.py to replace with proper imports

* feat: add PipelineHandle to package exports

* feat: add PipelineHandle import for protocols

* feat: update PipelineResourceProtocol to return PipelineHandle

* feat: import PipelineHandle in resource module

* feat: update list method to return PipelineHandle instances

* feat: update get method to return PipelineHandle instance

* fix: avoid circular import by using TYPE_CHECKING for PipelineHandle import

* fix: move PipelineHandle import inside list method to avoid circular import

* fix: move PipelineHandle import inside get method to avoid circular import

* feat: add PipelineHandle import to test module

* fix: update test to check for PipelineHandle instead of DeepsetPipeline

* fix: update get test to check for PipelineHandle instead of DeepsetPipeline

* fix: update get without yaml test to check for PipelineHandle

* fix: update special characters test to check for PipelineHandle

* feat: add PipelineHandle import to integration tests

* fix: update integration test to expect PipelineHandle for create test

* fix: update integration test to expect PipelineHandle for list test

* fix: update integration test pagination to expect PipelineHandle

* fix: update integration get test to expect PipelineHandle

* fix: update integration update test to expect PipelineHandle

* fix: update list_pipelines tool to work with PipelineHandle

* fix: update get_pipeline tool to work with PipelineHandle

* fix: update update_pipeline tool to work with PipelineHandle

* feat: add PipelineHandle import to tools tests

* fix: update FakePipelineResource to return PipelineHandle instances

* fix: no changes needed for get_pipeline test - it should work with new handle

* feat: add comprehensive tests for PipelineHandle class

* fix: tests

Author: mathislucka

src/deepset_mcp/api/pipeline/__init__.py

@@ -0,0 +1,12 @@
+from .handle import PipelineHandle
+from .models import DeepsetPipeline, NoContentResponse, PipelineValidationResult, ValidationError
+from .resource import PipelineResource
+
+__all__ = [
+    "PipelineHandle",
+    "DeepsetPipeline",
+    "NoContentResponse",
+    "PipelineValidationResult",
+    "ValidationError",
+    "PipelineResource",
+]

src/deepset_mcp/api/pipeline/handle.py

@@ -0,0 +1,49 @@
+from typing import TYPE_CHECKING
+
+from deepset_mcp.api.pipeline.models import DeepsetPipeline
+
+if TYPE_CHECKING:
+    from deepset_mcp.api.protocols import PipelineResourceProtocol
+
+
+class PipelineHandle:
+    """Handle for performing operations on a specific pipeline.
+
+    This class provides a convenient interface for interacting with a pipeline,
+    combining the pipeline data model with resource operations. It allows both
+    direct access to pipeline attributes and future operations like getting logs
+    or deployment management.
+    """
+
+    def __init__(self, pipeline: DeepsetPipeline, resource: "PipelineResourceProtocol") -> None:
+        """Initialize a PipelineHandle.
+
+        :param pipeline: The pipeline data model containing all pipeline information.
+        :param resource: The resource interface for performing operations on this pipeline.
+        """
+        self._pipeline = pipeline
+        self._resource = resource
+
+    def __getattr__(self, name: str) -> object:
+        """Proxy attribute access to the underlying pipeline."""
+        return getattr(self._pipeline, name)
+
+    if TYPE_CHECKING:
+        # Help type checkers understand available attributes from the underlying pipeline
+        id: str
+        name: str
+        status: str
+        service_level: str
+        created_at: object
+        last_updated_at: object | None
+        created_by: object
+        last_updated_by: object | None
+        yaml_config: str | None
+
+    @property
+    def pipeline(self) -> DeepsetPipeline:
+        """Access the full pipeline data model.
+
+        :returns: The underlying DeepsetPipeline instance with all pipeline data.
+        """
+        return self._pipeline

src/deepset_mcp/api/pipeline/resource.py

@@ -1,4 +1,4 @@
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from deepset_mcp.api.exceptions import UnexpectedAPIError
 from deepset_mcp.api.pipeline.models import (
@@ -7,16 +7,19 @@
     PipelineValidationResult,
     ValidationError,
 )
-from deepset_mcp.api.protocols import AsyncClientProtocol, PipelineResourceProtocol
 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
 
-class PipelineResource(PipelineResourceProtocol):
+
+class PipelineResource:
     """Manages interactions with the deepset pipeline API."""
 
     def __init__(
         self,
-        client: AsyncClientProtocol,
+        client: "AsyncClientProtocol",
         workspace: str,
     ) -> None:
         """Initializes a PipelineResource instance."""
@@ -64,14 +67,16 @@ async def list(
         self,
         page_number: int = 1,
         limit: int = 10,
-    ) -> list[DeepsetPipeline]:
+    ) -> list["PipelineHandle"]:
         """
         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: PipelineListResponse containing `data`, `has_more`, and `total`.
+        :return: List of PipelineHandle instances.
         """
+        from deepset_mcp.api.pipeline.handle import PipelineHandle
+
         params: dict[str, Any] = {
             "page_number": page_number,
             "limit": limit,
@@ -92,10 +97,12 @@ async def list(
         else:
             pipelines = []
 
-        return pipelines
+        return [PipelineHandle(pipeline=pipeline, resource=self) for pipeline in pipelines]
 
-    async def get(self, pipeline_name: str, include_yaml: bool = True) -> DeepsetPipeline:
+    async def get(self, pipeline_name: str, include_yaml: bool = True) -> "PipelineHandle":
         """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)
 
@@ -111,7 +118,7 @@ async def get(self, pipeline_name: str, include_yaml: bool = True) -> DeepsetPip
             if yaml_response.json is not None:
                 pipeline.yaml_config = yaml_response.json["query_yaml"]
 
-        return pipeline
+        return PipelineHandle(pipeline=pipeline, resource=self)
 
     async def create(self, name: str, yaml_config: str) -> NoContentResponse:
         """Create a new pipeline with a name and YAML config."""

src/deepset_mcp/api/protocols.py

@@ -1,11 +1,14 @@
 from types import TracebackType
-from typing import Any, Protocol, Self, TypeVar, overload
+from typing import TYPE_CHECKING, Any, Protocol, Self, TypeVar, overload
 
 from deepset_mcp.api.indexes.models import Index, IndexList
-from deepset_mcp.api.pipeline.models import DeepsetPipeline, NoContentResponse, PipelineValidationResult
+from deepset_mcp.api.pipeline.models import NoContentResponse, 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."""
@@ -148,15 +151,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) -> DeepsetPipeline:
+    async def get(self, pipeline_name: str, include_yaml: bool = True) -> "PipelineHandle":
         """Fetch a single pipeline by its name."""
         ...
 
     async def list(
         self,
         page_number: int = 1,
         limit: int = 10,
-    ) -> list[DeepsetPipeline]:
+    ) -> list["PipelineHandle"]:
         """List pipelines in the configured workspace with optional pagination."""
         ...
 

src/deepset_mcp/tools/pipeline.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(p) for p in response]
+    formatted_pipelines = [pipeline_to_llm_readable_string(h.pipeline) for h 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)
+    return pipeline_to_llm_readable_string(response.pipeline)
 
 
 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_pipeline = await client.pipelines(workspace=workspace).get(pipeline_name=pipeline_name)
+        original_handle = 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_pipeline.yaml_config is None:
+    if original_handle.yaml_config is None:
         raise ValueError("The pipeline does not have a YAML configuration.")
 
-    occurrences = original_pipeline.yaml_config.count(original_config_snippet)
+    occurrences = original_handle.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_pipeline.yaml_config.replace(
+    updated_yaml_configuration = original_handle.yaml_config.replace(
         original_config_snippet, replacement_config_snippet, 1
     )
 

test/integration/test_integration_pipeline_resource.py

@@ -2,6 +2,7 @@
 
 from deepset_mcp.api.client import AsyncDeepsetClient
 from deepset_mcp.api.exceptions import ResourceNotFoundError
+from deepset_mcp.api.pipeline.handle import PipelineHandle
 from deepset_mcp.api.pipeline.models import DeepsetPipeline
 from deepset_mcp.api.pipeline.resource import PipelineResource
 
@@ -71,10 +72,11 @@ async def test_create_pipeline(
     await pipeline_resource.create(name=pipeline_name, yaml_config=sample_yaml_config)
 
     # Verify the pipeline was created by retrieving it
-    pipeline: DeepsetPipeline = await pipeline_resource.get(pipeline_name=pipeline_name)
+    handle: PipelineHandle = await pipeline_resource.get(pipeline_name=pipeline_name)
 
-    assert pipeline.name == pipeline_name
-    assert pipeline.yaml_config == sample_yaml_config
+    assert handle.name == pipeline_name
+    assert handle.yaml_config == sample_yaml_config
+    assert isinstance(handle.pipeline, DeepsetPipeline)
 
 
 @pytest.mark.asyncio
@@ -91,23 +93,30 @@ async def test_list_pipelines(
         await pipeline_resource.create(name=pipeline_name, yaml_config=sample_yaml_config)
 
     # Test listing without pagination
-    pipelines = await pipeline_resource.list(limit=10)
-    assert len(pipelines) == 3
+    handles = await pipeline_resource.list(limit=10)
+    assert len(handles) == 3
 
     # Verify our created pipelines are in the list
-    retrieved_names = [p.name for p in pipelines]
+    retrieved_names = [h.name for h in handles]
     for name in pipeline_names:
         assert name in retrieved_names
 
+    # Verify all are PipelineHandle instances
+    for handle in handles:
+        assert isinstance(handle, PipelineHandle)
+        assert isinstance(handle.pipeline, DeepsetPipeline)
+
     # Test pagination
-    if len(pipelines) > 1:
+    if len(handles) > 1:
         # Get the first page with 1 item
         first_page = await pipeline_resource.list(limit=1)
         assert len(first_page) == 1
+        assert isinstance(first_page[0], PipelineHandle)
 
         # Get the second page
         second_page = await pipeline_resource.list(page_number=2, limit=1)
         assert len(second_page) == 1
+        assert isinstance(second_page[0], PipelineHandle)
 
         # Verify they're different pipelines
         assert first_page[0].id != second_page[0].id
@@ -125,16 +134,16 @@ async def test_get_pipeline(
     await pipeline_resource.create(name=pipeline_name, yaml_config=sample_yaml_config)
 
     # Test getting with YAML config
-    pipeline_with_yaml: DeepsetPipeline = await pipeline_resource.get(pipeline_name=pipeline_name, include_yaml=True)
-    assert pipeline_with_yaml.name == pipeline_name
-    assert pipeline_with_yaml.yaml_config == sample_yaml_config
+    handle_with_yaml: PipelineHandle = await pipeline_resource.get(pipeline_name=pipeline_name, include_yaml=True)
+    assert handle_with_yaml.name == pipeline_name
+    assert handle_with_yaml.yaml_config == sample_yaml_config
+    assert isinstance(handle_with_yaml.pipeline, DeepsetPipeline)
 
     # Test getting without YAML config
-    pipeline_without_yaml: DeepsetPipeline = await pipeline_resource.get(
-        pipeline_name=pipeline_name, include_yaml=False
-    )
-    assert pipeline_without_yaml.name == pipeline_name
-    assert pipeline_without_yaml.yaml_config is None
+    handle_without_yaml: PipelineHandle = await pipeline_resource.get(pipeline_name=pipeline_name, include_yaml=False)
+    assert handle_without_yaml.name == pipeline_name
+    assert handle_without_yaml.yaml_config is None
+    assert isinstance(handle_without_yaml.pipeline, DeepsetPipeline)
 
 
 @pytest.mark.asyncio
@@ -156,8 +165,9 @@ async def test_update_pipeline(
     )
 
     # Verify the name was updated
-    updated_pipeline: DeepsetPipeline = await pipeline_resource.get(pipeline_name=updated_name)
-    assert updated_pipeline.name == updated_name
+    updated_handle: PipelineHandle = await pipeline_resource.get(pipeline_name=updated_name)
+    assert updated_handle.name == updated_name
+    assert isinstance(updated_handle.pipeline, DeepsetPipeline)
 
     # Update the pipeline config
     modified_yaml = sample_yaml_config.replace("temperature: 0.1", "temperature: 0.2")
@@ -167,8 +177,9 @@ async def test_update_pipeline(
     )
 
     # Verify the config was updated
-    updated_pipeline = await pipeline_resource.get(pipeline_name=updated_name)
-    assert updated_pipeline.yaml_config == modified_yaml
+    updated_handle = await pipeline_resource.get(pipeline_name=updated_name)
+    assert updated_handle.yaml_config == modified_yaml
+    assert isinstance(updated_handle.pipeline, DeepsetPipeline)
 
 
 @pytest.mark.asyncio