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

@@ -7,7 +7,8 @@ async def get_component_definition(client: AsyncClientProtocol, component_type:
 
     Args:
         client: The API client to use
-        component_type: Fully qualified component type (e.g. haystack.components.routers.conditional_router.ConditionalRouter)
+        component_type: Fully qualified component type
+            (e.g. haystack.components.routers.conditional_router.ConditionalRouter)
 
     Returns:
         A formatted string containing the component definition
@@ -34,15 +35,15 @@ async def get_component_definition(client: AsyncClientProtocol, 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:"
+        "\nInitialization Parameters:",
     ]
 
     if not init_params:

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
 
 
@@ -35,7 +35,7 @@ def haystack_service(self) -> FakeHaystackServiceResource:
 async def test_get_component_definition_success() -> None:
     # Sample component definition similar to the example provided
     component_type = "haystack.components.converters.xlsx.XLSXToDocument"
-    response = {
+    response: dict[str, Any] = {
         "component_schema": {
             "definitions": {
                 "Components": {
@@ -46,33 +46,33 @@ async def test_get_component_definition_success() -> None:
                             "type": {
                                 "const": component_type,
                                 "family": "converters",
-                                "family_description": "Convert data into a format your pipeline can query."
+                                "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
+                                        "default": None,
                                     },
                                     "table_format": {
                                         "_annotation": "str",
                                         "description": "The format to convert the Excel file to.",
-                                        "default": "csv"
-                                    }
+                                        "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
@@ -85,13 +85,7 @@ async def test_get_component_definition_success() -> None:
 
 @pytest.mark.asyncio
 async def test_get_component_definition_not_found() -> None:
-    response = {
-        "component_schema": {
-            "definitions": {
-                "Components": {}
-            }
-        }
-    }
+    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")