feat: add workspace resource API with CRUD operations (#129)

* feat: add workspace api module init file

* feat: add workspace models

* feat: add workspace resource protocol

* feat: add workspace resource implementation

* feat: add workspace protocol import

* feat: add workspaces method to AsyncClientProtocol

* feat: add workspace resource import

* feat: add workspaces method to AsyncDeepsetClient

* feat: add workspace protocol import to test conftest

* feat: add workspaces method to BaseFakeClient

* feat: add workspace unit test module init

* feat: add comprehensive unit tests for workspace resource

* feat: add integration tests for workspace resource

* fix: tests, format, add integration

Author: mathislucka

REPO.md

@@ -86,7 +86,7 @@ You would need to perform the following steps:
 4. refer to `src/deepset_mcp/tools/pipeline.py` as a good example for tool implementations
 5. extract model or response serialization into reusable helper functions
 6. once you added a tool, import it in `src/deepset_mcp/tool_factory.py` and add it to the tool registry with the appropriate config
-7the docstring of the tool will serve as the prompt for the large language model calling the tool, make sure it has good instructions on when to use the tool, how to best use it, and what kind of answer to expect.
+7. the docstring of the tool will serve as the prompt for the large language model calling the tool, make sure it has good instructions on when to use the tool, how to best use it, and what kind of answer to expect.
 
 #### Testing the tool
 

src/deepset_mcp/api/client.py

@@ -18,6 +18,7 @@
     TransportResponse,
 )
 from deepset_mcp.api.user.resource import UserResource
+from deepset_mcp.api.workspace.resource import WorkspaceResource
 
 T = TypeVar("T")
 
@@ -265,3 +266,7 @@ def users(self) -> UserResource:
     def secrets(self) -> SecretResource:
         """Resource to interact with secrets."""
         return SecretResource(client=self)
+
+    def workspaces(self) -> WorkspaceResource:
+        """Resource to interact with workspaces."""
+        return WorkspaceResource(client=self)

src/deepset_mcp/api/protocols.py

@@ -10,6 +10,7 @@
 from deepset_mcp.api.secrets.protocols import SecretResourceProtocol
 from deepset_mcp.api.transport import StreamingResponse, TransportResponse
 from deepset_mcp.api.user.protocols import UserResourceProtocol
+from deepset_mcp.api.workspace.protocols import WorkspaceResourceProtocol
 
 T = TypeVar("T")
 
@@ -110,3 +111,7 @@ def users(self) -> "UserResourceProtocol":
     def secrets(self) -> "SecretResourceProtocol":
         """Access secrets."""
         ...
+
+    def workspaces(self) -> "WorkspaceResourceProtocol":
+        """Access workspaces."""
+        ...

src/deepset_mcp/api/workspace/__init__.py

@@ -0,0 +1,7 @@
+"""Workspace API module."""
+
+from .models import Workspace, WorkspaceList
+from .protocols import WorkspaceResourceProtocol
+from .resource import WorkspaceResource
+
+__all__ = ["Workspace", "WorkspaceList", "WorkspaceResourceProtocol", "WorkspaceResource"]

src/deepset_mcp/api/workspace/models.py

@@ -0,0 +1,23 @@
+"""Models for workspace API responses."""
+
+from typing import Any
+from uuid import UUID
+
+from pydantic import BaseModel
+
+
+class Workspace(BaseModel):
+    """Model representing a workspace on the deepset platform."""
+
+    name: str
+    workspace_id: UUID
+    languages: dict[str, Any]
+    default_idle_timeout_in_seconds: int
+
+
+class WorkspaceList(BaseModel):
+    """Model representing a list of workspaces."""
+
+    data: list[Workspace]
+    has_more: bool = False
+    total: int

src/deepset_mcp/api/workspace/protocols.py

@@ -0,0 +1,41 @@
+"""Protocols for workspace resources."""
+
+from typing import Protocol
+
+from deepset_mcp.api.shared_models import NoContentResponse
+from deepset_mcp.api.workspace.models import Workspace, WorkspaceList
+
+
+class WorkspaceResourceProtocol(Protocol):
+    """Protocol defining the interface for workspace resources."""
+
+    async def list(self) -> WorkspaceList:
+        """List all workspaces.
+
+        :returns: A WorkspaceList containing all workspaces.
+        """
+        ...
+
+    async def get(self, workspace_name: str) -> Workspace:
+        """Get a specific workspace by name.
+
+        :param workspace_name: Name of the workspace to fetch.
+        :returns: A Workspace instance.
+        """
+        ...
+
+    async def create(self, name: str) -> NoContentResponse:
+        """Create a new workspace.
+
+        :param name: Name of the new workspace.
+        :returns: NoContentResponse indicating successful creation.
+        """
+        ...
+
+    async def delete(self, workspace_name: str) -> NoContentResponse:
+        """Delete a workspace.
+
+        :param workspace_name: Name of the workspace to delete.
+        :returns: NoContentResponse indicating successful deletion.
+        """
+        ...

src/deepset_mcp/api/workspace/resource.py

