Add admin-only Homebox MCP connector

Load an allowlisted subset of homebox-mcp tools (read + safe writes, no
deletes) over stdio and compile them into a separate admin agent, so
non-admins never see them. Each tool also re-checks is_admin per call.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tests/unit/test_homebox_mcp.py

@@ -0,0 +1,122 @@
+"""Tests for the admin-only Homebox MCP tool wrapping and agent routing."""
+
+import os
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from langchain_core.tools import StructuredTool
+from quart import Quart
+
+from utils.homebox_mcp import (
+    ADMIN_ONLY_MESSAGE,
+    HOMEBOX_TOOL_ALLOWLIST,
+    load_homebox_tools,
+    select_homebox_tools,
+)
+
+ADMIN = {"configurable": {"is_admin": True}}
+NON_ADMIN = {"configurable": {"is_admin": False}}
+
+
+def _mcp_tool(name: str, result=None, error: Exception | None = None):
+    """Build a stub shaped like a langchain-mcp-adapters tool."""
+    calls = []
+
+    async def call(runtime=None, **arguments):
+        calls.append(arguments)
+        if error:
+            raise error
+        return (
+            result if result is not None else ([{"type": "text", "text": "ok"}], None)
+        )
+
+    tool = StructuredTool(
+        name=name,
+        description=f"{name} description",
+        args_schema={
+            "type": "object",
+            "properties": {"query": {"type": "string"}},
+        },
+        coroutine=call,
+        response_format="content_and_artifact",
+    )
+    return tool, calls
+
+
+class TestSelectHomeboxTools:
+    def test_filters_to_allowlist_and_prefixes(self):
+        stubs = [_mcp_tool(n)[0] for n in ("search_items", "delete_item", "get_item")]
+        tools = select_homebox_tools(stubs)
+        assert [t.name for t in tools] == ["homebox_search_items", "homebox_get_item"]
+
+    def test_never_exposes_deletes(self):
+        assert not any(name.startswith("delete_") for name in HOMEBOX_TOOL_ALLOWLIST)
+
+    def test_truncates_long_descriptions(self):
+        stub, _ = _mcp_tool("search_items")
+        stub.description = "word " * 200
+        [tool] = select_homebox_tools([stub])
+        assert len(tool.description) <= 300
+
+
+class TestAdminGuard:
+    async def test_admin_call_delegates_and_flattens_content(self):
+        stub, calls = _mcp_tool(
+            "search_items",
+            result=(
+                [{"type": "text", "text": "Drill"}, {"type": "text", "text": "Garage"}],
+                None,
+            ),
+        )
+        [tool] = select_homebox_tools([stub])
+        out = await tool.ainvoke({"query": "drill"}, config=ADMIN)
+        assert out == "Drill\nGarage"
+        assert calls == [{"query": "drill"}]
+
+    async def test_non_admin_is_refused_without_calling_server(self):
+        stub, calls = _mcp_tool("search_items")
+        [tool] = select_homebox_tools([stub])
+        assert (
+            await tool.ainvoke({"query": "drill"}, config=NON_ADMIN)
+            == ADMIN_ONLY_MESSAGE
+        )
+        assert await tool.ainvoke({"query": "drill"}) == ADMIN_ONLY_MESSAGE
+        assert calls == []
+
+    async def test_server_errors_are_returned_as_text(self):
+        stub, _ = _mcp_tool("search_items", error=RuntimeError("cannot reach Homebox"))
+        [tool] = select_homebox_tools([stub])
+        out = await tool.ainvoke({"query": "drill"}, config=ADMIN)
+        assert out == "Homebox error: cannot reach Homebox"
+
+    def test_config_not_exposed_in_schema(self):
+        [tool] = select_homebox_tools([_mcp_tool("search_items")[0]])
+        assert list(tool.tool_call_schema["properties"]) == ["query"]
+
+
+class TestLoadHomeboxTools:
+    async def test_returns_empty_when_unconfigured(self):
+        with patch.dict(os.environ, {"HOMEBOX_URL": "", "HOMEBOX_TOKEN": ""}):
+            assert await load_homebox_tools() == []
+
+    async def test_returns_empty_when_server_fails(self):
+        env = {
+            "HOMEBOX_URL": "http://homebox",
+            "HOMEBOX_TOKEN": "hb_x",
+            "HOMEBOX_MCP_COMMAND": "/nonexistent/homebox-mcp",
+        }
+        with patch.dict(os.environ, env):
+            assert await load_homebox_tools() == []
+
+
+class TestGetAgentFor:
+    async def test_routes_admin_to_admin_agent(self):
+        from blueprints.conversation.logic import get_agent_for
+
+        app = Quart(__name__)
+        app.main_agent, app.admin_agent = "main", "admin"
+        admin = SimpleNamespace(is_admin=lambda: True)
+        user = SimpleNamespace(is_admin=lambda: False)
+        async with app.app_context():
+            assert get_agent_for(admin) == "admin"
+            assert get_agent_for(user) == "main"