import os import json import logging 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: str = "") -> 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: JSON string of parameters to send in request body 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"}' """ try: parsed_params = json.loads(params) if params and params.strip() else {} except json.JSONDecodeError as e: return json.dumps({"ok": False, "error": f"Invalid JSON params: {str(e)}"}) 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.""" if not OUTLINE_API_URL or not OUTLINE_API_TOKEN: return JSONResponse( { "status": "degraded", "error": "Missing OUTLINE_API_URL or OUTLINE_API_TOKEN", }, status_code=503, ) healthy, message = await outline_client.check_health() if healthy: return JSONResponse({"status": "ok"}) return JSONResponse({"status": "degraded", "error": message}, status_code=503) # --- 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)