feat: add pipeline validation

Author: mathislucka

pyproject.toml

@@ -27,7 +27,8 @@ packages = ["src/deepset_mcp"]
 dev = [
     "pytest",
     "pytest-asyncio",
-    "python-dotenv"
+    "python-dotenv",
+    "pyyaml",
 ]
 lint = [
     "ruff",

src/deepset_mcp/api/client.py

@@ -1,38 +1,11 @@
 import os
 from types import TracebackType
-from typing import Any, Protocol, Self
+from typing import Any, Self
 
+from deepset_mcp.api.pipeline.resource import PipelineResource
 from deepset_mcp.api.transport import AsyncTransport, TransportProtocol, TransportResponse
 
 
-class AsyncClientProtocol(Protocol):
-    """Protocol defining the implementation for AsyncClient."""
-
-    async def request(
-        self,
-        endpoint: str,
-        method: str = "GET",
-        data: dict[str, Any] | None = None,
-        headers: dict[str, str] | None = None,
-    ) -> TransportResponse:
-        """Make a request to the API."""
-        ...
-
-    async def close(self) -> None:
-        """Close underlying transport resources."""
-        ...
-
-    async def __aenter__(self) -> Self:
-        """Enter the AsyncContextManager."""
-        ...
-
-    async def __aexit__(
-        self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None
-    ) -> bool:
-        """Exit the AsyncContextmanager and clean up resources."""
-        ...
-
-
 class AsyncDeepsetClient:
     """Async Client for interacting with the deepset API."""
 
@@ -115,3 +88,7 @@ async def __aexit__(
         """Exit the AsyncContextmanager and clean up resources."""
         await self.close()
         return False
+
+    def pipelines(self, workspace: str) -> PipelineResource:
+        """Resource to interact with pipelines in the specified workspace."""
+        return PipelineResource(client=self, workspace=workspace)

src/deepset_mcp/api/exceptions.py

@@ -0,0 +1,42 @@
+from typing import Any
+
+
+class DeepsetAPIError(Exception):
+    """Base exception for all deepset API errors."""
+
+    def __init__(self, status_code: int | None = None, message: Any | None = None, detail: Any | None = None) -> None:
+        """Initialize the exception."""
+        self.status_code = status_code
+        self.message = message
+        self.detail = detail
+        super().__init__(self.message)
+
+    def __str__(self) -> str:
+        """Return a string representation of the exception."""
+        return f"{self.message} (Status Code: {self.status_code})"
+
+
+class ResourceNotFoundError(DeepsetAPIError):
+    """Exception raised when a resource is not found (HTTP 404)."""
+
+    def __init__(self, message: Any = "Resource not found", detail: Any | None = None) -> None:
+        """Initialize the exception."""
+        super().__init__(status_code=404, message=message, detail=detail)
+
+
+class BadRequestError(DeepsetAPIError):
+    """Exception raised for invalid requests (HTTP 400)."""
+
+    def __init__(self, message: Any = "Bad request", detail: Any | None = None) -> None:
+        """Initialize the exception."""
+        super().__init__(status_code=400, message=message, detail=detail)
+
+
+class UnexpectedAPIError(DeepsetAPIError):
+    """Catch-all exception for unexpected API errors."""
+
+    def __init__(
+        self, status_code: int | None = None, message: Any = "Unexpected API error", detail: Any | None = None
+    ):
+        """Initialize the exception."""
+        super().__init__(status_code=status_code, message=message, detail=detail)

src/deepset_mcp/api/pipeline/models.py

@@ -1,6 +1,5 @@
 from datetime import datetime
 from enum import StrEnum
-from typing import List
 
 from pydantic import BaseModel, Field
 
@@ -59,4 +58,4 @@ class PipelineValidationResult(BaseModel):
     """Result of validating a pipeline configuration."""
 
     valid: bool
-    errors: List[ValidationError] = []
+    errors: list[ValidationError] = []

src/deepset_mcp/api/pipeline/resource.py

@@ -1,7 +1,9 @@
 from typing import Any
 
-from deepset_mcp.api.client import AsyncClientProtocol
+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
 
 
 class PipelineResource:
@@ -15,48 +17,44 @@ def __init__(
         """Initializes a PipelineResource instance."""
         self._client = client
         self._workspace = workspace
-        
+
     async def validate(self, yaml_config: str) -> PipelineValidationResult:
         """
         Validate a pipeline's YAML configuration against the API.
-        
+
         Args:
             yaml_config: The YAML configuration string to validate
-            
+
         Returns:
             PipelineValidationResult containing validation status and any errors
-            
+
         Raises:
             ValueError: If the YAML is not valid (422 error) or contains syntax errors
         """
-        data = {
-            "deepset_cloud_version": "v2",
-            "query_yaml": yaml_config
-        }
-        
+        data = {"query_yaml": yaml_config}
+
         resp = await self._client.request(
             endpoint=f"v1/workspaces/{self._workspace}/pipeline_validations",
             method="POST",
             data=data,
         )
-        
+
         # If successful (status 200), the YAML is valid
         if resp.success:
             return PipelineValidationResult(valid=True)
-            
+
         # If 400 error, we have validation errors to process
-        if resp.status_code == 400 and resp.json is not None and "errors" in resp.json:
-            errors = [ValidationError(code=error["code"], message=error["message"]) 
-                      for error in resp.json["errors"]]
+        if resp.status_code == 400 and resp.json is not None and "details" in resp.json:
+            errors = [ValidationError(code=error["code"], message=error["message"]) for error in resp.json["details"]]
+
+            return PipelineValidationResult(valid=False, errors=errors)
+
+        if resp.status_code == 422:
+            errors = [ValidationError(code="YAML_ERROR", message="Syntax error in YAML")]
+
             return PipelineValidationResult(valid=False, errors=errors)
-            
-        # For other errors, return as an exception
-        # (like 422 for invalid YAML syntax)
-        if not resp.success:
-            if resp.json and "detail" in resp.json:
-                raise ValueError(f"Pipeline validation failed: {resp.json['detail']}")
-            else:
-                raise ValueError(f"Pipeline validation failed with status {resp.status_code}")
+
+        raise UnexpectedAPIError(status_code=resp.status_code, message=resp.text, detail=resp.json)
 
     async def list(
         self,
@@ -81,6 +79,8 @@ async def list(
             method="GET",
         )
 
+        raise_for_status(resp)
+
         response = resp.json
 
         if response is not None:
@@ -93,13 +93,17 @@ async def list(
     async def get(self, pipeline_name: str, include_yaml: bool = True) -> DeepsetPipeline:
         """Fetch a single pipeline by its name."""
         resp = await self._client.request(endpoint=f"v1/workspaces/{self._workspace}/pipelines/{pipeline_name}")
+        raise_for_status(resp)
+
         pipeline = DeepsetPipeline.model_validate(resp.json)
 
         if include_yaml:
             yaml_response = await self._client.request(
                 endpoint=f"v1/workspaces/{self._workspace}/pipelines/{pipeline_name}/yaml"
             )
 
+            raise_for_status(yaml_response)
+
             if yaml_response.json is not None:
                 pipeline.yaml_config = yaml_response.json["query_yaml"]
 
@@ -108,12 +112,14 @@ async def get(self, pipeline_name: str, include_yaml: bool = True) -> DeepsetPip
     async def create(self, name: str, yaml_config: str) -> None:
         """Create a new pipeline with a name and YAML config."""
         data = {"name": name, "query_yaml": yaml_config}
-        await self._client.request(
+        resp = await self._client.request(
             endpoint=f"v1/workspaces/{self._workspace}/pipelines",
             method="POST",
             data=data,
         )
 
+        raise_for_status(resp)
+
     async def update(
         self,
         pipeline_name: str,
@@ -123,23 +129,21 @@ async def update(
         """Update name and/or YAML config of an existing pipeline."""
         # Handle name update first if any
         if updated_pipeline_name is not None:
-            await self._client.request(
+            name_resp = await self._client.request(
                 endpoint=f"v1/workspaces/{self._workspace}/pipelines/{pipeline_name}",
                 method="PATCH",
                 data={"name": updated_pipeline_name},
             )
 
+            raise_for_status(name_resp)
+
             pipeline_name = updated_pipeline_name
 
         if yaml_config is not None:
-            await self._client.request(
+            yaml_resp = await self._client.request(
                 endpoint=f"v1/workspaces/{self._workspace}/pipelines/{pipeline_name}/yaml",
                 method="PUT",
                 data={"query_yaml": yaml_config},
             )
 
-
-# async with AsyncDeepsetClient() as client:
-#    await client.pipelines("default").list()
-#    await client.pipelines("default").get("hello")
-#    await client.pipelines("default").update(yaml_config="blabla")
+            raise_for_status(yaml_resp)

src/deepset_mcp/api/protocols.py

@@ -0,0 +1,32 @@
+from types import TracebackType
+from typing import Any, Protocol, Self
+
+from deepset_mcp.api.transport import TransportResponse
+
+
+class AsyncClientProtocol(Protocol):
+    """Protocol defining the implementation for AsyncClient."""
+
+    async def request(
+        self,
+        endpoint: str,
+        method: str = "GET",
+        data: dict[str, Any] | None = None,
+        headers: dict[str, str] | None = None,
+    ) -> TransportResponse:
+        """Make a request to the API."""
+        ...
+
+    async def close(self) -> None:
+        """Close underlying transport resources."""
+        ...
+
+    async def __aenter__(self) -> Self:
+        """Enter the AsyncContextManager."""
+        ...
+
+    async def __aexit__(
+        self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None
+    ) -> bool:
+        """Exit the AsyncContextmanager and clean up resources."""
+        ...

src/deepset_mcp/api/transport.py

@@ -4,6 +4,8 @@
 
 import httpx
 
+from deepset_mcp.api.exceptions import BadRequestError, ResourceNotFoundError, UnexpectedAPIError
+
 
 @dataclass
 class TransportResponse:
@@ -12,13 +14,41 @@ class TransportResponse:
     text: str
     status_code: int
     json: dict[str, Any] | None = None
-    
+
     @property
     def success(self) -> bool:
         """Returns True if the response status code indicates success (< 400)."""
         return self.status_code < 400
 
 
+def raise_for_status(response: TransportResponse) -> None:
+    """Raises the appropriate exception based on the response status code."""
+    if response.success:
+        return
+
+    # Map status codes to exception classes
+    exception_map = {
+        400: BadRequestError,
+        404: ResourceNotFoundError,
+    }
+
+    # Extract error details from response if available
+    detail = response.json.get("details") if response.json else None
+    message = response.json.get("message") if response.json else response.text
+
+    # Get exception class
+    exception_class = exception_map.get(response.status_code)
+
+    if exception_class:
+        # For specific exceptions (BadRequestError, ResourceNotFoundError)
+        raise exception_class(message=message, detail=detail)
+    else:
+        # For the catch-all case, include the status code
+        raise UnexpectedAPIError(
+            status_code=response.status_code, message=message or "Unexpected API error", detail=detail
+        )
+
+
 class TransportProtocol(Protocol):
     """Protocol for HTTP transport."""