CubeSandbox 快照、回滚与克隆实战指南：cubesandbox Python SDK 端到端示例详解

CubeSandbox 快照、回滚与克隆实战指南：cubesandbox Python SDK 端到端示例详解

【免费下载链接】CubeSandboxInstant, Concurrent, Secure & Lightweight Sandbox for AI Agents.项目地址: https://gitcode.com/GitHub_Trending/cu/CubeSandbox

快照、回滚与克隆是 CubeSandbox 面向 AI Agent 场景提供的三组高级能力：create_snapshot()把运行中沙箱的完整状态（内存 + 文件系统）持久化为可复用镜像，clone()从运行中沙箱一行代码派生 N 个相互隔离的副本，rollback()将沙箱原地还原到历史快照并继续执行。本文以仓库中的端到端示例目录 examples/snapshot-rollback-clone 为骨架，逐脚本讲解每个 API 的用法、运行方式和验证点，并结合 SDK 源码与 CubeAPI 后端实现说明其底层原理，帮助你把这些能力直接落地到自己的 Agent 工作流中。

能力总览：三个 API 分别解决什么问题

在 AI Agent 的迭代式执行中，经常需要在"状态可回溯、结果可复制、分支可并行"之间取得平衡。CubeSandbox 用三个接口完成了这一闭环：

三者的协作关系可以用下面的流程图概括：

快照是三者共同的枢纽：clone()内部通过"快照一次、创建 N 次"实现扇出，rollback()则把沙箱进程从快照镜像重新启动来达成原地还原。完整的概念阐述可参考 docs/guide/snapshot-rollback-clone.md。

需要说明的是，快照、回滚与克隆是 CubeSandbox 独占能力——E2B 原版 SDK 没有对应接口。cubesandboxSDK 与 E2B SDK 兼容，可以作为 drop-in 替换使用，同时额外提供这些高级特性。

环境准备与运行方式

所有示例脚本都依赖cubesandboxPython SDK（0.2.0 及以上版本），可以直接安装或通过目录下的 requirements.txt（内容为cubesandbox>=0.2.0）安装：

pip install "cubesandbox>=0.2.0" # 或： pip install -r requirements.txt export CUBE_API_URL=http://127.0.0.1:3000 export CUBE_TEMPLATE_ID=tpl-xxxxxxxxxxxxxxxxxxxxxxxx

两个环境变量的用途：

• CUBE_API_URL：CubeAPI 服务的地址，本地部署时默认为http://127.0.0.1:3000；
• CUBE_TEMPLATE_ID：创建沙箱所用的基础模板 ID，格式为tpl-开头的字符串。

共享环境辅助脚本 env.py 统一从环境变量读取配置：未设置CUBE_TEMPLATE_ID时它会输出错误并以退出码 2 终止，并提示可用cubemastercli tpl list查询模板；当 CubeAPI 走 HTTPS 时，还可以通过可选的SSL_CERT_FILE指定集群根 CA 证书路径。

运行某个示例非常简单——每个脚本都是独立、自包含、可直接运行的：

python 01_create_snapshot.py python 04_state_preserved.py python 09_rollback.py # ...

目录中还附带clone_demo.py、rollback_demo.py以及一组bench_*.py并发基准脚本（如bench_snapshot_concurrency.py、bench_rollback_concurrency.py、bench_clone_concurrency.py），方便在真实集群上评估各操作的并发吞吐。

示例清单

快照：创建、列出与删除

创建快照（Demo 01）

sb.create_snapshot()会短暂暂停沙箱，捕获其完整状态（内存 + 文件系统）后恢复运行，并返回一个包含snapshot_id的SnapshotInfo对象：

from cubesandbox import Sandbox from env import TEMPLATE_ID with Sandbox.create(template=TEMPLATE_ID) as sb: print(f"sandbox: {sb.sandbox_id}") snapshot = sb.create_snapshot() print(f"snapshot created: {snapshot.snapshot_id}") print(f"use as template: Sandbox.create(template='{snapshot.snapshot_id}')") # Cleanup Sandbox.delete_snapshot(snapshot.snapshot_id) print("snapshot deleted")

