feat: add get_component_input_output method to HaystackServiceResource (#29)

* feat: add get_component_input_output method to HaystackServiceResource

* feat: add get_component_input_output to FakeHaystackServiceResource

* test: add unit tests for get_component_input_output

* test: add integration test for get_component_input_output

* fix: typing issues; tests

Author: mathislucka

src/deepset_mcp/api/client.py

@@ -1,13 +1,15 @@
 import os
 from types import TracebackType
-from typing import Any, Self
+from typing import Any, Self, TypeVar, overload
 
 from deepset_mcp.api.haystack_service.resource import HaystackServiceResource
 from deepset_mcp.api.pipeline.resource import PipelineResource
 from deepset_mcp.api.pipeline_template.resource import PipelineTemplateResource
 from deepset_mcp.api.protocols import AsyncClientProtocol
 from deepset_mcp.api.transport import AsyncTransport, TransportProtocol, TransportResponse
 
+T = TypeVar("T")
+
 
 class AsyncDeepsetClient(AsyncClientProtocol):
     """Async Client for interacting with the deepset API."""
@@ -46,13 +48,37 @@ def __init__(
                 config=transport_config,
             )
 
+    @overload
+    async def request(
+        self,
+        endpoint: str,
+        *,
+        response_type: type[T],
+        method: str = "GET",
+        data: dict[str, Any] | None = None,
+        headers: dict[str, str] | None = None,
+    ) -> TransportResponse[T]: ...
+
+    @overload
+    async def request(
+        self,
+        endpoint: str,
+        *,
+        response_type: None = None,
+        method: str = "GET",
+        data: dict[str, Any] | None = None,
+        headers: dict[str, str] | None = None,
+    ) -> TransportResponse[Any]: ...
+
     async def request(
         self,
         endpoint: str,
+        *,
         method: str = "GET",
         data: dict[str, Any] | None = None,
         headers: dict[str, str] | None = None,
-    ) -> TransportResponse:
+        response_type: type[T] | None = None,
+    ) -> TransportResponse[Any]:
         """Make a request to the deepset API."""
         if not endpoint.startswith("/"):
             endpoint = f"/{endpoint}"
@@ -75,6 +101,7 @@ async def request(
             url,
             json=data,
             headers=request_headers,
+            response_type=response_type,
         )
 
     async def close(self) -> None:

src/deepset_mcp/api/haystack_service/resource.py

@@ -1,5 +1,6 @@
 from typing import Any
 
+from deepset_mcp.api.exceptions import ResourceNotFoundError
 from deepset_mcp.api.protocols import AsyncClientProtocol, HaystackServiceProtocol
 from deepset_mcp.api.transport import raise_for_status
 
@@ -27,3 +28,26 @@ async def get_component_schemas(self) -> dict[str, Any]:
         raise_for_status(resp)
 
         return resp.json if resp.json is not None else {}
+
+    async def get_component_input_output(self, component_name: str) -> dict[str, Any]:
+        """Fetch the component input and output schema from the API.
+
+        Args:
+            component_name: The name of the component to fetch the input/output schema for
+
+        Returns:
+            The component input/output schema as a dictionary
+        """
+        resp = await self._client.request(
+            endpoint=f"v1/haystack/components/input-output?domain=deepset-cloud&names={component_name}",
+            method="GET",
+            headers={"accept": "application/json"},
+            response_type=list[dict[str, Any]],
+        )
+
+        raise_for_status(resp)
+
+        if resp.json is None or len(resp.json) == 0:
+            raise ResourceNotFoundError(f"Component '{component_name}' not found.")
+
+        return resp.json[0] if resp.json is not None else {}

src/deepset_mcp/api/pipeline/resource.py

@@ -48,8 +48,7 @@ async def validate(self, yaml_config: str) -> PipelineValidationResult:
         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:
+        if resp.status_code == 400 and resp.json is not None and isinstance(resp.json, dict) 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)

src/deepset_mcp/api/protocols.py

@@ -1,5 +1,5 @@
 from types import TracebackType
-from typing import Any, Protocol, Self
+from typing import Any, Protocol, Self, TypeVar, overload
 
 from deepset_mcp.api.pipeline.models import DeepsetPipeline, NoContentResponse, PipelineValidationResult
 from deepset_mcp.api.pipeline_template.models import PipelineTemplate
@@ -13,17 +13,48 @@ async def get_component_schemas(self) -> dict[str, Any]:
         """Fetch the component schema from the API."""
         ...
 
