feat: add HaystackServiceResource to API SDK (#21)

* feat: add HaystackServiceProtocol imports

* feat: add HaystackServiceProtocol and extend AsyncClientProtocol

* feat: add HaystackComponentSchema model

* feat: add HaystackServiceResource implementation

* feat: import HaystackServiceResource in client

* feat: add haystack_service method to AsyncDeepsetClient

* test: add unit tests for HaystackServiceResource

* feat: wrapup haystack service; update test structure

Author: mathislucka

src/deepset_mcp/api/client.py

@@ -2,6 +2,7 @@
 from types import TracebackType
 from typing import Any, Self
 
+from deepset_mcp.api.haystack_service.resource import HaystackServiceResource
 from deepset_mcp.api.pipeline.resource import PipelineResource
 from deepset_mcp.api.protocols import AsyncClientProtocol
 from deepset_mcp.api.transport import AsyncTransport, TransportProtocol, TransportResponse
@@ -93,3 +94,7 @@ async def __aexit__(
     def pipelines(self, workspace: str) -> PipelineResource:
         """Resource to interact with pipelines in the specified workspace."""
         return PipelineResource(client=self, workspace=workspace)
+
+    def haystack_service(self) -> HaystackServiceResource:
+        """Resource to interact with the Haystack service API."""
+        return HaystackServiceResource(client=self)

src/deepset_mcp/api/haystack_service/__init__.py

@@ -0,0 +1,29 @@
+from typing import Any
+
+from deepset_mcp.api.protocols import AsyncClientProtocol, HaystackServiceProtocol
+from deepset_mcp.api.transport import raise_for_status
+
+
+class HaystackServiceResource(HaystackServiceProtocol):
+    """Manages interactions with the deepset Haystack service API."""
+
+    def __init__(self, client: AsyncClientProtocol) -> None:
+        """Initializes a HaystackServiceResource instance."""
+        self._client = client
+
+    async def get_component_schemas(self) -> dict[str, Any]:
+        """Fetch the component schema from the API.
+
+        Returns:
+            The component schema as a dictionary
+        """
+        resp = await self._client.request(
+            endpoint="v1/haystack/components",
+            method="GET",
+            headers={"accept": "application/json"},
+            data={"domain": "deepset-cloud"},
+        )
+
+        raise_for_status(resp)
+
+        return resp.json if resp.json is not None else {}

src/deepset_mcp/api/haystack_service/resource.py

@@ -5,6 +5,14 @@
 from deepset_mcp.api.transport import TransportResponse
 
 
+class HaystackServiceProtocol(Protocol):
+    """Protocol defining the implementation for HaystackService."""
+
+    async def get_component_schemas(self) -> dict[str, Any]:
+        """Fetch the component schema from the API."""
+        ...
+
+
 class AsyncClientProtocol(Protocol):
     """Protocol defining the implementation for AsyncClient."""
 
@@ -36,6 +44,10 @@ def pipelines(self, workspace: str) -> "PipelineResourceProtocol":
         """Access pipelines in the specified workspace."""
         ...
 
+    def haystack_service(self) -> "HaystackServiceProtocol":
+        """Access the Haystack service."""
+        ...
+
 
 class PipelineResourceProtocol(Protocol):
     """Protocol defining the implementation for PipelineResource."""

src/deepset_mcp/api/protocols.py

@@ -1,4 +1,5 @@
 import os
+from typing import Any
 
 from mcp.server.fastmcp import FastMCP
 
@@ -86,20 +87,16 @@ async def update_pipeline(
     return response
 
 
-#
-#
-# @mcp.tool()
-# async def get_component_schemas() -> Any:
-#     """Retrieves the schemas for all available Haystack components from the deepset API.
-#
-#     These schemas define the expected input and output parameters for each component type, which is useful for
-#     constructing or validating componets in a pipeline YAML.
-#     """
-#     async with AsyncDeepsetClient() as client:
-#         response = await client.request(endpoint="v1/haystack/components", method="GET")
-#         return response.json
-#
-#
+@mcp.tool()
+async def get_component_schemas() -> Any:
+    """Retrieves the schemas for all available Haystack components from the deepset API.
+
+    These schemas define the expected input and output parameters for each component type, which is useful for
+    constructing or validating componets in a pipeline YAML.
+    """
+    async with AsyncDeepsetClient() as client:
+        response = await client.request(endpoint="v1/haystack/components", method="GET")
+        return response.json
 
 
 @mcp.tool()

src/deepset_mcp/main.py

@@ -0,0 +1,54 @@
+import os
+import uuid
+from collections.abc import AsyncGenerator
+from datetime import datetime
+
+import pytest
+from dotenv import load_dotenv
+
+from deepset_mcp.api.client import AsyncDeepsetClient
+
+load_dotenv()
+
+
+@pytest.fixture
+def test_workspace_name() -> str:
+    """Create a unique workspace name for testing."""
+    timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
+    return f"test-workspace-{timestamp}-{uuid.uuid4().hex[:8]}"
+
+
+@pytest.fixture
+async def client() -> AsyncGenerator[AsyncDeepsetClient, None]:
+    """Create and configure the deepset client."""
+    api_key = os.environ.get("DEEPSET_API_KEY")
+    if not api_key:
+        pytest.skip("DEEPSET_API_KEY environment variable not set")
+
+    async with AsyncDeepsetClient(api_key=api_key) as client:
+        yield client
+
+
+@pytest.fixture
+async def test_workspace(
+    client: AsyncDeepsetClient,
+    test_workspace_name: str,
+) -> AsyncGenerator[str, None]:
+    """Create a test workspace and clean it up after tests."""
+    # Create a test workspace
+    await client.request(
+        endpoint="v1/workspaces",
+        method="POST",
+        data={"name": test_workspace_name},
+    )
+
+    yield test_workspace_name
+
+    # Clean up the workspace after tests
+    try:
+        await client.request(
+            endpoint=f"v1/workspaces/{test_workspace_name}",
+            method="DELETE",
+        )
+    except Exception as e:
+        print(f"Failed to delete test workspace: {e}")

test/__init__.py

@@ -0,0 +1,25 @@
+import pytest
+
+from deepset_mcp.api.client import AsyncDeepsetClient
+from deepset_mcp.api.haystack_service.resource import HaystackServiceResource
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture
+async def haystack_service_resource(
+    client: AsyncDeepsetClient,
+) -> HaystackServiceResource:
+    """Create a PipelineResource instance for testing."""
+    return HaystackServiceResource(client=client)
+
+
+@pytest.mark.asyncio
+async def test_get_component_schemas(
+    haystack_service_resource: HaystackServiceResource,
+) -> None:
+    """Test for getting component schemas."""
+    response = await haystack_service_resource.get_component_schemas()
+
+    assert isinstance(response, dict)
+    assert "component_schema" in response

test/integration/__init__.py

@@ -1,61 +1,11 @@
-import os
-import uuid
-from collections.abc import AsyncGenerator
-from datetime import datetime
-
 import pytest
-from dotenv import load_dotenv
 
 from deepset_mcp.api.client import AsyncDeepsetClient
 from deepset_mcp.api.exceptions import ResourceNotFoundError
 from deepset_mcp.api.pipeline.models import DeepsetPipeline
 from deepset_mcp.api.pipeline.resource import PipelineResource
 
 pytestmark = pytest.mark.integration
-load_dotenv()
-
-
-@pytest.fixture
-def test_workspace_name() -> str:
-    """Create a unique workspace name for testing."""
-    timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
-    return f"test-workspace-{timestamp}-{uuid.uuid4().hex[:8]}"
-
-
-@pytest.fixture
-async def client() -> AsyncGenerator[AsyncDeepsetClient, None]:
-    """Create and configure the deepset client."""
-    api_key = os.environ.get("DEEPSET_API_KEY")
-    if not api_key:
-        pytest.skip("DEEPSET_API_KEY environment variable not set")
-
-    async with AsyncDeepsetClient(api_key=api_key) as client:
-        yield client
-
-
-@pytest.fixture
-async def test_workspace(
-    client: AsyncDeepsetClient,
-    test_workspace_name: str,
-) -> AsyncGenerator[str, None]:
-    """Create a test workspace and clean it up after tests."""
-    # Create a test workspace
-    await client.request(
-        endpoint="v1/workspaces",
-        method="POST",
-        data={"name": test_workspace_name},
-    )
-
-    yield test_workspace_name
-
-    # Clean up the workspace after tests
-    try:
-        await client.request(
-            endpoint=f"v1/workspaces/{test_workspace_name}",
-            method="DELETE",
-        )
-    except Exception as e:
-        print(f"Failed to delete test workspace: {e}")
 
 
 @pytest.fixture