Google Workspace MCP Server

Instructions

Implementation Reference

• gdrive/drive_tools.py:234-302 (handler)

  The core handler function implementing the create_drive_file tool. It creates a new file in the user's Google Drive (including shared drives) either from provided content string or by fetching from a URL. Includes decorators for MCP server registration (@server.tool()), HTTP error handling, and Google service authentication.
  @server.tool() @handle_http_errors("create_drive_file", service_type="drive") @require_google_service("drive", "drive_file") async def create_drive_file( service, user_google_email: str, file_name: str, content: Optional[str] = None, # Now explicitly Optional folder_id: str = 'root', mime_type: str = 'text/plain', fileUrl: Optional[str] = None, # Now explicitly Optional ) -> str: """ Creates a new file in Google Drive, supporting creation within shared drives. Accepts either direct content or a fileUrl to fetch the content from. Args: user_google_email (str): The user's Google email address. Required. file_name (str): The name for the new file. content (Optional[str]): If provided, the content to write to the file. folder_id (str): The ID of the parent folder. Defaults to 'root'. For shared drives, this must be a folder ID within the shared drive. mime_type (str): The MIME type of the file. Defaults to 'text/plain'. fileUrl (Optional[str]): If provided, fetches the file content from this URL. Returns: str: Confirmation message of the successful file creation with file link. """ logger.info(f"[create_drive_file] Invoked. Email: '{user_google_email}', File Name: {file_name}, Folder ID: {folder_id}, fileUrl: {fileUrl}") if not content and not fileUrl: raise Exception("You must provide either 'content' or 'fileUrl'.") file_data = None # Prefer fileUrl if both are provided if fileUrl: logger.info(f"[create_drive_file] Fetching file from URL: {fileUrl}") async with httpx.AsyncClient() as client: resp = await client.get(fileUrl) if resp.status_code != 200: raise Exception(f"Failed to fetch file from URL: {fileUrl} (status {resp.status_code})") file_data = await resp.aread() # Try to get MIME type from Content-Type header content_type = resp.headers.get("Content-Type") if content_type and content_type != "application/octet-stream": mime_type = content_type logger.info(f"[create_drive_file] Using MIME type from Content-Type header: {mime_type}") elif content: file_data = content.encode('utf-8') file_metadata = { 'name': file_name, 'parents': [folder_id], 'mimeType': mime_type } media = io.BytesIO(file_data) created_file = await asyncio.to_thread( service.files().create( body=file_metadata, media_body=MediaIoBaseUpload(media, mimetype=mime_type, resumable=True), fields='id, name, webViewLink', supportsAllDrives=True ).execute ) link = created_file.get('webViewLink', 'No link available') confirmation_message = f"Successfully created file '{created_file.get('name', file_name)}' (ID: {created_file.get('id', 'N/A')}) in folder '{folder_id}' for {user_google_email}. Link: {link}" logger.info(f"Successfully created file. Link: {link}") return confirmation_message
• gdrive/drive_tools.py:234-234 (registration)

  MCP tool registration decorator @server.tool() applied to the create_drive_file handler.
  @server.tool()