企业官网建设流程全解析

如何在 Composio 远程沙箱中运行 Python 批量处理并通过 /mnt/files 挂载交换文件

【免费下载链接】composioComposio powers 1000+ toolkits, tool search, context management, authentication, and a sandboxed workbench to help you build AI agents that turn intent into action.项目地址: https://gitcode.com/GitHub_Trending/co/composio

Composio 的远程沙箱（remote sandbox）是一个持久化的 Python 环境：agent 通过COMPOSIO_REMOTE_WORKBENCH元工具在其中执行 Python 代码，沙箱自带对 Composio 工具的编程访问和一组辅助函数。当任务涉及批量处理、数据转换或多步逻辑，而单次工具调用装不下时，就用它。本文的操作路径是：创建一个开启沙箱的 session，把输入文件上传到持久挂载点/mnt/files/，在沙箱里分步运行 Python 处理，最后把结果文件下载回本地。

前置条件：

• 一个 Composio 项目 API key，以及 Python SDKcomposio（若要通过sandbox.sandbox_size选择更大算力档位，需要composio≥0.12.1，旧版本会静默丢弃该参数）；
• 一个本地输入文件（下文示例用./sales.csv）；
• 文档出处：远程沙箱、配置 session、Remote Workbench 元工具。

创建一个开启沙箱的 session

按 配置 session 的说明，session 默认就包含沙箱，会提供COMPOSIO_REMOTE_WORKBENCH和COMPOSIO_REMOTE_BASH_TOOL两个元工具，所以普通创建即可：

session = composio.sessions.create(user_id="user_123")

只有明确不需要代码执行时才传sandbox={"enable": False}；被禁用的 session 直接调用沙箱会收到 400 错误。sandbox是推荐的配置键名，旧的workbench键仍是受支持的别名。

批量任务如果涉及大 DataFrame 或内存量较大的中间结果，可以在创建 session 时指定更大的算力档位（默认standard为 1 vCPU / 1 GB RAM，可选medium2 vCPU / 2 GB、large4 vCPU / 4 GB、xlarge8 vCPU / 8 GB）：

session = composio.sessions.create( user_id="user_123", sandbox={ "sandbox_size": "large", }, )

注意两点：一是改sandbox_size会重建沙箱，内存状态（变量、import）会丢失，但/mnt/files/挂载会保留；二是沙箱目前不计费，但文档声明后续会按档位和运行时长计费。

把输入文件上传到 /mnt/files/

沙箱有一个持久文件挂载点/mnt/files/，代码读写这里，且挂载在沙箱重启后仍然存在。用session.experimental.files在应用和挂载之间搬文件。该 API 位于experimental命名空间下，文档明确提示其接口面可能在后续版本变化。

uploaded = session.experimental.files.upload("./sales.csv") print(uploaded.sandbox_mount_prefix, uploaded.mount_relative_path) # /mnt/files sales.csv

文档示例输出为：前缀/mnt/files、相对路径sales.csv，即沙箱内代码在/mnt/files/sales.csv读到这个文件。

挂载提供四个方法，批量场景常用前三个：

RemoteFile携带文件字节和一个带签名的downloadUrl，可用text()/buffer()读取，或save(path)写盘。它的expiresAt是下载链接的过期时间，不是文件本身的 TTL——挂载点上的文件没有你设置的有效期。所有文件都落在 session 默认的files挂载上，mountId参数用于按 ID 指定挂载点，但目前没有创建自定义挂载的 SDK 调用，实际操作中只用files。

在沙箱中分步执行 Python 批量处理

COMPOSIO_REMOTE_WORKBENCH接收一段code_to_execute（必填的 Python 代码字符串），沙箱是持久 Jupyter notebook：变量、import、文件和内存状态在多次调用之间保留。执行时有一条硬性约束：每个 cell 有 3 分钟（180 秒）的执行上限。因此批量处理要拆成多个小步骤，中间结果落盘到/mnt/files/作为检查点，超时可从最后完成的步骤恢复。

执行时还需要遵守文档给出的编码规则：

