Dropbox Toolkit (#332)

Basic tools to list items in a folder, search for files / folders, and
download a file content.
This commit is contained in:
Renato Byrro 2025-03-27 18:34:10 -03:00 committed by GitHub
parent 13a01254df
commit c1e8fc795a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
19 changed files with 1465 additions and 0 deletions

View file

@ -0,0 +1,18 @@
files: ^./
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: "v4.4.0"
hooks:
- id: check-case-conflict
- id: check-merge-conflict
- id: check-toml
- id: check-yaml
- id: end-of-file-fixer
- id: trailing-whitespace
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.6.7
hooks:
- id: ruff
args: [--fix]
- id: ruff-format

View file

@ -0,0 +1,46 @@
target-version = "py39"
line-length = 100
fix = true
[lint]
select = [
# flake8-2020
"YTT",
# flake8-bandit
"S",
# flake8-bugbear
"B",
# flake8-builtins
"A",
# flake8-comprehensions
"C4",
# flake8-debugger
"T10",
# flake8-simplify
"SIM",
# isort
"I",
# mccabe
"C90",
# pycodestyle
"E", "W",
# pyflakes
"F",
# pygrep-hooks
"PGH",
# pyupgrade
"UP",
# ruff
"RUF",
# tryceratops
"TRY",
]
[lint.per-file-ignores]
"*" = ["TRY003", "B904"]
"**/tests/*" = ["S101", "E501"]
"**/evals/*" = ["S101", "E501"]
[format]
preview = true
skip-magic-trailing-comma = false

21
toolkits/dropbox/LICENSE Normal file
View file

@ -0,0 +1,21 @@
MIT License
Copyright (c) 2025 Arcade
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

53
toolkits/dropbox/Makefile Normal file
View file

@ -0,0 +1,53 @@
.PHONY: help
help:
@echo "🛠️ dropbox Commands:\n"
@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) | sort | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m%-30s\033[0m %s\n", $$1, $$2}'
.PHONY: install
install: ## Install the poetry environment and install the pre-commit hooks
@echo "📦 Checking if Poetry is installed"
@if ! command -v poetry &> /dev/null; then \
echo "📦 Installing Poetry with pip"; \
pip install poetry==1.8.5; \
else \
echo "📦 Poetry is already installed"; \
fi
@echo "🚀 Installing package in development mode with all extras"
poetry install --all-extras
.PHONY: build
build: clean-build ## Build wheel file using poetry
@echo "🚀 Creating wheel file"
poetry build
.PHONY: clean-build
clean-build: ## clean build artifacts
@echo "🗑️ Cleaning dist directory"
rm -rf dist
.PHONY: test
test: ## Test the code with pytest
@echo "🚀 Testing code: Running pytest"
@poetry run pytest -W ignore -v --cov --cov-config=pyproject.toml --cov-report=xml
.PHONY: coverage
coverage: ## Generate coverage report
@echo "coverage report"
coverage report
@echo "Generating coverage report"
coverage html
.PHONY: bump-version
bump-version: ## Bump the version in the pyproject.toml file
@echo "🚀 Bumping version in pyproject.toml"
poetry version patch
.PHONY: check
check: ## Run code quality tools.
@echo "🚀 Checking Poetry lock file consistency with 'pyproject.toml': Running poetry check"
@poetry check
@echo "🚀 Linting code: Running pre-commit"
@poetry run pre-commit run -a
@echo "🚀 Static type checking: Running mypy"
@poetry run mypy --config-file=pyproject.toml

View file

@ -0,0 +1,34 @@
from enum import Enum
class EndpointType(Enum):
API = "api"
CONTENT = "content"
class Endpoint(Enum):
LIST_FOLDER = "/files/list_folder"
SEARCH_FILES = "/files/search"
DOWNLOAD_FILE = "/files/download"
class ItemCategory(Enum):
IMAGE = "image"
DOCUMENT = "document"
PDF = "pdf"
SPREADSHEET = "spreadsheet"
PRESENTATION = "presentation"
AUDIO = "audio"
VIDEO = "video"
FOLDER = "folder"
PAPER = "paper"
API_BASE_URL = "https://{endpoint_type}.dropboxapi.com"
API_VERSION = "2"
ENDPOINT_URL_MAP = {
Endpoint.LIST_FOLDER: (EndpointType.API, "files/list_folder"),
Endpoint.SEARCH_FILES: (EndpointType.API, "files/search_v2"),
Endpoint.DOWNLOAD_FILE: (EndpointType.CONTENT, "files/download"),
}
MAX_RESPONSE_BODY_SIZE = 10 * 1024 * 1024 # 10 MiB

View file

@ -0,0 +1,34 @@
from typing import Any
from arcade.sdk.eval import BinaryCritic
class DropboxPathCritic(BinaryCritic):
def evaluate(self, expected: Any, actual: Any) -> dict[str, float | bool]:
"""
Ignores leading slash in the actual value when comparing to the expected value.
Note: sometimes the LLM won't start the path with a slash, so this critic ignores it when
comparing. Dropbox tools will add the slash, when needed, so no worries about API errors.
Args:
expected: The expected value.
actual: The actual value to compare, cast to the type of expected.
Returns:
dict: A dictionary containing the match status and score.
"""
try:
actual_casted = self.cast_actual(expected, actual)
# TODO log or something better here
except TypeError:
actual_casted = actual
if isinstance(expected, str):
expected = expected.lstrip("/")
if isinstance(actual_casted, str):
actual_casted = actual_casted.lstrip("/")
match = expected == actual_casted
return {"match": match, "score": self.weight if match else 0.0}

View file

@ -0,0 +1,18 @@
from typing import Optional
class DropboxApiError(Exception):
def __init__(
self,
status_code: int,
error_summary: str,
user_message: Optional[str],
):
if "path/not_found" in error_summary:
self.message = "The specified path was not found by Dropbox"
elif "unsupported_file" in error_summary:
self.message = "The specified file is not supported for the requested operation"
else:
self.message = user_message or error_summary
self.status_code = status_code

View file

@ -0,0 +1,138 @@
from typing import Annotated, Optional
from arcade.sdk import ToolContext, tool
from arcade.sdk.auth import Dropbox
from arcade.sdk.errors import ToolExecutionError
from arcade_dropbox.constants import Endpoint, ItemCategory
from arcade_dropbox.exceptions import DropboxApiError
from arcade_dropbox.utils import (
build_dropbox_json,
clean_dropbox_entries,
parse_dropbox_path,
send_dropbox_request,
)
@tool(
requires_auth=Dropbox(
scopes=["files.metadata.read"],
)
)
async def list_items_in_folder(
context: ToolContext,
folder_path: Annotated[
str,
"The path to the folder to list the contents of. E.g. '/AcmeInc/Reports'. "
"Defaults to an empty string (list items in the Dropbox root folder).",
] = "",
limit: Annotated[
int,
"The maximum number of items to return. Defaults to 100. Maximum allowed is 2000.",
] = 100,
cursor: Annotated[
Optional[str],
"The cursor token for the next page of results. "
"Defaults to None (returns the first page of results).",
] = None,
) -> Annotated[
dict, "Dictionary containing the list of files and folders in the specified folder path"
]:
"""Provides a dictionary containing the list of items in the specified folder path.
Note 1: when paginating, it is not necessary to provide any other argument besides the cursor.
Note 2: when paginating, any given item (file or folder) may be returned in multiple pages.
"""
limit = min(limit, 2000)
try:
result = await send_dropbox_request(
context.get_auth_token_or_empty(),
endpoint=Endpoint.LIST_FOLDER,
path=parse_dropbox_path(folder_path),
limit=limit,
cursor=cursor,
)
except DropboxApiError as api_error:
return {"error": api_error.message}
return {
"items": clean_dropbox_entries(result["entries"]),
"cursor": result.get("cursor"),
"has_more": result.get("has_more", False),
}
@tool(
requires_auth=Dropbox(
scopes=["files.metadata.read"],
)
)
async def search_files_and_folders(
context: ToolContext,
keywords: Annotated[
str,
"The keywords to search for. E.g. 'quarterly report'. "
"Maximum length allowed by the Dropbox API is 1000 characters. ",
],
search_in_folder_path: Annotated[
Optional[str],
"Restricts the search to the specified folder path. E.g. '/AcmeInc/Reports'. "
"Defaults to None (search in the entire Dropbox).",
] = None,
filter_by_category: Annotated[
Optional[list[ItemCategory]],
"Restricts the search to the specified category(ies) of items. "
"Provide None, one or multiple, if needed. Defaults to None (returns all categories).",
] = None,
limit: Annotated[
int,
"The maximum number of items to return. Defaults to 100. Maximum allowed is 1000.",
] = 100,
cursor: Annotated[
Optional[str],
"The cursor token for the next page of results. Defaults to None (first page of results).",
] = None,
) -> Annotated[dict, "List of items in the specified folder path matching the search criteria"]:
"""Returns a list of items in the specified folder path matching the search criteria.
Note 1: the Dropbox API will return up to 10,000 (ten thousand) items cumulatively across
multiple pagination requests using the cursor token.
Note 2: when paginating, it is not necessary to provide any other argument besides the cursor.
Note 3: when paginating, any given item (file or folder) may be returned in multiple pages.
"""
if len(keywords) > 1000:
raise ToolExecutionError(
"The keywords argument must be a string with up to 1000 characters."
)
limit = min(limit, 1000)
filter_by_category = filter_by_category or []
options = build_dropbox_json(
file_status="active",
filename_only=False,
path=parse_dropbox_path(search_in_folder_path),
max_results=limit,
file_categories=[category.value for category in filter_by_category],
)
try:
result = await send_dropbox_request(
context.get_auth_token_or_empty(),
endpoint=Endpoint.SEARCH_FILES,
query=keywords,
options=options,
cursor=cursor,
)
except DropboxApiError as api_error:
return {"error": api_error.message}
return {
"items": clean_dropbox_entries([
match["metadata"]["metadata"] for match in result["matches"]
]),
"cursor": result.get("cursor"),
"has_more": result.get("has_more", False),
}

View file

@ -0,0 +1,47 @@
from typing import Annotated, Optional
from arcade.sdk import ToolContext, tool
from arcade.sdk.auth import Dropbox
from arcade.sdk.errors import ToolExecutionError
from arcade_dropbox.constants import Endpoint
from arcade_dropbox.exceptions import DropboxApiError
from arcade_dropbox.utils import parse_dropbox_path, send_dropbox_request
@tool(
requires_auth=Dropbox(
scopes=["files.content.read"],
)
)
async def download_file(
context: ToolContext,
file_path: Annotated[
Optional[str],
"The path of the file to download. E.g. '/AcmeInc/Reports/Q1_2025.txt'. Defaults to None.",
] = None,
file_id: Annotated[
Optional[str],
"The ID of the file to download. E.g. 'id:a4ayc_80_OEAAAAAAAAAYa'. Defaults to None.",
] = None,
) -> Annotated[dict, "Contents of the specified file"]:
"""Downloads the specified file.
Note: either one of `file_path` or `file_id` must be provided.
"""
if not file_path and not file_id:
raise ToolExecutionError("Either `file_path` or `file_id` must be provided.")
if file_path and file_id:
raise ToolExecutionError("Only one of `file_path` or `file_id` can be provided.")
try:
result = await send_dropbox_request(
context.get_auth_token_or_empty(),
endpoint=Endpoint.DOWNLOAD_FILE,
path=parse_dropbox_path(file_path) or file_id,
)
except DropboxApiError as api_error:
return {"error": api_error.message}
return {"file": result}

View file

@ -0,0 +1,106 @@
import json
from typing import Any, Optional
import httpx
from arcade_dropbox.constants import (
API_BASE_URL,
API_VERSION,
ENDPOINT_URL_MAP,
Endpoint,
EndpointType,
)
from arcade_dropbox.exceptions import DropboxApiError
def build_dropbox_url(endpoint_type: EndpointType, endpoint_path: str) -> str:
base_url = API_BASE_URL.format(endpoint_type=endpoint_type.value)
return f"{base_url}/{API_VERSION}/{endpoint_path.strip('/')}"
def build_dropbox_headers(token: Optional[str]) -> dict[str, str]:
return {"Authorization": f"Bearer {token}"} if token else {}
def build_dropbox_json(**kwargs: Any) -> dict:
return {key: value for key, value in kwargs.items() if value is not None}
async def send_dropbox_request(
authorization_token: Optional[str],
endpoint: Endpoint,
**kwargs: Any,
) -> Any:
endpoint_type, endpoint_path = ENDPOINT_URL_MAP[endpoint]
url = build_dropbox_url(endpoint_type, endpoint_path)
headers = build_dropbox_headers(authorization_token)
json_data = build_dropbox_json(**kwargs)
if json_data.get("cursor"):
url += "/continue"
# If cursor is provided, every other argument must be ignored to avoid API error
json_data = {"cursor": json_data["cursor"]}
if endpoint_type == EndpointType.CONTENT:
headers["Dropbox-API-Arg"] = json.dumps(json_data)
json_data = {}
async with httpx.AsyncClient() as client:
request_args: dict[str, Any] = {"url": url, "headers": headers}
if json_data:
request_args["json"] = json_data
response = await client.post(**request_args)
try:
data = response.json()
except Exception:
data = {}
if response.status_code != 200:
raise DropboxApiError(
status_code=response.status_code,
error_summary=data.get("error_summary", response.text),
user_message=data.get("user_message"),
)
if endpoint_type == EndpointType.CONTENT:
data = json.loads(response.headers["Dropbox-API-Result"])
data = clean_dropbox_entry(data, default_type="file")
data["content"] = response.text
return data
return response.json()
def clean_dropbox_entry(entry: dict, default_type: Optional[str] = None) -> dict:
return {
"type": entry.get(".tag", default_type),
"id": entry.get("id"),
"name": entry.get("name"),
"path": entry.get("path_display"),
"size_in_bytes": entry.get("size"),
"modified_datetime": entry.get("server_modified"),
}
def clean_dropbox_entries(entries: list[dict]) -> list[dict]:
return [clean_dropbox_entry(entry) for entry in entries]
def parse_dropbox_path(path: Optional[str]) -> Optional[str]:
if not isinstance(path, str):
return ""
if not path:
return ""
if path in ["/", "\\"]:
return ""
# Normalize windows-style paths to unix-style paths
path = path.replace("\\", "/")
# Dropbox expects the path to always start with a slash
return "/" + path.strip("/")

View file

@ -0,0 +1,45 @@
from unittest.mock import patch
import pytest
from arcade.sdk import ToolAuthorizationContext, ToolContext
@pytest.fixture
def mock_context():
mock_auth = ToolAuthorizationContext(token="fake-token") # noqa: S106
return ToolContext(authorization=mock_auth)
@pytest.fixture
def mock_httpx_client(mocker):
with patch("arcade_dropbox.utils.httpx") as mock_httpx:
yield mock_httpx.AsyncClient().__aenter__.return_value
@pytest.fixture
def sample_folder_entry():
return {
".tag": "folder",
"name": "test.txt",
"path_display": "/TestFolder",
"path_lower": "/testfolder",
"id": "1234567890",
"client_modified": "2025-01-01T00:00:00Z",
"server_modified": "2025-01-01T00:00:00Z",
"rev": "1234567890",
}
@pytest.fixture
def sample_file_entry():
return {
".tag": "file",
"name": "test.txt",
"path_display": "/TestFile.txt",
"path_lower": "/testfile.txt",
"id": "1234567890",
"client_modified": "2025-01-01T00:00:00Z",
"server_modified": "2025-01-01T00:00:00Z",
"rev": "1234567890",
"size": 1024,
}

View file

@ -0,0 +1,88 @@
from arcade.sdk import ToolCatalog
from arcade.sdk.eval import (
BinaryCritic,
EvalRubric,
EvalSuite,
ExpectedToolCall,
tool_eval,
)
import arcade_dropbox
from arcade_dropbox.critics import DropboxPathCritic
from arcade_dropbox.tools.files import download_file
rubric = EvalRubric(
fail_threshold=0.8,
warn_threshold=0.9,
)
catalog = ToolCatalog()
catalog.add_module(arcade_dropbox)
@tool_eval()
def download_file_eval_suite() -> EvalSuite:
"""Create an evaluation suite for the download_file tool."""
suite = EvalSuite(
name="download_file",
system_message="You are an AI assistant that can interact with files and folders in Dropbox using the provided tools.",
catalog=catalog,
rubric=rubric,
)
suite.add_case(
name="Download file in the root folder by file path",
user_message="Download the file test.txt from Dropbox",
expected_tool_calls=[
ExpectedToolCall(
func=download_file,
args={
"file_path": "test.txt",
"file_id": None,
},
),
],
critics=[
DropboxPathCritic(critic_field="file_path", weight=0.5),
BinaryCritic(critic_field="file_id", weight=0.5),
],
)
suite.add_case(
name="Download file with a sub-folder structure",
user_message="Download the file Q1report.ppt in the folder AcmeInc/Reports from Dropbox",
expected_tool_calls=[
ExpectedToolCall(
func=download_file,
args={
"file_path": "/AcmeInc/Reports/Q1report.ppt",
"file_id": None,
},
),
],
critics=[
DropboxPathCritic(critic_field="file_path", weight=0.5),
BinaryCritic(critic_field="file_id", weight=0.5),
],
)
suite.add_case(
name="Download file by ID",
user_message="Download the file id:a4ayc_80_OEAAAAAAAAAYa from Dropbox",
expected_tool_calls=[
ExpectedToolCall(
func=download_file,
args={
"file_path": None,
"file_id": "id:a4ayc_80_OEAAAAAAAAAYa",
},
),
],
critics=[
BinaryCritic(critic_field="file_path", weight=0.5),
BinaryCritic(critic_field="file_id", weight=0.5),
],
)
return suite

View file

@ -0,0 +1,94 @@
from arcade.sdk import ToolCatalog
from arcade.sdk.eval import (
BinaryCritic,
EvalRubric,
EvalSuite,
ExpectedToolCall,
tool_eval,
)
import arcade_dropbox
from arcade_dropbox.critics import DropboxPathCritic
from arcade_dropbox.tools.browse import list_items_in_folder
rubric = EvalRubric(
fail_threshold=0.8,
warn_threshold=0.9,
)
catalog = ToolCatalog()
catalog.add_module(arcade_dropbox)
@tool_eval()
def list_items_in_folder_eval_suite() -> EvalSuite:
"""Create an evaluation suite for the list_items_in_folder tool."""
suite = EvalSuite(
name="list_items_in_folder",
system_message="You are an AI assistant that can interact with files and folders in Dropbox using the provided tools.",
catalog=catalog,
rubric=rubric,
)
suite.add_case(
name="List items in the Dropbox root folder",
user_message="List the items in the Dropbox root folder",
expected_tool_calls=[
ExpectedToolCall(
func=list_items_in_folder,
args={
"folder_path": "",
"limit": 100,
"cursor": None,
},
),
],
critics=[
DropboxPathCritic(critic_field="folder_path", weight=0.6),
BinaryCritic(critic_field="limit", weight=0.2),
BinaryCritic(critic_field="cursor", weight=0.2),
],
)
suite.add_case(
name="List items in a sub-folder",
user_message="List the items in the folder AcmeInc/Reports",
expected_tool_calls=[
ExpectedToolCall(
func=list_items_in_folder,
args={
"folder_path": "/AcmeInc/Reports",
"limit": 100,
"cursor": None,
},
),
],
critics=[
DropboxPathCritic(critic_field="folder_path", weight=0.6),
BinaryCritic(critic_field="limit", weight=0.2),
BinaryCritic(critic_field="cursor", weight=0.2),
],
)
suite.add_case(
name="List items in a sub-folder with custom limit",
user_message="List the first 50 items in the folder AcmeInc/Reports",
expected_tool_calls=[
ExpectedToolCall(
func=list_items_in_folder,
args={
"folder_path": "/AcmeInc/Reports",
"limit": 50,
"cursor": None,
},
),
],
critics=[
DropboxPathCritic(critic_field="folder_path", weight=0.4),
BinaryCritic(critic_field="limit", weight=0.4),
BinaryCritic(critic_field="cursor", weight=0.2),
],
)
return suite

View file

@ -0,0 +1,131 @@
from arcade.sdk import ToolCatalog
from arcade.sdk.eval import (
BinaryCritic,
EvalRubric,
EvalSuite,
ExpectedToolCall,
tool_eval,
)
import arcade_dropbox
from arcade_dropbox.constants import ItemCategory
from arcade_dropbox.critics import DropboxPathCritic
from arcade_dropbox.tools.browse import search_files_and_folders
rubric = EvalRubric(
fail_threshold=0.8,
warn_threshold=0.9,
)
catalog = ToolCatalog()
catalog.add_module(arcade_dropbox)
@tool_eval()
def search_files_and_folders_eval_suite() -> EvalSuite:
"""Create an evaluation suite for the search_files_and_folders tool."""
suite = EvalSuite(
name="list_items_in_folder",
system_message="You are an AI assistant that can interact with files and folders in Dropbox using the provided tools.",
catalog=catalog,
rubric=rubric,
)
suite.add_case(
name="Search for files about 'quarterly report' in my Dropbox",
user_message="Search for files about 'quarterly report' in my Dropbox",
expected_tool_calls=[
ExpectedToolCall(
func=search_files_and_folders,
args={
"keywords": "quarterly report",
"search_in_folder_path": None,
"filter_by_category": None,
"limit": 100,
"cursor": None,
},
),
],
critics=[
BinaryCritic(critic_field="keywords", weight=0.6),
BinaryCritic(critic_field="search_in_folder_path", weight=0.1),
BinaryCritic(critic_field="filter_by_category", weight=0.1),
BinaryCritic(critic_field="limit", weight=0.1),
BinaryCritic(critic_field="cursor", weight=0.1),
],
)
suite.add_case(
name="Search for files about 'quarterly report' in a sub-folder",
user_message="Search for files about 'quarterly report' in the folder AcmeInc/Reports",
expected_tool_calls=[
ExpectedToolCall(
func=search_files_and_folders,
args={
"keywords": "quarterly report",
"search_in_folder_path": "/AcmeInc/Reports",
"filter_by_category": None,
"limit": 100,
"cursor": None,
},
),
],
critics=[
BinaryCritic(critic_field="keywords", weight=0.35),
DropboxPathCritic(critic_field="search_in_folder_path", weight=0.35),
BinaryCritic(critic_field="filter_by_category", weight=0.1),
BinaryCritic(critic_field="limit", weight=0.1),
BinaryCritic(critic_field="cursor", weight=0.1),
],
)
suite.add_case(
name="Search for PDF files about 'quarterly report' in a sub-folder",
user_message="Search for PDF files about 'quarterly report' in the folder AcmeInc/Reports",
expected_tool_calls=[
ExpectedToolCall(
func=search_files_and_folders,
args={
"keywords": "quarterly report",
"search_in_folder_path": "/AcmeInc/Reports",
"filter_by_category": [ItemCategory.PDF.value],
"limit": 100,
"cursor": None,
},
),
],
critics=[
BinaryCritic(critic_field="keywords", weight=0.25),
DropboxPathCritic(critic_field="search_in_folder_path", weight=0.25),
BinaryCritic(critic_field="filter_by_category", weight=0.25),
BinaryCritic(critic_field="limit", weight=0.125),
BinaryCritic(critic_field="cursor", weight=0.125),
],
)
suite.add_case(
name="Search for PDF files about 'quarterly report' in a sub-folder",
user_message="Return the first 10 PDF files about 'quarterly report' in the folder AcmeInc/Reports",
expected_tool_calls=[
ExpectedToolCall(
func=search_files_and_folders,
args={
"keywords": "quarterly report",
"search_in_folder_path": "/AcmeInc/Reports",
"filter_by_category": [ItemCategory.PDF.value],
"limit": 10,
"cursor": None,
},
),
],
critics=[
BinaryCritic(critic_field="keywords", weight=0.2),
DropboxPathCritic(critic_field="search_in_folder_path", weight=0.2),
BinaryCritic(critic_field="filter_by_category", weight=0.2),
BinaryCritic(critic_field="limit", weight=0.2),
BinaryCritic(critic_field="cursor", weight=0.2),
],
)
return suite

View file

@ -0,0 +1,42 @@
[tool.poetry]
name = "arcade_dropbox"
version = "0.1.0"
description = "Arcade tools designed for LLMs to interact with Dropbox"
authors = ["Arcade <dev@arcade.dev>"]
[tool.poetry.dependencies]
python = "^3.10"
arcade-ai = ">=1.0.5,<2.0"
dropbox = ">=12.0.0,<13.0"
[tool.poetry.dev-dependencies]
pytest = "^8.3.0"
pytest-cov = "^4.0.0"
pytest-asyncio = "^0.24.0"
pytest-mock = "^3.11.1"
mypy = "^1.5.1"
pre-commit = "^3.4.0"
tox = "^4.11.1"
ruff = "^0.7.4"
[build-system]
requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"
[tool.mypy]
files = ["arcade_dropbox/**/*.py"]
python_version = "3.10"
disallow_untyped_defs = "True"
disallow_any_unimported = "True"
no_implicit_optional = "True"
check_untyped_defs = "True"
warn_return_any = "True"
warn_unused_ignores = "True"
show_error_codes = "True"
ignore_missing_imports = "True"
[tool.pytest.ini_options]
testpaths = ["tests"]
[tool.coverage.report]
skip_empty = true

View file

@ -0,0 +1,119 @@
import json
from unittest.mock import MagicMock
import httpx
import pytest
from arcade_dropbox.tools.files import download_file
from arcade_dropbox.utils import clean_dropbox_entry
@pytest.fixture
def file_content_response_header():
return json.dumps({
"id": "123",
"name": "test.txt",
"path_display": "/test.txt",
"size": 1024,
"server_modified": "2021-01-01T00:00:00Z",
"content_hash": "1234567890",
"is_downloadable": True,
"rev": "a1c10ce0dd78",
"sharing_info": {
"modified_by": "dbid:AAH4f99T0taONIb-OurWxbNQ6ywGRopQngc",
"parent_shared_folder_id": "84528192421",
"read_only": True,
},
})
@pytest.mark.asyncio
async def test_download_file_success(
mock_context,
mock_httpx_client,
file_content_response_header,
):
file_content = "test file content"
mock_httpx_response = MagicMock(spec=httpx.Response)
mock_httpx_response.status_code = 200
mock_httpx_response.json.side_effect = ValueError("not json")
mock_httpx_response.headers = {"Dropbox-API-Result": file_content_response_header}
mock_httpx_response.text = file_content
mock_httpx_client.post.return_value = mock_httpx_response
tool_response = await download_file(
context=mock_context,
file_path="test.txt",
)
expected_response = clean_dropbox_entry(json.loads(file_content_response_header))
expected_response["content"] = file_content
expected_response["type"] = "file"
assert tool_response == {"file": expected_response}
@pytest.mark.asyncio
async def test_download_file_path_not_found(
mock_context,
mock_httpx_client,
):
mock_httpx_response = MagicMock(spec=httpx.Response)
mock_httpx_response.status_code = 409
mock_httpx_response.json.return_value = {"error_summary": "path/not_found"}
mock_httpx_client.post.return_value = mock_httpx_response
tool_response = await download_file(
context=mock_context,
file_path="test.txt",
)
assert tool_response == {
"error": "The specified path was not found by Dropbox",
}
@pytest.mark.asyncio
async def test_download_file_unsupported_file(
mock_context,
mock_httpx_client,
):
mock_httpx_response = MagicMock(spec=httpx.Response)
mock_httpx_response.status_code = 409
mock_httpx_response.json.return_value = {"error_summary": "unsupported_file/not_supported"}
mock_httpx_client.post.return_value = mock_httpx_response
tool_response = await download_file(
context=mock_context,
file_path="test.txt",
)
assert tool_response == {
"error": "The specified file is not supported for the requested operation",
}
@pytest.mark.asyncio
async def test_download_file_server_error(
mock_context,
mock_httpx_client,
):
mock_httpx_response = MagicMock(spec=httpx.Response)
mock_httpx_response.status_code = 500
mock_httpx_response.text = "500 Internal server error"
mock_httpx_response.json.side_effect = ValueError("not json")
mock_httpx_client.post.return_value = mock_httpx_response
tool_response = await download_file(
context=mock_context,
file_path="test.txt",
)
assert tool_response == {
"error": "500 Internal server error",
}

View file

@ -0,0 +1,145 @@
from unittest.mock import MagicMock
import httpx
import pytest
from arcade_dropbox.tools.browse import list_items_in_folder
from arcade_dropbox.utils import clean_dropbox_entries
@pytest.mark.asyncio
async def test_list_items_success_empty_folder(
mock_context,
mock_httpx_client,
):
mock_httpx_response = MagicMock(spec=httpx.Response)
mock_httpx_response.status_code = 200
mock_httpx_response.json.return_value = {"entries": [], "cursor": None, "has_more": False}
mock_httpx_client.post.return_value = mock_httpx_response
tool_response = await list_items_in_folder(
context=mock_context,
folder_path="/path/to/folder",
)
assert tool_response == {
"items": [],
"cursor": None,
"has_more": False,
}
@pytest.mark.asyncio
async def test_list_items_success_with_folder_entries(
mock_context,
mock_httpx_client,
sample_folder_entry,
sample_file_entry,
):
entries = [sample_folder_entry, sample_file_entry]
mock_httpx_response = MagicMock(spec=httpx.Response)
mock_httpx_response.status_code = 200
mock_httpx_response.json.return_value = {"entries": entries, "cursor": None, "has_more": False}
mock_httpx_client.post.return_value = mock_httpx_response
tool_response = await list_items_in_folder(
context=mock_context,
folder_path="/path/to/folder",
)
assert tool_response == {
"items": clean_dropbox_entries(entries),
"cursor": None,
"has_more": False,
}
@pytest.mark.asyncio
async def test_list_items_success_with_more_items_to_paginate(
mock_context,
mock_httpx_client,
sample_folder_entry,
sample_file_entry,
):
entries = [sample_folder_entry, sample_file_entry]
mock_httpx_response = MagicMock(spec=httpx.Response)
mock_httpx_response.status_code = 200
mock_httpx_response.json.return_value = {
"entries": entries,
"cursor": "cursor",
"has_more": True,
}
mock_httpx_client.post.return_value = mock_httpx_response
tool_response = await list_items_in_folder(
context=mock_context,
folder_path="/path/to/folder",
)
assert tool_response == {
"items": clean_dropbox_entries(entries),
"cursor": "cursor",
"has_more": True,
}
@pytest.mark.asyncio
async def test_list_items_success_providing_cursor(
mock_context,
mock_httpx_client,
sample_folder_entry,
sample_file_entry,
):
entries = [sample_folder_entry, sample_file_entry]
mock_httpx_response = MagicMock(spec=httpx.Response)
mock_httpx_response.status_code = 200
mock_httpx_response.json.return_value = {
"entries": entries,
"cursor": "cursor2",
"has_more": True,
}
mock_httpx_client.post.return_value = mock_httpx_response
tool_response = await list_items_in_folder(
context=mock_context,
folder_path="/path/to/folder",
cursor="cursor1",
limit=2,
)
assert tool_response == {
"items": clean_dropbox_entries(entries),
"cursor": "cursor2",
"has_more": True,
}
# Check that the request was made with the cursor and not the other arguments
mock_httpx_client.post.assert_called_with(
url="https://api.dropboxapi.com/2/files/list_folder/continue",
headers={"Authorization": "Bearer fake-token"},
json={"cursor": "cursor1"},
)
@pytest.mark.asyncio
async def test_list_items_path_not_found(
mock_context,
mock_httpx_client,
):
mock_httpx_response = MagicMock(spec=httpx.Response)
mock_httpx_response.status_code = 409
mock_httpx_response.json.return_value = {"error_summary": "path/not_found"}
mock_httpx_client.post.return_value = mock_httpx_response
tool_response = await list_items_in_folder(
context=mock_context,
folder_path="/not/exist/folder",
)
assert tool_response == {
"error": "The specified path was not found by Dropbox",
}

View file

@ -0,0 +1,286 @@
from unittest.mock import MagicMock
import httpx
import pytest
from arcade_dropbox.constants import ItemCategory
from arcade_dropbox.tools.browse import search_files_and_folders
from arcade_dropbox.utils import clean_dropbox_entry
@pytest.fixture
def sample_folder_match(sample_folder_entry):
return {
"metadata": {
".tag": "metadata",
"metadata": sample_folder_entry,
}
}
@pytest.fixture
def sample_file_match(sample_file_entry):
return {
"metadata": {
".tag": "metadata",
"metadata": sample_file_entry,
}
}
@pytest.mark.asyncio
async def test_search_files_success_empty_results(
mock_context,
mock_httpx_client,
):
mock_httpx_response = MagicMock(spec=httpx.Response)
mock_httpx_response.status_code = 200
mock_httpx_response.json.return_value = {"matches": [], "cursor": None, "has_more": False}
mock_httpx_client.post.return_value = mock_httpx_response
tool_response = await search_files_and_folders(
context=mock_context,
keywords="do not match anything",
)
assert tool_response == {
"items": [],
"cursor": None,
"has_more": False,
}
@pytest.mark.asyncio
async def test_search_files_success_with_matches(
mock_context,
mock_httpx_client,
sample_file_match,
sample_folder_match,
sample_file_entry,
sample_folder_entry,
):
matches = [
sample_file_match,
sample_folder_match,
]
mock_httpx_response = MagicMock(spec=httpx.Response)
mock_httpx_response.status_code = 200
mock_httpx_response.json.return_value = {"matches": matches, "cursor": None, "has_more": False}
mock_httpx_client.post.return_value = mock_httpx_response
tool_response = await search_files_and_folders(
context=mock_context,
keywords="test",
)
assert tool_response == {
"items": [
clean_dropbox_entry(sample_file_entry),
clean_dropbox_entry(sample_folder_entry),
],
"cursor": None,
"has_more": False,
}
@pytest.mark.asyncio
async def test_search_files_success_with_path_missing_leading_slash(
mock_context,
mock_httpx_client,
sample_file_match,
sample_folder_match,
sample_file_entry,
sample_folder_entry,
):
matches = [
sample_file_match,
sample_folder_match,
]
mock_httpx_response = MagicMock(spec=httpx.Response)
mock_httpx_response.status_code = 200
mock_httpx_response.json.return_value = {"matches": matches, "cursor": None, "has_more": False}
mock_httpx_client.post.return_value = mock_httpx_response
tool_response = await search_files_and_folders(
context=mock_context,
keywords="test",
search_in_folder_path="TestFolder",
)
assert tool_response == {
"items": [
clean_dropbox_entry(sample_file_entry),
clean_dropbox_entry(sample_folder_entry),
],
"cursor": None,
"has_more": False,
}
mock_httpx_client.post.assert_called_once_with(
url="https://api.dropboxapi.com/2/files/search_v2",
headers={"Authorization": "Bearer fake-token"},
json={
"query": "test",
"options": {
"file_categories": [],
"path": "/TestFolder",
"file_status": "active",
"filename_only": False,
"max_results": 100,
},
},
)
@pytest.mark.asyncio
async def test_search_files_success_with_more_results_to_paginate(
mock_context,
mock_httpx_client,
sample_file_match,
sample_folder_match,
sample_file_entry,
sample_folder_entry,
):
matches = [
sample_file_match,
sample_folder_match,
]
mock_httpx_response = MagicMock(spec=httpx.Response)
mock_httpx_response.status_code = 200
mock_httpx_response.json.return_value = {
"matches": matches,
"cursor": "cursor",
"has_more": True,
}
mock_httpx_client.post.return_value = mock_httpx_response
tool_response = await search_files_and_folders(
context=mock_context,
keywords="test",
)
assert tool_response == {
"items": [
clean_dropbox_entry(sample_file_entry),
clean_dropbox_entry(sample_folder_entry),
],
"cursor": "cursor",
"has_more": True,
}
@pytest.mark.asyncio
async def test_search_files_success_providing_pagination_cursor(
mock_context,
mock_httpx_client,
sample_file_match,
sample_folder_match,
sample_file_entry,
sample_folder_entry,
):
matches = [
sample_file_match,
sample_folder_match,
]
mock_httpx_response = MagicMock(spec=httpx.Response)
mock_httpx_response.status_code = 200
mock_httpx_response.json.return_value = {"matches": matches, "cursor": None, "has_more": False}
mock_httpx_client.post.return_value = mock_httpx_response
tool_response = await search_files_and_folders(
context=mock_context,
keywords="test",
cursor="cursor",
)
assert tool_response == {
"items": [
clean_dropbox_entry(sample_file_entry),
clean_dropbox_entry(sample_folder_entry),
],
"cursor": None,
"has_more": False,
}
# Assert that the request was made with the correct cursor and not other arguments
mock_httpx_client.post.assert_called_once_with(
url="https://api.dropboxapi.com/2/files/search_v2/continue",
headers={"Authorization": "Bearer fake-token"},
json={"cursor": "cursor"},
)
@pytest.mark.asyncio
async def test_search_files_path_not_found(
mock_context,
mock_httpx_client,
):
mock_httpx_response = MagicMock(spec=httpx.Response)
mock_httpx_response.status_code = 409
mock_httpx_response.json.return_value = {"error_summary": "path/not_found"}
mock_httpx_client.post.return_value = mock_httpx_response
tool_response = await search_files_and_folders(
context=mock_context,
keywords="test",
search_in_folder_path="/not/exist/folder",
)
assert tool_response == {
"error": "The specified path was not found by Dropbox",
}
@pytest.mark.asyncio
async def test_search_files_success_filtering_by_category(
mock_context,
mock_httpx_client,
sample_file_match,
sample_folder_match,
sample_file_entry,
sample_folder_entry,
):
matches = [
sample_file_match,
sample_folder_match,
]
mock_httpx_response = MagicMock(spec=httpx.Response)
mock_httpx_response.status_code = 200
mock_httpx_response.json.return_value = {"matches": matches, "cursor": None, "has_more": False}
mock_httpx_client.post.return_value = mock_httpx_response
tool_response = await search_files_and_folders(
context=mock_context,
keywords="test",
filter_by_category=[ItemCategory.PDF],
)
assert tool_response == {
"items": [
clean_dropbox_entry(sample_file_entry),
clean_dropbox_entry(sample_folder_entry),
],
"cursor": None,
"has_more": False,
}
mock_httpx_client.post.assert_called_once_with(
url="https://api.dropboxapi.com/2/files/search_v2",
headers={"Authorization": "Bearer fake-token"},
json={
"query": "test",
"options": {
"path": "",
"file_status": "active",
"filename_only": False,
"max_results": 100,
"file_categories": [ItemCategory.PDF.value],
},
},
)