从 SDK 源码看，create_snapshot的实现位于 sdk/python/cubesandbox/sandbox.py：它向POST /sandboxes/:sandboxID/snapshots发送请求，返回的SnapshotInfo包含snapshot_id与names两个字段（定义见 sdk/python/cubesandbox/_models.py）。该方法还接受一个可选的name参数：当同名模板已存在时，新的快照构建会挂接到既有模板上而非新建模板。

在后端侧，CubeAPI 对应的处理逻辑位于 CubeAPI/src/handlers/snapshots.rs，创建成功返回 HTTP 201 与SnapshotInfo；沙箱不存在时返回 404。

快照的生命周期独立于沙箱（Demo 05）

快照创建后，其生命周期与源沙箱解耦——即使源沙箱被kill()掉，快照依然可用。Demo 05 完整演示了这一特性：

sb = Sandbox.create(template=TEMPLATE_ID) snap = sb.create_snapshot() snapshot_id = snap.snapshot_id sb.kill() print(f"sandbox killed: {sandbox_id}") # 遍历分页，确认快照仍然存在 all_ids = [] items, token = Sandbox.list_snapshots() while True: all_ids.extend(s.snapshot_id for s in items) if not token: break items, token = Sandbox.list_snapshots(next_token=token) if snapshot_id in all_ids: print(f"OK: snapshot {snapshot_id} still exists after sandbox kill")

这一点在 SDK 与后端文档中都有明确佐证：delete_snapshot的 docstring 指出"删除源沙箱不会级联删除其快照"（sandbox.py）；后端同样明确"快照以模板形式存储，必须显式删除"。

列出快照：全量、过滤与分页（Demo 02）

Sandbox.list_snapshots()返回(list[SnapshotInfo], next_token)二元组，支持三种用法：

# 2a. 全量遍历（分页） items, token = Sandbox.list_snapshots() while True: for snap in items: print(f" {snap.snapshot_id}") if not token: break items, token = Sandbox.list_snapshots(next_token=token) # 2b. 按 sandbox_id 过滤 items, _ = Sandbox.list_snapshots(sandbox_id=sandbox_id)

分页语义为：next_token为None表示没有更多页；非None则将其作为nextToken参数传回以获取下一页。从源码看（sandbox.py），该方法支持三个可选参数：sandbox_id（按源沙箱 ID 过滤，对应请求参数sandboxID）、limit（页大小，默认 100）、next_token（分页游标）。请求路径为GET /snapshots，下一页游标从响应头x-next-token中读取。

在后端 CubeAPI/src/handlers/snapshots.rs 中，分页令牌同样通过x-next-token响应头返回，与 SDK 的读取逻辑一一对应。

删除快照（Demo 11）

Sandbox.delete_snapshot(snapshot_id)是类方法，不需要某个特定沙箱实例：

sb = Sandbox.create(template=TEMPLATE_ID) snap = sb.create_snapshot() snapshot_id = snap.snapshot_id sb.kill() Sandbox.delete_snapshot(snapshot_id) # 验证快照已从列表中消失 all_ids = [] items, token = Sandbox.list_snapshots() while True: all_ids.extend(s.snapshot_id for s in items) if not token: break items, token = Sandbox.list_snapshots(next_token=token) assert snapshot_id not in all_ids print(f"OK: snapshot {snapshot_id} removed from list")

删除快照在 HTTP 层走的是DELETE /templates/:templateID而非独立的/snapshots/{id}端点——因为快照本质上以模板形式存储在系统中。后端代码 CubeAPI/src/handlers/snapshots.rs 对此有明确注释：不额外暴露/snapshots/{id}的 DELETE 端点，是为了避免两个入口返回不一致的响应形态，同时把"这个 ID 是不是快照"的判定收敛到delete_template一处。

从快照派生新沙箱：状态保留验证

用template=从快照启动沙箱（Demo 03）

快照 ID 可以直接当作模板 ID 传给Sandbox.create(template=...)，新沙箱将从快照捕获的内存 + 文件系统状态精确启动：