• 顶层 cell 不能写return（Jupyter 只允许在函数内），最终值用output或print(output)收尾；
• 所有辅助函数（run_composio_tool、invoke_llm、upload_local_file等）都已预加载，不要重新 import 或声明，且统一返回(result, error)元组，先检查error再用result；
• 不要在沙箱内用run_composio_tool调COMPOSIO_*元工具，它只用于应用工具。

沙箱预装了pandas、numpy、matplotlib、Pillow、PyTorch、reportlab；用到其他受支持但尚未安装的包时，沙箱会自动安装。

下面是一次典型的两段式批量处理。第一段读取上传的输入、做清洗并保存检查点：

import pandas as pd df = pd.read_csv("/mnt/files/sales.csv") df["revenue"] = df["qty"] * df["price"] df.to_csv("/mnt/files/sales_clean.csv", index=False) print("rows:", len(df))

第二段从检查点继续，生成汇总报告并写入挂载点。若单步耗时接近 3 分钟上限，把数据拆成更小的批次分到多个 cell 里跑：

import pandas as pd df = pd.read_csv("/mnt/files/sales_clean.csv") summary = df.groupby("category")["revenue"].sum().reset_index() summary.to_csv("/mnt/files/report.csv", index=False) print(summary.shape)

如果是批量调用 Composio 工具（例如对 100 封邮件逐个打标签），文档给出的模式是用ThreadPoolExecutor并行调用run_composio_tool，把并发数调到 3 分钟内能完成：

import concurrent.futures MAX_CONCURRENCY = 10 # 按任务量调整 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))

items必须从文件或工具响应里加载、循环构造，不要在代码里硬编码数据。文档还建议：用current_step（如FETCHING_EMAILS）和current_step_metric（格式"done/total units"，例如"10/100 emails"）参数标记当前进度，便于跨 cell 保持对齐。

结果过长时，返回值会被截断并给出文件路径：元工具响应里的results_file_path/stdout_file_path指向完整内容，沙箱侧的大产物落在/mnt/files/.composio/output（该目录同样是持久挂载）。若最终产物是沙箱文件，文档要求用upload_local_file生成可下载链接，不要把原始 workbench 文件路径暴露给用户。

把结果文件下载回本地

沙箱写出/mnt/files/report.csv后，从应用侧取回：

report = session.experimental.files.download("/mnt/files/report.csv") report.save("./report.csv")

取回前也可以先用list()确认文件已落盘：

entries = session.experimental.files.list(path="/mnt/files")

能列出report.csv且download(...).save(...)成功写入本地，即表示本次「上传输入 → 沙箱批量处理 → 下载结果」的闭环完成。注意RemoteFile的expiresAt只是下载链接的过期时间；要长期保留，及时save到磁盘。

限制与常见排查点

• 单 cell 3 分钟上限：超时后内存中未落盘的结果会丢。解法就是文档的主模式——把工作拆成多个 cell，每步把中间结果写入/mnt/files/，下次调用从检查点恢复。
• 换档位或沙箱重启：改sandbox_size在下次访问时重建沙箱，in-memory 状态清空，/mnt/files/不受影响。所以检查点要写挂载点，不要只留在变量里。
• experimental命名空间：session.experimental.files的接口面可能在未来版本变化，升级 SDK 后核对这四个方法是否仍在。
• 沙箱被禁用：若 session 建时传了sandbox={"enable": False}，沙箱调用返回 400，这是配置行为而非故障。
• 字段名不匹配：沙箱会对 agent 生成代码中的常见笔误做纠偏（例如把result["apiKey"]解析为实际字段api_key），但不能依赖它替代正确的 schema 检查；不确定run_composio_tool的返回结构时，先在沙箱外发一个简单请求或用invoke_llm查看。
• 如果你想在自有的环境里跑同样的工具调用（Composio 只保留鉴权和发现），可以看 Local sandbox。

【免费下载链接】composioComposio powers 1000+ toolkits, tool search, context management, authentication, and a sandboxed workbench to help you build AI agents that turn intent into action.项目地址: https://gitcode.com/GitHub_Trending/co/composio