.env.example

@@ -144,6 +144,13 @@ SIMBAKIT_URL=http://simbakit:8585
 HOMEBOX_URL=https://homebox.example.com
 HOMEBOX_TOKEN=
 
+# BookStack wiki (admin-only agent tools via bookstack-mcp: read, create, update; no deletes)
+# API token from BookStack Profile -> API Tokens (user needs the "Access System API" permission).
+# Leave empty to disable.
+BOOKSTACK_BASE_URL=https://bookstack.example.com
+BOOKSTACK_TOKEN_ID=
+BOOKSTACK_TOKEN_SECRET=
+
 # Obsidian Configuration (headless sync)
 # Obsidian account credentials (used for `ob login` on container startup)
 OBSIDIAN_EMAIL=your-obsidian-email

CLAUDE.md

@@ -90,6 +90,8 @@ make redeploy    # git pull + scripts/deploy.sh (build new color, wait healthy,
   - `request.py` - Paperless-NGX API client
   - `obsidian_service.py`, `mealie_service.py`, `ynab_service.py`, `s3_client.py`, `simbakit_service.py` - integrations with Obsidian notes, Mealie recipes, YNAB budgets, S3, and SimbaKit
   - `homebox_mcp.py` - Homebox inventory tools via the `homebox-mcp` stdio MCP server; admin-only (compiled into `app.admin_agent`, picked per user by `get_agent_for`)
+  - `bookstack_mcp.py` - BookStack wiki tools (read/create/update, no deletes) via the `bookstack-mcp` npm stdio server; admin-only like Homebox
+  - `mcp_tools.py` - shared allowlist + admin-gate wrapping for stdio MCP integrations
 - `scripts/` - Administrative and utility scripts
   - `add_user.py` - Create users manually
   - `user_message_stats.py` - User message statistics
@@ -156,7 +158,7 @@ See `.env.example`. Key ones:
 - `TWILIO_*`, `ALLOWED_IMESSAGE_NUMBERS`, `ALLOWED_WHATSAPP_NUMBERS` - iMessage/WhatsApp webhooks
 - `MAILGUN_*`, `EMAIL_HMAC_SECRET`, `FERNET_KEY` - email integration
 - `SENDBLUE_*` - SMS sending for scheduled messages
-- `YNAB_*`, `MEALIE_*`, `OBSIDIAN_*`, `IMMICH_*`, `SIMBAKIT_URL`, `HOMEBOX_URL`/`HOMEBOX_TOKEN` - household integrations
+- `YNAB_*`, `MEALIE_*`, `OBSIDIAN_*`, `IMMICH_*`, `SIMBAKIT_URL`, `HOMEBOX_URL`/`HOMEBOX_TOKEN`, `BOOKSTACK_BASE_URL`/`BOOKSTACK_TOKEN_ID`/`BOOKSTACK_TOKEN_SECRET` - household integrations
 
 ## Documentation
 

Dockerfile

@@ -18,6 +18,9 @@ ENV PATH="/root/.local/bin:$PATH"
 # Homebox MCP server (stdio), pinned so the tool surface can't drift
 RUN uv tool install homebox-mcp==1.1.0
 
+# BookStack MCP server (stdio), pinned for the same reason
+RUN npm install -g bookstack-mcp@5.6.0
+
 # Copy dependency files
 COPY pyproject.toml ./
 

app.py

@@ -30,6 +30,7 @@ import blueprints.users.models
 from config.db import TORTOISE_CONFIG
 from metrics import render_metrics
 from blueprints.conversation.agents import build_agent
+from utils.bookstack_mcp import load_bookstack_tools
 from utils.homebox_mcp import load_homebox_tools
 
 # Load environment variables
@@ -107,11 +108,9 @@ async def lifespan():
         checkpointer = AsyncSqliteSaver(conn)
         await checkpointer.setup()
         app.main_agent = build_agent(checkpointer)
-        homebox_tools = await load_homebox_tools()
+        admin_tools = await load_homebox_tools() + await load_bookstack_tools()
         app.admin_agent = (
-            build_agent(checkpointer, homebox_tools)
+            build_agent(checkpointer, admin_tools) if admin_tools else app.main_agent
-            if homebox_tools
-            else app.main_agent
         )
         logging.info(f"Agent checkpointer ready at {checkpoint_path}")
 

blueprints/conversation/agents.py