+    async def get_component_input_output(self, component_name: str) -> dict[str, Any]:
+        """Fetch input and output schema for a component from the API."""
+        ...
+
+
+T = TypeVar("T")
+
 
 class AsyncClientProtocol(Protocol):
     """Protocol defining the implementation for AsyncClient."""
 
+    @overload
+    async def request(
+        self,
+        endpoint: str,
+        *,
+        response_type: type[T],
+        method: str = "GET",
+        data: dict[str, Any] | None = None,
+        headers: dict[str, str] | None = None,
+    ) -> TransportResponse[T]: ...
+
+    @overload
+    async def request(
+        self,
+        endpoint: str,
+        *,
+        response_type: None = None,
+        method: str = "GET",
+        data: dict[str, Any] | None = None,
+        headers: dict[str, str] | None = None,
+    ) -> TransportResponse[Any]: ...
+
     async def request(
         self,
         endpoint: str,
+        *,
+        response_type: type[T] | None = None,
         method: str = "GET",
         data: dict[str, Any] | None = None,
         headers: dict[str, str] | None = None,
-    ) -> TransportResponse:
+    ) -> TransportResponse[Any]:
         """Make a request to the API."""
         ...
 

src/deepset_mcp/api/transport.py

@@ -1,27 +1,29 @@
 import json
 from dataclasses import dataclass
-from typing import Any, Protocol
+from typing import Any, Generic, Protocol, TypeVar, cast, overload
 
 import httpx
 
 from deepset_mcp.api.exceptions import BadRequestError, ResourceNotFoundError, UnexpectedAPIError
 
+T = TypeVar("T")
+
 
 @dataclass
-class TransportResponse:
-    """Response envelope for HTTP transport."""
+class TransportResponse(Generic[T]):
+    """Reponse envelope for HTTP transport."""
 
     text: str
     status_code: int
-    json: dict[str, Any] | None = None
+    json: T | None = None
 
     @property
     def success(self) -> bool:
-        """Returns True if the response status code indicates success (< 400)."""
+        """Check if the response was successful (status code < 400)."""
         return self.status_code < 400
 
 
-def raise_for_status(response: TransportResponse) -> None:
+def raise_for_status(response: TransportResponse[Any]) -> None:
     """Raises the appropriate exception based on the response status code."""
     if response.success:
         return
@@ -32,9 +34,12 @@ def raise_for_status(response: TransportResponse) -> None:
         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
+    if isinstance(response.json, dict):
+        detail = response.json.get("details") if response.json else None
+        message = response.json.get("message") if response.json else response.text
+    else:
+        detail = json.dumps(response.json) if response.json else None
+        message = response.text
 
     # Get exception class
     exception_class = exception_map.get(response.status_code)
@@ -52,8 +57,20 @@ def raise_for_status(response: TransportResponse) -> None:
 class TransportProtocol(Protocol):
     """Protocol for HTTP transport."""
 
-    async def request(self, method: str, url: str, **kwargs: Any) -> TransportResponse:
-        """Send an HTTP request and return the parsed JSON response."""
+    @overload
+    async def request(
+        self, method: str, url: str, *, response_type: type[T], **kwargs: Any
+    ) -> TransportResponse[T]: ...
+
+    @overload
+    async def request(
+        self, method: str, url: str, *, response_type: None = None, **kwargs: Any
+    ) -> TransportResponse[Any]: ...
+
+    async def request(
+        self, method: str, url: str, *, response_type: type[T] | None = None, **kwargs: Any
+    ) -> TransportResponse[Any]:
+        """Send an HTTP request and return the response."""
         ...
 
     async def close(self) -> None:
@@ -94,18 +111,34 @@ def __init__(
         }
         self._client = httpx.AsyncClient(**client_kwargs)
 
-    async def request(self, method: str, url: str, **kwargs: Any) -> TransportResponse:
+    @overload
+    async def request(
+        self, method: str, url: str, *, response_type: type[T], **kwargs: Any
+    ) -> TransportResponse[T]: ...
+
+    @overload
+    async def request(
+        self, method: str, url: str, *, response_type: None = None, **kwargs: Any
+    ) -> TransportResponse[Any]: ...
+
+    async def request(
+        self, method: str, url: str, *, response_type: type[T] | None = None, **kwargs: Any
+    ) -> TransportResponse[Any]:
         """Send an HTTP request and return the response."""
         response = await self._client.request(method, url, **kwargs)
 
