feat: extend get_component_definition to include input/output schema (#31)

* feat: extend get_component_definition to include input/output schema

* test: update test data with input/output schema information

* test: extend test assertions to cover input/output schema

* feat: enhanced output schema formatting in component definition

* test: update test with full output schema example

* test: update assertions to check for detailed output schema

* fix: linting

Author: mathislucka

src/deepset_mcp/tools/haystack_service.py

@@ -36,7 +36,7 @@ async def get_component_definition(client: AsyncClientProtocol, component_type:
     component_type_info = component_def["properties"]["type"]
     init_params = component_def["properties"].get("init_parameters", {}).get("properties", {})
 
-    # Format the output
+    # Format the basic component information
     parts = [
         f"Component: {component_type}",
         f"Name: {component_def.get('title', 'Unknown')}",
@@ -55,6 +55,71 @@ async def get_component_definition(client: AsyncClientProtocol, component_type:
             default = f" (default: {param_info['default']})" if "default" in param_info else ""
             parts.append(f"  {param_name}: {param_type}{default}\n    {param_desc}")
 
+    # Fetch and add input/output information
+    try:
+        # Extract component name from the full path
+        component_name = component_type.split(".")[-1]
+        io_info = await haystack_service.get_component_input_output(component_name)
+
+        # Add Input Schema
+        parts.append("\nInput Schema:")
+        if "input" in io_info:
+            input_props = io_info["input"].get("properties", {})
+            if not input_props:
+                parts.append("  No input parameters")
+            else:
+                required = io_info["input"].get("required", [])
+                for param_name, param_info in input_props.items():
+                    req_marker = " (required)" if param_name in required else ""
+                    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}{req_marker}{default}\n    {param_desc}")
+        else:
+            parts.append("  Input schema not available")
+
+        # Add Output Schema
+        parts.append("\nOutput Schema:")
+        if "output" in io_info and isinstance(io_info["output"], dict):
+            output_info = io_info["output"]
+            if "properties" in output_info:
+                output_props = output_info.get("properties", {})
+                if not output_props:
+                    parts.append("  No output parameters")
+                else:
+                    required = output_info.get("required", [])
+                    for param_name, param_info in output_props.items():
+                        req_marker = " (required)" if param_name in required else ""
+                        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}{req_marker}{default}\n    {param_desc}")
+
+                    # Include any definitions if they exist
+                    if "definitions" in output_info:
+                        parts.append("\n  Definitions:")
+                        for def_name, def_info in output_info["definitions"].items():
+                            parts.append(f"\n    {def_name}:")
+                            if "properties" in def_info:
+                                def_required = def_info.get("required", [])
+                                for prop_name, prop_info in def_info["properties"].items():
+                                    req_marker = " (required)" if prop_name in def_required else ""
+                                    prop_type = prop_info.get("_annotation", prop_info.get("type", "Unknown"))
+                                    prop_desc = prop_info.get("description", "No description available.")
+                                    default = f" (default: {prop_info['default']})" if "default" in prop_info else ""
+                                    parts.append(
+                                        f"      {prop_name}: {prop_type}{req_marker}{default}\n        {prop_desc}"
+                                    )
+            else:
+                # Simple output schema
+                desc = output_info.get("description", "No description available.")
+                output_type = output_info.get("type", "Unknown")
+                parts.append(f"  Type: {output_type}\n  {desc}")
+        else:
+            parts.append("  Output schema not available")
+    except Exception as e:
+        parts.append(f"\nFailed to fetch input/output schema: {str(e)}")
+
     return "\n".join(parts)
 
 

test/unit/tools/test_haystack_service.py

@@ -76,7 +76,43 @@ async def test_get_component_definition_success() -> None:
         }
     }
 
-    resource = FakeHaystackServiceResource(get_component_schemas_response=response)
+    io_response = {
+        "input": {
+            "properties": {
+                "file_path": {"_annotation": "str", "description": "Path to the XLSX file", "type": "string"}
+            },
+            "required": ["file_path"],
+            "type": "object",
+        },
+        "output": {
+            "properties": {
+                "documents": {
+                    "_annotation": "typing.List[haystack.dataclasses.document.Document]",
+                    "description": "List of documents",
+                    "type": "array",
+                    "items": {"$ref": "#/definitions/Document"},
+                }
+            },
+            "required": ["documents"],
+            "type": "object",
+            "definitions": {
+                "Document": {
+                    "type": "object",
+                    "properties": {
+                        "content": {"type": "string", "description": "The content of the document"},
+                        "meta": {"type": "object", "description": "Metadata about the document"},
+                    },
+                    "required": ["content"],
+                }
+            },
+        },
+    }
+
+    class FakeHaystackServiceResourceWithIO(FakeHaystackServiceResource):
+        async def get_component_input_output(self, component_name: str) -> dict[str, Any]:
+            return io_response
+
+    resource = FakeHaystackServiceResourceWithIO(get_component_schemas_response=response)
     client = FakeClient(resource)
     result = await get_component_definition(client, component_type)
 
@@ -89,6 +125,19 @@ async def test_get_component_definition_success() -> None:
     assert "table_format" in result
     assert "default: csv" in result
 
+    # Check input/output schema information
+    assert "Input Schema:" in result
+    assert "file_path" in result
+    assert "Path to the XLSX file" in result
+    assert "(required)" in result
+    assert "Output Schema:" in result
+    assert "List of documents" in result
+    assert "documents: typing.List[haystack.dataclasses.document.Document]" in result
+    assert "Definitions:" in result
+    assert "Document:" in result
+    assert "content: string (required)" in result
+    assert "meta: object" in result
+
 
 @pytest.mark.asyncio
 async def test_get_component_definition_not_found() -> None: