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()