from cubesandbox import Sandbox from env import TEMPLATE_ID with Sandbox.create(template=TEMPLATE_ID) as src: snapshot = src.create_snapshot() snapshot_id = snapshot.snapshot_id print(f"snapshot created: {snapshot_id}") with Sandbox.create(template=snapshot_id) as cloned: print(f"cloned sandbox: {cloned.sandbox_id}") Sandbox.delete_snapshot(snapshot_id)

状态保留的端到端验证（Demo 04）

Demo 04 用"写标记文件 → 快照 → 克隆 → 读标记"的方式，对状态保留做了断言式验证：

MARKER = "hello from snapshot" with Sandbox.create(template=TEMPLATE_ID) as src: src.run_code(f"open('/tmp/marker.txt','w').write('{MARKER}')") snapshot = src.create_snapshot() snapshot_id = snapshot.snapshot_id with Sandbox.create(template=snapshot_id) as cloned: result = cloned.run_code("print(open('/tmp/marker.txt').read())") content = result.logs.stdout[0].strip() if result.logs.stdout else "" assert content == MARKER, f"state not preserved: got {content!r}" print("OK: filesystem state preserved in cloned sandbox") Sandbox.delete_snapshot(snapshot_id)

注意这里快照捕获的不仅是文件系统，还包括内存状态——这意味着运行中的进程、已加载的模块、未刷盘的变量都处于同一快照语义下（create_snapshot的 docstring 明确说明"沙箱在快照创建期间会被临时暂停"）。如果你的 Agent 依赖内存态（如已训练的模型、已建立的会话），快照同样能保住这部分状态。

克隆：一行代码扇出 N 个沙箱

基础克隆（Demo 06）

sb.clone(n=N)是"快照 + 派生 + 清理"的一站式封装，调用后源沙箱保持运行，临时快照自动被清理（list_snapshots()不会再看到它）。其内部执行步骤如下（源码见 sandbox.py 的 docstring）：

1. self.create_snapshot()—— 捕获当前状态；
2. Sandbox.create(template=snapshot_id) × n—— 从该快照派生 N 个沙箱；
3. 将共享清理状态挂到每个克隆上，最后一个克隆被kill()时删除临时快照（best-effort）。

N = 3 src = Sandbox.create(template=TEMPLATE_ID) src.run_code("open('/tmp/shared.txt','w').write('shared state')") # ★ 一行克隆 —— SDK 内部处理 snapshot/create/delete clones = src.clone(n=N) print(f"cloned {len(clones)} sandboxes") for i, sb in enumerate(clones): result = sb.run_code("print(open('/tmp/shared.txt').read())") content = result.logs.stdout[0].strip() if result.logs.stdout else "" assert content == "shared state" print(f"OK: {N} sandboxes cloned via sb.clone(n={N}), all share initial state") src.kill() for sb in clones: sb.kill()

并发克隆（Demo 07）

默认情况下clone(n=N)串行创建子沙箱。对于大规模扇出场景（例如并行 Agent rollout），可以传入concurrency=C：

import os N = int(os.environ.get("FORK_N", "10")) CONCURRENCY = int(os.environ.get("FORK_CONCURRENCY", "5")) src = Sandbox.create(template=TEMPLATE_ID) src.run_code("open('/tmp/origin.txt','w').write('I am from sandbox a')") clones = src.clone(n=N, concurrency=CONCURRENCY) # 验证每个克隆都继承了源沙箱写入的标记 expect = "I am from sandbox a" ok = 0 for i, sb in enumerate(clones): r = sb.run_code("print(open('/tmp/origin.txt').read())") marker = r.logs.stdout[0].strip() if r.logs.stdout else "" if marker == expect: ok += 1 assert ok == N, "some clones failed to inherit state"

FORK_N与FORK_CONCURRENCY两个环境变量可调节扇出规模与并发度。并发语义上有几个关键点（均来自 SDK 源码实现）：