-        transport_response = TransportResponse(text=response.text, status_code=response.status_code)
+        if response_type is not None:
+            raw = response.json()
+            payload: T = cast(T, raw)
+            return TransportResponse(text=response.text, status_code=response.status_code, json=payload)
 
         try:
-            transport_response.json = response.json()
+            untyped_response = response.json()
         except json.decoder.JSONDecodeError:
+            untyped_response = None
             pass
 
-        return transport_response
+        return TransportResponse(text=response.text, status_code=response.status_code, json=untyped_response)
 
     async def close(self) -> None:
         """Clean up any resources (e.g., close connections)."""

src/deepset_mcp/main.py

@@ -134,84 +134,6 @@ async def validate_pipeline(yaml_configuration: str) -> str:
     return response
 
 
-# @mcp.tool()
-# async def list_pipeline_templates() -> str:
-#     """Retrieves a list of pipeline templates to build AI applications like RAG or Agents."""
-#     workspace = get_workspace()
-#     async with AsyncDeepsetClient() as client:
-#         resp = await client.request(
-#           endpoint=f"v1/workspaces/{workspace}/pipeline_templates?limit=100&page_number=1&field=created_at&order=DESC"
-#         )
-#
-#         response = cast(dict[str, Any], resp.json)
-#
-#     if isinstance(response, dict) and "error" in response:
-#         return f"Error retrieving pipeline templates: {response['error']}"
-#
-#     # Extract the template data
-#     templates = response.get("data", [])
-#
-#     if not templates:
-#         return "No pipeline templates found."
-#
-#     # Format the response as requested
-#     formatted_output = []
-#     for template in templates:
-#         name = template.get("pipeline_name", "Unnamed Template")
-#         description = template.get("description", "No description available")
-#
-#         formatted_output.append(f"<template>\n{name}\n\n{description}\n</template>\n")
-#
-#     # Join all template entries and return
-#     return "\n".join(formatted_output)
-#
-#
-# @mcp.tool()
-# async def get_pipeline_template(template_name: str) -> str:
-#     """Retrieves a specific pipeline template by name and returns its YAML configurations.
-#
-#     Parameters
-#     ----------
-#     template_name : str
-#         The name of the pipeline template to retrieve
-#
-#     Returns
-#     -------
-#     str
-#         A formatted string containing the template's query and indexing YAML (if available)
-#     """
-#     workspace = get_workspace()
-#
-#     async with AsyncDeepsetClient() as client:
-#         resp = await client.request(endpoint=f"v1/workspaces/{workspace}/pipeline_templates/{template_name}")
-#
-#         response = cast(dict[str, Any], resp.json)
-#
-#     if isinstance(response, dict) and "error" in response:
-#         return f"Error retrieving pipeline template: {response['error']}"
-#
-#     # Extract the YAML data
-#     query_yaml = response.get("query_yaml")
-#     indexing_yaml = response.get("indexing_yaml")
-#     template_name = response.get("name", "Unnamed Template")
-#
-#     # Format the response
-#     formatted_output = [f"# Pipeline Template: {template_name}\n"]
-#
-#     # Add query YAML if available
-#     if query_yaml:
-#         formatted_output.append("## QUERY PIPELINE YAML\n```yaml\n" + query_yaml + "\n```\n")
-#     else:
-#         formatted_output.append("## QUERY PIPELINE YAML\nNo query pipeline YAML available for this template.\n")
-#
-#     # Add indexing YAML if available
-#     if indexing_yaml:
-#         formatted_output.append("## INDEXING PIPELINE YAML\n```yaml\n" + indexing_yaml + "\n```\n")
-#     else:
-#         formatted_output.append("## INDEXING PIPELINE YAML\nNo indexing pipeline YAML available for this template.\n")
-#
-#     # Join all sections and return
-#     return "\n".join(formatted_output)
 #
 #
 # @mcp.tool()

test/integration/test_integration_haystack_service_resource.py

@@ -23,3 +23,17 @@ async def test_get_component_schemas(
 
     assert isinstance(response, dict)
     assert "component_schema" in response
+
+
+@pytest.mark.asyncio
+async def test_get_component_input_output(
+    haystack_service_resource: HaystackServiceResource,
+) -> None:
+    """Test for getting component input/output schema."""
+    response = await haystack_service_resource.get_component_input_output("Agent")
+
+    assert isinstance(response, dict)
+    assert "name" in response
+    assert response["name"] == "Agent"
+    assert "input" in response
+    assert "output" in response