Python SDK 参考
opendsec 是 OpenDsec 的异步 Python SDK,基于 gRPC 和 asyncio。这一页列出它的全部类和方法。概念(后端、状态、超时、会话、权限)见核心概念。
- 要求 Python 3.10 或以上,依赖
grpcio>=1.73.1、protobuf>=6.31,<8、googleapis-common-protos>=1.60。 - 许可证 Apache-2.0:调用 OpenDsec 的程序不受服务端 AGPL 约束。
- 所有方法都是协程,要在
async函数里await。
安装
# 从源码仓库
python3 -m venv .venv && .venv/bin/pip install ./sdk/python
# 从管理员给的 wheel
pip install opendsec-0.11.0-py3-none-any.whl离线环境:离线安装包的 wheels/ 目录里有 SDK 和它的全部依赖,用 pip install --no-index --find-links wheels opendsec。
快速示例
import asyncio
from opendsec import DSecClient, DSecRunArgs, DSecError
async def main():
async with DSecClient() as client: # 读取 OPENDSEC_* 环境变量
sb = await client.run_sandbox(DSecRunArgs(
backend="microvm", image="root/py:3.12",
cpu_cores_limit=1, memory_limit_mb=1024,
ttl_running_stop=600, # 空闲 10 分钟自动停止
network_rules={"pypi": True})) # 只放行 PyPI
try:
r = await sb.run_shell("pip install requests && python3 -c 'import requests; print(requests.__version__)'")
print(r.exit_code, r.stdout, r.stderr)
finally:
await sb.stop()
asyncio.run(main())连接:DSecClient
DSecClient(endpoint=None, token=None, project=None, ca_file=None,
server_name=None, retry_timeout=60)| 参数 | 环境变量 | 默认 | 说明 |
|---|---|---|---|
endpoint | OPENDSEC_ADDR | 127.0.0.1:7100 | apiserver 地址 host:port。多个地址用逗号分隔(10.0.0.11:7100,10.0.0.12:7100):请求分散到所有在线的 apiserver,一台宕机后自动避开 |
token | OPENDSEC_TOKEN | 空 | 身份凭证 dsk.<id>.<secret>,每个请求都带上 |
project | OPENDSEC_PROJECT | root | 默认项目,创建、列出、任务操作、镜像列表都用它 |
ca_file | OPENDSEC_CA | 无 | 集群 CA 证书。设置后使用 TLS;不设置则是明文,只用于开发集群 |
server_name | — | 无;多地址时为 api.opendsec.internal | 校验服务端证书时使用的名字。apiserver 证书都带有 api.opendsec.internal |
retry_timeout | — | 60 | 服务端不可达(unavailable,例如重启、切换)时自动重试的时长,秒。0 表示不重试 |
使用方式:
async with DSecClient() as client: # 推荐:进入时连接,退出时关闭
...
client = DSecClient(endpoint="10.0.0.10:7100", token=tok, project="root/team-a", ca_file="ca.crt")
await client.open() # 也可以手动打开和关闭
...
await client.close()连接参数(自动处理,通常不用关心):断线后 100 ms 起重连,最长退避 2 秒;每 10 秒发送一次 keepalive,静默断开的连接约 15 秒内被发现;单个消息最大 80 MB。
创建参数:DSecRunArgs
@dataclass
class DSecRunArgs:
backend: str = "sim"
image: str = ""
cpu_cores_limit: float = 1.0
memory_limit_mb: int = 2048
gpus: int = 0
gpu_mode: str = ""
disk_mb: int = 0
ttl_running_stop: int = 0
ttl_max: int = 0
network_rules: dict[str, bool] = {}
init_user: str = "root"
task_id: str = ""
qos: str = "LS"
labels: dict[str, str] = {}
project: str | None = None| 字段 | 说明 |
|---|---|
backend | microvm、container、fncall、fullvm 或 sim。默认是 sim(只模拟,不运行任何东西),实际使用时一定要指定 |
image | 镜像名 <项目>/<名字>:<标签>,例如 root/py:3.12。需要在镜像所属项目上有 layer.use |
cpu_cores_limit | CPU 核数,可以是小数 |
memory_limit_mb | 内存(MB)。用带内存的快照创建时,必须和快照时的值一致 |
gpus、gpu_mode | GPU 个数;gpu_mode 为 exclusive(独占)或 shared(共享),默认由节点决定 |
disk_mb | 可写盘大小(MB)。0 表示用后端默认值 |
ttl_running_stop | 空闲超时(秒)。0 表示集群默认(通常 1 小时) |
ttl_max | 最长寿命(秒),从创建时算,暂停时间也计入。0 表示集群默认(通常 1 天),上限通常 7 天 |
network_rules | 网络规则,例如 {"pypi": True}、{"*": True}、{"github.com:443": True}。见网络策略 |
init_user | sandbox 里默认的用户,也是会话的默认用户 |
task_id | 任务名,用于 pause_task 等批量操作 |
qos | LS(延迟敏感,默认)或 BE(尽力而为) |
labels | 任意键值标签,原样保存在 sandbox 的规格里 |
project | 创建到哪个项目,默认用客户端的 project |
DSecContainerRunArgs 是它的子类,默认 backend="container",并多一个 container_image 字段(与论文里的写法一致),设置了就代替 image:
DSecContainerRunArgs(container_image="root/py:3.12", memory_limit_mb=4096, cpu_cores_limit=4)创建与查询
run_sandbox(args, timeout=120) -> Sandbox
创建一个 sandbox,等它进入 RUNNING 后返回。
timeout:整个创建过程的时限(秒),None表示不限。- 服务端不可达时在
retry_timeout内自动重试。每次创建带一个随机的幂等键:即使第一次的应答丢了,重试拿到的也是第一次创建的那个 sandbox,不会多建一个。 - 需要:项目上的
sandbox.create;镜像所属项目上的layer.use。 - 常见错误:
quota_exceeded、no_capacity、invalid(例如镜像不存在、后端不存在、超时超过上限)、forbidden。
run_container(args, timeout=120) -> Sandbox
与 run_sandbox 相同,参数通常是 DSecContainerRunArgs。
run_sandboxes(args, count, timeout=600) -> list[Sandbox | DSecError]
批量创建 count 个相同规格的 sandbox,一次最多 32768 个。
- 返回列表按序号排列;成功的是
Sandbox,失败的是DSecError(不会抛出)。 - 服务端按限流速率排队,大批量不会因限流部分失败。
- 断线后自动重试,只补齐还没有创建的部分。
- 批量请求本身失败(例如没有权限)时直接抛出
DSecError。
res = await client.run_sandboxes(DSecRunArgs(backend="microvm", image="root/py:3.12", task_id="job-7"), 500)
ok = [x for x in res if not isinstance(x, DSecError)]
failed = [x for x in res if isinstance(x, DSecError)]run_containers(args, count, timeout=600)
与 run_sandboxes 相同。
get_sandbox(sandbox_id) -> Sandbox
按 id 取 sandbox。需要它所在项目上的 sandbox.operate。已结束的 sandbox 在节点上保留 10 分钟,之后返回 not_found。
list_sandboxes(task_id="", include_ended=False, project=None) -> list[Sandbox]
列出项目(含全部子项目)里的 sandbox。task_id 只列出这个任务的;include_ended=True 包括最近结束的。需要 sandbox.operate。
cluster_status() -> Snapshot
集群快照:每个 edge 的地址、健康、是否排空、后端、容量、已分配资源、各后端的 sandbox 数、内存和磁盘余量、压力原因等(protobuf Snapshot,字段见 proto/opendsec/v1/cluster.proto 和 types.proto 的 EdgeState)。
sandbox 句柄:Sandbox
run_sandbox、get_sandbox 等返回 Sandbox 对象。
| 属性 | 说明 |
|---|---|
id | sandbox id |
state | 状态名字符串,例如 "RUNNING"。这是最后一次看到的状态,调用 refresh() 更新 |
edge_id | 所在节点 |
info | 完整记录(protobuf Sandbox):project、principal、spec、created_at、last_active_at、ended_at、message |
生命周期
所有生命周期方法都返回 Sandbox 本身(info 已更新),并在服务端不可达时自动重试。需要 sandbox.operate。
| 方法 | 说明 |
|---|---|
refresh() | 重新获取记录 |
stop() | 停止并清理。配额立即归还 |
pause() | 暂停。microVM 拍快照后结束进程,内存全部释放;container 冻结进程 |
resume() | 恢复。不调用也可以:对暂停的 sandbox 发起运行时调用会自动恢复 |
touch() | 刷新空闲计时;对暂停的 sandbox 会恢复它 |
update_network(rules, replace=False) | 修改网络规则:默认合并进现有规则,replace=True 整体替换。收回的放行对新连接立即生效,已有连接也会被切断 |
执行命令
run_shell(cmd, timeout=None, session="default", env=None, cwd=None, user=None) -> ShellResult在会话里执行一条 shell 命令,等它结束后返回结果。
| 参数 | 说明 |
|---|---|
cmd | 整条 shell 命令,由会话的 shell 解释,可以用管道、&& 等 |
timeout | 秒。到期后杀掉这条命令启动的进程,会话保留,ShellResult.timed_out 为 True |
session | 会话名。同一会话里 cd、export 在命令之间保留;不同会话互不影响 |
env、cwd | 只对这一条命令生效的环境变量和工作目录 |
user | 会话第一次执行命令时确定的用户:名字,或 uid[:gid](例如 "1000")。之后对同一会话传不同的值不起作用 |
断线(edge 或 apiserver 重启)时自动重试,不会把命令执行两次。
r = await sb.run_shell("cd /srv && export X=1")
r = await sb.run_shell("pwd; echo $X") # /srv 1
r = await sb.run_shell("pytest -q", timeout=600, cwd="/srv/repo")
if not r.ok:
print(r.exit_code, r.stderr)ShellResult 的字段:
| 字段 | 说明 |
|---|---|
cmd_id | 命令在会话里的编号 |
exit_code | 退出码 |
stdout、stderr | 输出(UTF-8 解码,无法解码的字节替换为 �);原始字节在 stdout_bytes、stderr_bytes |
done | 是否已结束(后台命令可能为 False) |
timed_out | 是否因超时被杀 |
duration | 执行时长(秒) |
truncated | 输出超过了单次调用能返回的上限(每个流 16 MB),没有全部返回;完整的输出(每个流最多 64 MB)仍在记录里,用 get_output(cmd_id, stdout_offset=...) 按偏移量继续取 |
dropped_bytes | 超过每个流 64 MB 记录上限后丢弃的字节数 |
ok | done and exit_code == 0 and not timed_out |
后台命令
cmd_id = await sb.run_shell_background("python3 train.py", session="train")
...
r = await sb.get_output(cmd_id, session="train", wait=30) # 最多等 30 秒
print(r.done, r.stdout)run_shell_background(cmd, timeout=None, session="default", env=None, cwd=None, user=None) -> int:启动后立即返回命令编号。get_output(cmd_id, session="default", wait=0, stdout_offset=0, stderr_offset=0) -> ShellResult:取已记录的输出。wait秒内等命令结束;stdout_offset、stderr_offset从指定字节位置开始取,用于增量读取。
流式输出
async for kind, data in sb.stream_shell("make test", timeout=1800):
if kind == "stdout":
sys.stdout.buffer.write(data)
elif kind == "stderr":
sys.stderr.buffer.write(data)
elif kind == "exit":
print("exit code", data.exit_code) # data 是不含输出的 ShellResultstream_shell(cmd, timeout=None, session="default", env=None, cwd=None, user=None) 依次产出 ("stdout", bytes)、("stderr", bytes),最后是 ("exit", ShellResult)。连接中断时从断开的位置继续,命令在 sandbox 里一直在运行,输出不重复也不丢失。
会话
list_sessions() -> list[str]:sandbox 里现有的会话名。close_session(session):结束会话,杀掉它启动的所有进程。
文件
所有路径都是 sandbox 里的绝对路径;文件操作以会话的用户身份执行(session 参数,默认 default)。
| 方法 | 说明 |
|---|---|
read_file(path, offset=0, length=0) -> bytes | 读文件。length=0 读到末尾 |
write_file(path, data, mode=0, append=False) -> FileInfo | 写文件,data 是 bytes 或 str(按 UTF-8 编码)。mode 为文件权限(例如 0o755),0 表示默认;append=True 追加 |
list_dir(path) -> list[FileInfo] | 列目录 |
stat(path) -> FileInfo | 文件信息 |
remove(path, recursive=False) | 删除文件;recursive=True 删除目录及其内容 |
FileInfo 的字段:name、type(file、dir、symlink、other)、size、mode、mtime。
大小限制:
- 一次
read_file最多返回 64 MB,更大的文件用offset和length分段读; - 一次
write_file的数据建议不超过 64 MB(单个消息上限 80 MB),更大的文件用append=True分段写; - 较早的代码里,经过 apiserver 的单个消息上限是 4 MB,超过时报
quota_exceeded(实际是消息过大)。CHANGELOG 0.11.0「Fixed」第一条修复了它:把服务端、Go 客户端和 Python SDK 都放宽到 80 MB,这类错误报invalid。
# 分段读一个大文件
chunks, off = [], 0
while True:
part = await sb.read_file("/data/big.bin", offset=off, length=32 << 20)
if not part:
break
chunks.append(part)
off += len(part)sandbox 里的 HTTP 服务
r = await sb.http("http://127.0.0.1:8000/health", method="GET", headers=None, body=b"", timeout=30)
print(r.status, r.headers, r.body)在 sandbox 的网络命名空间里发起 HTTP 请求,用来访问 sandbox 里运行的服务。返回 HttpResult(status, headers, body)。需要直接的 TCP 连接(例如 WebSocket),用 API 的 ConnectPort,或通过 E2B 网关访问端口。
fork、快照与制作镜像
fork(count=1, ttl_running_stop=0, ttl_max=0, labels=None, task_id="", timeout=600) -> list[Sandbox | DSecError]把运行中的 sandbox 复制成 count 个(1 到 256)运行中的 sandbox。
- 源 sandbox 短暂暂停后继续运行,
id和超时都不变。 - 副本继承源的规格;
labels追加到副本的标签上,task_id替换任务名(默认沿用源的);ttl_running_stop、ttl_max是副本的超时(0 为集群默认)。 - microVM 的副本连同内存复制,进程从 fork 时的状态继续;container 只复制文件。
- 结果按序号返回,失败的是
DSecError。每个副本单独计入配额。断线后重试只补齐未完成的部分。 - 需要:项目上的
sandbox.create;源镜像所属项目上的layer.use。
create_snapshot(image, memory=False, overwrite=False) -> ImageInfo把 sandbox 原样保存为镜像 image(<项目>/<名字>:<标签>),不清理任何文件。
memory=True(只有 microVM):连内存一起保存,用这个镜像创建的 sandbox 直接从快照恢复,进程从保存时的状态继续运行。创建时memory_limit_mb要和快照时一致。- microVM 必须
memory=True。需要只含文件的镜像时,用 container 构建,再pack_diff。 overwrite=True覆盖同名镜像。- 需要镜像所属项目上的
layer.publish。
pack_diff(image, scrub=(), overwrite=False) -> ImageInfo把 sandbox 相对它启动镜像的改动保存为新镜像 image,叠在原镜像之上(只有 container 类后端)。
- 默认清理的路径:
/tmp/*、/var/tmp/*、/root/.cache、/root/.bash_history、/home/*/.bash_history、/root/.python_history。scrub追加更多(相对于/的 glob,例如["srv/repo/.git"])。 - 得到的镜像 container 和 microVM 都能用。
- 需要镜像所属项目上的
layer.publish。
sb = await client.run_container(DSecContainerRunArgs(container_image="root/rl/python:3.12",
network_rules={"pypi": True, "github": True}))
await sb.run_shell("git clone https://github.com/org/repo /srv/repo && pip install -e /srv/repo")
img = await sb.pack_diff("root/rl/repo-env:v1")
await sb.stop()
# 之后直接用 root/rl/repo-env:v1 创建,不必重复安装ImageInfo 的字段:name、layers(从底到顶)、env、workdir、user、parent、source、created_at、created_by、snapshot、memory。
任务操作
| 方法 | 说明 |
|---|---|
pause_task(task_id, project=None) | 暂停任务的所有 sandbox |
resume_task(task_id, project=None) | 恢复任务的所有 sandbox |
stop_task(task_id, project=None) | 停止任务的所有 sandbox |
返回 TaskActionResult:affected(成功个数)、failed(失败个数)、errors(失败原因)。需要 sandbox.operate。
镜像管理
| 方法 | 说明 | 权限 |
|---|---|---|
list_images(project=None) -> list[ImageInfo] | 列出项目及其子项目的镜像 | layer.use |
get_image(name) -> ImageInfo | 镜像详情 | layer.use |
compose_image(name, parts, env=None, workdir="", user="", overwrite=False) -> ImageInfo | 把 parts 里的镜像从底到顶叠成新镜像,不复制数据。env 追加环境变量,workdir、user 设置默认工作目录和用户 | 新镜像所属项目上 layer.publish,每个 part 上 layer.use |
delete_image(name) | 删除镜像的登记。它的层由 opendsecctl image gc 回收 | layer.publish |
await client.compose_image("root/rl/swe-4821:v1",
["root/rl/python:3.12", "root/rl/django-ws:a1b2", "root/tools/agent-kit:3"])从镜像仓库导入只能用命令行(opendsecctl image import),因为要在挂载了镜像库的机器上写入数据。
项目、身份与 token
这些方法和 opendsecctl 用的是同一套 API,人和 agent 都可以调用,权限规则相同(项目、身份与权限)。
| 方法 | 说明 | 需要 |
|---|---|---|
whoami() -> str | 当前 token 对应的身份 id | — |
create_project(parent, name, quota=None) -> Project | 在 parent 下建子项目。quota 例如 {"sandboxes": 100, "cpu_cores": 400},不写的键继承上级 | parent 上的 project.admin |
get_project(project=None) -> Project | 项目详情:path、quota、usage(当前用量)、policies(每个身份的权限)、children | 项目上的任意权限 |
set_quota(project, quota) -> Project | 设置配额(整体替换),不能超过任何祖先 | 上级项目的 project.admin |
delete_project(project) | 删除空项目(没有子项目、没有用量、没有属于它的身份) | 上级项目的 project.admin |
create_principal(project, principal_id, kind="agent", ttl=0) -> str | 在项目里建身份,返回它的第一个 token(只返回这一次)。ttl 是 token 有效期(秒),0 为服务端默认 | project.admin |
delete_principal(principal) | 删除身份和它的全部 token | 身份所属项目的 project.admin |
grant(project, principal, perms) -> Project | 授予权限,例如 ["sandbox.create", "sandbox.operate"]。只能授予自己拥有的权限 | project.admin 及要授予的每个权限 |
revoke(project, principal, perms=()) -> Project | 收回权限;perms 为空时收回全部 | project.admin |
create_token(principal=None, ttl=0, description="") -> (str, TokenInfo) | 签发新 token,默认给自己 | 给自己:无;给别人:其所属项目的 project.admin |
list_tokens(principal=None) -> list[TokenInfo] | 列出 token(不含密钥):id、created_at、expires_at(0 为永不过期)、description、created_by | 同上 |
revoke_token(principal, token_id) | 撤销 token,约 1 秒内在所有 apiserver 上失效 | 同上 |
get_usage(project=None, start="", end="") -> UsageReport | 项目及其子项目按 UTC 日期的用量,start、end 为 YYYY-MM-DD(含) | project.admin |
UsageReport.rows 的每一行:project、day、sandboxes_started、sandbox_hours、cpu_core_hours、memory_gb_hours、gpu_hours。
错误:DSecError
所有失败都抛出 DSecError(批量创建和 fork 的单个失败项除外,它们放在结果列表里)。
from opendsec import DSecError, QUOTA_EXCEEDED, NO_CAPACITY
try:
sb = await client.run_sandbox(args)
except DSecError as e:
if e.reason in (QUOTA_EXCEEDED, NO_CAPACITY):
await asyncio.sleep(5) # 稍后重试
else:
raisee.reason:稳定的原因字符串,模块里有同名常量:UNAUTHENTICATED、FORBIDDEN、NOT_FOUND、QUOTA_EXCEEDED、RATE_LIMITED、NO_CAPACITY、INVALID、UNAVAILABLE、TIMEOUT、INTERNAL。各自的含义和处理方法见核心概念 §9。e.message:服务端给出的说明。
自动重试的范围:只有 unavailable(服务端不可达)会在 retry_timeout 内自动重试,适用于创建、批量创建、fork、生命周期操作、执行命令、读文件、列目录、会话和流式输出。write_file、remove、http、update_network、pack_diff、create_snapshot 和管理类方法不自动重试,失败时由调用方决定。
常见用法
强化学习 rollout
async def rollout(client, step_image, episode):
sb = await client.run_sandbox(DSecRunArgs(backend="microvm", image=step_image,
cpu_cores_limit=2, memory_limit_mb=2048,
task_id="ppo-run-42", ttl_max=3600))
try:
for action in episode:
r = await sb.run_shell(action, session="agent", user="1000", timeout=120)
...
score = await sb.run_shell("python3 /grader/grade.py", session="grader") # root 会话
return float(score.stdout)
finally:
await sb.stop()
# 训练被抢占
await client.pause_task("ppo-run-42")
# 恢复后,对每个 sandbox 的下一次调用会自动恢复它;也可以:
await client.resume_task("ppo-run-42")从同一个状态分支探索
base = await client.run_sandbox(DSecRunArgs(backend="microvm", image="root/rl/swe-4821:v1"))
await base.run_shell("cd /srv/repo && git apply /tmp/partial.patch")
branches = await base.fork(count=8, task_id="search-17")
results = await asyncio.gather(*(b.run_shell(f"python3 try.py --variant {i}")
for i, b in enumerate(branches) if not isinstance(b, DSecError)))预热好的环境
sb = await client.run_sandbox(DSecRunArgs(backend="microvm", image="root/rl/model-server:v1",
memory_limit_mb=4096))
# 启动一个常驻的服务:加载模型(慢)后在 8000 端口等待请求
await sb.run_shell("(nohup python3 /srv/serve.py > /tmp/serve.log 2>&1 &)")
await sb.run_shell("until curl -sf http://127.0.0.1:8000/ready; do sleep 1; done", timeout=600)
await sb.create_snapshot("root/rl/model-server:warm", memory=True)
await sb.stop()
# 之后从这个快照创建:服务进程和已加载的模型都在,约 55 ms 就能接请求
fast = await client.run_sandbox(DSecRunArgs(backend="microvm", image="root/rl/model-server:warm",
memory_limit_mb=4096))更多示例在仓库的 sdk/python/examples/:quickstart.py、test_features.py(逐项检查各功能)、load_test.py(压测)、decoupled_rollout.py(解耦 rollout)。