声明式用法(@sandbox 装饰器)
@sandbox 装饰器让你以声明式方式在远程沙箱中执行 Python 函数,无需手动管理沙箱生命周期。
基本用法
from agent_sandbox.declarative import sandbox
@sandbox(template="code-interpreter-v1")
def analyze(data):
import pandas as pd
df = pd.DataFrame(data)
return df.describe().to_dict()
# 调用时自动:创建沙箱 → 上传函数 → 执行 → 返回结果 → 销毁沙箱
result = analyze({"col1": [1, 2, 3], "col2": [4, 5, 6]})
print(result)
参数配置
@sandbox(
template="code-interpreter-v1", # 模板名称(默认 "code-interpreter-v1")
timeout=300, # 沙箱超时秒数(默认 300)
envs={"MY_KEY": "value"}, # 环境变量
packages=["pandas", "numpy"], # 远程预安装的 pip 包
serializer="json", # 序列化方式(默认 "json")
sandbox_id=None, # 复用已有沙箱 ID(不为 None 时跳过创建)
keep_alive=False, # True 时执行后不销毁沙箱
api_key=None, # API Key 覆盖
api_url=None, # 平台 API URL 覆盖
domain=None, # 域名覆盖
cpu=None, # CPU 核数
memory=None, # 内存 MB
)
def my_func():
pass
参数详解
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
template |
str |
"code-interpreter-v1" |
沙箱模板名称 |
timeout |
int |
300 |
沙箱超时秒数 |
envs |
dict |
None |
注入的环境变量(envd 作用域;详见环境变量指南) |
packages |
list[str] |
None |
远程预安装的 pip 包 |
serializer |
str |
"json" |
序列化模式(仅支持 json) |
sandbox_id |
str |
None |
复用已有沙箱 ID |
keep_alive |
bool |
False |
执行后不销毁沙箱 |
api_key |
str |
None |
API Key 覆盖 |
api_url |
str |
None |
平台 URL 覆盖 |
domain |
str |
None |
平台域名 |
cpu |
int |
None |
CPU 核数 |
memory |
int |
None |
内存 MB |
序列化规则
serializer 参数控制函数参数和返回值的序列化方式。当前仅支持 JSON 模式。
| 模式 | 额外依赖 | 支持类型 | 说明 |
|---|---|---|---|
json(默认) |
无 | JSON 可序列化类型 | 最安全,跨语言兼容 |
JSON 模式:参数通过 json.dumps 序列化,结果通过 json.loads 反序列化。不支持的类型会使用 default=str 回退。
与 async 结合
装饰器会自动检测被装饰函数是否为异步函数:
@sandbox(template="code-interpreter-v1")
async def async_analyze(data):
import pandas as pd
df = pd.DataFrame(data)
return df.describe().to_dict()
# 异步调用
result = await async_analyze({"col": [1, 2, 3]})
对于同步函数,装饰器生成同步包装器。若在已有事件循环中调用同步包装器,会通过 ThreadPoolExecutor 避免阻塞。
复用沙箱
# 创建沙箱并保持运行
@sandbox(template="code-interpreter-v1", keep_alive=True)
def setup():
import subprocess
subprocess.run(["pip", "install", "flask"])
return "ready"
sandbox_id = "sbx-xxxx" # 获取沙箱 ID
# 复用已有沙箱
@sandbox(sandbox_id=sandbox_id)
def use_flask():
from flask import Flask
return "Flask imported successfully"
注册式命令
除了装饰器模式,sandbox 对象还支持注册式命令:
from agent_sandbox.declarative import sandbox
@sandbox.register
def demo(x: int, y: str) -> str:
"""简单的演示命令"""
return f"{y}={x}"
# 同步调用
result = sandbox.run("demo", x=1, y="hello")
print(result) # "hello=1"
# 异步调用
result = await sandbox.run_async("demo", x=2, y="world")
# 查看已注册的命令
commands = sandbox.list_registered()
注册函数的参数仅支持标量类型:str、int、float、bool。
这是命名命令的唯一机制。注册在模板 commands.py 里(或用 @sandbox.register)的命令由沙箱内 server 提供,普通 Sandbox 用 await sandbox.run("demo", x=1, y="hello") 调用(POST /commands/demo)。template.yaml 没有 custom_commands 键,客户端会拒绝它。
文件传输
sandbox.register 只注册命名命令。文件上传和下载是 server file_ops 组的端点(默认开启),客户端用 sandbox.files 或 ServerClient.files。
启动 HTTP Server
sandbox.server.start(port=9000, host="0.0.0.0")
限制
- 参数序列化:函数参数和返回值必须是 JSON 可序列化的(不支持的类型会回退为
str) - 函数源码:装饰器通过
inspect.getsource()获取函数源码,Lambda 和动态生成的函数不支持 - 导入:远程沙箱中需要的库必须通过
packages参数预安装,或已包含在模板中 - 注册命令参数类型:
@sandbox.register仅支持标量类型(str/int/float/bool),不支持复杂类型 - 闭包变量:函数中引用的外部变量不会被自动序列化,请使用显式参数传递
- 执行环境隔离:被装饰的函数在独立的远程 Python 进程中执行,不共享本地进程的状态