feat: use memory with custom component tools (#115)

Author: mathislucka

src/deepset_mcp/api/custom_components/models.py

@@ -2,6 +2,8 @@
 
 from pydantic import BaseModel
 
+from deepset_mcp.api.shared_models import DeepsetUser
+
 
 class CustomComponentInstallation(BaseModel):
     """Model representing a custom component installation."""
@@ -12,6 +14,7 @@ class CustomComponentInstallation(BaseModel):
     created_by_user_id: str
     logs: list[dict[str, Any]]
     organization_id: str
+    user_info: DeepsetUser | None = None
 
 
 class CustomComponentInstallationList(BaseModel):

src/deepset_mcp/tool_factory.py

@@ -195,11 +195,11 @@ def get_workspace_from_env() -> str:
     ),
     "list_custom_component_installations": (
         list_custom_component_installations_tool,
-        ToolConfig(needs_client=True, needs_workspace=True),
+        ToolConfig(needs_client=True, needs_workspace=True, memory_type=MemoryType.EXPLORABLE),
     ),
     "get_latest_custom_component_installation_logs": (
         get_latest_custom_component_installation_logs_tool,
-        ToolConfig(needs_client=True, needs_workspace=True),
+        ToolConfig(needs_client=True, needs_workspace=True, memory_type=MemoryType.EXPLORABLE),
     ),
     # Non-workspace tools
     "list_component_families": (

src/deepset_mcp/tools/custom_components.py

@@ -1,13 +1,16 @@
+from deepset_mcp.api.custom_components.models import CustomComponentInstallationList
 from deepset_mcp.api.protocols import AsyncClientProtocol
 
 
-async def list_custom_component_installations(client: AsyncClientProtocol, workspace: str) -> str:
+async def list_custom_component_installations(
+    client: AsyncClientProtocol, workspace: str
+) -> CustomComponentInstallationList | str:
     """List custom component installations.
 
     :param client: The API client to use.
     :param workspace: The workspace to operate in.
 
-    :returns: A formatted string containing installation information.
+    :returns: Custom component installations or error message.
     """
     custom_components = client.custom_components(workspace)
     users = client.users()
@@ -17,58 +20,17 @@ async def list_custom_component_installations(client: AsyncClientProtocol, works
     except Exception as e:
         return f"Failed to retrieve custom component installations: {e}"
 
-    if not installations.data:
-        return "No custom component installations found."
-
-    # Format the response
-    formatted_output = [
-        f"# Custom Component Installations (showing {len(installations.data)} of {installations.total})\n"
-    ]
-
-    for install in installations.data:
-        # Try to fetch user information
-        user_info = "Unknown"
-        if install.created_by_user_id:
+    # Enrich installations with user information
+    for installation in installations.data:
+        if installation.created_by_user_id:
             try:
-                user = await users.get(install.created_by_user_id)
-                given_name = user.given_name or ""
-                family_name = user.family_name or ""
-                email = user.email or ""
-                user_info = f"{given_name} {family_name} ({email})" if email else f"{given_name} {family_name}"
-                user_info = user_info.strip()
-                if not user_info:
-                    user_info = "Unknown"
+                user = await users.get(installation.created_by_user_id)
+                installation.user_info = user
             except Exception:
-                user_info = "Unknown"
-
-        # Format installation details
-        install_details = [
-            f"## Installation {install.custom_component_id[:8]}...",
-            f"- **Status**: {install.status}",
-            f"- **Version**: {install.version}",
-            f"- **Installed by**: {user_info}",
-        ]
-
-        # Add logs if available
-        if install.logs:
-            install_details.append("\n### Recent Logs:")
-            for log in install.logs[:5]:  # Show only the first 5 logs
-                level = log.get("level", "INFO")
-                msg = log.get("msg", "No message")
-                install_details.append(f"- [{level}] {msg}")
-
-            if len(install.logs) > 5:
-                install_details.append(f"- ... and {len(install.logs) - 5} more log entries")
-
-        formatted_output.append("\n".join(install_details) + "\n")
+                # If user fetch fails, user_info remains None
+                pass
 
-    if installations.has_more:
-        formatted_output.append(
-            "*Note: There are more installations available. This listing shows only the most recent.*"
-        )
-
-    # Join all sections and return
-    return "\n".join(formatted_output)
+    return installations
 
 
 async def get_latest_custom_component_installation_logs(client: AsyncClientProtocol, workspace: str) -> str:
@@ -77,13 +39,14 @@ async def get_latest_custom_component_installation_logs(client: AsyncClientProto
     :param client: The API client to use.
     :param workspace: The workspace to operate in.
 
-    :returns: A formatted string containing the latest installation logs.
+    :returns: The latest installation logs or error message.
     """
     custom_components = client.custom_components(workspace)
 
-    logs = await custom_components.get_latest_installation_logs()
-
-    if not logs:
-        return "No installation logs found."
-
-    return f"Latest custom component installation logs:\n\n{logs}"
+    try:
+        logs = await custom_components.get_latest_installation_logs()
+        if not logs:
+            return "No installation logs found."
+        return logs
+    except Exception as e:
+        return f"Failed to retrieve latest installation logs: {e}"

test/unit/tools/test_custom_components.py

@@ -130,18 +130,35 @@ async def test_list_custom_component_installations() -> None:
 
     result = await list_custom_component_installations(client, "test-workspace")
 
-    assert "# Custom Component Installations (showing 2 of 2)" in result
-    assert "## Installation comp_123..." in result
-    assert "## Installation comp_456..." in result
-    assert "**Status**: installed" in result
-    assert "**Status**: failed" in result
-    assert "**Version**: 1.0.0" in result
-    assert "**Version**: 0.9.0" in result
-    assert "**Installed by**: John Doe (john.doe@example.com)" in result
-    assert "**Installed by**: Jane Smith (jane.smith@example.com)" in result
-    assert "[INFO] Installation complete" in result
-    assert "[ERROR] Installation failed" in result
-    assert "[DEBUG] Debug info" in result
+    assert isinstance(result, CustomComponentInstallationList)
+    assert len(result.data) == 2
+    assert result.total == 2
+    assert result.has_more is False
+
+    # Check first installation
+    first_install = result.data[0]
+    assert first_install.custom_component_id == "comp_123"
+    assert first_install.status == "installed"
+    assert first_install.version == "1.0.0"
+    assert first_install.created_by_user_id == "user_123"
+    assert len(first_install.logs) == 1
+    assert first_install.logs[0]["level"] == "INFO"
+    assert first_install.user_info is not None
+    assert first_install.user_info.given_name == "John"
+    assert first_install.user_info.family_name == "Doe"
+    assert first_install.user_info.email == "john.doe@example.com"
+
+    # Check second installation
+    second_install = result.data[1]
+    assert second_install.custom_component_id == "comp_456"
+    assert second_install.status == "failed"
+    assert second_install.version == "0.9.0"
+    assert second_install.created_by_user_id == "user_456"
+    assert len(second_install.logs) == 2
+    assert second_install.user_info is not None
+    assert second_install.user_info.given_name == "Jane"
+    assert second_install.user_info.family_name == "Smith"
+    assert second_install.user_info.email == "jane.smith@example.com"
 
 
 @pytest.mark.asyncio
@@ -162,7 +179,10 @@ async def test_list_custom_component_installations_empty() -> None:
 
     result = await list_custom_component_installations(client, "test-workspace")
 
-    assert result == "No custom component installations found."
+    assert isinstance(result, CustomComponentInstallationList)
+    assert len(result.data) == 0
+    assert result.total == 0
+    assert result.has_more is False
 
 
 @pytest.mark.asyncio
@@ -192,7 +212,10 @@ async def test_list_custom_component_installations_user_fetch_error() -> None:
 
     result = await list_custom_component_installations(client, "test-workspace")
 
-    assert "**Installed by**: Unknown" in result
+    assert isinstance(result, CustomComponentInstallationList)
+    assert len(result.data) == 1
+    assert result.data[0].created_by_user_id == "user_unknown"
+    assert result.data[0].user_info is None  # User fetch failed, so user_info should be None
 
 
 @pytest.mark.asyncio
@@ -220,7 +243,7 @@ async def test_get_latest_custom_component_installation_logs() -> None:
 
     result = await get_latest_custom_component_installation_logs(client, "test-workspace")
 
-    assert result == f"Latest custom component installation logs:\n\n{mock_logs}"
+    assert result == mock_logs
 
 
 @pytest.mark.asyncio
@@ -242,5 +265,5 @@ async def test_get_latest_custom_component_installation_logs_api_error() -> None
     )
     client = FakeClient(custom_components_resource=custom_components_resource)
 
-    with pytest.raises(UnexpectedAPIError):
-        await get_latest_custom_component_installation_logs(client, "test-workspace")
+    result = await get_latest_custom_component_installation_logs(client, "test-workspace")
+    assert result == "Failed to retrieve latest installation logs: API Error (Status Code: 500)"