@@ -1172,7 +1172,7 @@ def build_agent(checkpointer=None, extra_tools=()):
     """Compile and return the LangGraph agent.
 
     ``extra_tools`` are appended to the shared tool list, e.g. admin-only
-    Homebox tools for the admin agent.
+    Homebox and BookStack tools for the admin agent.
     """
     return create_react_agent(
         model=model_with_fallback,

blueprints/conversation/logic.py

@@ -25,7 +25,7 @@ def tool_result_text(tool_name: str, output) -> str:
 
 
 def get_agent_for(user: blueprints.users.models.User):
-    """Admins get the agent with admin-only tools (Homebox); everyone else the main one."""
+    """Admins get the agent with admin-only tools (Homebox, BookStack); everyone else the main one."""
     if user.is_admin():
         return getattr(current_app, "admin_agent", current_app.main_agent)
     return current_app.main_agent

blueprints/conversation/prompts.py

@@ -52,6 +52,9 @@ Always use these tools when asked about meals, recipes, what's for dinner, or sh
 HOME INVENTORY (Homebox Integration):
 If homebox_* tools are available, use them for questions about household items: where something is stored, what's in a location, warranties, and maintenance. Search before creating items to avoid duplicates.
 
+WIKI (BookStack Integration):
+If bookstack_* tools are available, use them to look up and write household documentation in the BookStack wiki. Search before creating books, chapters, or pages to avoid duplicates, and prefer updating an existing page over creating a near-copy. Write page content as markdown.
+
 NOTES & RESEARCH (Obsidian Integration):
 You have access to Ryan's Obsidian vault through the Obsidian integration. When users ask about research, personal notes, or information that might be stored in markdown files, use the appropriate Obsidian tools:
 - Use obsidian_search_notes to search through your vault for relevant information

docker-compose.yml

@@ -65,6 +65,9 @@ x-raggr-common: &raggr-common
     - CRONICLE_API_KEY=${CRONICLE_API_KEY}
     - HOMEBOX_URL=${HOMEBOX_URL}
     - HOMEBOX_TOKEN=${HOMEBOX_TOKEN}
+    - BOOKSTACK_BASE_URL=${BOOKSTACK_BASE_URL}
+    - BOOKSTACK_TOKEN_ID=${BOOKSTACK_TOKEN_ID}
+    - BOOKSTACK_TOKEN_SECRET=${BOOKSTACK_TOKEN_SECRET}
   depends_on:
     postgres:
       condition: service_healthy

tests/unit/test_bookstack_mcp.py

@@ -0,0 +1,103 @@
+"""Tests for the admin-only BookStack MCP tool wrapping."""
+
+import os
+from unittest.mock import patch
+
+from langchain_core.tools import StructuredTool
+
+from utils.bookstack_mcp import (
+    ADMIN_ONLY_MESSAGE,
+    BOOKSTACK_TOOL_ALLOWLIST,
+    load_bookstack_tools,
+    select_bookstack_tools,
+)
+
+ADMIN = {"configurable": {"is_admin": True}}
+NON_ADMIN = {"configurable": {"is_admin": False}}
+
+
+def _mcp_tool(name: str, 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 ([{"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 TestSelectBookstackTools:
+    def test_filters_to_allowlist_and_prefixes(self):
+        stubs = [
+            _mcp_tool(n)[0] for n in ("search_content", "delete_page", "update_page")
+        ]
+        tools = select_bookstack_tools(stubs)
+        assert [t.name for t in tools] == [
+            "bookstack_search_content",
+            "bookstack_update_page",
+        ]
+
+    def test_never_exposes_destructive_tools(self):
+        for name in BOOKSTACK_TOOL_ALLOWLIST:
+            assert not name.startswith("delete_")
+            assert name not in ("permanently_delete", "restore_deleted")
+
+    def test_allows_reads_and_writes(self):
+        assert "get_page" in BOOKSTACK_TOOL_ALLOWLIST
+        assert "create_page" in BOOKSTACK_TOOL_ALLOWLIST
+        assert "update_page" in BOOKSTACK_TOOL_ALLOWLIST
+
+
+class TestAdminGuard:
+    async def test_admin_call_delegates(self):
+        stub, calls = _mcp_tool("search_content")
+        [tool] = select_bookstack_tools([stub])
+        assert await tool.ainvoke({"query": "wifi"}, config=ADMIN) == "ok"
+        assert calls == [{"query": "wifi"}]
+
+    async def test_non_admin_is_refused_without_calling_server(self):
+        stub, calls = _mcp_tool("search_content")
+        [tool] = select_bookstack_tools([stub])
+        assert (
+            await tool.ainvoke({"query": "wifi"}, config=NON_ADMIN)
+            == ADMIN_ONLY_MESSAGE
+        )
+        assert ADMIN_ONLY_MESSAGE == "BookStack access is restricted to admin users."
+        assert calls == []
+
+    async def test_server_errors_are_returned_as_text(self):
+        stub, _ = _mcp_tool("search_content", error=RuntimeError("401 Unauthorized"))
+        [tool] = select_bookstack_tools([stub])
+        out = await tool.ainvoke({"query": "wifi"}, config=ADMIN)
+        assert out == "BookStack error: 401 Unauthorized"
+
+
+class TestLoadBookstackTools:
+    async def test_returns_empty_when_unconfigured(self):
+        env = {
+            "BOOKSTACK_BASE_URL": "https://wiki",
+            "BOOKSTACK_TOKEN_ID": "id",
+            "BOOKSTACK_TOKEN_SECRET": "",
+        }
+        with patch.dict(os.environ, env):
+            assert await load_bookstack_tools() == []
+
+    async def test_returns_empty_when_server_fails(self):
+        env = {
+            "BOOKSTACK_BASE_URL": "https://wiki",
+            "BOOKSTACK_TOKEN_ID": "id",
+            "BOOKSTACK_TOKEN_SECRET": "secret",
+            "BOOKSTACK_MCP_COMMAND": "/nonexistent/bookstack-mcp",
+        }
+        with patch.dict(os.environ, env):
+            assert await load_bookstack_tools() == []

utils/bookstack_mcp.py

@@ -0,0 +1,68 @@
+"""BookStack wiki tools, loaded from the bookstack-mcp server over stdio.
+
+Admin-only: the tools are only compiled into the admin agent, and each call
+re-checks ``is_admin`` in the run config as a second line of defense.
+"""
+
+import logging
+import os
+
+from langchain_core.tools import BaseTool, StructuredTool
+
+from utils.mcp_tools import admin_only_message, load_stdio_tools, select_tools
+
+logger = logging.getLogger(__name__)
+
+# Read + create/update only. The server exposes 43 tools with writes enabled;
+# exposing all of them would blow the local model's 32k context and allow
+# deletes, shelf/attachment management, and recycle-bin purges.
+BOOKSTACK_TOOL_ALLOWLIST = (
+    "search_content",
+    "get_books",
+    "get_book",
+    "get_chapters",
+    "get_pages",
+    "get_page",
+    "get_recent_changes",
+    "create_book",
+    "create_chapter",
+    "create_page",
+    "update_page",
+    "create_comment",
+)
+
+LABEL = "BookStack"
+ADMIN_ONLY_MESSAGE = admin_only_message(LABEL)
+
+
+def select_bookstack_tools(mcp_tools: list[BaseTool]) -> list[StructuredTool]:
+    """Keep allowlisted tools (in allowlist order) and wrap them admin-only."""
+    return select_tools(mcp_tools, BOOKSTACK_TOOL_ALLOWLIST, "bookstack", LABEL)
+
+
+async def load_bookstack_tools() -> list[StructuredTool]:
+    """Start bookstack-mcp and return the wrapped tools, or [] if unavailable."""
+    url = os.getenv("BOOKSTACK_BASE_URL")
+    token_id = os.getenv("BOOKSTACK_TOKEN_ID")
+    token_secret = os.getenv("BOOKSTACK_TOKEN_SECRET")
+    if not url or not token_id or not token_secret:
+        logger.info(
+            "BookStack MCP not configured "
+            "(BOOKSTACK_BASE_URL/BOOKSTACK_TOKEN_ID/BOOKSTACK_TOKEN_SECRET unset)"
+        )
+        return []
+
+    return await load_stdio_tools(
+        server="bookstack",
+        command=os.getenv("BOOKSTACK_MCP_COMMAND", "bookstack-mcp"),
+        env={
+            "BOOKSTACK_BASE_URL": url,
+            "BOOKSTACK_TOKEN_ID": token_id,
+            "BOOKSTACK_TOKEN_SECRET": token_secret,
+            # Server-side write gate; the allowlist above keeps deletes out.
+            "BOOKSTACK_ENABLE_WRITE": "true",
+        },
+        allowlist=BOOKSTACK_TOOL_ALLOWLIST,
+        prefix="bookstack",
+        label=LABEL,
+    )

utils/homebox_mcp.py

@@ -7,9 +7,10 @@ re-checks ``is_admin`` in the run config as a second line of defense.
 import logging
 import os
 
-from langchain_core.runnables import RunnableConfig
 from langchain_core.tools import BaseTool, StructuredTool
 
+from utils.mcp_tools import admin_only_message, load_stdio_tools, select_tools
+
 logger = logging.getLogger(__name__)
 
 # Read + safe writes only. The server exposes ~39 tools; exposing all of them
@@ -28,65 +29,13 @@ HOMEBOX_TOOL_ALLOWLIST = (
     "log_maintenance",
 )
 
-DESCRIPTION_MAX_CHARS = 300
+LABEL = "Homebox"
-ADMIN_ONLY_MESSAGE = "Homebox access is restricted to admin users."
+ADMIN_ONLY_MESSAGE = admin_only_message(LABEL)
-
-
-def _content_to_text(content) -> str:
-    """Flatten MCP content blocks into plain text for the LLM and persistence."""
-    if isinstance(content, str):
-        return content
-    if isinstance(content, (list, tuple)):
-        parts = []
-        for block in content:
-            if isinstance(block, dict):
-                parts.append(block.get("text") or f"[{block.get('type', 'content')}]")
-            else:
-                parts.append(str(block))
-        return "\n".join(parts)
-    return str(content)
-
-
-def _short_description(description: str) -> str:
-    description = " ".join((description or "").split())
-    if len(description) > DESCRIPTION_MAX_CHARS:
-        description = description[: DESCRIPTION_MAX_CHARS - 3].rstrip() + "..."
-    return description
-
-
-def wrap_admin_only(inner: BaseTool) -> StructuredTool:
-    """Rename an MCP tool to ``homebox_<name>`` and gate it on ``is_admin``."""
-
-    async def call(config: RunnableConfig, **kwargs) -> str:
-        if not (config or {}).get("configurable", {}).get("is_admin"):
-            return ADMIN_ONLY_MESSAGE
-        try:
-            result = await inner.coroutine(**kwargs)
-        except Exception as e:
-            return f"Homebox error: {e}"
-        content = result[0] if isinstance(result, tuple) else result
-        return _content_to_text(content)
-
-    return StructuredTool(
-        name=f"homebox_{inner.name}",
-        description=_short_description(inner.description),
-        args_schema=inner.args_schema,
-        coroutine=call,
-        metadata=inner.metadata,
-    )
 
 
 def select_homebox_tools(mcp_tools: list[BaseTool]) -> list[StructuredTool]:
     """Keep allowlisted tools (in allowlist order) and wrap them admin-only."""
-    by_name = {t.name: t for t in mcp_tools}
+    return select_tools(mcp_tools, HOMEBOX_TOOL_ALLOWLIST, "homebox", LABEL)
-    missing = [name for name in HOMEBOX_TOOL_ALLOWLIST if name not in by_name]
-    if missing:
-        logger.warning(f"Homebox MCP: server is missing tools {missing}")
-    return [
-        wrap_admin_only(by_name[name])
-        for name in HOMEBOX_TOOL_ALLOWLIST
-        if name in by_name
-    ]
 
 
 async def load_homebox_tools() -> list[StructuredTool]:
@@ -97,27 +46,11 @@ async def load_homebox_tools() -> list[StructuredTool]:
         logger.info("Homebox MCP not configured (HOMEBOX_URL/HOMEBOX_TOKEN unset)")
         return []
 
-    try:
+    return await load_stdio_tools(
-        from langchain_mcp_adapters.client import MultiServerMCPClient
+        server="homebox",
-
+        command=os.getenv("HOMEBOX_MCP_COMMAND", "homebox-mcp"),
-        client = MultiServerMCPClient(
+        env={"HOMEBOX_URL": url, "HOMEBOX_TOKEN": token},
-            {
+        allowlist=HOMEBOX_TOOL_ALLOWLIST,
-                "homebox": {
+        prefix="homebox",
-                    "transport": "stdio",
+        label=LABEL,
-                    "command": os.getenv("HOMEBOX_MCP_COMMAND", "homebox-mcp"),
+    )
-                    "args": [],
-                    "env": {
-                        "HOMEBOX_URL": url,
-                        "HOMEBOX_TOKEN": token,
-                        "PATH": os.environ.get("PATH", ""),
-                    },
-                }
-            }
-        )
-        tools = select_homebox_tools(await client.get_tools())
-    except Exception:
-        logger.exception("Homebox MCP failed to load; continuing without it")
-        return []
-
-    logger.info(f"Homebox MCP: loaded {len(tools)} tools")
-    return tools

utils/mcp_tools.py

@@ -0,0 +1,111 @@
+"""Shared helpers for loading admin-only agent tools from stdio MCP servers.
+
+Each integration (Homebox, BookStack, ...) picks an allowlist of server tools;
+they are renamed ``<prefix>_<name>``, get trimmed descriptions, and re-check
+``is_admin`` in the run config on every call as a second line of defense
+(the tools are only compiled into the admin agent in the first place).
+"""
+
+import logging
+import os
+
+from langchain_core.runnables import RunnableConfig
+from langchain_core.tools import BaseTool, StructuredTool
+
+logger = logging.getLogger(__name__)
+
+DESCRIPTION_MAX_CHARS = 300
+
+
+def content_to_text(content) -> str:
+    """Flatten MCP content blocks into plain text for the LLM and persistence."""
+    if isinstance(content, str):
+        return content
+    if isinstance(content, (list, tuple)):
+        parts = []
+        for block in content:
+            if isinstance(block, dict):
+                parts.append(block.get("text") or f"[{block.get('type', 'content')}]")
+            else:
+                parts.append(str(block))
+        return "\n".join(parts)
+    return str(content)
+
+
+def short_description(description: str) -> str:
+    description = " ".join((description or "").split())
+    if len(description) > DESCRIPTION_MAX_CHARS:
+        description = description[: DESCRIPTION_MAX_CHARS - 3].rstrip() + "..."
+    return description
+
+
+def admin_only_message(label: str) -> str:
+    return f"{label} access is restricted to admin users."
+
+
+def wrap_admin_only(inner: BaseTool, prefix: str, label: str) -> StructuredTool:
+    """Rename an MCP tool to ``<prefix>_<name>`` and gate it on ``is_admin``."""
+
+    async def call(config: RunnableConfig, **kwargs) -> str:
+        if not (config or {}).get("configurable", {}).get("is_admin"):
+            return admin_only_message(label)
+        try:
+            result = await inner.coroutine(**kwargs)
+        except Exception as e:
+            return f"{label} error: {e}"
+        content = result[0] if isinstance(result, tuple) else result
+        return content_to_text(content)
+
+    return StructuredTool(
+        name=f"{prefix}_{inner.name}",
+        description=short_description(inner.description),
+        args_schema=inner.args_schema,
+        coroutine=call,
+        metadata=inner.metadata,
+    )
+
+
+def select_tools(
+    mcp_tools: list[BaseTool], allowlist: tuple[str, ...], prefix: str, label: str
+) -> list[StructuredTool]:
+    """Keep allowlisted tools (in allowlist order) and wrap them admin-only."""
+    by_name = {t.name: t for t in mcp_tools}
+    missing = [name for name in allowlist if name not in by_name]
+    if missing:
+        logger.warning(f"{label} MCP: server is missing tools {missing}")
+    return [
+        wrap_admin_only(by_name[name], prefix, label)
+        for name in allowlist
+        if name in by_name
+    ]
+
+
+async def load_stdio_tools(
+    server: str,
+    command: str,
+    env: dict[str, str],
+    allowlist: tuple[str, ...],
+    prefix: str,
+    label: str,
+) -> list[StructuredTool]:
+    """Start a stdio MCP server and return its wrapped tools, or [] on failure."""
+    try:
+        from langchain_mcp_adapters.client import MultiServerMCPClient
+
+        client = MultiServerMCPClient(
+            {
+                server: {
+                    "transport": "stdio",
+                    "command": command,
+                    "args": [],
+                    "env": {**env, "PATH": os.environ.get("PATH", "")},
+                }
+            }
+        )
+        tools = select_tools(await client.get_tools(), allowlist, prefix, label)
+    except Exception:
+        logger.exception(f"{label} MCP failed to load; continuing without it")
+        return []
+
+    logger.info(f"{label} MCP: loaded {len(tools)} tools")
+    return tools