""" title: Dropbox Tools author: William Hogben description: | Give your AI Agent the ability to interact with your Dropbox account. ## Tools: - **get_file_info(path):** Gets the info for the provided file path. - **get_folder_info(path):** Gets the info for the provided folder path. - **list_path(path, recursive):** Lists the contents of the provided path. - **read_text(path):** Reads the text content of the file at path. - **write_text(path, text):** Writes the text content to the file at path. - **append_text(path, text):** Appends the text content to the file at path. - **replace_text(path, text_to_replace, replacement_text):** Replaces all occurances of text_to_replace in the file with replacement_text. - **remove_file(path):** Removes (erases) the file at path. - **create_folder(path):** Creates a folder at the specified path. - **remove_folder(path):** Removes (erases) the folder at path. - **move_file(from_path, to_path):** Moves (or just renames) the file. - **move_folder(from_path, to_path):** Moves (or just renames) the folder. - **search(query, path, max_results):** Searches for files and folders in the user's Dropbox. - **get_sharable_link(path):** Gets a sharable link for the file or folder at the given path. - **get_private_link(path):** Gets a temporary, private link to a file that can be used to stream its contents. - **start_oauth_flow():** Starts the OAuth flow for Dropbox authorization. - **complete_oauth_flow(auth_code):** Completes the OAuth flow after the user has authorized the application. - **erase_user_secrets():** Erases the user's Dropbox secrets. ## Authentication: - Supports multiple users with separate Dropbox accounts. - The AI will prompt each user to authenticate on first use. - Uses Dropbox OAuth2, admins will need to create a Dropbox app at https://www.dropbox.com/developers/apps - Admins then specify the App Key and App Secret in the tool's Valves. - User credentials are stored in the Open WebUI backend at /app/backend/data/owoho/dropbox_tool.json required_open_webui_version: 0.6.0 requirements: dropbox version: 1.0.0 licence: MIT """ import os import dropbox as dropbox_sdk import math from datetime import datetime from pydantic import BaseModel, Field from pydantic.fields import FieldInfo from typing import Optional, List import json import fcntl # Implementation notes: # - Rules for Dropbox Paths: # - Use "" for root instead of "/" # - Start paths with a "/" when they are not the root # - Do not use a trailing "/" in paths # - Clean model paths when they're input to match above rules # - convert all timestamps to YYYY-MM-DD HH:MM:SS in in UTC # - use the client updated time for updated timestamps (not the server updated time) # - When reporting file sizes give maximum 3 digits of precision class Tools: def __init__(self): self.valves = self.Valves(app_key="", app_secret="") class Valves(BaseModel): app_key: str = Field( ..., description="The Dropbox app key.", ) app_secret: str = Field( ..., description="The Dropbox app secret.", ) secret_storage_path: str = Field( default="/app/backend/data/owoho/dropbox_secrets.json", description="The path to the file where user secrets are stored.", ) def get_file_info( self, path: str = Field(..., description="/the/full_path/to/the_file.ext"), __user__: dict = {}, ) -> str: """ Gets the info for the provided file path. Return format: File at <path>: - XYZ kb - Updated at 2025-05-02 21:24:21 Or: File Not Found. """ dbx = _get_authorized_client(self.valves, __user__["id"]) if isinstance(dbx, str): return dbx if path == "/": path = "" elif path and not path.startswith("/"): path = f"/{path}" try: metadata = dbx.files_get_metadata(path) if isinstance(metadata, dropbox_sdk.files.FileMetadata): size_str = _format_size(metadata.size) date_str = metadata.client_modified.strftime("%Y-%m-%d %H:%M:%S") return f"File at {metadata.path_display}:\n- {size_str}\n- Updated at {date_str}" else: return f"Path '{path}' is a folder, not a file." except dropbox_sdk.exceptions.ApiError as err: if ( isinstance(err.error, dropbox_sdk.files.GetMetadataError) and err.error.is_path() and err.error.get_path().is_not_found() ): return "File Not Found." return f"Error getting file info for '{path}': {err}" def get_folder_info( self, path: str = Field(..., description="/the/full_path/to/the/folder"), __user__: dict = {}, ) -> str: """ Gets the info for the provided folder path. Return format: Folder at <path>: - id: ... Or: Folder Not Found. """ dbx = _get_authorized_client(self.valves, __user__["id"]) if isinstance(dbx, str): return dbx if path == "/": path = "" elif path and not path.startswith("/"): path = f"/{path}" try: metadata = dbx.files_get_metadata(path) if isinstance(metadata, dropbox_sdk.files.FolderMetadata): return ( "Folder at " + metadata.path_display + ":\n" + "\t- id: " + metadata.id ) else: return f"Path '{path}' is a file, not a folder." except dropbox_sdk.exceptions.ApiError as err: if ( isinstance(err.error, dropbox_sdk.files.GetMetadataError) and err.error.is_path() and err.error.get_path().is_not_found() ): return "Folder Not Found." return f"Error getting folder info for '{path}': {err}" def list_path( self, path: str = Field( "", description='The path to the dir to list. Use "" for root instead of "/".', ), recursive: bool = Field( False, description="Whether to list ancestor directories' contents." ), __user__: dict = {}, ) -> str: """ Lists the contents of the provided path. If recursive is specified, lists the contents of all subpaths. Return format: Contents of <path> (<including / excluding> subpaths): - example_filename.ext - 21 kb - 2025-05-02 21:24:21 - example_subdirname/ - subdir_filename.ext - 1.2 mb - 2025-03-14 02:12:53 """ dbx = _get_authorized_client(self.valves, __user__["id"]) if isinstance(dbx, str): return dbx if isinstance(path, FieldInfo): path = "" if isinstance(recursive, FieldInfo): recursive = False if path == "/": path = "" elif path and not path.startswith("/"): path = f"/{path}" def format_entry(entry, indent_level): indent = " " * indent_level if isinstance(entry, dropbox_sdk.files.FolderMetadata): return f"{indent}- {entry.name}/" elif isinstance(entry, dropbox_sdk.files.FileMetadata): size_str = _format_size(entry.size) date_str = entry.client_modified.strftime("%Y-%m-%d %H:%M:%S") return f"{indent}- {entry.name} - {size_str} - {date_str}" return "" def list_folder_recursive(current_path, indent_level): try: result = dbx.files_list_folder(current_path, recursive=False) lines = [] for entry in result.entries: lines.append(format_entry(entry, indent_level)) if recursive and isinstance( entry, dropbox_sdk.files.FolderMetadata ): lines.extend( list_folder_recursive(entry.path_lower, indent_level + 1) ) return lines except dropbox_sdk.exceptions.ApiError as err: return [f"Error listing path '{current_path}': {err}"] subpath_str = "including" if recursive else "excluding" header = f"Contents of {path or '/'} ({subpath_str} subpaths):" output_lines = list_folder_recursive(path, 0) return header + "\n" + "\n".join(output_lines) def read_text( self, path: str = Field(..., description="/the/full_path/to/the_file.ext"), __user__: dict = {}, ) -> str: """ Reads the text content of the file at path. Return format: Contents of <path> (updated 2025-05-02 21:24:21): <content> """ dbx = _get_authorized_client(self.valves, __user__["id"]) if isinstance(dbx, str): return dbx if path == "/": path = "" elif path and not path.startswith("/"): path = f"/{path}" try: metadata, res = dbx.files_download(path) content = res.content.decode("utf-8") date_str = metadata.client_modified.strftime("%Y-%m-%d %H:%M:%S") return ( f"Contents of {metadata.path_display} (updated {date_str}):\n{content}" ) except dropbox_sdk.exceptions.ApiError as err: return f"Error reading file '{path}': {err}" def write_text( self, path: str = Field(..., description="/the/full_path/to/the_file.ext"), text: str = Field(..., description="A string to be written to the path."), __user__: dict = {}, ) -> str: """ Writes the text content to the file at path, creating it if it does not exist. Return format: Wrote text to <path>. """ dbx = _get_authorized_client(self.valves, __user__["id"]) if isinstance(dbx, str): return dbx if path == "/": path = "" elif path and not path.startswith("/"): path = f"/{path}" try: dbx.files_upload( text.encode("utf-8"), path, mode=dropbox_sdk.files.WriteMode("overwrite"), ) return f"Wrote text to {path}." except dropbox_sdk.exceptions.ApiError as err: return f"Error writing to file '{path}': {err}" def append_text( self, path: str = Field(..., description="/the/full_path/to/the_file.ext"), text: str = Field(..., description="A string to be written to the path."), __user__: dict = {}, ) -> str: """ Appends the text content to the file at path, creating it if it does not exist. Return format: Appended text to <path>. Remember that to add a line, if the note does not already have a newline at the end, add a newline at the beginning of the text you add. """ dbx = _get_authorized_client(self.valves, __user__["id"]) if isinstance(dbx, str): return dbx if path == "/": path = "" elif path and not path.startswith("/"): path = f"/{path}" try: _, res = dbx.files_download(path) existing_content = res.content.decode("utf-8") new_content = existing_content + text dbx.files_upload( new_content.encode("utf-8"), path, mode=dropbox_sdk.files.WriteMode("overwrite"), ) return f"Appended text to {path}." except dropbox_sdk.exceptions.ApiError as err: if ( isinstance(err.error, dropbox_sdk.files.DownloadError) and err.error.is_path() and err.error.get_path().is_not_found() ): dbx.files_upload( text.encode("utf-8"), path, mode=dropbox_sdk.files.WriteMode("add") ) return f"Created file and appended text to {path}." return f"Error appending to file '{path}': {err}" def replace_text( self, path: str = Field(..., description="/the/full_path/to/the_file.ext"), text_to_replace: str = Field( ..., description="A string that must be exact matched in the file." ), replacement_text: str = Field( ..., description="A string that will replace the text_to_replace." ), __user__: dict = {}, ) -> str: """ Replaces all occurances of text_to_replace in the file with replacement_text. Return format: Replaced text in <path>. """ dbx = _get_authorized_client(self.valves, __user__["id"]) if isinstance(dbx, str): return dbx if path == "/": path = "" elif path and not path.startswith("/"): path = f"/{path}" try: _, res = dbx.files_download(path) existing_content = res.content.decode("utf-8") new_content = existing_content.replace(text_to_replace, replacement_text) dbx.files_upload( new_content.encode("utf-8"), path, mode=dropbox_sdk.files.WriteMode("overwrite"), ) return f"Replaced text in {path}." except dropbox_sdk.exceptions.ApiError as err: return f"Error replacing text in file '{path}': {err}" def remove_file( self, path: str = Field(..., description="/the/full_path/to/the_file.ext"), __user__: dict = {}, ) -> str: """ Removes (erases) the file at path. Don't do this lightly. """ dbx = _get_authorized_client(self.valves, __user__["id"]) if isinstance(dbx, str): return dbx if path == "/": path = "" elif path and not path.startswith("/"): path = f"/{path}" try: dbx.files_delete_v2(path) return f"Removed file at {path}." except dropbox_sdk.exceptions.ApiError as err: return f"Error removing file at '{path}': {err}" def create_folder( self, path: str = Field(..., description="/the/full_path/to/the/folder"), __user__: dict = {}, ) -> str: """ Creates a folder at the specified path. Return format: Created folder at <path>. """ dbx = _get_authorized_client(self.valves, __user__["id"]) if isinstance(dbx, str): return dbx if path == "/": return "Cannot create a folder at the root." if not path.startswith("/"): path = f"/{path}" try: dbx.files_create_folder_v2(path) return f"Created folder at {path}." except dropbox_sdk.exceptions.ApiError as err: if ( isinstance(err.error, dropbox_sdk.files.CreateFolderError) and err.error.is_path() and err.error.get_path().is_conflict() ): return f"Error creating folder: A folder or file already exists at '{path}'." return f"Error creating folder at '{path}': {err}" def remove_folder( self, path: str = Field(..., description="/the/full_path/to/the/folder"), __user__: dict = {}, ) -> str: """ Removes (erases) the folder at path. This is permanent. """ dbx = _get_authorized_client(self.valves, __user__["id"]) if isinstance(dbx, str): return dbx if path == "/": return "Cannot remove the root folder." if not path.startswith("/"): path = f"/{path}" try: dbx.files_delete_v2(path) return f"Removed folder at {path}." except dropbox_sdk.exceptions.ApiError as err: return f"Error removing folder at '{path}': {err}" def move_file( self, from_path: str = Field( ..., description="/the/full_path/to/the_before_file.ext" ), to_path: str = Field(..., description="/the/full_path/to/the_after_final.ext"), __user__: dict = {}, ) -> str: """ Moves (or just renames) the file. """ dbx = _get_authorized_client(self.valves, __user__["id"]) if isinstance(dbx, str): return dbx if from_path == "/": from_path = "" elif from_path and not from_path.startswith("/"): from_path = f"/{from_path}" if to_path == "/": to_path = "" elif to_path and not to_path.startswith("/"): to_path = f"/{to_path}" try: dbx.files_move_v2(from_path, to_path) return f"Moved file from {from_path} to {to_path}." except dropbox_sdk.exceptions.ApiError as err: if ( isinstance(err.error, dropbox_sdk.files.RelocationError) and err.error.is_to() and isinstance(err.error.get_to(), dropbox_sdk.files.WriteError) and err.error.get_to().is_conflict() ): return f"Error moving file: A file already exists at the destination path '{to_path}'." return f"Error moving file from '{from_path}' to '{to_path}': {err}" def move_folder( self, from_path: str = Field(..., description="/the/full_path/to/the/before_folder"), to_path: str = Field(..., description="/the/full_path/to/the/after_folder"), __user__: dict = {}, ) -> str: """ Moves (or just renames) the folder. """ dbx = _get_authorized_client(self.valves, __user__["id"]) if isinstance(dbx, str): return dbx if from_path == "/": from_path = "" elif from_path and not from_path.startswith("/"): from_path = f"/{from_path}" if to_path == "/": to_path = "" elif to_path and not to_path.startswith("/"): to_path = f"/{to_path}" try: dbx.files_move_v2(from_path, to_path) return f"Moved folder from {from_path} to {to_path}." except dropbox_sdk.exceptions.ApiError as err: if ( isinstance(err.error, dropbox_sdk.files.RelocationError) and err.error.is_to() and isinstance(err.error.get_to(), dropbox_sdk.files.WriteError) and err.error.get_to().is_conflict() ): return f"Error moving folder: A folder already exists at the destination path '{to_path}'." return f"Error moving folder from '{from_path}' to '{to_path}': {err}" def search( self, query: str = Field( ..., description="The search query string. Matches file/folder names and, if supported by your Dropbox plan, file contents.", ), path: str = Field( "", description='The path to search in. Use "" for root instead of "/".', ), max_results: int = Field( 25, description="The maximum number of results to return. The default is 25, and the maximum is 100.", ), __user__: dict = {}, ) -> str: """ Searches for files and folders in the user's Dropbox. Return format: Search results for "<query>" in <path>: - example_filename.ext - 21 kb - 2025-05-02 21:24:21 - example_subdirname/ """ dbx = _get_authorized_client(self.valves, __user__["id"]) if isinstance(dbx, str): return dbx if isinstance(path, FieldInfo): path = "" if isinstance(query, FieldInfo): return "Error: A search query must be provided." if isinstance(max_results, FieldInfo): max_results = 25 if path == "/": path = "" elif path and not path.startswith("/"): path = f"/{path}" try: results = dbx.files_search_v2( query, options=dropbox_sdk.files.SearchOptions( path=path, max_results=max_results ), ) lines = [] for match in results.matches: entry = match.metadata.get_metadata() if isinstance(entry, dropbox_sdk.files.FolderMetadata): lines.append(f"- {entry.name}/") elif isinstance(entry, dropbox_sdk.files.FileMetadata): size_str = _format_size(entry.size) date_str = entry.client_modified.strftime("%Y-%m-%d %H:%M:%S") lines.append(f"- {entry.name} - {size_str} - {date_str}") path_str = f"'{path}'" if path else "the entire Dropbox" return f'Search results for "{query}" in {path_str}:\n' + "\n".join(lines) except dropbox_sdk.exceptions.ApiError as err: return f"Error searching for '{query}': {err}" def get_sharable_link(self, path: str, __user__: dict = {}) -> str: """ Gets a sharable link for the file or folder at the given path. """ dbx = _get_authorized_client(self.valves, __user__["id"]) if isinstance(dbx, str): return dbx try: links = dbx.sharing_list_shared_links(path=path, direct_only=True).links if links: return links[0].url else: link = dbx.sharing_create_shared_link_with_settings(path=path).url return link except dropbox_sdk.exceptions.ApiError as err: return f"Error getting sharable link for '{path}': {err}" def get_private_link(self, path: str, __user__: dict = {}) -> str: """ Gets a temporary, private link to a file that can be used to stream its contents. """ dbx = _get_authorized_client(self.valves, __user__["id"]) if isinstance(dbx, str): return dbx try: link = dbx.files_get_temporary_link(path=path).link return link except dropbox_sdk.exceptions.ApiError as err: return f"Error getting private link for '{path}': {err}" def start_oauth_flow(self, __user__: dict = {}) -> str: oauth_flow = dropbox_sdk.DropboxOAuth2Flow( self.valves.app_key, self.valves.app_secret, redirect_uri=f"{__user__['host']}/tools/callbacks/dropbox", session={"user_id": __user__["id"]}, csrf_token_session_key="dropbox-auth-csrf-token", token_access_type="offline", ) auth_url = oauth_flow.start() _set_secrets( self.valves, __user__["id"], {"oauth_flow": oauth_flow.to_json()}, ) return f"Please authorize this application by visiting this URL: {auth_url}" def complete_oauth_flow( self, auth_code: str = Field(..., description="The authorization code from Dropbox."), __user__: dict = {}, ) -> str: """ Completes the OAuth flow after the user has authorized the application. """ try: secrets = _get_secrets(self.valves, __user__["id"]) oauth_flow_json = secrets.get("oauth_flow") if not oauth_flow_json: return "Error: OAuth flow not started." oauth_flow = dropbox_sdk.DropboxOAuth2Flow.from_json(oauth_flow_json) oauth_result = oauth_flow.finish( {"code": auth_code}, ) _set_secrets( self.valves, __user__["id"], { "oauth_flow": None, "access_token": oauth_result.access_token, "refresh_token": oauth_result.refresh_token, "expires_at": ( oauth_result.expires_at.isoformat() if oauth_result.expires_at else None ), }, ) return "Successfully authorized with Dropbox." except Exception as e: return f"Error completing OAuth flow: {e}" def erase_user_secrets(self, __user__: dict) -> str: _set_secrets(self.valves, __user__["id"], {}) return "Your Dropbox secrets have been erased." def _get_secrets(valves: Tools.Valves, user_id: str) -> dict: if not os.path.exists(valves.secret_storage_path): if "/" in valves.secret_storage_path: os.makedirs(os.path.dirname(valves.secret_storage_path), exist_ok=True) with open(valves.secret_storage_path, "w") as f: json.dump({"users": {}}, f) with open(valves.secret_storage_path, "r") as f: secrets = json.load(f) return secrets.get("users", {}).get(user_id, {}) def _set_secrets(valves: Tools.Valves, user_id: str, user_secrets: dict) -> None: if not os.path.exists(valves.secret_storage_path): if "/" in valves.secret_storage_path: os.makedirs(os.path.dirname(valves.secret_storage_path), exist_ok=True) with open(valves.secret_storage_path, "w") as f: json.dump({"users": {}}, f) with open(valves.secret_storage_path, "r+") as f: fcntl.flock(f, fcntl.LOCK_EX) secrets = json.load(f) if "users" not in secrets: secrets["users"] = {} secrets["users"][user_id] = user_secrets f.seek(0) json.dump(secrets, f) f.truncate() fcntl.flock(f, fcntl.LOCK_UN) def _get_authorized_client( valves: Tools.Valves, user_id: str ) -> dropbox_sdk.Dropbox | str: secrets = _get_secrets(valves, user_id) refresh_token = secrets.get("refresh_token") if not refresh_token: return "Not authorized. Please start the OAuth flow." try: dbx = dropbox_sdk.Dropbox( oauth2_refresh_token=refresh_token, app_key=valves.app_key, app_secret=valves.app_secret, ) dbx.users_get_current_account() # Test the connection return dbx except dropbox_sdk.exceptions.AuthError as e: return f"Authorization failed: {e}. Please re-authorize." def _format_size(size_bytes: int) -> str: if size_bytes == 0: return "0 bytes" size_name = ("bytes", "kb", "mb", "gb", "tb") i = int(math.floor(math.log(size_bytes, 1024))) p = math.pow(1024, i) s = round(size_bytes / p, 2) return f"{s} {size_name[i]}"