"""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) # Two different things that are easy to conflate: # scopes -> what we ASK the IdP for # required_scopes -> what must be present on every incoming token # FastMCP's issued token only carries the scopes the IdP echoes back in # its token response. If a requested scope isn't echoed (Authelia does # not always echo offline_access), enforcing the full request list # rejects every call with invalid_token, and the client loops # refreshing. So validation defaults to nothing unless explicitly set. required = settings.oidc_required_scopes or [] oauth = OIDCProxy( config_url=settings.oidc_config_url, extra_authorize_params={"scope": " ".join(scopes)}, client_id=settings.oidc_client_id, client_secret=settings.oidc_client_secret, base_url=settings.public_url, redirect_path="/auth/callback", required_scopes=required, # 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, # RFC 8707 resource indicators. Authelia validates the forwarded # 'resource' against the client's audience whitelist and returns # invalid_target if it isn't listed. Either whitelist it there or # set OIDC_FORWARD_RESOURCE=false to stop sending it. forward_resource=settings.oidc_forward_resource, ) 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)