MCP Notion Upload Server

Instructions

Implementation Reference

• mcp_notion_upload.py:10-96 (handler)

  The core handler function decorated with @mcp.tool() for registration. Handles file validation, Notion API authentication, and performs the three-step file upload process to obtain the file_upload_id.
  @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_notion_upload.py:10-10 (registration)

  The @mcp.tool() decorator registers the upload_file_to_notion function as an MCP tool.
  @mcp.tool()
• mcp_notion_upload.py:11-14 (schema)

  Function signature defining the input schema (parameters) and output type for the tool.
  async def upload_file_to_notion( file_path: str, notion_token: Optional[str] = None, file_name: Optional[str] = None