"""MkDocs MCP Server implementation."""
import logging
import sys
from pathlib import Path
from typing import Any, Sequence
import anyio
from mcp.server import Server
from mcp.server.models import InitializationOptions
from mcp.server.stdio import stdio_server
from mcp.types import (
Resource,
Tool,
TextContent,
ImageContent,
EmbeddedResource,
CallToolResult,
ListResourcesResult,
ListToolsResult,
ReadResourceResult,
)
from .resources import DocumentationResourceManager
from .tools import DocumentationToolManager
# Configure logging to stderr to avoid interfering with STDIO transport
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
stream=sys.stderr,
)
logger = logging.getLogger(__name__)
class MkDocsMCPServer:
"""MCP server for MkDocs documentation."""
def __init__(self, docs_path: Path | None = None) -> None:
"""Initialize the MkDocs MCP server.
Args:
docs_path: Path to the documentation directory. Defaults to ../docs
"""
self.server = Server("mkdocs-mcp")
# Set default docs path relative to the server location
if docs_path is None:
server_dir = Path(__file__).parent.parent.parent.parent
docs_path = server_dir / "docs"
self.docs_path = Path(docs_path)
logger.info(f"Using docs path: {self.docs_path}")
# Initialize managers
self.resource_manager = DocumentationResourceManager(self.docs_path)
self.tool_manager = DocumentationToolManager(self.docs_path)
# Set up handlers
self._setup_handlers()
def _setup_handlers(self) -> None:
"""Set up MCP message handlers."""
@self.server.list_resources()
async def handle_list_resources() -> ListResourcesResult:
"""List available documentation resources."""
logger.info("Listing resources")
resources = await self.resource_manager.list_resources()
return ListResourcesResult(resources=resources)
@self.server.read_resource()
async def handle_read_resource(uri: str) -> ReadResourceResult:
"""Read a documentation resource."""
logger.info(f"Reading resource: {uri}")
content = await self.resource_manager.read_resource(uri)
return ReadResourceResult(contents=[content])
@self.server.list_tools()
async def handle_list_tools() -> ListToolsResult:
"""List available documentation tools."""
logger.info("Listing tools")
tools = await self.tool_manager.list_tools()
return ListToolsResult(tools=tools)
@self.server.call_tool()
async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> CallToolResult:
"""Call a documentation tool."""
logger.info(f"Calling tool: {name} with args: {arguments}")
result = await self.tool_manager.call_tool(name, arguments or {})
return result
async def run(self) -> None:
"""Run the MCP server using STDIO transport."""
logger.info("Starting MkDocs MCP server")
# Verify docs directory exists
if not self.docs_path.exists():
logger.error(f"Documentation directory not found: {self.docs_path}")
raise FileNotFoundError(f"Documentation directory not found: {self.docs_path}")
if not self.docs_path.is_dir():
logger.error(f"Documentation path is not a directory: {self.docs_path}")
raise NotADirectoryError(f"Documentation path is not a directory: {self.docs_path}")
# Start the server
async with stdio_server() as (read_stream, write_stream):
logger.info("MCP server connected via STDIO")
await self.server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="mkdocs-mcp",
server_version="0.1.0",
capabilities=self.server.get_capabilities(
notification_options=None,
experimental_capabilities={}
),
),
)
def main() -> None:
"""Main entry point for the MCP server."""
try:
server = MkDocsMCPServer()
anyio.run(server.run)
except KeyboardInterrupt:
logger.info("Server stopped by user")
except Exception as e:
logger.error(f"Server error: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