@@ -0,0 +1,94 @@
+"""Resource implementation for workspace API."""
+
+import logging
+from typing import TYPE_CHECKING
+
+from deepset_mcp.api.shared_models import NoContentResponse
+from deepset_mcp.api.transport import raise_for_status
+from deepset_mcp.api.workspace.models import Workspace, WorkspaceList
+from deepset_mcp.api.workspace.protocols import WorkspaceResourceProtocol
+
+logger = logging.getLogger(__name__)
+
+if TYPE_CHECKING:
+    from deepset_mcp.api.protocols import AsyncClientProtocol
+
+
+class WorkspaceResource(WorkspaceResourceProtocol):
+    """Manages interactions with the deepset workspace API."""
+
+    def __init__(self, client: "AsyncClientProtocol") -> None:
+        """Initialize a WorkspaceResource instance.
+
+        :param client: The async client protocol instance.
+        """
+        self._client = client
+
+    async def list(self) -> WorkspaceList:
+        """List all workspaces.
+
+        :returns: A WorkspaceList containing all workspaces.
+        """
+        resp = await self._client.request(
+            endpoint="v1/workspaces",
+            method="GET",
+        )
+
+        raise_for_status(resp)
+
+        if resp.json is not None and isinstance(resp.json, list):
+            workspaces = [Workspace.model_validate(item) for item in resp.json]
+            return WorkspaceList(
+                data=workspaces,
+                has_more=False,
+                total=len(workspaces),
+            )
+        else:
+            return WorkspaceList(data=[], has_more=False, total=0)
+
+    async def get(self, workspace_name: str) -> Workspace:
+        """Get a specific workspace by name.
+
+        :param workspace_name: Name of the workspace to fetch.
+        :returns: A Workspace instance.
+        """
+        resp = await self._client.request(
+            endpoint=f"v1/workspaces/{workspace_name}",
+            method="GET",
+        )
+
+        raise_for_status(resp)
+
+        return Workspace.model_validate(resp.json)
+
+    async def create(self, name: str) -> NoContentResponse:
+        """Create a new workspace.
+
+        :param name: Name of the new workspace.
+        :returns: NoContentResponse indicating successful creation.
+        """
+        data = {"name": name}
+        resp = await self._client.request(
+            endpoint="v1/workspaces",
+            method="POST",
+            data=data,
+        )
+
+        raise_for_status(resp)
+
+        return NoContentResponse(message="Workspace created successfully.")
+
+    async def delete(self, workspace_name: str) -> NoContentResponse:
+        """Delete a workspace.
+
+        :param workspace_name: Name of the workspace to delete.
+        :returns: NoContentResponse indicating successful deletion.
+        """
+        resp = await self._client.request(
+            endpoint=f"v1/workspaces/{workspace_name}",
+            method="DELETE",
+        )
+
+        raise_for_status(resp)
+
+        return NoContentResponse(message="Workspace deleted successfully.")

test/integration/test_integration_workspace_resource.py

@@ -0,0 +1,64 @@
+"""Integration tests for WorkspaceResource."""
+
+import uuid
+
+import pytest
+
+from deepset_mcp.api.client import AsyncDeepsetClient
+from deepset_mcp.api.exceptions import ResourceNotFoundError
+from deepset_mcp.api.workspace.models import Workspace, WorkspaceList
+
+pytestmark = pytest.mark.integration
+
+
+class TestWorkspaceResourceIntegration:
+    """Integration tests for WorkspaceResource."""
+
+    @pytest.mark.asyncio
+    async def test_list_workspaces(self) -> None:
+        """Test listing workspaces."""
+        async with AsyncDeepsetClient() as client:
+            workspaces = await client.workspaces().list()
+            assert isinstance(workspaces, WorkspaceList)
+            assert isinstance(workspaces.data, list)
+            assert workspaces.total >= 0
+
+            # If we have workspaces, verify their structure
+            if workspaces.data:
+                workspace = workspaces.data[0]
+                assert isinstance(workspace, Workspace)
+                assert isinstance(workspace.name, str)
+                assert isinstance(workspace.workspace_id, uuid.UUID)
+                assert isinstance(workspace.languages, dict)
+                assert isinstance(workspace.default_idle_timeout_in_seconds, int)
+
+    @pytest.mark.asyncio
+    async def test_get_workspace_not_found(self) -> None:
+        """Test getting a non-existent workspace."""
+        async with AsyncDeepsetClient() as client:
+            with pytest.raises(ResourceNotFoundError):
+                await client.workspaces().get("definitely-does-not-exist-workspace")
+
+    @pytest.mark.asyncio
+    async def test_create_get_and_delete_workspace(self) -> None:
+        """Tests creating, getting and deleting a workspace."""
+        workspace_name = f"test-workspace-{uuid.uuid4()}"
+        async with AsyncDeepsetClient() as client:
+            # Create a new workspace
+            create_response = await client.workspaces().create(workspace_name)
+            assert create_response.success is True
+            assert create_response.message == "Workspace created successfully."
+
+            # Get the workspace
+            workspace = await client.workspaces().get(workspace_name)
+            assert isinstance(workspace, Workspace)
+            assert workspace.name == workspace_name
+
+            # Delete the workspace
+            delete_response = await client.workspaces().delete(workspace_name)
+            assert delete_response.success is True
+            assert delete_response.message == "Workspace deleted successfully."
+
+            # Verify the workspace is deleted
+            with pytest.raises(ResourceNotFoundError):
+                await client.workspaces().get(workspace_name)

test/unit/api/workspace/__init__.py

@@ -0,0 +1 @@
+"""Unit tests for workspace API."""