feat: use pipeline tools in MCP

Author: mathislucka

pyproject.toml

@@ -11,6 +11,7 @@ dependencies = [
     "uvicorn>=0.34.2",
     "httpx",
     "pydantic>=2.0.0",
+    "pyyaml",
 ]
 
 [project.scripts]
@@ -28,15 +29,15 @@ dev = [
     "pytest",
     "pytest-asyncio",
     "python-dotenv",
-    "pyyaml",
 ]
 lint = [
     "ruff",
 ]
 
 types = [
     "mypy",
-    "types-requests"
+    "types-requests",
+    "types-PyYAML",
 ]
 
 [tool.pytest.ini_options]

src/deepset_mcp/api/pipeline/resource.py

@@ -3,7 +3,7 @@
 from deepset_mcp.api.exceptions import UnexpectedAPIError
 from deepset_mcp.api.pipeline.models import DeepsetPipeline, PipelineValidationResult, ValidationError
 from deepset_mcp.api.protocols import AsyncClientProtocol
-from deepset_mcp.api.transport import raise_for_status
+from deepset_mcp.api.transport import TransportResponse, raise_for_status
 
 
 class PipelineResource:
@@ -109,7 +109,7 @@ async def get(self, pipeline_name: str, include_yaml: bool = True) -> DeepsetPip
 
         return pipeline
 
-    async def create(self, name: str, yaml_config: str) -> None:
+    async def create(self, name: str, yaml_config: str) -> TransportResponse:
         """Create a new pipeline with a name and YAML config."""
         data = {"name": name, "query_yaml": yaml_config}
         resp = await self._client.request(
@@ -120,12 +120,14 @@ async def create(self, name: str, yaml_config: str) -> None:
 
         raise_for_status(resp)
 
+        return resp
+
     async def update(
         self,
         pipeline_name: str,
         updated_pipeline_name: str | None = None,
         yaml_config: str | None = None,
-    ) -> None:
+    ) -> TransportResponse:
         """Update name and/or YAML config of an existing pipeline."""
         # Handle name update first if any
         if updated_pipeline_name is not None:
@@ -139,6 +141,9 @@ async def update(
 
             pipeline_name = updated_pipeline_name
 
+            if yaml_config is None:
+                return name_resp
+
         if yaml_config is not None:
             yaml_resp = await self._client.request(
                 endpoint=f"v1/workspaces/{self._workspace}/pipelines/{pipeline_name}/yaml",
@@ -147,3 +152,7 @@ async def update(
             )
 
             raise_for_status(yaml_resp)
+
+            return yaml_resp
+
+        raise ValueError("Either `updated_pipeline_name` or `yaml_config` must be provided.")

src/deepset_mcp/api/protocols.py

@@ -1,6 +1,7 @@
 from types import TracebackType
 from typing import Any, Protocol, Self
 
+from deepset_mcp.api.pipeline.models import DeepsetPipeline, PipelineValidationResult
 from deepset_mcp.api.transport import TransportResponse
 
 
@@ -30,3 +31,44 @@ async def __aexit__(
     ) -> bool:
         """Exit the AsyncContextmanager and clean up resources."""
         ...
+
+    def pipelines(self, workspace: str) -> "PipelineResourceProtocol":
+        """Access pipelines in the specified workspace."""
+        ...
+
+
+class PipelineResourceProtocol(Protocol):
+    """Protocol defining the implementation for PipelineResource."""
+
+    def __init__(self, client: AsyncClientProtocol, workspace: str) -> None:
+        """Initialize a PipelineResource."""
+        ...
+
+    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:
+        """Fetch a single pipeline by its name."""
+        ...
+
+    async def list(
+        self,
+        page_number: int = 1,
+        limit: int = 10,
+    ) -> list[DeepsetPipeline]:
+        """List pipelines in the configured workspace with optional pagination."""
+        ...
+
+    async def create(self, name: str, yaml_config: str) -> Any:
+        """Create a new pipeline with a name and YAML config."""
+        ...
+
+    async def update(
+        self,
+        pipeline_name: str,
+        updated_pipeline_name: str | None = None,
+        yaml_config: str | None = None,
+    ) -> Any:
+        """Update name and/or YAML config of an existing pipeline."""
+        ...