• 线程池：concurrency > 1时通过concurrent.futures.ThreadPoolExecutor分发，实际工作线程数为min(n, concurrency)，所以传一个大于 N 的值是无害的；
• 串行与并发等价：concurrency=1（默认）不启动任何线程，行为与串行循环逐字节一致；
• 快照只做一次：快照创建与删除各执行一次，只有"创建 N 个子沙箱"这一步并行；
• 全有或全无的失败语义：任一子任务失败时，所有已成功创建的克隆都会被自动kill()，随后抛出首个异常——调用方要么拿到恰好 N 个沙箱，要么拿到一个异常，不会产生孤儿资源。实现上（sandbox.py）会先排空所有 in-flight 的 future，收集成功结果与首个异常，再决定清理策略；
• 返回顺序：concurrency > 1时列表顺序不确定（按后端 create 调用返回顺序排列）；concurrency == 1时保持提交顺序。

继承性、隔离性与连续性（Demo 08）

Demo 08 用a.clone(n=2)产生b、c两个克隆，对三个性质逐一断言验证：

a = Sandbox.create(template=TEMPLATE_ID) a.run_code("open('/tmp/origin.txt','w').write('from a')") b, c = a.clone(n=2) # 继承性：b 和 c 都能看到 fork 之前 a 写入的文件 for sb, name in [(b, "b"), (c, "c")]: r = sb.run_code("print(open('/tmp/origin.txt').read())") assert marker == "from a" # 隔离性：b 的写入对 c 不可见 b.run_code("open('/tmp/b_only.txt','w').write('b')") r = c.run_code("import os; print(os.path.exists('/tmp/b_only.txt'))") assert leaked == "False", "isolation violated" # 连续性：a 仍在运行且状态不受影响 r = a.run_code("print(open('/tmp/origin.txt').read())") assert still == "from a"

三个性质的完整语义如下表：

回滚：原地还原并继续执行

基础回滚（Demo 09）

sb.rollback(snapshot_id)将沙箱原地还原到指定快照状态：文件系统被完整重置，sandbox_id 保持不变，sb对象在回滚后依然可用，无需重新创建连接。Demo 09 通过"v0 → 检查点 v1 → 写入 v2 → 回滚到 v1"的完整时间线验证：

# Step 1: 创建基础快照 v0 with Sandbox.create(template=TEMPLATE_ID) as src: src.run_code("open('/tmp/v.txt','w').write('v0')") base = src.create_snapshot() base_id = base.snapshot_id # Step 2: 从基础快照启动沙箱 sb = Sandbox.create(template=base_id) # Step 3: 写入 v1，打检查点 sb.run_code("open('/tmp/v.txt','w').write('v1')") checkpoint = sb.create_snapshot() checkpoint_id = checkpoint.snapshot_id # Step 4: 写入 v2，确认生效 sb.run_code("open('/tmp/v.txt','w').write('v2')") before = sb.run_code("print(open('/tmp/v.txt').read())").logs.stdout assert before[0].strip() == "v2" # Step 5: 回滚到 v1 检查点 sb.rollback(checkpoint_id) # Step 6: 验证状态回到 v1 after = sb.run_code("print(open('/tmp/v.txt').read())").logs.stdout assert after[0].strip() == "v1" print("OK: rollback restored state to checkpoint (v1)")

回滚的底层机制值得展开说明（源码见 sandbox.py）：

• HTTP 请求为POST /sandboxes/:sandboxID/rollback，请求体为{"snapshotID": ...}，成功后返回形如{"sandboxID": ..., "snapshotID": ..., "status": "success"}的响应；
• 回滚后沙箱进程从快照镜像重新启动，这会使此前对该沙箱保持的 TCP 连接全部失效（包括沙箱内的 jupyter-server 与 CubeAPI 的 keep-alive 连接池）；
• 为避免下一次run_code()与半关闭的 socket 竞争，SDK 会主动调用_reset_connections()关闭底层 HTTP 客户端池，使其在下一次使用时惰性重建——调用方不需要做任何事，回滚后直接sb.run_code(...)即可；
• _reset_connections()是幂等且 best-effort 的，即使客户端从未构建（从未调用过 run_code）或close()抛错也不会影响回滚本身。这一行为在 sdk/python/tests/test_sandbox.py 的Sandbox.rollback测试族中有完整覆盖（如test_rollback_closes_httpx_client_so_run_code_rebuilds、test_rollback_when_client_never_built_is_safe）。

