Merge pull request #14 from deepset-ai/feat/pipeline_resource

feat: add pipeline resource

Author: mathislucka

.github/workflows/ci.yml

@@ -25,7 +25,7 @@ jobs:
         run: uv sync --locked --all-extras --dev
 
       - name: Run tests
-        run: uv run pytest
+        run: uv run --dev pytest -m "not integration"
 
   lint:
     name: Run Linting

.gitignore

@@ -8,3 +8,9 @@ wheels/
 
 # Virtual environments
 .venv
+
+# env files
+.env
+.env.prod
+.env.dev
+.env.test

pyproject.toml

@@ -10,6 +10,7 @@ dependencies = [
     "requests>=2.32.3",
     "uvicorn>=0.34.2",
     "httpx",
+    "pydantic>=2.0.0",
 ]
 
 [project.scripts]
@@ -26,6 +27,7 @@ packages = ["src/deepset_mcp"]
 dev = [
     "pytest",
     "pytest-asyncio",
+    "python-dotenv"
 ]
 lint = [
     "ruff",
@@ -38,6 +40,12 @@ types = [
 
 [tool.pytest.ini_options]
 testpaths = ["test"]
+asyncio_mode = "auto"
+asyncio_default_fixture_loop_scope = "function"
+
+markers = [
+    "integration: marks tests that interact with external resources (e.g. deepset API)."
+]
 
 [tool.ruff]
 line-length = 120

src/deepset_mcp/api/__init__.py

@@ -1,8 +1,36 @@
 import os
 from types import TracebackType
-from typing import Any
+from typing import Any, Protocol, Self
 
-from deepset_mcp.transport import AsyncTransport, TransportProtocol
+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:
@@ -48,7 +76,7 @@ async def request(
         method: str = "GET",
         data: dict[str, Any] | None = None,
         headers: dict[str, str] | None = None,
-    ) -> Any:
+    ) -> TransportResponse:
         """Make a request to the deepset API."""
         if not endpoint.startswith("/"):
             endpoint = f"/{endpoint}"
@@ -77,7 +105,7 @@ async def close(self) -> None:
         """Close underlying transport resources."""
         await self._transport.close()
 
-    async def __aenter__(self) -> "AsyncDeepsetClient":
+    async def __aenter__(self) -> Self:
         """Enter the AsyncContextManager."""
         return self
 

src/deepset_mcp/client.py src/deepset_mcp/api/client.pysrc/deepset_mcp/client.py renamed to src/deepset_mcp/api/client.py

@@ -0,0 +1,47 @@
+from datetime import datetime
+from enum import StrEnum
+
+from pydantic import BaseModel, Field
+
+
+class PipelineServiceLevel(StrEnum):
+    """Describes the service level of a pipeline."""
+
+    PRODUCTION = "PRODUCTION"
+    DEVELOPMENT = "DEVELOPMENT"
+    DRAFT = "DRAFT"
+
+
+class DeepsetUser(BaseModel):
+    """Model representing a user on the deepset platform."""
+
+    id: str = Field(alias="user_id")
+    given_name: str | None = None
+    family_name: str | None = None
+    email: str | None = None
+
+
+class DeepsetPipeline(BaseModel):
+    """Model representing a pipeline on the deepset platform."""
+
+    id: str = Field(alias="pipeline_id")
+    name: str
+    status: str
+    service_level: PipelineServiceLevel
+
+    created_at: datetime
+    last_updated_at: datetime | None = Field(None, alias="last_edited_at")  # Map API's last_edited_at
+
+    created_by: DeepsetUser
+    last_updated_by: DeepsetUser | None = Field(None, alias="last_edited_by")  # Map API's last_edited_by
+
+    yaml_config: str | None = None
+
+    class Config:
+        """Configuration for serialization and deserialization."""
+
+        populate_by_name = True  # Allow both alias and model field names
+        json_encoders = {
+            # When serializing back to JSON, convert datetimes to ISO format
+            datetime: lambda dt: dt.isoformat()
+        }

src/deepset_mcp/api/pipeline/__init__.py

@@ -0,0 +1,103 @@
+from typing import Any
+
+from deepset_mcp.api.client import AsyncClientProtocol
+from deepset_mcp.api.pipeline.models import DeepsetPipeline
+
+
+class PipelineResource:
+    """Manages interactions with the deepset pipeline API."""
+
+    def __init__(
+        self,
+        client: AsyncClientProtocol,
+        workspace: str,
+    ) -> None:
+        """Initializes a PipelineResource instance."""
+        self._client = client
+        self._workspace = workspace
+
+    async def list(
+        self,
+        page_number: int = 1,
+        limit: int = 10,
+    ) -> 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: PipelineListResponse containing `data`, `has_more`, and `total`.
+        """
+        params: dict[str, Any] = {
+            "page_number": page_number,
+            "limit": limit,
+        }
+
+        query = "?" + "&".join(f"{k}={v}" for k, v in params.items()) if params else ""
+        resp = await self._client.request(
+            endpoint=f"v1/workspaces/{self._workspace}/pipelines{query}",
+            method="GET",
+        )
+
+        response = resp.json
+
+        if response is not None:
+            pipelines = [DeepsetPipeline.model_validate(item) for item in response.get("data", [])]
+        else:
+            pipelines = []
+
+        return pipelines
+
+    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}")
+        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"
+            )
+
+            if yaml_response.json is not None:
+                pipeline.yaml_config = yaml_response.json["query_yaml"]
+
+        return pipeline
+
+    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(
+            endpoint=f"v1/workspaces/{self._workspace}/pipelines",
+            method="POST",
+            data=data,
+        )
+
+    async def update(
+        self,
+        pipeline_name: str,
+        updated_pipeline_name: str | None = None,
+        yaml_config: str | None = None,
+    ) -> None:
+        """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(
+                endpoint=f"v1/workspaces/{self._workspace}/pipelines/{pipeline_name}",
+                method="PATCH",
+                data={"name": updated_pipeline_name},
+            )
+
+            pipeline_name = updated_pipeline_name
+
+        if yaml_config is not None:
+            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")

src/deepset_mcp/api/pipeline/models.py

@@ -1,12 +1,23 @@
+import json
+from dataclasses import dataclass
 from typing import Any, Protocol
 
 import httpx
 
 
+@dataclass
+class TransportResponse:
+    """Response envelope for HTTP transport."""
+
+    text: str
+    status_code: int
+    json: dict[str, Any] | None = None
+
+
 class TransportProtocol(Protocol):
     """Protocol for HTTP transport."""
 
-    async def request(self, method: str, url: str, **kwargs: Any) -> Any:
+    async def request(self, method: str, url: str, **kwargs: Any) -> TransportResponse:
         """Send an HTTP request and return the parsed JSON response."""
         ...
 
@@ -15,7 +26,7 @@ async def close(self) -> None:
         ...
 
 
-class AsyncTransport(TransportProtocol):
+class AsyncTransport:
     """Asynchronous HTTP transport using httpx.AsyncClient."""
 
     def __init__(
@@ -48,12 +59,20 @@ def __init__(
         }
         self._client = httpx.AsyncClient(**client_kwargs)
 
-    async def request(self, method: str, url: str, **kwargs: Any) -> Any:
-        """Send an HTTP request and return the parsed JSON response."""
-        resp = await self._client.request(method, url, **kwargs)
-        resp.raise_for_status()
+    async def request(self, method: str, url: str, **kwargs: Any) -> TransportResponse:
+        """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:
+            transport_response.json = response.json()
+        except json.decoder.JSONDecodeError:
+            pass
 
-        return resp.json()
+        return transport_response
 
     async def close(self) -> None:
         """Clean up any resources (e.g., close connections)."""