"""MCP Server for Docmost."""
import os
import asyncio
from typing import Any
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from .client import DocmostClient, DocmostConfig
server = Server("docmost")
client: DocmostClient | None = None
def get_client() -> DocmostClient:
"""Get or create Docmost client."""
global client
if client is None:
config = DocmostConfig(
base_url=os.environ.get("DOCMOST_URL", "https://your-docmost-instance.com"),
auth_token=os.environ["DOCMOST_AUTH_TOKEN"],
)
client = DocmostClient(config)
return client
def format_page_info(page: dict) -> str:
"""Format page info for display."""
lines = [
f"📄 **{page.get('title', 'Untitled')}**",
f"- ID: {page.get('id')}",
f"- Slug: {page.get('slugId')}",
f"- Space ID: {page.get('spaceId')}",
f"- Created: {page.get('createdAt')}",
f"- Updated: {page.get('updatedAt')}",
]
if page.get("parentPageId"):
lines.append(f"- Parent: {page.get('parentPageId')}")
return "\n".join(lines)
def extract_text_from_content(content: dict) -> str:
"""Extract plain text from Docmost content structure."""
if not content:
return ""
def extract_node(node: dict) -> str:
text_parts = []
if node.get("text"):
text_parts.append(node["text"])
if node.get("content"):
for child in node["content"]:
text_parts.append(extract_node(child))
return " ".join(text_parts)
return extract_node(content)
@server.list_tools()
async def list_tools() -> list[Tool]:
"""List available tools."""
return [
Tool(
name="docmost_get_page",
description="Get page information and content by ID or slug. Returns title, metadata, and full content.",
inputSchema={
"type": "object",
"properties": {
"page_id": {
"type": "string",
"description": "Page ID (UUID) or slug ID (e.g., 'yQQmTbt8it')",
},
"include_content": {
"type": "boolean",
"description": "Whether to include page content (default: true)",
"default": True,
},
},
"required": ["page_id"],
},
),
Tool(
name="docmost_list_pages",
description="List pages in a space or under a parent page.",
inputSchema={
"type": "object",
"properties": {
"space_id": {
"type": "string",
"description": "Space ID to list pages from",
},
"parent_page_id": {
"type": "string",
"description": "Optional: Parent page ID to list children",
},
"page": {
"type": "integer",
"description": "Page number for pagination (default: 1)",
"default": 1,
},
},
"required": [],
},
),
Tool(
name="docmost_recent_pages",
description="Get recently updated pages in a space.",
inputSchema={
"type": "object",
"properties": {
"space_id": {
"type": "string",
"description": "Space ID to get recent pages from",
},
},
"required": ["space_id"],
},
),
Tool(
name="docmost_create_page",
description="Create a new page in Docmost.",
inputSchema={
"type": "object",
"properties": {
"title": {"type": "string", "description": "Page title"},
"space_id": {"type": "string", "description": "Space ID where the page will be created"},
"parent_page_id": {"type": "string", "description": "Optional: Parent page ID for nested pages"},
"content_text": {"type": "string", "description": "Optional: Initial page content as plain text"},
},
"required": ["title", "space_id"],
},
),
Tool(
name="docmost_update_page",
description="Update an existing page's title or content.",
inputSchema={
"type": "object",
"properties": {
"page_id": {"type": "string", "description": "Page ID to update"},
"title": {"type": "string", "description": "New page title"},
"content_text": {"type": "string", "description": "New page content as plain text"},
},
"required": ["page_id"],
},
),
Tool(
name="docmost_delete_page",
description="Delete a page from Docmost.",
inputSchema={
"type": "object",
"properties": {
"page_id": {"type": "string", "description": "Page ID to delete"},
},
"required": ["page_id"],
},
),
Tool(
name="docmost_list_spaces",
description="List all spaces in the workspace.",
inputSchema={
"type": "object",
"properties": {
"limit": {"type": "integer", "description": "Maximum results (default: 50)", "default": 50},
"page": {"type": "integer", "description": "Page number (default: 1)", "default": 1},
},
"required": [],
},
),
Tool(
name="docmost_get_space",
description="Get space information by ID or slug.",
inputSchema={
"type": "object",
"properties": {
"space_id": {"type": "string", "description": "Space ID or slug"},
},
"required": ["space_id"],
},
),
Tool(
name="docmost_get_current_user",
description="Get information about the currently authenticated user.",
inputSchema={"type": "object", "properties": {}, "required": []},
),
Tool(
name="docmost_list_members",
description="List all members in the workspace.",
inputSchema={
"type": "object",
"properties": {
"limit": {"type": "integer", "description": "Maximum results (default: 50)", "default": 50},
},
"required": [],
},
),
Tool(
name="docmost_list_comments",
description="List comments on a page.",
inputSchema={
"type": "object",
"properties": {
"page_id": {"type": "string", "description": "Page ID to get comments from"},
"limit": {"type": "integer", "description": "Maximum results (default: 50)", "default": 50},
},
"required": ["page_id"],
},
),
Tool(
name="docmost_add_comment",
description="Add a comment to a page.",
inputSchema={
"type": "object",
"properties": {
"page_id": {"type": "string", "description": "Page ID to comment on"},
"content": {"type": "string", "description": "Comment text"},
"parent_comment_id": {"type": "string", "description": "Optional: Parent comment ID for replies"},
},
"required": ["page_id", "content"],
},
),
Tool(
name="docmost_get_page_history",
description="Get revision history of a page.",
inputSchema={
"type": "object",
"properties": {
"page_id": {"type": "string", "description": "Page ID"},
"limit": {"type": "integer", "description": "Maximum revisions to return (default: 20)", "default": 20},
},
"required": ["page_id"],
},
),
]
@server.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
"""Handle tool calls."""
docmost = get_client()
try:
# ==================== Page Tools ====================
if name == "docmost_get_page":
page_id = arguments["page_id"]
include_content = arguments.get("include_content", True)
resp = await docmost.get_page_info(page_id)
page_data = resp.get("data", resp)
output = format_page_info(page_data)
if include_content and page_data.get("content"):
content_text = extract_text_from_content(page_data["content"])
output += f"\n\n**Content:**\n{content_text}"
return [TextContent(type="text", text=output)]
elif name == "docmost_list_pages":
space_id = arguments.get("space_id")
parent_page_id = arguments.get("parent_page_id")
page_num = arguments.get("page", 1)
if not space_id:
return [TextContent(type="text", text="❌ space_id is required")]
resp = await docmost.get_sidebar_pages(space_id, parent_page_id, page_num)
pages = resp.get("data", {}).get("items", [])
if not pages:
return [TextContent(type="text", text="No pages found.")]
output = "📚 Pages:\n\n"
for i, pg in enumerate(pages, 1):
output += f"{i}. **{pg.get('title', 'Untitled')}** (Slug: {pg.get('slugId')})\n"
if pg.get("hasChildren"):
output += " └─ Has children\n"
return [TextContent(type="text", text=output)]
elif name == "docmost_recent_pages":
space_id = arguments["space_id"]
resp = await docmost.get_recent_pages(space_id)
pages = resp.get("data", {}).get("items", resp.get("data", []))
if not pages:
return [TextContent(type="text", text="No recent pages found.")]
output = "🕐 Recent Pages:\n\n"
for i, pg in enumerate(pages, 1):
output += f"{i}. **{pg.get('title', 'Untitled')}** (Slug: {pg.get('slugId')})\n"
output += f" Updated: {pg.get('updatedAt')}\n"
return [TextContent(type="text", text=output)]
elif name == "docmost_create_page":
title = arguments["title"]
space_id = arguments["space_id"]
parent_page_id = arguments.get("parent_page_id")
content_text = arguments.get("content_text")
content = None
if content_text:
content = {
"type": "doc",
"content": [{"type": "paragraph", "content": [{"type": "text", "text": content_text}]}]
}
resp = await docmost.create_page(title, space_id, parent_page_id, content)
page = resp.get("data", resp)
return [TextContent(type="text", text=f"✅ Page created!\n{format_page_info(page)}")]
elif name == "docmost_update_page":
page_id = arguments["page_id"]
title = arguments.get("title")
content_text = arguments.get("content_text")
content = None
if content_text:
content = {
"type": "doc",
"content": [{"type": "paragraph", "content": [{"type": "text", "text": content_text}]}]
}
await docmost.update_page(page_id, title, content)
return [TextContent(type="text", text="✅ Page updated!")]
elif name == "docmost_delete_page":
page_id = arguments["page_id"]
await docmost.delete_page(page_id)
return [TextContent(type="text", text=f"✅ Page '{page_id}' deleted.")]
# ==================== Space Tools ====================
elif name == "docmost_list_spaces":
limit = arguments.get("limit", 50)
page_num = arguments.get("page", 1)
resp = await docmost.list_spaces(limit, page_num)
spaces = resp.get("data", {}).get("items", [])
if not spaces:
return [TextContent(type="text", text="No spaces found.")]
output = "🗂️ Spaces:\n\n"
for i, space in enumerate(spaces, 1):
output += f"{i}. **{space.get('name')}** (Slug: {space.get('slug')})\n"
output += f" - ID: {space.get('id')}\n"
if space.get("description"):
output += f" - {space.get('description')}\n"
return [TextContent(type="text", text=output)]
elif name == "docmost_get_space":
space_id = arguments["space_id"]
resp = await docmost.get_space_info(space_id)
space = resp.get("data", resp)
output = f"🗂️ **{space.get('name')}**\n"
output += f"- ID: {space.get('id')}\n"
output += f"- Slug: {space.get('slug')}\n"
if space.get("description"):
output += f"- Description: {space.get('description')}\n"
return [TextContent(type="text", text=output)]
# ==================== User Tools ====================
elif name == "docmost_get_current_user":
resp = await docmost.get_current_user()
user = resp.get("data", resp)
output = f"👤 **{user.get('name', user.get('email'))}**\n"
output += f"- ID: {user.get('id')}\n"
output += f"- Email: {user.get('email')}\n"
if user.get("role"):
output += f"- Role: {user.get('role')}\n"
return [TextContent(type="text", text=output)]
elif name == "docmost_list_members":
limit = arguments.get("limit", 50)
resp = await docmost.list_workspace_members(limit)
members = resp.get("data", {}).get("items", [])
if not members:
return [TextContent(type="text", text="No members found.")]
output = "👥 Members:\n\n"
for i, m in enumerate(members, 1):
output += f"{i}. **{m.get('name', m.get('email'))}** - {m.get('role', 'member')}\n"
return [TextContent(type="text", text=output)]
# ==================== Comment Tools ====================
elif name == "docmost_list_comments":
page_id = arguments["page_id"]
limit = arguments.get("limit", 50)
resp = await docmost.list_comments(page_id, limit)
comments = resp.get("data", {}).get("items", [])
if not comments:
return [TextContent(type="text", text="No comments.")]
output = "💬 Comments:\n\n"
for i, c in enumerate(comments, 1):
output += f"{i}. {c.get('content', '')}\n"
output += f" - At: {c.get('createdAt')}\n"
return [TextContent(type="text", text=output)]
elif name == "docmost_add_comment":
page_id = arguments["page_id"]
content = arguments["content"]
parent_id = arguments.get("parent_comment_id")
await docmost.create_comment(page_id, content, parent_id)
return [TextContent(type="text", text="✅ Comment added!")]
# ==================== History ====================
elif name == "docmost_get_page_history":
page_id = arguments["page_id"]
limit = arguments.get("limit", 20)
resp = await docmost.get_page_history(page_id, limit)
history = resp.get("data", {}).get("items", [])
if not history:
return [TextContent(type="text", text="No history.")]
output = "📜 History:\n\n"
for i, h in enumerate(history, 1):
output += f"{i}. {h.get('createdAt')}\n"
return [TextContent(type="text", text=output)]
else:
return [TextContent(type="text", text=f"Unknown tool: {name}")]
except Exception as e:
return [TextContent(type="text", text=f"❌ Error: {str(e)}")]
async def run():
"""Run the MCP server."""
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream, server.create_initialization_options())
def main():
"""Entry point."""
asyncio.run(run())
if __name__ == "__main__":
main()
