"""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.github_allowed_users: from .authz import GitHubAllowlistMiddleware mcp.add_middleware(GitHubAllowlistMiddleware(settings.github_allowed_users)) # ------------------------------------------------------------------ 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) } ) if "github" not in mode: return verifier from fastmcp.server.auth.providers.github import GitHubProvider github = 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"], ) if verifier is None: return github from fastmcp.server.auth import MultiAuth return MultiAuth(server=github, verifiers=[verifier], base_url=settings.public_url)