Compare commits

...

1 Commits

Author SHA1 Message Date
Iris Clawd
0000239b3c feat: add google_drive/upload_from_file action for token-efficient file uploads
Add a new CrewAIPlatformFileUploadTool that reads files directly from disk
and uploads to Google Drive via the platform API, bypassing the LLM context
window entirely. This solves two problems:

1. File content no longer consumes LLM tokens
2. Binary/large files no longer risk exceeding the 128k context limit

The new action accepts a file_path parameter instead of content. It handles:
- Auto-detection of MIME type from file extension
- Optional custom file name (defaults to local filename)
- File size validation (50 MB limit for simple uploads)
- Base64 encoding of file content before sending to API

The existing google_drive/upload_file action is unchanged — full backwards
compatibility with the 10k+ existing executions.

Changes:
- New tool: CrewAIPlatformFileUploadTool
- Builder auto-injects the tool when apps include google_drive or
  google_drive/upload_from_file
- 14 unit tests covering upload, error handling, MIME detection, SSL
- Updated docs with new action reference and usage examples
2026-04-03 20:30:34 +00:00
5 changed files with 612 additions and 0 deletions

View File

@@ -86,6 +86,22 @@ CREWAI_PLATFORM_INTEGRATION_TOKEN=your_enterprise_token
</Accordion>
<Accordion title="google_drive/upload_from_file">
**Description:** Upload a file from a local path to Google Drive. The file is read directly from disk — its content never passes through the LLM context window, making this efficient for large or binary files.
<Note>
Unlike `google_drive/upload_file` which requires passing file content as a parameter (consuming LLM tokens), this action takes a file path and reads the file directly. Use this for binary files, large documents, or any case where you want to avoid token overhead.
</Note>
**Parameters:**
- `file_path` (string, required): Path to the local file to upload.
- `name` (string, optional): Name for the file in Google Drive (defaults to the local filename).
- `mime_type` (string, optional): MIME type of the file (auto-detected from file extension if not provided).
- `parent_folder_id` (string, optional): ID of the parent folder where the file should be created.
- `description` (string, optional): Description of the file.
</Accordion>
<Accordion title="google_drive/download_file">
**Description:** Download a file from Google Drive.
@@ -205,6 +221,42 @@ crew = Crew(
crew.kickoff()
```
### Uploading Files from Disk (Token-Efficient)
```python
from crewai import Agent, Task, Crew
# Use upload_from_file to avoid passing file content through the LLM context
upload_agent = Agent(
role="File Uploader",
goal="Upload local files to Google Drive efficiently",
backstory="An AI assistant that uploads files without wasting context tokens.",
apps=[
'google_drive/upload_from_file',
'google_drive/list_files'
]
)
# The agent only needs to specify the file path — the file is read directly
# from disk and never passes through the LLM context window
upload_task = Task(
description="Upload the file at /data/reports/quarterly_report.pdf to Google Drive in the Reports folder",
agent=upload_agent,
expected_output="File uploaded successfully with file ID"
)
crew = Crew(
agents=[upload_agent],
tasks=[upload_task]
)
crew.kickoff()
```
<Tip>
Use `google_drive/upload_from_file` instead of `google_drive/upload_file` when uploading existing files from disk. This is especially important for binary files (PDFs, images, etc.) or large files that would otherwise consume significant LLM context tokens or exceed context limits.
</Tip>
### Advanced File Management
```python

View File

@@ -7,6 +7,9 @@ through the CrewAI platform API.
from crewai_tools.tools.crewai_platform_tools.crewai_platform_action_tool import (
CrewAIPlatformActionTool,
)
from crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool import (
CrewAIPlatformFileUploadTool,
)
from crewai_tools.tools.crewai_platform_tools.crewai_platform_tool_builder import (
CrewaiPlatformToolBuilder,
)
@@ -17,6 +20,7 @@ from crewai_tools.tools.crewai_platform_tools.crewai_platform_tools import (
__all__ = [
"CrewAIPlatformActionTool",
"CrewAIPlatformFileUploadTool",
"CrewaiPlatformToolBuilder",
"CrewaiPlatformTools",
]

View File

@@ -0,0 +1,165 @@
"""CrewAI Platform File Upload Tool.
Uploads a file from disk to Google Drive without passing file content
through the LLM context window. This avoids token waste and context
limit issues with large or binary files.
"""
import base64
import json
import logging
import mimetypes
import os
from pathlib import Path
from typing import Any, Optional
from crewai.tools import BaseTool
from pydantic import Field, create_model
from crewai_tools.tools.crewai_platform_tools.misc import (
get_platform_api_base_url,
get_platform_integration_token,
)
import requests
logger = logging.getLogger(__name__)
# Google Drive simple upload limit
MAX_FILE_SIZE_BYTES = 50 * 1024 * 1024 # 50 MB
_UploadFromFileSchema = create_model(
"UploadFromFileSchema",
file_path=(str, Field(description="Path to the local file to upload")),
name=(
Optional[str],
Field(
default=None,
description="Name for the file in Google Drive (defaults to the local filename)",
),
),
mime_type=(
Optional[str],
Field(
default=None,
description="MIME type of the file (auto-detected if not provided)",
),
),
parent_folder_id=(
Optional[str],
Field(
default=None,
description="ID of the parent folder where the file should be created",
),
),
description=(
Optional[str],
Field(default=None, description="Description of the file"),
),
)
class CrewAIPlatformFileUploadTool(BaseTool):
"""Upload a file from disk to Google Drive.
Reads the file locally and sends it to the platform API, bypassing
the LLM context window entirely. Supports auto-detection of MIME type
and optional file naming.
"""
name: str = "google_drive_upload_from_file"
description: str = (
"Upload a file from a local path to Google Drive. "
"The file is read directly from disk — its content never passes "
"through the LLM context, making this efficient for large or binary files."
)
args_schema: type = _UploadFromFileSchema
def _run(
self,
file_path: str,
name: str | None = None,
mime_type: str | None = None,
parent_folder_id: str | None = None,
description: str | None = None,
**kwargs: Any,
) -> str:
try:
path = Path(file_path).expanduser().resolve()
if not path.exists():
return f"Error: File not found: {file_path}"
if not path.is_file():
return f"Error: Path is not a file: {file_path}"
file_size = path.stat().st_size
if file_size > MAX_FILE_SIZE_BYTES:
return (
f"Error: File size ({file_size / (1024 * 1024):.1f} MB) exceeds "
f"the 50 MB limit for simple uploads. Consider splitting the file "
f"or using a resumable upload method."
)
# Read and encode file content
content_bytes = path.read_bytes()
content_b64 = base64.b64encode(content_bytes).decode("utf-8")
# Auto-detect MIME type if not provided
if mime_type is None:
guessed_type, _ = mimetypes.guess_type(str(path))
mime_type = guessed_type or "application/octet-stream"
# Use filename if name not provided
upload_name = name or path.name
# Build payload matching the existing upload_file action format
payload_data: dict[str, Any] = {
"name": upload_name,
"content": content_b64,
}
if mime_type:
payload_data["mime_type"] = mime_type
if parent_folder_id:
payload_data["parent_folder_id"] = parent_folder_id
if description:
payload_data["description"] = description
api_url = (
f"{get_platform_api_base_url()}/actions/GOOGLE_DRIVE_SAVE_FILE/execute"
)
token = get_platform_integration_token()
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
}
payload = {"integration": payload_data}
response = requests.post(
url=api_url,
headers=headers,
json=payload,
timeout=120, # Longer timeout for file uploads
verify=os.environ.get("CREWAI_FACTORY", "false").lower() != "true",
)
data = response.json()
if not response.ok:
if isinstance(data, dict):
error_info = data.get("error", {})
if isinstance(error_info, dict):
error_message = error_info.get("message", json.dumps(data))
else:
error_message = str(error_info)
else:
error_message = str(data)
return f"API request failed: {error_message}"
return json.dumps(data, indent=2)
except PermissionError:
return f"Error: Permission denied reading file: {file_path}"
except Exception as e:
return f"Error uploading file: {e!s}"

View File

@@ -11,11 +11,22 @@ import requests
from crewai_tools.tools.crewai_platform_tools.crewai_platform_action_tool import (
CrewAIPlatformActionTool,
)
from crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool import (
CrewAIPlatformFileUploadTool,
)
from crewai_tools.tools.crewai_platform_tools.misc import (
get_platform_api_base_url,
get_platform_integration_token,
)
# Apps that have client-side local tools (e.g. file-path-based upload)
_LOCAL_TOOL_APPS = {
"google_drive": [CrewAIPlatformFileUploadTool],
}
_LOCAL_TOOL_ACTIONS = {
"google_drive/upload_from_file": CrewAIPlatformFileUploadTool,
}
logger = logging.getLogger(__name__)
@@ -95,6 +106,23 @@ class CrewaiPlatformToolBuilder:
tools.append(tool)
# Inject client-side local tools based on requested apps
added_local_tools: set[type] = set()
for app in self._apps:
# Check for specific action (e.g. "google_drive/upload_from_file")
if app in _LOCAL_TOOL_ACTIONS:
tool_cls = _LOCAL_TOOL_ACTIONS[app]
if tool_cls not in added_local_tools:
tools.append(tool_cls())
added_local_tools.add(tool_cls)
# Check for full app (e.g. "google_drive") — inject all local tools
app_base = app.split("/")[0]
if app_base in _LOCAL_TOOL_APPS:
for tool_cls in _LOCAL_TOOL_APPS[app_base]:
if tool_cls not in added_local_tools:
tools.append(tool_cls())
added_local_tools.add(tool_cls)
self._tools = tools
def __enter__(self) -> list[BaseTool]:

View File

@@ -0,0 +1,363 @@
"""Tests for CrewAIPlatformFileUploadTool."""
import base64
import json
import os
import tempfile
from pathlib import Path
from unittest.mock import Mock, patch
import pytest
from crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool import (
CrewAIPlatformFileUploadTool,
MAX_FILE_SIZE_BYTES,
)
class TestCrewAIPlatformFileUploadTool:
"""Test suite for the file upload tool."""
def setup_method(self):
self.tool = CrewAIPlatformFileUploadTool()
def test_tool_name_and_description(self):
assert self.tool.name == "google_drive_upload_from_file"
assert "local path" in self.tool.description.lower()
@patch.dict(
"os.environ",
{"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"},
clear=True,
)
@patch(
"crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool.requests.post"
)
def test_successful_upload(self, mock_post):
"""Test uploading a file successfully."""
mock_response = Mock()
mock_response.ok = True
mock_response.json.return_value = {
"id": "file123",
"name": "test.txt",
}
mock_post.return_value = mock_response
with tempfile.NamedTemporaryFile(
mode="w", suffix=".txt", delete=False
) as f:
f.write("Hello, Google Drive!")
temp_path = f.name
try:
result = self.tool._run(file_path=temp_path)
assert "file123" in result
mock_post.assert_called_once()
call_kwargs = mock_post.call_args
payload = call_kwargs.kwargs["json"]["integration"]
assert payload["name"] == Path(temp_path).name
assert payload["mime_type"] == "text/plain"
# Verify content is base64-encoded
decoded = base64.b64decode(payload["content"]).decode("utf-8")
assert decoded == "Hello, Google Drive!"
finally:
os.unlink(temp_path)
@patch.dict(
"os.environ",
{"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"},
clear=True,
)
@patch(
"crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool.requests.post"
)
def test_custom_name_override(self, mock_post):
"""Test that a custom name overrides the filename."""
mock_response = Mock()
mock_response.ok = True
mock_response.json.return_value = {"id": "file123"}
mock_post.return_value = mock_response
with tempfile.NamedTemporaryFile(
mode="w", suffix=".txt", delete=False
) as f:
f.write("content")
temp_path = f.name
try:
self.tool._run(file_path=temp_path, name="custom_report.txt")
payload = mock_post.call_args.kwargs["json"]["integration"]
assert payload["name"] == "custom_report.txt"
finally:
os.unlink(temp_path)
@patch.dict(
"os.environ",
{"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"},
clear=True,
)
@patch(
"crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool.requests.post"
)
def test_mime_type_auto_detection(self, mock_post):
"""Test MIME type is auto-detected from file extension."""
mock_response = Mock()
mock_response.ok = True
mock_response.json.return_value = {"id": "file123"}
mock_post.return_value = mock_response
with tempfile.NamedTemporaryFile(
mode="w", suffix=".json", delete=False
) as f:
f.write("{}")
temp_path = f.name
try:
self.tool._run(file_path=temp_path)
payload = mock_post.call_args.kwargs["json"]["integration"]
assert payload["mime_type"] == "application/json"
finally:
os.unlink(temp_path)
@patch.dict(
"os.environ",
{"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"},
clear=True,
)
@patch(
"crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool.requests.post"
)
def test_custom_mime_type(self, mock_post):
"""Test that a custom MIME type overrides auto-detection."""
mock_response = Mock()
mock_response.ok = True
mock_response.json.return_value = {"id": "file123"}
mock_post.return_value = mock_response
with tempfile.NamedTemporaryFile(
mode="w", suffix=".txt", delete=False
) as f:
f.write("content")
temp_path = f.name
try:
self.tool._run(file_path=temp_path, mime_type="application/pdf")
payload = mock_post.call_args.kwargs["json"]["integration"]
assert payload["mime_type"] == "application/pdf"
finally:
os.unlink(temp_path)
def test_file_not_found(self):
"""Test error when file does not exist."""
result = self.tool._run(file_path="/nonexistent/path/file.txt")
assert "Error: File not found" in result
def test_path_is_directory(self):
"""Test error when path is a directory."""
with tempfile.TemporaryDirectory() as tmpdir:
result = self.tool._run(file_path=tmpdir)
assert "Error: Path is not a file" in result
def test_file_too_large(self):
"""Test error when file exceeds 50 MB limit."""
with tempfile.NamedTemporaryFile(delete=False) as f:
temp_path = f.name
# Create a sparse file that reports as > 50 MB
f.seek(MAX_FILE_SIZE_BYTES + 1)
f.write(b"\0")
try:
result = self.tool._run(file_path=temp_path)
assert "exceeds" in result
assert "50 MB" in result
finally:
os.unlink(temp_path)
@patch.dict(
"os.environ",
{"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"},
clear=True,
)
@patch(
"crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool.requests.post"
)
def test_optional_params_included(self, mock_post):
"""Test that optional params are included in payload when provided."""
mock_response = Mock()
mock_response.ok = True
mock_response.json.return_value = {"id": "file123"}
mock_post.return_value = mock_response
with tempfile.NamedTemporaryFile(
mode="w", suffix=".txt", delete=False
) as f:
f.write("content")
temp_path = f.name
try:
self.tool._run(
file_path=temp_path,
parent_folder_id="folder456",
description="My report",
)
payload = mock_post.call_args.kwargs["json"]["integration"]
assert payload["parent_folder_id"] == "folder456"
assert payload["description"] == "My report"
finally:
os.unlink(temp_path)
@patch.dict(
"os.environ",
{"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"},
clear=True,
)
@patch(
"crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool.requests.post"
)
def test_optional_params_excluded_when_none(self, mock_post):
"""Test that optional params are NOT in payload when not provided."""
mock_response = Mock()
mock_response.ok = True
mock_response.json.return_value = {"id": "file123"}
mock_post.return_value = mock_response
with tempfile.NamedTemporaryFile(
mode="w", suffix=".txt", delete=False
) as f:
f.write("content")
temp_path = f.name
try:
self.tool._run(file_path=temp_path)
payload = mock_post.call_args.kwargs["json"]["integration"]
assert "parent_folder_id" not in payload
assert "description" not in payload
finally:
os.unlink(temp_path)
@patch.dict(
"os.environ",
{"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"},
clear=True,
)
@patch(
"crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool.requests.post"
)
def test_api_error_response(self, mock_post):
"""Test handling of API error responses."""
mock_response = Mock()
mock_response.ok = False
mock_response.json.return_value = {
"error": {"message": "Unauthorized"}
}
mock_post.return_value = mock_response
with tempfile.NamedTemporaryFile(
mode="w", suffix=".txt", delete=False
) as f:
f.write("content")
temp_path = f.name
try:
result = self.tool._run(file_path=temp_path)
assert "API request failed" in result
assert "Unauthorized" in result
finally:
os.unlink(temp_path)
@patch.dict(
"os.environ",
{
"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token",
"CREWAI_FACTORY": "true",
},
clear=True,
)
@patch(
"crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool.requests.post"
)
def test_ssl_verification_disabled_for_factory(self, mock_post):
"""Test that SSL verification is disabled when CREWAI_FACTORY=true."""
mock_response = Mock()
mock_response.ok = True
mock_response.json.return_value = {"id": "file123"}
mock_post.return_value = mock_response
with tempfile.NamedTemporaryFile(
mode="w", suffix=".txt", delete=False
) as f:
f.write("content")
temp_path = f.name
try:
self.tool._run(file_path=temp_path)
assert mock_post.call_args.kwargs["verify"] is False
finally:
os.unlink(temp_path)
@patch.dict(
"os.environ",
{"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"},
clear=True,
)
@patch(
"crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool.requests.post"
)
def test_ssl_verification_enabled_by_default(self, mock_post):
"""Test that SSL verification is enabled by default."""
mock_response = Mock()
mock_response.ok = True
mock_response.json.return_value = {"id": "file123"}
mock_post.return_value = mock_response
with tempfile.NamedTemporaryFile(
mode="w", suffix=".txt", delete=False
) as f:
f.write("content")
temp_path = f.name
try:
self.tool._run(file_path=temp_path)
assert mock_post.call_args.kwargs["verify"] is True
finally:
os.unlink(temp_path)
@patch.dict(
"os.environ",
{"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"},
clear=True,
)
@patch(
"crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool.requests.post"
)
def test_binary_file_upload(self, mock_post):
"""Test uploading a binary file (e.g. PNG)."""
mock_response = Mock()
mock_response.ok = True
mock_response.json.return_value = {"id": "file123"}
mock_post.return_value = mock_response
binary_content = bytes(range(256))
with tempfile.NamedTemporaryFile(
suffix=".png", delete=False
) as f:
f.write(binary_content)
temp_path = f.name
try:
self.tool._run(file_path=temp_path)
payload = mock_post.call_args.kwargs["json"]["integration"]
assert payload["mime_type"] == "image/png"
decoded = base64.b64decode(payload["content"])
assert decoded == binary_content
finally:
os.unlink(temp_path)