后端侧，回滚处理逻辑位于 CubeAPI/src/handlers/snapshots.rs，沙箱或快照不存在时返回 404；CubeAPI 与 CubeMaster 之间的回滚协议（同步终端结果）在 CubeAPI/src/cubemaster/mod.rs 有对应客户端实现。

回滚后继续执行并再打快照（Demo 10）

回滚后的沙箱依然可写，可以继续执行、写入新状态，并在新分支上再次创建快照。这正是 Agent 重试循环的理想模式：在每个关键决策点打检查点，失败时回滚重试另一条分支。Demo 10 的时间线为v1（检查点）→ v2 → 回滚到 v1 → 写入 v3 → 再次快照：

sb = Sandbox.create(template=TEMPLATE_ID) # 写入 v1 并打检查点 sb.run_code("open('/tmp/v.txt','w').write('v1')") checkpoint = sb.create_snapshot() # 推进到 v2，然后回滚到 v1 sb.run_code("open('/tmp/v.txt','w').write('v2')") sb.rollback(checkpoint.snapshot_id) assert got == "v1" # 在回滚后的分支上继续：写入 v3 sb.run_code("open('/tmp/v.txt','w').write('v3')") assert got == "v3" # 从该分支再打新快照，并通过克隆验证 new_snap = sb.create_snapshot() with Sandbox.create(template=new_snap.snapshot_id) as forked: assert got == "v3" print("OK: rollback + continue + re-snapshot all consistent")

最佳实践

结合官方指南（docs/guide/snapshot-rollback-clone.md）与示例代码中的清理模式，总结出以下四条实践经验：

1. 快照不是免费的。每个快照都对应持久化存储中的一份完整镜像。不需要时应及时删除，或定期运行list_snapshots()盘点清理，避免存储膨胀。
2. with块不会删除快照。上下文管理器只负责kill()沙箱本身；用create_snapshot()创建的快照必须显式调用Sandbox.delete_snapshot()清理。观察所有示例可以发现，每个脚本在结束时都严格执行delete_snapshot，这正是推荐的习惯用法。
3. clone()会自动清理内部临时快照。对于临时扇出场景，无需手动管理快照生命周期，直接调用clone()即可，且失败时 SDK 保证不残留孤儿沙箱。
4. 大规模扇出优先使用clone(n=N, concurrency=C)。SDK 内部处理失败清理与临时快照，避免并发创建时出现部分成功、部分失败导致的资源泄漏；concurrency=1时行为与串行循环完全一致，可以放心从默认值开始调参。

API 速查

from cubesandbox import Sandbox sb = Sandbox.create(template=TEMPLATE_ID) # 快照 snap = sb.create_snapshot() # → SnapshotInfo(snapshot_id=...) items, token = Sandbox.list_snapshots() # 分页；token 为 None 表示结束 Sandbox.delete_snapshot(snap.snapshot_id) # 克隆（从运行中的沙箱一对多派生） clones = sb.clone(n=5) # 串行 clones = sb.clone(n=10, concurrency=4) # 通过线程池并发 # 回滚（原地，sandbox_id 保持不变） sb.rollback(snap.snapshot_id) # 把快照当作模板启动新沙箱 fresh = Sandbox.create(template=snap.snapshot_id)

延伸阅读

• 指南：快照、回滚与克隆 —— 本文对应的官方完整指南；
• 跨节点快照 —— S3 后端、remote_status=ready以及 Resume / FromSnap 跨节点调度的调度器规则；
• SDK 源码：sdk/python/cubesandbox（核心实现集中在 sandbox.py，模型定义见 _models.py）；
• SDK 测试：sdk/python/tests/test_sandbox.py ——clone/rollback/ 连接重置行为的单元测试覆盖；
• 后端处理：CubeAPI/src/handlers/snapshots.rs —— 快照创建、列表、回滚三个 HTTP 端点的 axum 实现；
• 基础用法示例：examples/code-sandbox-quickstart —— E2B 兼容的基础流程。

【免费下载链接】CubeSandboxInstant, Concurrent, Secure & Lightweight Sandbox for AI Agents.项目地址: https://gitcode.com/GitHub_Trending/cu/CubeSandbox