Add admin-only BookStack MCP connector

Load an allowlisted subset of bookstack-mcp tools (search/read plus
create/update books, chapters, pages and comments; no deletes) over stdio
and add them to the admin agent alongside Homebox. Extract the shared
allowlist + is_admin wrapping into utils/mcp_tools.py.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-12 16:54:01 -04:00
parent e03c10cde5
commit 0b8fe7b2df
12 changed files with 319 additions and 87 deletions
+13 -80
View File
@@ -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
ADMIN_ONLY_MESSAGE = "Homebox access is restricted to admin users."
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,
)
LABEL = "Homebox"
ADMIN_ONLY_MESSAGE = admin_only_message(LABEL)
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}
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
]
return select_tools(mcp_tools, HOMEBOX_TOOL_ALLOWLIST, "homebox", LABEL)
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:
from langchain_mcp_adapters.client import MultiServerMCPClient
client = MultiServerMCPClient(
{
"homebox": {
"transport": "stdio",
"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
return await load_stdio_tools(
server="homebox",
command=os.getenv("HOMEBOX_MCP_COMMAND", "homebox-mcp"),
env={"HOMEBOX_URL": url, "HOMEBOX_TOKEN": token},
allowlist=HOMEBOX_TOOL_ALLOWLIST,
prefix="homebox",
label=LABEL,
)