Fix issue 26 (#27)

* feat: add get_component_definition function

* test: add tests for get_component_definition function

* feat: add mcp tool for component definition

Author: mathislucka

src/deepset_mcp/main.py

@@ -3,7 +3,10 @@
 from mcp.server.fastmcp import FastMCP
 
 from deepset_mcp.api.client import AsyncDeepsetClient
-from deepset_mcp.tools.haystack_service import list_component_families as list_component_families_tool
+from deepset_mcp.tools.haystack_service import (
+    get_component_definition as get_component_definition_tool,
+    list_component_families as list_component_families_tool,
+)
 from deepset_mcp.tools.pipeline import (
     create_pipeline as create_pipeline_tool,
     get_pipeline as get_pipeline_tool,
@@ -100,6 +103,20 @@ async def list_component_families() -> str:
     return response
 
 
+@mcp.tool()
+async def get_component_definition(component_type: str) -> str:
+    """Use this to get the full definition of a specific component.
+
+    The component type is the fully qualified import path of the component class.
+    For example: haystack.components.converters.xlsx.XLSXToDocument
+    The component definition contains a description, parameters, and example usage of the component.
+    """
+    async with AsyncDeepsetClient() as client:
+        response = await get_component_definition_tool(client, component_type)
+
+    return response
+
+
 @mcp.tool()
 async def validate_pipeline(yaml_configuration: str) -> str:
     """

src/deepset_mcp/tools/haystack_service.py

@@ -2,6 +2,62 @@
 from deepset_mcp.api.protocols import AsyncClientProtocol
 
 
+async def get_component_definition(client: AsyncClientProtocol, component_type: str) -> str:
+    """Returns the definition of a specific Haystack component.
+
+    Args:
+        client: The API client to use
+        component_type: Fully qualified component type
+            (e.g. haystack.components.routers.conditional_router.ConditionalRouter)
+
+    Returns:
+        A formatted string containing the component definition
+    """
+    haystack_service = client.haystack_service()
+
+    try:
+        response = await haystack_service.get_component_schemas()
+    except UnexpectedAPIError as e:
+        return f"Failed to retrieve component definition: {e}"
+
+    components = response["component_schema"]["definitions"]["Components"]
+
+    # Find the component by its type
+    component_def = None
+    for comp in components.values():
+        if comp["properties"]["type"].get("const") == component_type:
+            component_def = comp
+            break
+
+    if not component_def:
+        return f"Component not found: {component_type}"
+
+    # Extract relevant information
+    component_type_info = component_def["properties"]["type"]
+    init_params = component_def["properties"].get("init_parameters", {}).get("properties", {})
+
+    # Format the output
+    parts = [
+        f"Component: {component_type}",
+        f"Name: {component_def.get('title', 'Unknown')}",
+        f"Family: {component_type_info.get('family', 'Unknown')}",
+        f"Family Description: {component_type_info.get('family_description', 'No description available.')}",
+        f"\nDescription:\n{component_def.get('description', 'No description available.')}\n",
+        "\nInitialization Parameters:",
+    ]
+
+    if not init_params:
+        parts.append("  No initialization parameters")
+    else:
+        for param_name, param_info in init_params.items():
+            param_type = param_info.get("_annotation", param_info.get("type", "Unknown"))
+            param_desc = param_info.get("description", "No description available.")
+            default = f" (default: {param_info['default']})" if "default" in param_info else ""
+            parts.append(f"  {param_name}: {param_type}{default}\n    {param_desc}")
+
+    return "\n".join(parts)
+
+
 async def list_component_families(client: AsyncClientProtocol) -> str:
     """Lists all Haystack component families that are available on deepset."""
     haystack_service = client.haystack_service()

test/unit/tools/test_haystack_service.py

@@ -3,7 +3,7 @@
 import pytest
 
 from deepset_mcp.api.exceptions import UnexpectedAPIError
-from deepset_mcp.tools.haystack_service import list_component_families
+from deepset_mcp.tools.haystack_service import get_component_definition, list_component_families
 from test.unit.conftest import BaseFakeClient
 
 
@@ -31,6 +31,76 @@ def haystack_service(self) -> FakeHaystackServiceResource:
         return self._resource
 
 
+@pytest.mark.asyncio
+async def test_get_component_definition_success() -> None:
+    # Sample component definition similar to the example provided
+    component_type = "haystack.components.converters.xlsx.XLSXToDocument"
+    response: dict[str, Any] = {
+        "component_schema": {
+            "definitions": {
+                "Components": {
+                    "XLSXToDocument": {
+                        "title": "XLSXToDocument",
+                        "description": "Converts XLSX files into Documents.",
+                        "properties": {
+                            "type": {
+                                "const": component_type,
+                                "family": "converters",
+                                "family_description": "Convert data into a format your pipeline can query.",
+                            },
+                            "init_parameters": {
+                                "properties": {
+                                    "sheet_name": {
+                                        "_annotation": "typing.Union[str, int, list, None]",
+                                        "description": "The name of the sheet to read.",
+                                        "default": None,
+                                    },
+                                    "table_format": {
+                                        "_annotation": "str",
+                                        "description": "The format to convert the Excel file to.",
+                                        "default": "csv",
+                                    },
+                                }
+                            },
+                        },
+                    }
+                }
+            }
+        }
+    }
+
+    resource = FakeHaystackServiceResource(get_component_schemas_response=response)
+    client = FakeClient(resource)
+    result = await get_component_definition(client, component_type)
+
+    # Check that all required information is present
+    assert component_type in result
+    assert "XLSXToDocument" in result
+    assert "converters" in result
+    assert "Convert data into a format" in result
+    assert "sheet_name" in result
+    assert "table_format" in result
+    assert "default: csv" in result
+
+
+@pytest.mark.asyncio
+async def test_get_component_definition_not_found() -> None:
+    response: dict[str, Any] = {"component_schema": {"definitions": {"Components": {}}}}
+    resource = FakeHaystackServiceResource(get_component_schemas_response=response)
+    client = FakeClient(resource)
+    result = await get_component_definition(client, "nonexistent.component")
+    assert "Component not found" in result
+
+
+@pytest.mark.asyncio
+async def test_get_component_definition_api_error() -> None:
+    resource = FakeHaystackServiceResource(exception=UnexpectedAPIError(status_code=500, message="API Error"))
+    client = FakeClient(resource)
+    result = await get_component_definition(client, "some.component")
+    assert "Failed to retrieve component definition" in result
+    assert "API Error" in result
+
+
 @pytest.mark.asyncio
 async def test_list_component_families_no_families() -> None:
     response: dict[str, Any] = {"component_schema": {"definitions": {"Components": {}}}}