MCP Notion Upload Server

from typing import Any, Optional, Dict import httpx import os from pathlib import Path from mcp.server.fastmcp import FastMCP # Initialize FastMCP server mcp = FastMCP("mcp_notion_upload") @mcp.tool() async def upload_file_to_notion( file_path: str, notion_token: Optional[str] = None, file_name: Optional[str] = None ) -> str: """ Upload a file to Notion using the Notion API. Args: file_path: Path to the file to upload notion_token: Notion API token for authentication (optional if NOTION_API_TOKEN env var is set) file_name: Optional custom filename (defaults to original filename) Returns: The file upload ID from Notion (use upload_and_attach_file_to_page for URL) Raises: Exception: If file doesn't exist, exceeds size limit, or API call fails """ # Get token from parameter or environment variable token = notion_token or os.environ.get('NOTION_API_TOKEN') if not token: raise Exception("Notion API token is required. Pass it as a parameter or set NOTION_API_TOKEN environment variable") # Validate file exists file_path_obj = Path(file_path) if not file_path_obj.exists(): raise Exception(f"File not found: {file_path}") # Check file size (max 20MB) file_size = file_path_obj.stat().st_size max_size = 20 * 1024 * 1024 # 20MB in bytes if file_size > max_size: raise Exception(f"File size ({file_size / 1024 / 1024:.2f}MB) exceeds 20MB limit") # Use custom filename if provided, otherwise use original upload_filename = file_name or file_path_obj.name # Notion API headers headers = { "Authorization": f"Bearer {token}", "Notion-Version": "2022-06-28" } async with httpx.AsyncClient() as client: try: # Step 1: Create file upload object create_upload_response = await client.post( "https://api.notion.com/v1/file_uploads", headers={**headers, "Content-Type": "application/json"}, json={"name": upload_filename} ) create_upload_response.raise_for_status() upload_data = create_upload_response.json() file_upload_id = upload_data.get("id") upload_url = upload_data.get("upload_url") if not file_upload_id or not upload_url: raise Exception("Failed to get upload ID or URL from Notion API") # Step 2: Upload file contents to the upload URL with open(file_path, 'rb') as f: files = {"file": (upload_filename, f)} upload_response = await client.post( upload_url, headers=headers, files=files ) upload_response.raise_for_status() # Step 3: Complete the upload and get the final file object complete_response = await client.get( f"https://api.notion.com/v1/file_uploads/{file_upload_id}", headers=headers ) complete_response.raise_for_status() final_data = complete_response.json() # Return the file_upload_id instead of URL (URL is only available after attachment to page) return file_upload_id except httpx.HTTPStatusError as e: raise Exception(f"Notion API error: {e.response.status_code} - {e.response.text}") except Exception as e: raise Exception(f"Upload failed: {str(e)}") @mcp.tool() async def upload_and_attach_file_to_page( file_path: str, page_id: str, notion_token: Optional[str] = None, file_name: Optional[str] = None, caption: Optional[str] = None ) -> Dict[str, Any]: """ Upload a file to Notion and attach it to a specific page, returning download URL. This function performs the complete workflow: 1. Upload file to Notion 2. Create file block and attach to specified page 3. Extract and return download URL with metadata Args: file_path: Path to the file to upload page_id: Notion page ID to attach the file to (with or without dashes) notion_token: Notion API token for authentication (optional if NOTION_API_TOKEN env var is set) file_name: Optional custom filename (defaults to original filename) caption: Optional caption for the file block Returns: Dictionary containing: - file_upload_id: The uploaded file ID - file_block_id: The created file block ID - download_url: Temporary download URL (expires in 1 hour) - expiry_time: ISO timestamp when download URL expires - filename: Name of the file - content_type: MIME type of the file - file_size: Size of the file in bytes Raises: Exception: If file doesn't exist, exceeds size limit, or API call fails Example: result = await upload_and_attach_file_to_page( "document.pdf", "2785bbc0e5c281f48dfae9a48f53f6a6", notion_token="ntn_xxx" ) download_url = result["download_url"] """ # Get token from parameter or environment variable token = notion_token or os.environ.get('NOTION_API_TOKEN') if not token: raise Exception("Notion API token is required. Pass it as a parameter or set NOTION_API_TOKEN environment variable") # Validate file exists file_path_obj = Path(file_path) if not file_path_obj.exists(): raise Exception(f"File not found: {file_path}") # Check file size (max 20MB) file_size = file_path_obj.stat().st_size max_size = 20 * 1024 * 1024 # 20MB in bytes if file_size > max_size: raise Exception(f"File size ({file_size / 1024 / 1024:.2f}MB) exceeds 20MB limit") # Use custom filename if provided, otherwise use original upload_filename = file_name or file_path_obj.name # Notion API headers headers = { "Authorization": f"Bearer {token}", "Notion-Version": "2022-06-28" } async with httpx.AsyncClient() as client: try: # Step 1: Upload file to Notion # Create file upload object create_upload_response = await client.post( "https://api.notion.com/v1/file_uploads", headers={**headers, "Content-Type": "application/json"}, json={"name": upload_filename} ) create_upload_response.raise_for_status() upload_data = create_upload_response.json() file_upload_id = upload_data.get("id") upload_url = upload_data.get("upload_url") if not file_upload_id or not upload_url: raise Exception("Failed to get upload ID or URL from Notion API") # Upload file contents to the upload URL with open(file_path, 'rb') as f: files = {"file": (upload_filename, f)} upload_response = await client.post( upload_url, headers=headers, files=files ) upload_response.raise_for_status() # Step 2: Create file block and attach to page file_block_payload = { "children": [ { "object": "block", "type": "file", "file": { "type": "file_upload", "file_upload": { "id": file_upload_id }, "caption": [ { "type": "text", "text": { "content": caption or upload_filename } } ] if caption or upload_filename else [] } } ] } # Normalize page ID (remove dashes) normalized_page_id = page_id.replace("-", "") attach_response = await client.patch( f"https://api.notion.com/v1/blocks/{normalized_page_id}/children", headers={**headers, "Content-Type": "application/json"}, json=file_block_payload ) attach_response.raise_for_status() # Step 3: Extract file block information and download URL attach_data = attach_response.json() if not attach_data.get("results"): raise Exception("Failed to create file block") file_block = attach_data["results"][0] file_block_id = file_block.get("id") # Extract file information file_info = file_block.get("file", {}) file_data = file_info.get("file", {}) download_url = file_data.get("url") expiry_time = file_data.get("expiry_time") content_type = file_data.get("content_type", "application/octet-stream") if not download_url: raise Exception("File block created but no download URL available") # Return comprehensive file information return { "file_upload_id": file_upload_id, "file_block_id": file_block_id, "download_url": download_url, "expiry_time": expiry_time, "filename": upload_filename, "content_type": content_type, "file_size": file_size, "status": "success", "message": f"File '{upload_filename}' uploaded and attached successfully" } except httpx.HTTPStatusError as e: error_detail = "" try: error_detail = e.response.json().get("message", e.response.text) except: error_detail = e.response.text raise Exception(f"Notion API error: {e.response.status_code} - {error_detail}") except Exception as e: raise Exception(f"Upload and attach failed: {str(e)}") # Initialize and run the server if __name__ == "__main__": mcp.run(transport='stdio')