Execute Code remotely in work bench
Process **REMOTE FILES** or script BULK TOOL EXECUTIONS using Python code IN A REMOTE SANDBOX. If you can see the data in chat, DON'T USE THIS TOOL.
**ONLY** use this when processing **data stored in a remote file** or when scripting bulk tool executions.
DO NOT USE
- When the complete response is already inline/in-memory, or you only need quick parsing, summarization, or basic math.
USE IF
- To parse/analyze tool outputs saved by COMPOSIO_MULTI_EXECUTE_TOOL to a remote file in the sandbox or to script multi-tool chains there.
- For bulk or repeated executions of known Composio tools (e.g., add a label to 100 emails).
- To call APIs via proxy_execute when no Composio tool exists for that API.
OUTPUTS
- Returns a compact result or, if too long, artifacts under `/mnt/files/.composio/output` (cloud-backed FUSE mount, persisted across sandbox restarts).
IMPORTANT CODING RULES:
1. Stepwise Execution: Split work into small steps. Save intermediate outputs to `/mnt/files/` (cloud-backed, persisted across failures/timeouts) or variables. Call COMPOSIO_REMOTE_WORKBENCH again for the next step.
2. Notebook Persistence: This is a persistent Jupyter notebook cell: variables, functions, imports, and in-memory state persist across executions. Helper functions are preloaded.
3. Top-level cells: Do not use `return`; Jupyter only allows it inside functions. For final values, end with `output` or `print(output)`, not `return output`.
4. Parallelism & Timeout (CRITICAL): There is a **hard 3-minute (180s) execution limit** per cell. Always prioritize PARALLEL execution using ThreadPoolExecutor for bulk operations - e.g., call run_composio_tool or invoke_llm across rows. If data is large, split it into smaller batches across cells.
5. Checkpoints: Save checkpoints to `/mnt/files/` so that long runs can be resumed from the last completed step, even after a timeout or sandbox restart.
6. Schema Safety: Never assume the response schema for run_composio_tool if not known already from previous tools. To inspect schema, either run a simple request **outside** the workbench or use invoke_llm helper.
7. LLM Helpers: Always use invoke_llm helper for summary, analysis, or field extraction on results; prefer it for much better results over ad hoc filtering.
8. Avoid Meta Loops: Do not use run_composio_tool to call COMPOSIO_* meta tools. Only use it for app tools.
9. Pagination: Use when data spans multiple pages. Continue fetching pages with the returned next_page_token or cursor until none remains. Parallelize page fetches when the tool supports page_number.
10. No Hardcoding: Never hardcode data. Load it from files or tool responses, iterating to construct intermediate or final inputs/outputs.
11. If the final output is in a workbench file, use upload_local_file to download it - never expose the raw workbench file path to the user. Prefer to download useful artifacts after task is complete.
ENV & HELPERS:
- Home directory: `/home/user`.
- NOTE: Helper functions already initialized in the workbench - DO NOT import or redeclare them:
-
`run_composio_tool(tool_slug: str, arguments: dict) -> tuple[Dict[str, Any], str]`: Execute a known Composio **app** tool. Do not invent names; match the tool input schema. Use for loops/parallel/bulk calls.
i) run_composio_tool returns JSON with top-level "data". Parse carefully—structure may be nested.
-
`invoke_llm(query: str) -> tuple[str, str]`: Invoke an LLM for semantic tasks. Pass MAX 200k characters.
i) NOTE Prompting guidance: When building prompts for invoke_llm, prefer f-strings (or concatenation) so literal braces stay intact. If using str.format, escape braces by doubling them ({{ }}).
ii) Define the exact JSON schema you want and batch items into smaller groups to stay within token limit.
- `upload_local_file(*file_paths) -> tuple[Dict[str, Any], str]`: Upload sandbox files to Composio S3/R2 storage for user-downloadable artifacts.
- `proxy_execute(method, endpoint, toolkit, query_params=None, body=None, headers=None) -> tuple[Any, str]`: Call a toolkit API directly when no Composio tool exists. Only one toolkit can be invoked with proxy_execute per workbench call
- `web_search(query: str) -> tuple[str, str]`: Search the web for information.
- `smart_file_extract(sandbox_file_path: str, show_preview: bool = True) -> tuple[str, str]`: Extracts text from files in the sandbox (e.g., PDF, image).
All helper functions return a tuple (result, error). Always check error before using result.
## Python Helper Functions for LLM Scripting
### run_composio_tool
Executes a known Composio tool via backend API. Do NOT call COMPOSIO_* meta tools to avoid cycles.
def run_composio_tool(tool_slug: str, arguments: Dict[str, Any]) -> tuple[Dict[str, Any], str]
# Returns: (tool_response_dict, error_message)
# Success: ({"data": {actual_data}}, "") - Note the top-level data
# Error: ({}, "error_message") or (response_data, "error_message")
result, error = run_composio_tool("GMAIL_FETCH_EMAILS", {"max_results": 1, "user_id": "me"})
if error:
print("GMAIL_FETCH_EMAILS error:", error)
else:
email_data = result.get("data", {})
print("Fetched:", email_data)
### invoke_llm
Calls LLM for reasoning, analysis, and semantic tasks. Pass MAX 200k characters.
# Returns: (llm_response, error_message)
# Example: analyze tool response with LLM
tool_resp, err = run_composio_tool("GMAIL_FETCH_EMAILS", {"max_results": 5, "user_id": "me"})
if not err:
parsed = tool_resp.get("data", {})
resp, err2 = invoke_llm(f"Summarize these emails: {parsed}")
if not err2:
print(resp)
# TIP: batch prompts to reduce LLM calls.
### upload_local_file
Uploads sandbox files to Composio S3/R2 storage for upload/download requests involving generated sandbox artifacts. Single files upload directly; multiple files are auto-zipped.
# Returns: (result_dict, error_string)
# Success: ({"s3_url": str, "uploaded_file": str, "type": str, "id": str, "s3key": str, "message": str}, "")
# Error: ({}, "error_message")
# Single file
result, error = upload_local_file("/path/to/report.pdf")
# Multiple files are auto-zipped
result, error = upload_local_file("/home/user/doc1.txt", "/home/user/doc2.txt")
if not error:
print("Uploaded:", result["s3_url"])
### proxy_execute
Direct API call to a connected toolkit service.
def proxy_execute(
method: Literal["GET","POST","PUT","DELETE","PATCH"],
endpoint: str,
toolkit: str,
query_params: Optional[Dict[str, str]] = None,
body: Optional[object] = None,
headers: Optional[Dict[str, str]] = None,
) -> tuple[Any, str]
# Returns: (response_data, error_message)
# Example: GET request with query parameters
query_params = {"q": "is:unread", "maxResults": "10"}
data, error = proxy_execute("GET", "/gmail/v1/users/me/messages", "gmail", query_params=query_params)
if not error:
print("Success:", data)
### web_search
Searches the web via Exa AI.
# Returns: (search_results_text, error_message)
results, error = web_search("latest developments in AI")
if not error:
print("Results:", results)
## Best Practices
### Error-first pattern and Defensive parsing (print keys while narrowing)
res, err = run_composio_tool("GMAIL_FETCH_EMAILS", {"max_results": 5})
if err:
print("error:", err)
elif isinstance(res, dict):
print("res keys:", list(res.keys()))
data = res.get("data") or {}
print("data keys:", list(data.keys()))
msgs = data.get("messages") or []
print("messages count:", len(msgs))
for m in msgs:
print("subject:", m.get("subject", "<missing>"))
### Parallelize within the 3-minute cell timeout
Adjust concurrency so all tasks finish within 3 minutes.
import concurrent.futures
MAX_CONCURRENCY = 10 # Adjust as needed
def process_one(item):
result, error = run_composio_tool("GMAIL_SEND_EMAIL", item)
if error:
return {"status": "failed", "error": error}
return {"status": "ok", "data": result}
with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_CONCURRENCY) as ex:
results = list(ex.map(process_one, items))