Merge pull request #16 from deepset-ai/fix-issue-15

feat: add pipeline config 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

@@ -45,3 +45,17 @@ class Config:
             # When serializing back to JSON, convert datetimes to ISO format
             datetime: lambda dt: dt.isoformat()
         }
+
+
+class ValidationError(BaseModel):
+    """Model representing a validation error from the pipeline validation API."""
+
+    code: str
+    message: str
+
+
+class PipelineValidationResult(BaseModel):
+    """Result of validating a pipeline configuration."""
+
+    valid: bool
+    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.pipeline.models import DeepsetPipeline
+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:
@@ -16,6 +18,44 @@ def __init__(
         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 = {"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 "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)
+
+        raise UnexpectedAPIError(status_code=resp.status_code, message=resp.text, detail=resp.json)
+
     async def list(
         self,
         page_number: int = 1,
@@ -39,6 +79,8 @@ async def list(
             method="GET",
         )
 
+        raise_for_status(resp)
+
         response = resp.json
 
         if response is not None:
@@ -51,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"]
 
@@ -66,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,
@@ -81,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:
@@ -13,6 +15,39 @@ class TransportResponse:
     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."""
@@ -63,8 +98,6 @@ async def request(self, method: str, url: str, **kwargs: Any) -> TransportRespon
         """Send an HTTP request and return the response."""
         response = await self._client.request(method, url, **kwargs)
 
-        response.raise_for_status()
-
         transport_response = TransportResponse(text=response.text, status_code=response.status_code)
 
         try: