From ca0f89a671699a2ab4682d627210aa42482a681a Mon Sep 17 00:00:00 2001 From: Chewbaccalakis Date: Sun, 16 Aug 2026 22:47:26 -0700 Subject: [PATCH] update --- QUICKSTART.md | 101 ++++++ bookstack_mcp/__init__.py | 0 bookstack_mcp/__main__.py | 48 +++ bookstack_mcp/authz.py | 97 +++++ bookstack_mcp/client.py | 121 +++++++ bookstack_mcp/config.py | 143 ++++++++ bookstack_mcp/server.py | 412 ++++++++++++++++++++++ deploy/authelia/configuration.snippet.yml | 120 +++++++ deploy/lxc/bookstack-mcp.service | 43 +++ deploy/lxc/install.sh | 150 ++++++++ pyproject.toml | 17 + test_smoke.py | 179 ++++++++++ 12 files changed, 1431 insertions(+) create mode 100644 QUICKSTART.md create mode 100644 bookstack_mcp/__init__.py create mode 100644 bookstack_mcp/__main__.py create mode 100644 bookstack_mcp/authz.py create mode 100644 bookstack_mcp/client.py create mode 100644 bookstack_mcp/config.py create mode 100644 bookstack_mcp/server.py create mode 100644 deploy/authelia/configuration.snippet.yml create mode 100644 deploy/lxc/bookstack-mcp.service create mode 100755 deploy/lxc/install.sh create mode 100644 pyproject.toml create mode 100644 test_smoke.py diff --git a/QUICKSTART.md b/QUICKSTART.md new file mode 100644 index 0000000..49224b0 --- /dev/null +++ b/QUICKSTART.md @@ -0,0 +1,101 @@ +# Quickstart (Proxmox / LXD container) + +Full detail is in `README.md`. This is the short path. + +## 0. Where to put this + +Unzip **anywhere except `/opt/bookstack-mcp`** — that path is the install +*target*, and the installer overwrites it. `/opt/src` is a good home: + +```bash +cd /opt/src # or wherever you unzipped +unzip bookstack-mcp.zip +cd bookstack-mcp +``` + +You should now be able to run this with no errors — all three paths must exist: + +```bash +ls pyproject.toml bookstack_mcp/server.py deploy/lxc/install.sh +``` + +## 1. Install + +Requires Debian 12+ or Ubuntu 24.04+ (Python 3.11 or newer). + +```bash +bash deploy/lxc/install.sh +``` + +This creates the `bookstack-mcp` service user, a venv at +`/opt/bookstack-mcp/venv`, config at `/etc/bookstack-mcp/env`, and a systemd +unit. Re-run it any time to upgrade; your config file is preserved. + +## 2. Get a BookStack API token + +Your instance uses OIDC, so the service user never needs to log in: + +1. Settings → Roles → create a role with **Access System API**. +2. Settings → Users → create a user, assign that role. +3. Edit that user as an admin → **API Tokens** section at the bottom → + **Create Token**. Copy the Token ID and Token Secret. + +If you have `OIDC_REMOVE_FROM_GROUPS=true`, do not log this user in via SSO — +group sync will strip the role. See README step 1. + +## 3. Configure + +```bash +nano /etc/bookstack-mcp/env +``` + +Minimum to get running with Claude Code only: + +```ini +BOOKSTACK_URL=https://wiki.yourdomain.com +BOOKSTACK_TOKEN_ID=... +BOOKSTACK_TOKEN_SECRET=... +BOOKSTACK_READ_ONLY=true +MCP_AUTH_MODE=token +MCP_STATIC_TOKENS= +``` + +Add claude.ai later by switching to `MCP_AUTH_MODE=oidc+token` and filling in +the OIDC block — see `deploy/authelia/configuration.snippet.yml`. + +## 4. Start + +```bash +systemctl start bookstack-mcp +systemctl status bookstack-mcp +journalctl -u bookstack-mcp -f +``` + +Verify locally. A 401 with a `www-authenticate` header is the correct answer: + +```bash +curl -i http://127.0.0.1:8080/mcp +``` + +## 5. Reverse proxy + +Point `https://mcp.yourdomain.com` at this container on **port 8080**. + +Three things that will otherwise cost you an evening: + +- Proxy the **whole hostname**, not just `/mcp`. OAuth discovery lives at + `/.well-known/*` on the domain root. +- **Disable buffering** (`proxy_buffering off`, long `proxy_read_timeout`). + Streamable HTTP holds a stream open. +- **Bypass Authelia forward-auth** for this host. Claude can't hold session + cookies; the MCP server does its own OAuth. + +## 6. Connect + +```bash +claude mcp add --transport http bookstack https://mcp.yourdomain.com/mcp \ + --header "Authorization: Bearer YOUR_STATIC_TOKEN" --scope user +``` + +For claude.ai: Settings → Connectors → Add custom connector → +`https://mcp.yourdomain.com/mcp`. Requires `MCP_AUTH_MODE=oidc+token`. diff --git a/bookstack_mcp/__init__.py b/bookstack_mcp/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/bookstack_mcp/__main__.py b/bookstack_mcp/__main__.py new file mode 100644 index 0000000..b28bbe7 --- /dev/null +++ b/bookstack_mcp/__main__.py @@ -0,0 +1,48 @@ +"""Entrypoint: python -m bookstack_mcp""" + +from __future__ import annotations + +import logging +import sys + +from .config import Settings +from .server import create_server + + +def build_app(): + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s %(message)s", + ) + settings = Settings.from_env() + mcp = create_server(settings) + kwargs = {} + if settings.allowed_hosts: + kwargs["allowed_hosts"] = settings.allowed_hosts + kwargs["host_origin_protection"] = "auto" + return mcp.http_app(path=settings.mcp_path, **kwargs), settings + + +def main() -> None: + import uvicorn + + try: + app, settings = build_app() + except RuntimeError as exc: + print(f"Configuration error: {exc}", file=sys.stderr) + raise SystemExit(2) from exc + + logging.getLogger("bookstack-mcp").info( + "Serving BookStack MCP on http://%s:%s%s (auth=%s, read_only=%s)", + settings.host, + settings.port, + settings.mcp_path, + settings.auth_mode, + settings.read_only, + ) + uvicorn.run(app, host=settings.host, port=settings.port, proxy_headers=True, + forwarded_allow_ips="*", access_log=False) + + +if __name__ == "__main__": + main() diff --git a/bookstack_mcp/authz.py b/bookstack_mcp/authz.py new file mode 100644 index 0000000..f184614 --- /dev/null +++ b/bookstack_mcp/authz.py @@ -0,0 +1,97 @@ +"""Restrict access to specific users or groups from the identity provider.""" + +from __future__ import annotations + +import logging +from typing import Any + +from fastmcp.exceptions import ToolError +from fastmcp.server.dependencies import get_access_token +from fastmcp.server.middleware import Middleware, MiddlewareContext + +logger = logging.getLogger("bookstack-mcp.authz") + +# Claims that different providers use for the human-readable username. +# GitHub -> login; Keycloak/Authentik/Authelia/Zitadel -> preferred_username. +USERNAME_CLAIMS = ("login", "preferred_username", "username", "email", "sub") + + +def _dig(claims: dict[str, Any], path: str) -> Any: + """Look up a possibly nested claim, e.g. 'resource_access.mcp.roles'.""" + value: Any = claims + for part in path.split("."): + if not isinstance(value, dict) or part not in value: + return None + value = value[part] + return value + + +class IdentityAllowlistMiddleware(Middleware): + """Reject requests from identities outside the configured allowlist. + + Two independent checks, both optional: + * allowed_users -- username must appear in this list + * required_groups -- at least one group/role claim must match + + Static machine tokens carry no identity claims. Those are already trusted + because the operator minted them by hand, so they pass through untouched. + """ + + def __init__( + self, + allowed_users: list[str] | None = None, + required_groups: list[str] | None = None, + groups_claim: str = "groups", + ) -> None: + self.allowed_users = {u.lower() for u in (allowed_users or [])} + self.required_groups = {g.lower() for g in (required_groups or [])} + self.groups_claim = groups_claim + + @staticmethod + def _username(claims: dict[str, Any]) -> str | None: + for claim in USERNAME_CLAIMS: + value = claims.get(claim) + if isinstance(value, str) and value: + return value + return None + + def _groups(self, claims: dict[str, Any]) -> set[str]: + raw = _dig(claims, self.groups_claim) + if isinstance(raw, str): + raw = [raw] + if not isinstance(raw, list): + return set() + return {str(g).lower() for g in raw} + + def _check(self) -> None: + token = get_access_token() + if token is None: + return + + claims = token.claims or {} + username = self._username(claims) + if username is None: + return # static token, no identity to check + + if self.allowed_users and username.lower() not in self.allowed_users: + logger.warning("Denied BookStack MCP access to user %r (not in allowlist)", username) + raise ToolError(f"User '{username}' is not authorised to use this BookStack server.") + + if self.required_groups: + groups = self._groups(claims) + if not (groups & self.required_groups): + logger.warning( + "Denied BookStack MCP access to user %r (groups %s, need one of %s)", + username, sorted(groups), sorted(self.required_groups), + ) + raise ToolError( + f"User '{username}' is not in a group permitted to use this BookStack server." + ) + + async def on_call_tool(self, context: MiddlewareContext, call_next): + self._check() + return await call_next(context) + + async def on_list_tools(self, context: MiddlewareContext, call_next): + self._check() + return await call_next(context) diff --git a/bookstack_mcp/client.py b/bookstack_mcp/client.py new file mode 100644 index 0000000..2b7f310 --- /dev/null +++ b/bookstack_mcp/client.py @@ -0,0 +1,121 @@ +"""Thin async wrapper around the BookStack REST API.""" + +from __future__ import annotations + +import re +from html import unescape +from typing import Any + +import httpx + +from .config import Settings + + +class BookStackError(RuntimeError): + """Raised when BookStack returns an error response.""" + + +_TAG_RE = re.compile(r"<[^>]+>") + + +def strip_html(value: str | None) -> str: + """Turn BookStack's preview HTML into plain text.""" + if not value: + return "" + text = _TAG_RE.sub(" ", value) + return re.sub(r"\s+", " ", unescape(text)).strip() + + +class BookStackClient: + def __init__(self, settings: Settings) -> None: + self._settings = settings + self._client = httpx.AsyncClient( + base_url=settings.api_base, + timeout=settings.request_timeout, + verify=settings.bookstack_verify_tls, + headers={ + "Authorization": ( + f"Token {settings.bookstack_token_id}:{settings.bookstack_token_secret}" + ), + "Accept": "application/json", + "User-Agent": "bookstack-mcp/1.0", + }, + follow_redirects=True, + ) + + async def aclose(self) -> None: + await self._client.aclose() + + async def request( + self, + method: str, + path: str, + *, + params: dict[str, Any] | None = None, + json: dict[str, Any] | None = None, + raw: bool = False, + ) -> Any: + try: + response = await self._client.request( + method, path, params=params, json=json + ) + except httpx.HTTPError as exc: # network/DNS/TLS failures + raise BookStackError(f"Could not reach BookStack at {self._settings.bookstack_url}: {exc}") from exc + + if response.status_code == 401: + raise BookStackError( + "BookStack rejected the API token (401). Check BOOKSTACK_TOKEN_ID / " + "BOOKSTACK_TOKEN_SECRET and that the token's user has the " + "'Access System API' permission." + ) + if response.status_code == 403: + raise BookStackError( + "BookStack denied access (403). The token's user lacks permission for " + "this content or action." + ) + if response.status_code == 404: + raise BookStackError("Not found in BookStack (404).") + if response.status_code == 429: + raise BookStackError( + "BookStack rate limit hit (429). Default is 180 requests/minute per user." + ) + if response.status_code >= 400: + detail = response.text[:500] + raise BookStackError(f"BookStack returned HTTP {response.status_code}: {detail}") + + if raw: + return response.text + if not response.content: + return {} + return response.json() + + # ---------- listing helpers ---------- + + async def list_endpoint( + self, + endpoint: str, + *, + count: int = 25, + offset: int = 0, + sort: str | None = None, + filters: dict[str, str] | None = None, + ) -> dict[str, Any]: + params: dict[str, Any] = {"count": min(max(count, 1), 500), "offset": max(offset, 0)} + if sort: + params["sort"] = sort + for key, value in (filters or {}).items(): + params[f"filter[{key}]"] = value + return await self.request("GET", endpoint, params=params) + + async def search(self, query: str, *, count: int, page: int) -> dict[str, Any]: + return await self.request( + "GET", + "/search", + params={"query": query, "count": min(max(count, 1), 100), "page": max(page, 1)}, + ) + + async def export(self, kind: str, item_id: int, fmt: str) -> str: + return await self.request("GET", f"/{kind}/{item_id}/export/{fmt}", raw=True) + + def web_url(self, path: str) -> str: + return f"{self._settings.bookstack_url}/{path.lstrip('/')}" diff --git a/bookstack_mcp/config.py b/bookstack_mcp/config.py new file mode 100644 index 0000000..b91ec2e --- /dev/null +++ b/bookstack_mcp/config.py @@ -0,0 +1,143 @@ +"""Configuration, loaded from environment variables.""" + +from __future__ import annotations + +import os +from dataclasses import dataclass, field + + +def _bool(name: str, default: bool = False) -> bool: + raw = os.environ.get(name) + if raw is None: + return default + return raw.strip().lower() in {"1", "true", "yes", "on"} + + +def _list(name: str) -> list[str]: + raw = os.environ.get(name, "") + return [item.strip() for item in raw.split(",") if item.strip()] + + +def _require(name: str) -> str: + value = os.environ.get(name, "").strip() + if not value: + raise RuntimeError(f"Missing required environment variable: {name}") + return value + + +@dataclass(frozen=True) +class Settings: + # --- BookStack --- + bookstack_url: str + bookstack_token_id: str + bookstack_token_secret: str + bookstack_verify_tls: bool = True + request_timeout: float = 30.0 + + # --- Behaviour --- + read_only: bool = True + allow_delete: bool = False + max_content_chars: int = 40_000 + + # --- Transport --- + host: str = "0.0.0.0" + port: int = 8080 + mcp_path: str = "/mcp" + + # --- Auth (client -> this MCP server) --- + auth_mode: str = "none" # none | token | github | github+token + static_tokens: list[str] = field(default_factory=list) + public_url: str = "" + github_client_id: str = "" + github_client_secret: str = "" + allowed_hosts: list[str] = field(default_factory=list) + + # --- OIDC (your own IdP: Authentik, Keycloak, Authelia, Zitadel, ...) --- + oidc_config_url: str = "" + oidc_client_id: str = "" + oidc_client_secret: str = "" + oidc_verify_id_token: bool = False + oidc_groups_claim: str = "groups" + oidc_scopes: list[str] = field(default_factory=list) + + # --- Who may use the server (applies to both GitHub and OIDC) --- + allowed_users: list[str] = field(default_factory=list) + required_groups: list[str] = field(default_factory=list) + + @classmethod + def from_env(cls) -> "Settings": + auth_mode = os.environ.get("MCP_AUTH_MODE", "none").strip().lower() + valid_modes = { + "none", "token", + "github", "github+token", + "oidc", "oidc+token", + } + if auth_mode not in valid_modes: + raise RuntimeError( + f"MCP_AUTH_MODE must be one of {'|'.join(sorted(valid_modes))}, got {auth_mode!r}" + ) + + settings = cls( + bookstack_url=_require("BOOKSTACK_URL").rstrip("/"), + bookstack_token_id=_require("BOOKSTACK_TOKEN_ID"), + bookstack_token_secret=_require("BOOKSTACK_TOKEN_SECRET"), + bookstack_verify_tls=_bool("BOOKSTACK_VERIFY_TLS", True), + request_timeout=float(os.environ.get("BOOKSTACK_TIMEOUT", "30")), + read_only=_bool("BOOKSTACK_READ_ONLY", True), + allow_delete=_bool("BOOKSTACK_ALLOW_DELETE", False), + max_content_chars=int(os.environ.get("MCP_MAX_CONTENT_CHARS", "40000")), + host=os.environ.get("MCP_HOST", "0.0.0.0"), + port=int(os.environ.get("MCP_PORT", "8080")), + mcp_path=os.environ.get("MCP_PATH", "/mcp"), + auth_mode=auth_mode, + static_tokens=_list("MCP_STATIC_TOKENS"), + public_url=os.environ.get("MCP_PUBLIC_URL", "").rstrip("/"), + github_client_id=os.environ.get("GITHUB_CLIENT_ID", ""), + github_client_secret=os.environ.get("GITHUB_CLIENT_SECRET", ""), + allowed_hosts=_list("MCP_ALLOWED_HOSTS"), + oidc_config_url=os.environ.get("OIDC_CONFIG_URL", "").strip(), + oidc_client_id=os.environ.get("OIDC_CLIENT_ID", ""), + oidc_client_secret=os.environ.get("OIDC_CLIENT_SECRET", ""), + oidc_verify_id_token=_bool("OIDC_VERIFY_ID_TOKEN", False), + oidc_groups_claim=os.environ.get("OIDC_GROUPS_CLAIM", "groups"), + oidc_scopes=_list("OIDC_SCOPES"), + # GITHUB_ALLOWED_USERS is the old name, still honoured. + allowed_users=[ + u.lower() for u in (_list("MCP_ALLOWED_USERS") or _list("GITHUB_ALLOWED_USERS")) + ], + required_groups=[g.lower() for g in _list("MCP_REQUIRED_GROUPS")], + ) + settings.validate() + return settings + + def validate(self) -> None: + if "token" in self.auth_mode and not self.static_tokens: + raise RuntimeError("MCP_AUTH_MODE includes 'token' but MCP_STATIC_TOKENS is empty") + if "github" in self.auth_mode: + if not (self.github_client_id and self.github_client_secret): + raise RuntimeError( + "MCP_AUTH_MODE includes 'github' but GITHUB_CLIENT_ID / " + "GITHUB_CLIENT_SECRET are not set" + ) + if "oidc" in self.auth_mode: + if not self.oidc_config_url: + raise RuntimeError( + "MCP_AUTH_MODE includes 'oidc' but OIDC_CONFIG_URL is not set. " + "This is your provider's discovery document, e.g. " + "https://auth.example.com/application/o/bookstack-mcp/.well-known/openid-configuration" + ) + if not (self.oidc_client_id and self.oidc_client_secret): + raise RuntimeError( + "MCP_AUTH_MODE includes 'oidc' but OIDC_CLIENT_ID / " + "OIDC_CLIENT_SECRET are not set" + ) + if ("github" in self.auth_mode or "oidc" in self.auth_mode) and not self.public_url: + raise RuntimeError( + "MCP_PUBLIC_URL is required for OAuth (e.g. https://mcp.example.com)" + ) + if not self.mcp_path.startswith("/"): + raise RuntimeError("MCP_PATH must start with '/'") + + @property + def api_base(self) -> str: + return f"{self.bookstack_url}/api" diff --git a/bookstack_mcp/server.py b/bookstack_mcp/server.py new file mode 100644 index 0000000..8e45ef0 --- /dev/null +++ b/bookstack_mcp/server.py @@ -0,0 +1,412 @@ +"""BookStack MCP server (streamable HTTP).""" + +from __future__ import annotations + +import logging +from contextlib import asynccontextmanager +from typing import Annotated, Any, Literal + +from fastmcp import FastMCP +from fastmcp.exceptions import ToolError +from pydantic import Field + +from .client import BookStackClient, BookStackError, strip_html +from .config import Settings + +logger = logging.getLogger("bookstack-mcp") + +INSTRUCTIONS = """\ +Tools for reading and writing a BookStack wiki. + +BookStack organises content as: shelves > books > chapters > pages. Pages are the +only items that hold text. + +Start with `search_content` for almost any question. It accepts BookStack search +syntax, e.g.: + firewall rules plain terms + {type:page} vpn restrict to pages + [server=prod] filter by tag name/value + "exact phrase" exact match + {updated_by:me} pages the API user last touched + +`search_content` returns ids; pass a page id to `get_page` to read the full text. +""" + + +def _truncate(text: str, limit: int) -> str: + if len(text) <= limit: + return text + return ( + text[:limit] + + f"\n\n[... truncated: {len(text) - limit} more characters. " + "Read the page in BookStack for the full content.]" + ) + + +def _tags(raw: list[dict[str, Any]] | None) -> list[str]: + out = [] + for tag in raw or []: + name, value = tag.get("name"), tag.get("value") + out.append(f"{name}={value}" if value else str(name)) + return out + + +def _tag_payload(tags: list[str] | None) -> list[dict[str, str]] | None: + if not tags: + return None + payload = [] + for tag in tags: + name, _, value = tag.partition("=") + payload.append({"name": name.strip(), "value": value.strip()}) + return payload + + +def create_server(settings: Settings) -> FastMCP: + client = BookStackClient(settings) + + @asynccontextmanager + async def lifespan(_server: FastMCP): + try: + yield + finally: + await client.aclose() + + auth = _build_auth(settings) + + mcp = FastMCP( + name="bookstack", + instructions=INSTRUCTIONS, + version="1.0.0", + auth=auth, + lifespan=lifespan, + ) + + if settings.allowed_users or settings.required_groups: + from .authz import IdentityAllowlistMiddleware + + mcp.add_middleware( + IdentityAllowlistMiddleware( + allowed_users=settings.allowed_users, + required_groups=settings.required_groups, + groups_claim=settings.oidc_groups_claim, + ) + ) + elif settings.auth_mode not in {"none", "token"}: + logger.warning( + "OAuth is enabled but MCP_ALLOWED_USERS and MCP_REQUIRED_GROUPS are both " + "empty: anyone who can log in to your identity provider can use this server." + ) + + # ------------------------------------------------------------------ read + + @mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": True}) + async def search_content( + query: Annotated[str, Field(description="BookStack search query. Supports filters like {type:page} and [tag=value].")], + count: Annotated[int, Field(description="Results per page (1-100).", ge=1, le=100)] = 15, + page: Annotated[int, Field(description="1-based page number.", ge=1)] = 1, + ) -> dict[str, Any]: + """Search across all BookStack content the API user can see. + + This is the main entry point: use it to locate books, chapters, and pages + before reading them with get_page / get_book / get_chapter. + """ + data = await _guard(client.search(query, count=count, page=page)) + results = [] + for item in data.get("data", []): + preview = item.get("preview_html") or {} + results.append( + { + "type": item.get("type"), + "id": item.get("id"), + "name": strip_html(preview.get("name")) or item.get("name"), + "excerpt": strip_html(preview.get("content"))[:400], + "url": item.get("url"), + "book_id": item.get("book_id"), + "chapter_id": item.get("chapter_id"), + "tags": _tags(item.get("tags")), + "updated_at": item.get("updated_at"), + } + ) + return {"total": data.get("total", len(results)), "page": page, "results": results} + + @mcp.tool(annotations={"readOnlyHint": True}) + async def get_page( + page_id: Annotated[int, Field(description="Numeric page id, from search_content or get_book.")], + format: Annotated[Literal["markdown", "plaintext", "html"], Field(description="Content format to return.")] = "markdown", + ) -> dict[str, Any]: + """Read the full content of a single page.""" + meta = await _guard(client.request("GET", f"/pages/{page_id}")) + try: + content = await client.export("pages", page_id, format) + except BookStackError: + content = meta.get("markdown") or meta.get("html") or "" + return { + "id": meta.get("id"), + "name": meta.get("name"), + "book_id": meta.get("book_id"), + "chapter_id": meta.get("chapter_id"), + "slug": meta.get("slug"), + "url": client.web_url(f"books/{meta.get('book_slug', '')}/page/{meta.get('slug', '')}"), + "tags": _tags(meta.get("tags")), + "updated_at": meta.get("updated_at"), + "format": format, + "content": _truncate(content, settings.max_content_chars), + } + + @mcp.tool(annotations={"readOnlyHint": True}) + async def list_books( + count: Annotated[int, Field(description="How many books to return.", ge=1, le=100)] = 50, + offset: Annotated[int, Field(description="Skip this many books.", ge=0)] = 0, + sort: Annotated[str, Field(description="Sort field, e.g. name, -updated_at, -created_at.")] = "name", + ) -> dict[str, Any]: + """List the books in the wiki. Useful for getting oriented.""" + data = await _guard(client.list_endpoint("/books", count=count, offset=offset, sort=sort)) + return { + "total": data.get("total"), + "books": [ + { + "id": b.get("id"), + "name": b.get("name"), + "description": (b.get("description") or "")[:300], + "updated_at": b.get("updated_at"), + } + for b in data.get("data", []) + ], + } + + @mcp.tool(annotations={"readOnlyHint": True}) + async def get_book( + book_id: Annotated[int, Field(description="Numeric book id.")], + ) -> dict[str, Any]: + """Get a book with its full table of contents (chapters and pages).""" + data = await _guard(client.request("GET", f"/books/{book_id}")) + contents = [] + for node in data.get("contents", []): + entry = { + "type": node.get("type"), + "id": node.get("id"), + "name": node.get("name"), + } + if node.get("type") == "chapter": + entry["pages"] = [ + {"id": p.get("id"), "name": p.get("name")} for p in node.get("pages", []) + ] + contents.append(entry) + return { + "id": data.get("id"), + "name": data.get("name"), + "description": data.get("description"), + "tags": _tags(data.get("tags")), + "updated_at": data.get("updated_at"), + "contents": contents, + } + + @mcp.tool(annotations={"readOnlyHint": True}) + async def get_chapter( + chapter_id: Annotated[int, Field(description="Numeric chapter id.")], + ) -> dict[str, Any]: + """Get a chapter and the list of pages inside it.""" + data = await _guard(client.request("GET", f"/chapters/{chapter_id}")) + return { + "id": data.get("id"), + "name": data.get("name"), + "book_id": data.get("book_id"), + "description": data.get("description"), + "tags": _tags(data.get("tags")), + "pages": [{"id": p.get("id"), "name": p.get("name")} for p in data.get("pages", [])], + } + + @mcp.tool(annotations={"readOnlyHint": True}) + async def list_shelves( + count: Annotated[int, Field(description="How many shelves to return.", ge=1, le=100)] = 50, + ) -> dict[str, Any]: + """List bookshelves, the top level of the BookStack hierarchy.""" + data = await _guard(client.list_endpoint("/shelves", count=count, sort="name")) + return { + "total": data.get("total"), + "shelves": [ + {"id": s.get("id"), "name": s.get("name"), "description": (s.get("description") or "")[:300]} + for s in data.get("data", []) + ], + } + + @mcp.tool(annotations={"readOnlyHint": True}) + async def list_recent_pages( + count: Annotated[int, Field(description="How many pages to return.", ge=1, le=100)] = 25, + ) -> dict[str, Any]: + """List the most recently updated pages. Good for 'what changed lately'.""" + data = await _guard(client.list_endpoint("/pages", count=count, sort="-updated_at")) + return { + "pages": [ + { + "id": p.get("id"), + "name": p.get("name"), + "book_id": p.get("book_id"), + "chapter_id": p.get("chapter_id"), + "updated_at": p.get("updated_at"), + } + for p in data.get("data", []) + ] + } + + # ----------------------------------------------------------------- write + + if not settings.read_only: + + @mcp.tool(annotations={"readOnlyHint": False, "destructiveHint": False}) + async def create_page( + name: Annotated[str, Field(description="Title of the new page.")], + markdown: Annotated[str, Field(description="Page body in Markdown.")], + book_id: Annotated[int | None, Field(description="Book to create the page in. Provide this or chapter_id.")] = None, + chapter_id: Annotated[int | None, Field(description="Chapter to create the page in. Provide this or book_id.")] = None, + tags: Annotated[list[str] | None, Field(description="Tags as 'name' or 'name=value' strings.")] = None, + ) -> dict[str, Any]: + """Create a new page in a book or chapter.""" + if not book_id and not chapter_id: + raise ToolError("Provide either book_id or chapter_id.") + payload: dict[str, Any] = {"name": name, "markdown": markdown} + if chapter_id: + payload["chapter_id"] = chapter_id + else: + payload["book_id"] = book_id + if tag_payload := _tag_payload(tags): + payload["tags"] = tag_payload + data = await _guard(client.request("POST", "/pages", json=payload)) + return {"id": data.get("id"), "name": data.get("name"), "slug": data.get("slug")} + + @mcp.tool(annotations={"readOnlyHint": False, "destructiveHint": True, "idempotentHint": True}) + async def update_page( + page_id: Annotated[int, Field(description="Page to update.")], + name: Annotated[str | None, Field(description="New title. Omit to leave unchanged.")] = None, + markdown: Annotated[str | None, Field(description="New body in Markdown. This REPLACES the whole page body.")] = None, + tags: Annotated[list[str] | None, Field(description="Replacement tag list. Omit to leave unchanged.")] = None, + ) -> dict[str, Any]: + """Update an existing page. Supplying markdown replaces the entire body.""" + payload: dict[str, Any] = {} + if name is not None: + payload["name"] = name + if markdown is not None: + payload["markdown"] = markdown + if tag_payload := _tag_payload(tags): + payload["tags"] = tag_payload + if not payload: + raise ToolError("Nothing to update: provide name, markdown, or tags.") + data = await _guard(client.request("PUT", f"/pages/{page_id}", json=payload)) + return {"id": data.get("id"), "name": data.get("name"), "updated_at": data.get("updated_at")} + + @mcp.tool(annotations={"readOnlyHint": False, "destructiveHint": False}) + async def create_book( + name: Annotated[str, Field(description="Title of the new book.")], + description: Annotated[str, Field(description="Short description.")] = "", + ) -> dict[str, Any]: + """Create a new book.""" + data = await _guard( + client.request("POST", "/books", json={"name": name, "description": description}) + ) + return {"id": data.get("id"), "name": data.get("name")} + + @mcp.tool(annotations={"readOnlyHint": False, "destructiveHint": False}) + async def create_chapter( + book_id: Annotated[int, Field(description="Book that will contain the chapter.")], + name: Annotated[str, Field(description="Title of the new chapter.")], + description: Annotated[str, Field(description="Short description.")] = "", + ) -> dict[str, Any]: + """Create a new chapter inside a book.""" + data = await _guard( + client.request( + "POST", + "/chapters", + json={"book_id": book_id, "name": name, "description": description}, + ) + ) + return {"id": data.get("id"), "name": data.get("name")} + + if settings.allow_delete: + + @mcp.tool(annotations={"readOnlyHint": False, "destructiveHint": True}) + async def delete_page( + page_id: Annotated[int, Field(description="Page to send to the recycle bin.")], + ) -> dict[str, Any]: + """Move a page to the BookStack recycle bin (recoverable).""" + await _guard(client.request("DELETE", f"/pages/{page_id}")) + return {"deleted": page_id, "note": "Moved to the recycle bin; recoverable in BookStack."} + + return mcp + + +async def _guard(awaitable): + """Convert BookStack API errors into clean MCP tool errors.""" + try: + return await awaitable + except BookStackError as exc: + raise ToolError(str(exc)) from exc + + +def _build_auth(settings: Settings): + mode = settings.auth_mode + if mode == "none": + logger.warning( + "MCP_AUTH_MODE=none: this server is unauthenticated. Only run it this way " + "on a trusted network or behind another authenticating proxy." + ) + return None + + from fastmcp.server.auth.providers.jwt import StaticTokenVerifier + + verifier = None + if "token" in mode: + verifier = StaticTokenVerifier( + tokens={ + token: {"client_id": f"static-{i}", "scopes": ["bookstack"]} + for i, token in enumerate(settings.static_tokens) + } + ) + + oauth = None + + if "github" in mode: + from fastmcp.server.auth.providers.github import GitHubProvider + + oauth = GitHubProvider( + client_id=settings.github_client_id, + client_secret=settings.github_client_secret, + base_url=settings.public_url, + redirect_path="/auth/callback", + required_scopes=["read:user"], + ) + + elif "oidc" in mode: + from fastmcp.server.auth.oidc_proxy import OIDCProxy + + # OIDC_SCOPES overrides this entirely when a provider needs something + # specific. Otherwise: identify the user, ask for groups only when group + # checks are configured (some providers reject unknown scopes), and + # request offline_access so the connection survives token expiry. + if settings.oidc_scopes: + scopes = settings.oidc_scopes + else: + scopes = ["openid", "profile", "email", "offline_access"] + if settings.required_groups: + scopes.insert(3, settings.oidc_groups_claim) + + oauth = OIDCProxy( + config_url=settings.oidc_config_url, + client_id=settings.oidc_client_id, + client_secret=settings.oidc_client_secret, + base_url=settings.public_url, + redirect_path="/auth/callback", + required_scopes=scopes, + # Some providers (Authelia, Okta) issue opaque access tokens that + # can't be validated as JWTs. Setting OIDC_VERIFY_ID_TOKEN=true + # verifies the ID token instead, which is always a signed JWT. + verify_id_token=settings.oidc_verify_id_token, + ) + + if oauth is None: + return verifier + if verifier is None: + return oauth + + from fastmcp.server.auth import MultiAuth + + return MultiAuth(server=oauth, verifiers=[verifier], base_url=settings.public_url) diff --git a/deploy/authelia/configuration.snippet.yml b/deploy/authelia/configuration.snippet.yml new file mode 100644 index 0000000..18db792 --- /dev/null +++ b/deploy/authelia/configuration.snippet.yml @@ -0,0 +1,120 @@ +# Authelia configuration for the BookStack MCP server. +# +# Merge these blocks into your existing configuration.yml. Tested against +# Authelia 4.39+, where ID Token claims changed (see the claims_policies note). +# +# Generate the client id and secret first: +# +# docker run --rm authelia/authelia:latest \ +# authelia crypto rand --length 72 --charset rfc3986 +# +# docker run --rm authelia/authelia:latest \ +# authelia crypto hash generate pbkdf2 --variant sha512 \ +# --random --random.length 72 --random.charset rfc3986 +# +# The second command prints BOTH a "Random Password" and a "Digest". +# Digest -> client_secret below +# Random Password -> OIDC_CLIENT_SECRET in /etc/bookstack-mcp/env +# Do not put the digest in the MCP server's env file. + +identity_providers: + oidc: + # ----------------------------------------------------------------------- + # 1. Claims policy + # + # Authelia 4.39 stopped putting non-standard claims in the ID Token by + # default. The MCP server reads identity from the ID Token (because + # Authelia issues opaque access tokens), so groups and preferred_username + # have to be put back explicitly. Without this, MCP_REQUIRED_GROUPS will + # never match and every tool call gets denied. + # ----------------------------------------------------------------------- + claims_policies: + mcp_claims: + id_token: + - 'groups' + - 'preferred_username' + - 'email' + - 'email_verified' + - 'name' + + # ----------------------------------------------------------------------- + # 2. Lifespan + # + # Authelia's default refresh_token lifespan is 90m. A claude.ai connector + # that sits idle longer than that has to be re-authorised by hand, which + # gets old fast. A long refresh token with a short access token is the + # right shape for a background integration. + # ----------------------------------------------------------------------- + lifespans: + custom: + mcp: + access_token: '1h' + id_token: '1h' + refresh_token: '30d' + + clients: + - client_id: 'REPLACE_WITH_GENERATED_ID' + client_name: 'BookStack MCP' + client_secret: 'REPLACE_WITH_PBKDF2_DIGEST' + public: false + + # one_factor if you don't want to be prompted for 2FA on reconnect. + authorization_policy: 'two_factor' + + claims_policy: 'mcp_claims' + lifespan: 'mcp' + + # Claude re-authorises on token refresh. Explicit consent every time is + # painful; pre-configured remembers the grant for the given duration. + consent_mode: 'pre-configured' + pre_configured_consent_duration: '1 month' + + require_pkce: true + pkce_challenge_method: 'S256' + + redirect_uris: + - 'https://mcp.example.com/auth/callback' + + # offline_access is what gets you a refresh token. Without it the + # connector dies when the access token expires. + scopes: + - 'openid' + - 'profile' + - 'email' + - 'groups' + - 'offline_access' + + grant_types: + - 'authorization_code' + - 'refresh_token' + + response_types: + - 'code' + + # Authelia's default and what the spec requires. Leave it alone unless + # you see 'invalid_client' at the token endpoint, in which case try + # 'client_secret_post'. + token_endpoint_auth_method: 'client_secret_basic' + + # Both are already the default; stated explicitly because they matter. + # 'none' means access tokens stay opaque rather than being JWTs, which + # is exactly why the MCP server needs OIDC_VERIFY_ID_TOKEN=true. + access_token_signed_response_alg: 'none' + userinfo_signed_response_alg: 'none' + +# --------------------------------------------------------------------------- +# 3. Access control bypass <-- the one everybody misses +# +# If mcp.example.com is behind Authelia's forward-auth on your reverse proxy, +# Claude will be served an HTML login page instead of the MCP endpoint and the +# connector will fail with a parse error or a redirect loop. Claude cannot do +# cookie-based forward auth; the MCP server runs its own OAuth flow against +# Authelia instead. So the host must bypass forward-auth entirely. +# +# This is not a security hole: the MCP server refuses every unauthenticated +# request itself, and Authelia is still the thing deciding who gets a token. +# --------------------------------------------------------------------------- +access_control: + rules: + - domain: 'mcp.example.com' + policy: 'bypass' diff --git a/deploy/lxc/bookstack-mcp.service b/deploy/lxc/bookstack-mcp.service new file mode 100644 index 0000000..2e44645 --- /dev/null +++ b/deploy/lxc/bookstack-mcp.service @@ -0,0 +1,43 @@ +[Unit] +Description=BookStack MCP Server +Documentation=https://github.com/your/bookstack-mcp +After=network-online.target +Wants=network-online.target + +[Service] +Type=exec +User=bookstack-mcp +Group=bookstack-mcp + +EnvironmentFile=/etc/bookstack-mcp/env +WorkingDirectory=/opt/bookstack-mcp +ExecStart=/opt/bookstack-mcp/venv/bin/python -m bookstack_mcp + +Restart=on-failure +RestartSec=5s +TimeoutStopSec=20s + +# --- Hardening --- +# These all work in an unprivileged Proxmox/LXD container. If the service fails +# with a 226/NAMESPACE or 227/SECCOMP status on an older kernel, comment out the +# block below and it will start fine -- it only reduces defence in depth. +NoNewPrivileges=yes +PrivateTmp=yes +ProtectSystem=strict +ProtectHome=yes +ProtectControlGroups=yes +RestrictSUIDSGID=yes +RemoveIPC=yes +LockPersonality=yes +MemoryDenyWriteExecute=yes +RestrictAddressFamilies=AF_INET AF_INET6 AF_UNIX +CapabilityBoundingSet= +AmbientCapabilities= + +# Journald captures stdout/stderr: journalctl -u bookstack-mcp -f +StandardOutput=journal +StandardError=journal +SyslogIdentifier=bookstack-mcp + +[Install] +WantedBy=multi-user.target diff --git a/deploy/lxc/install.sh b/deploy/lxc/install.sh new file mode 100755 index 0000000..bb0103e --- /dev/null +++ b/deploy/lxc/install.sh @@ -0,0 +1,150 @@ +#!/usr/bin/env bash +# +# Install the BookStack MCP server into a Debian/Ubuntu LXC container. +# Run as root, from inside the container, in the directory containing pyproject.toml: +# +# bash deploy/lxc/install.sh +# +set -euo pipefail + +APP_DIR=/opt/bookstack-mcp +CONF_DIR=/etc/bookstack-mcp +SVC_USER=bookstack-mcp +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" + +log() { printf '\033[1;34m==>\033[0m %s\n' "$*"; } +die() { printf '\033[1;31mERROR:\033[0m %s\n' "$*" >&2; exit 1; } + +# Find the project root. Normally this script lives at deploy/lxc/install.sh so +# the root is two levels up, but people often copy the tree around flat, so +# check the obvious candidates rather than assuming. +find_src() { + local candidate + for candidate in "$SCRIPT_DIR/../.." "$SCRIPT_DIR/.." "$SCRIPT_DIR" "$PWD"; do + if [[ -f "$candidate/pyproject.toml" && -d "$candidate/bookstack_mcp" ]]; then + (cd "$candidate" && pwd) + return 0 + fi + done + return 1 +} + +if ! SRC_DIR="$(find_src)"; then + die "Can't find the project source (needs pyproject.toml and a bookstack_mcp/ directory). + Looked in: $SCRIPT_DIR/../.. , $SCRIPT_DIR/.. , $SCRIPT_DIR , $PWD + + Get the whole project into the container, then run the script from its root: + git clone /opt/src/bookstack-mcp + cd /opt/src/bookstack-mcp + bash deploy/lxc/install.sh" +fi + +[[ $EUID -eq 0 ]] || die "Run this as root (or with sudo)." + +# The script copies SRC_DIR into APP_DIR, so they must not be the same place. +if [[ "$SRC_DIR" == "$APP_DIR" ]]; then + die "The source is $APP_DIR, which is also the install target. + Keep the source somewhere else and let the installer copy it in: + mkdir -p /opt/src && mv $APP_DIR /opt/src/bookstack-mcp + cd /opt/src/bookstack-mcp && bash deploy/lxc/install.sh" +fi + +log "Installing from $SRC_DIR" + +# --- Python version check ------------------------------------------------- +# Needs 3.11+. Debian 12/13 and Ubuntu 24.04 are fine; Debian 11 and +# Ubuntu 22.04 ship something older and need a newer python installed first. +log "Checking Python version" +if ! command -v python3 >/dev/null; then + die "python3 not installed. Try: apt install -y python3 python3-venv" +fi +PY_OK=$(python3 -c 'import sys; print(1 if sys.version_info >= (3,11) else 0)') +if [[ "$PY_OK" != "1" ]]; then + die "Python $(python3 -V | cut -d' ' -f2) found, but 3.11+ is required. + Debian 12+/Ubuntu 24.04+ work out of the box. On older releases install a + newer python3 first (e.g. via deadsnakes) and re-run." +fi + +# --- Dependencies --------------------------------------------------------- +log "Installing system packages" +export DEBIAN_FRONTEND=noninteractive +# A single broken third-party repo shouldn't abort the whole install, so the +# update is advisory; only the install itself is allowed to fail the script. +apt-get update -qq || log "apt update reported errors, continuing" +apt-get install -y -qq python3-venv python3-pip ca-certificates >/dev/null + +# --- Service user --------------------------------------------------------- +if ! id -u "$SVC_USER" >/dev/null 2>&1; then + log "Creating service user $SVC_USER" + useradd --system --home-dir "$APP_DIR" --shell /usr/sbin/nologin "$SVC_USER" +fi + +# --- Application ---------------------------------------------------------- +log "Installing application to $APP_DIR" +mkdir -p "$APP_DIR" +rm -rf "$APP_DIR/bookstack_mcp" +cp -r "$SRC_DIR/bookstack_mcp" "$APP_DIR/" +cp "$SRC_DIR/pyproject.toml" "$APP_DIR/" + +if [[ ! -d "$APP_DIR/venv" ]]; then + log "Creating virtualenv" + python3 -m venv "$APP_DIR/venv" +fi + +log "Installing Python dependencies (this takes a minute)" +"$APP_DIR/venv/bin/pip" install --quiet --upgrade pip +"$APP_DIR/venv/bin/pip" install --quiet "$APP_DIR" + +chown -R "$SVC_USER:$SVC_USER" "$APP_DIR" + +# --- Configuration -------------------------------------------------------- +mkdir -p "$CONF_DIR" +if [[ ! -f "$CONF_DIR/env" ]]; then + log "Creating $CONF_DIR/env from the template" + # systemd EnvironmentFile does not understand quotes or inline comments the + # way a shell does, so strip comments and blank lines out of the example. + grep -vE '^\s*(#|$)' "$SRC_DIR/.env.example" > "$CONF_DIR/env" + NEW_CONFIG=1 +else + log "Keeping existing $CONF_DIR/env" + NEW_CONFIG=0 +fi +chown root:"$SVC_USER" "$CONF_DIR/env" +chmod 640 "$CONF_DIR/env" + +# --- systemd -------------------------------------------------------------- +log "Installing systemd unit" +UNIT_SRC="" +for candidate in "$SRC_DIR/deploy/lxc/bookstack-mcp.service" "$SCRIPT_DIR/bookstack-mcp.service"; do + [[ -f "$candidate" ]] && { UNIT_SRC="$candidate"; break; } +done +[[ -n "$UNIT_SRC" ]] || die "Can't find bookstack-mcp.service" +install -m 644 "$UNIT_SRC" /etc/systemd/system/bookstack-mcp.service +systemctl daemon-reload +systemctl enable bookstack-mcp >/dev/null + +echo +log "Install complete." +echo +if [[ "$NEW_CONFIG" == "1" ]]; then + cat <=3.4,<4", + "httpx>=0.27", + "uvicorn>=0.30", +] + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["bookstack_mcp"] diff --git a/test_smoke.py b/test_smoke.py new file mode 100644 index 0000000..97df460 --- /dev/null +++ b/test_smoke.py @@ -0,0 +1,179 @@ +"""Smoke test: run the MCP server against a fake BookStack API.""" + +import asyncio +import os +import threading +import time + +import uvicorn +from starlette.applications import Starlette +from starlette.responses import JSONResponse, PlainTextResponse +from starlette.routing import Route + +# ---------------- fake BookStack ---------------- + +EXPECTED_AUTH = "Token tok-id:tok-secret" + + +def check(request): + assert request.headers.get("authorization") == EXPECTED_AUTH, "bad auth header" + + +async def search(request): + check(request) + q = request.query_params.get("query") + return JSONResponse({ + "data": [{ + "id": 42, "name": "Firewall Rules", "type": "page", "book_id": 3, + "chapter_id": 0, "url": "https://wiki.test/books/net/page/firewall-rules", + "preview_html": {"name": "Firewall Rules", + "content": "Inbound 443 allowed & logged"}, + "tags": [{"name": "env", "value": "prod"}], + "updated_at": "2026-08-01T10:00:00Z", + }], + "total": 1, "query": q, + }) + + +async def page_detail(request): + check(request) + return JSONResponse({ + "id": 42, "name": "Firewall Rules", "book_id": 3, "chapter_id": 0, + "slug": "firewall-rules", "book_slug": "net", + "tags": [{"name": "env", "value": "prod"}], + "updated_at": "2026-08-01T10:00:00Z", "html": "

