import os import json import logging from typing import Any import httpx from fastmcp import FastMCP from starlette.applications import Starlette from starlette.responses import JSONResponse from starlette.routing import Route, Mount from dotenv import load_dotenv # Load environment variables load_dotenv() # Logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # --- Configuration --- OUTLINE_API_URL = os.getenv("OUTLINE_API_URL", "").rstrip("/") OUTLINE_API_TOKEN = os.getenv("OUTLINE_API_TOKEN", "") class OutlineClient: """HTTP client for Outline API with authentication.""" def __init__(self, base_url: str, api_token: str): self.base_url = base_url.rstrip("/") self.api_token = api_token self._client: httpx.AsyncClient | None = None @property def client(self) -> httpx.AsyncClient: if self._client is None or self._client.is_closed: self._client = httpx.AsyncClient( base_url=self.base_url, headers={ "Authorization": f"Bearer {self.api_token}", "Content-Type": "application/json", }, timeout=30.0, ) return self._client async def call(self, method: str, params: dict | None = None) -> dict: """Execute an Outline API call. Args: method: API method (e.g., 'documents.list', 'collections.info') params: Optional parameters to send in request body Returns: API response as dict """ if params is None: params = {} try: response = await self.client.post(f"/api/{method}", json=params) response.raise_for_status() return response.json() except httpx.HTTPStatusError as e: error_body = e.response.text try: error_json = e.response.json() return { "ok": False, "error": error_json.get("message", str(e)), "status": e.response.status_code, } except Exception: return { "ok": False, "error": error_body or str(e), "status": e.response.status_code, } except httpx.RequestError as e: return {"ok": False, "error": f"Request failed: {str(e)}"} async def check_health(self) -> tuple[bool, str]: """Check API connectivity by calling auth.info.""" try: result = await self.call("auth.info") if result.get("ok"): return True, "Connected" return False, result.get("error", "Unknown error") except Exception as e: return False, str(e) async def close(self): if self._client and not self._client.is_closed: await self._client.aclose() # Initialize client outline_client = OutlineClient(OUTLINE_API_URL, OUTLINE_API_TOKEN) # --- FastMCP Server --- mcp = FastMCP("Outline MCP") @mcp.tool() async def search_documents( query: str, collection_id: str = None, limit: int = 25, include_archived: bool = False, ) -> str: """Search for documents in Outline using full-text search. Args: query: Search query string collection_id: Optional collection ID to search within limit: Maximum number of results (default 25, max 100) include_archived: Include archived documents in results Returns: JSON string with search results containing document titles, IDs, and context """ params = { "query": query, "limit": min(limit, 100), "includeArchived": include_archived, } if collection_id: params["collectionId"] = collection_id result = await outline_client.call("documents.search", params) return json.dumps(result, indent=2) @mcp.tool() async def get_document(document_id: str, share_id: str = None) -> str: """Retrieve a document by ID, returning its full Markdown content. Args: document_id: The document UUID share_id: Optional share ID for accessing shared documents Returns: JSON string with document metadata and Markdown content """ params = {"id": document_id} if share_id: params["shareId"] = share_id result = await outline_client.call("documents.info", params) return json.dumps(result, indent=2) @mcp.tool() async def list_collections(limit: int = 25, offset: int = 0) -> str: """List all collections (folders) in the Outline workspace. Args: limit: Maximum number of results (default 25) offset: Pagination offset Returns: JSON string with collection names, IDs, and metadata """ params = {"limit": limit, "offset": offset} result = await outline_client.call("collections.list", params) return json.dumps(result, indent=2) @mcp.tool() async def list_collection_documents( collection_id: str, limit: int = 25, offset: int = 0 ) -> str: """List all documents within a specific collection. Args: collection_id: The collection UUID limit: Maximum number of results (default 25) offset: Pagination offset Returns: JSON string with document tree structure for the collection """ params = {"id": collection_id, "limit": limit, "offset": offset} result = await outline_client.call("collections.documents", params) return json.dumps(result, indent=2) @mcp.tool() async def outline_api_call(method: str, params: Any = None) -> str: """Execute a raw Outline API call for any endpoint. Use the 'outline://api-reference' resource to discover available methods and their parameters. Args: method: API method name (e.g., 'documents.create', 'users.list') params: Parameters as JSON string or dict (optional) Returns: JSON string with raw API response Examples: - method='documents.create', params='{"title": "New Doc", "collectionId": "uuid"}' - method='users.list', params={"limit": 10} - method='documents.delete', params='{"id": "document-uuid"}' """ # Handle params being string, dict, or None/empty if params is None or params == "" or params == {}: parsed_params = {} elif isinstance(params, dict): parsed_params = params elif isinstance(params, str): try: parsed_params = json.loads(params) if params.strip() else {} except json.JSONDecodeError as e: return json.dumps({"ok": False, "error": f"Invalid JSON params: {str(e)}"}) else: return json.dumps( { "ok": False, "error": f"params must be string or dict, got {type(params).__name__}", } ) result = await outline_client.call(method, parsed_params) return json.dumps(result, indent=2) # --- API Reference Resource --- API_REFERENCE = """# Outline API Reference Outline uses an RPC-style API where all requests are POST to /api/. All responses have format: {"ok": true/false, "data": ...} ## Authentication All requests require Bearer token: Authorization: Bearer Get your token from Outline Settings > API Tokens. ## Pagination Most list endpoints accept: limit (default 25, max 100), offset (default 0) Response includes: pagination.nextPath if more results exist. --- ## Documents ### documents.list List documents. Params: collectionId?, userId?, parentDocumentId?, sort?, direction?, limit?, offset? ### documents.info Get document by ID. Params: id (required), shareId? ### documents.search Full-text search. Params: query (required), collectionId?, userId?, dateFilter?, includeArchived?, includeDrafts?, limit?, offset? ### documents.create Create document. Params: title (required), collectionId (required), text?, parentDocumentId?, templateId?, template?, publish? ### documents.update Update document. Params: id (required), title?, text?, append?, publish?, done? ### documents.delete Delete document. Params: id (required), permanent? ### documents.move Move document. Params: id (required), collectionId?, parentDocumentId? ### documents.archive Archive document. Params: id (required) ### documents.unarchive Restore archived document. Params: id (required) ### documents.restore Restore deleted document. Params: id (required), revisionId? ### documents.export Export document. Params: id (required) Returns Markdown content. ### documents.import Import document. Multipart form: file, collectionId, parentDocumentId?, template?, publish? ### documents.star Star a document. Params: id (required) ### documents.unstar Remove star. Params: id (required) ### documents.templatize Convert to template. Params: id (required) ### documents.unpublish Unpublish document. Params: id (required) --- ## Collections ### collections.list List all collections. Params: limit?, offset? ### collections.info Get collection details. Params: id (required) ### collections.documents Get document structure for collection. Params: id (required), limit?, offset? ### collections.create Create collection. Params: name (required), description?, color?, permission?, sharing? ### collections.update Update collection. Params: id (required), name?, description?, color?, permission?, sharing? ### collections.delete Delete collection. Params: id (required) ### collections.add_user Add user to collection. Params: id (required), userId (required), permission? ### collections.remove_user Remove user from collection. Params: id (required), userId (required) ### collections.memberships List collection members. Params: id (required), query?, permission?, limit?, offset? ### collections.add_group Add group to collection. Params: id (required), groupId (required), permission? ### collections.remove_group Remove group. Params: id (required), groupId (required) ### collections.group_memberships List group memberships. Params: id (required), query?, permission?, limit?, offset? ### collections.export Export entire collection. Params: id (required), format? ### collections.export_all Export all collections. Params: format? --- ## Users ### users.list List workspace users. Params: query?, filter?, sort?, direction?, limit?, offset? Filter options: all, active, invited, suspended, admins ### users.info Get user by ID. Params: id (required) ### users.invite Invite users. Params: invites (array of {email, name, role}) ### users.update Update user. Params: id (required), name?, avatarUrl? ### users.delete Delete user. Params: id (required) ### users.suspend Suspend user. Params: id (required) ### users.activate Activate suspended user. Params: id (required) ### users.promote Promote to admin. Params: id (required) ### users.demote Remove admin. Params: id (required) --- ## Groups ### groups.list List groups. Params: query?, sort?, direction?, limit?, offset? ### groups.info Get group. Params: id (required) ### groups.create Create group. Params: name (required) ### groups.update Update group. Params: id (required), name (required) ### groups.delete Delete group. Params: id (required) ### groups.memberships List group members. Params: id (required), query?, limit?, offset? ### groups.add_user Add user to group. Params: id (required), userId (required) ### groups.remove_user Remove user from group. Params: id (required), userId (required) --- ## Comments ### comments.list List comments on document. Params: documentId (required), limit?, offset? ### comments.create Create comment. Params: documentId (required), data (required - ProseMirror JSON) ### comments.update Update comment. Params: id (required), data (required) ### comments.delete Delete comment. Params: id (required) --- ## Revisions ### revisions.list List document revisions. Params: documentId (required), limit?, offset? ### revisions.info Get revision. Params: id (required) --- ## Shares ### shares.list List document shares. Params: documentId?, sort?, direction?, limit?, offset? ### shares.info Get share by ID or documentId. Params: id?, documentId? ### shares.create Create share link. Params: documentId (required), published?, urlId?, includeChildDocuments? ### shares.update Update share. Params: id (required), published?, urlId?, includeChildDocuments? ### shares.revoke Revoke share. Params: id (required) --- ## Stars ### stars.list List starred documents. Params: limit?, offset? ### stars.create Star an item. Params: documentId?, collectionId? ### stars.delete Remove star. Params: id (required) --- ## Events ### events.list List audit events. Params: name?, actorId?, documentId?, collectionId?, auditLog?, sort?, direction?, limit?, offset? --- ## File Operations ### fileOperations.list List file operations. Params: type?, limit?, offset? ### fileOperations.info Get file operation status. Params: id (required) ### fileOperations.redirect Get download URL. Params: id (required) ### fileOperations.delete Delete file operation. Params: id (required) --- ## Attachments ### attachments.create Create attachment upload URL. Params: name (required), documentId (required), size (required), contentType (required) ### attachments.delete Delete attachment. Params: id (required) ### attachments.redirect Get attachment URL. Params: id (required) --- ## Auth ### auth.info Get current user and team info. No params. ### auth.config Get auth configuration. No params. --- ## Views ### views.list List document views. Params: documentId (required), limit?, offset? ### views.create Record a view. Params: documentId (required) --- ## Common Response Patterns Success: {"ok": true, "data": {...}, "pagination": {"nextPath": "...", "limit": 25, "offset": 0}} Error: {"ok": false, "error": "message", "status": 401} Rate Limiting: HTTP 429 with Retry-After header """ @mcp.resource("outline://api-reference") def get_api_reference() -> str: """Returns comprehensive Outline API documentation for use with outline_api_call tool.""" return API_REFERENCE # --- Health Check Endpoint --- async def health(request): """Health check endpoint for Docker/load balancer. Returns 200 if server is running. API connectivity status is included in response body but doesn't affect HTTP status code. """ response = {"status": "ok", "server": "running"} if not OUTLINE_API_URL or not OUTLINE_API_TOKEN: response["api"] = "not_configured" response["detail"] = "Missing OUTLINE_API_URL or OUTLINE_API_TOKEN" else: healthy, message = await outline_client.check_health() response["api"] = "connected" if healthy else "disconnected" if not healthy: response["detail"] = message return JSONResponse(response) # --- ASGI Application --- def create_app(): """Create the ASGI application with health check and MCP routes.""" mcp_app = mcp.http_app() routes = [ Route("/health", health, methods=["GET"]), Mount("/", app=mcp_app), ] return Starlette(routes=routes, lifespan=mcp_app.lifespan) # Create the app instance for uvicorn app = create_app() if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)