hi

", + }) + + +async def page_export(request): + check(request) + return PlainTextResponse("# Firewall Rules\n\nInbound 443 allowed.\n") + + +async def books(request): + check(request) + return JSONResponse({"data": [{"id": 3, "name": "Networking", + "description": "net docs", + "updated_at": "2026-08-01T10:00:00Z"}], "total": 1}) + + +async def book_detail(request): + check(request) + return JSONResponse({ + "id": 3, "name": "Networking", "description": "net docs", "tags": [], + "updated_at": "2026-08-01T10:00:00Z", + "contents": [ + {"type": "chapter", "id": 7, "name": "Edge", + "pages": [{"id": 42, "name": "Firewall Rules"}]}, + {"type": "page", "id": 43, "name": "Loose Page"}, + ], + }) + + +async def create_page(request): + check(request) + body = await request.json() + assert body["markdown"], "missing markdown" + return JSONResponse({"id": 99, "name": body["name"], "slug": "new-page"}) + + +async def forbidden(request): + return JSONResponse({"error": "nope"}, status_code=403) + + +fake = Starlette(routes=[ + Route("/api/search", search), + Route("/api/pages/42", page_detail), + Route("/api/pages/42/export/markdown", page_export), + Route("/api/pages", books, methods=["GET"]), + Route("/api/pages", create_page, methods=["POST"]), + Route("/api/books", books, methods=["GET"]), + Route("/api/books/3", book_detail), + Route("/api/shelves", forbidden), +]) + + +def serve(app, port): + uvicorn.run(app, host="127.0.0.1", port=port, log_level="error") + + +async def main(): + threading.Thread(target=serve, args=(fake, 9911), daemon=True).start() + + os.environ.update({ + "BOOKSTACK_URL": "http://127.0.0.1:9911", + "BOOKSTACK_TOKEN_ID": "tok-id", + "BOOKSTACK_TOKEN_SECRET": "tok-secret", + "BOOKSTACK_READ_ONLY": "false", + "MCP_AUTH_MODE": "token", + "MCP_STATIC_TOKENS": "secret-abc,secret-def", + "MCP_PORT": "9912", + }) + + from bookstack_mcp.__main__ import build_app + app, settings = build_app() + threading.Thread(target=serve, args=(app, 9912), daemon=True).start() + time.sleep(2.5) + + from fastmcp import Client + from fastmcp.client.auth import BearerAuth + + url = "http://127.0.0.1:9912/mcp" + + # 1. unauthenticated must fail + try: + async with Client(url) as c: + await c.list_tools() + print("FAIL: unauthenticated access was allowed") + except Exception as exc: + print(f"PASS auth required -> {type(exc).__name__}") + + # 2. authenticated + async with Client(url, auth=BearerAuth("secret-abc")) as c: + tools = sorted(t.name for t in await c.list_tools()) + print("PASS tools:", tools) + + r = await c.call_tool("search_content", {"query": "{type:page} firewall"}) + hit = r.data["results"][0] + assert hit["name"] == "Firewall Rules", hit + assert hit["excerpt"] == "Inbound 443 allowed & logged", hit["excerpt"] + assert hit["tags"] == ["env=prod"], hit["tags"] + print("PASS search_content ->", hit["name"], "|", hit["excerpt"], "|", hit["tags"]) + + r = await c.call_tool("get_page", {"page_id": 42}) + assert r.data["content"].startswith("# Firewall Rules"), r.data + print("PASS get_page ->", repr(r.data["content"][:30]), "url:", r.data["url"]) + + r = await c.call_tool("get_book", {"book_id": 3}) + assert r.data["contents"][0]["pages"][0]["id"] == 42 + print("PASS get_book ->", r.data["contents"]) + + r = await c.call_tool("create_page", { + "name": "New", "markdown": "# hi", "book_id": 3, "tags": ["env=dev", "draft"]}) + assert r.data["id"] == 99 + print("PASS create_page ->", r.data) + + # error surfacing: shelves endpoint returns 403 + try: + await c.call_tool("list_shelves", {}) + print("FAIL: 403 was not surfaced") + except Exception as exc: + print("PASS 403 surfaced ->", str(exc)[:80]) + + # validation + try: + await c.call_tool("create_page", {"name": "x", "markdown": "y"}) + print("FAIL: missing parent not caught") + except Exception as exc: + print("PASS missing parent ->", str(exc)[:60]) + + # 3. wrong token rejected + try: + async with Client(url, auth=BearerAuth("wrong")) as c: + await c.list_tools() + print("FAIL: bad token accepted") + except Exception as exc: + print(f"PASS bad token rejected -> {type(exc).__name__}") + + +asyncio.run(main())