openai-agents-python 的 Daytona 沙箱后端:从客户端配置到云端工作区的完整实战指南
【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python
本指南系统讲解 openai-agents-python 中由agents.extensions.sandbox.daytona.sandbox模块实现的 Daytona 托管沙箱后端:如何用DaytonaSandboxClient与DaytonaSandboxClientOptions在云端创建隔离工作区,如何通过DaytonaSandboxSession执行命令、读写文件、运行 PTY 交互进程,如何借助persist_workspace/hydrate_workspace保存与恢复工作区,以及如何用DaytonaCloudBucketMountStrategy挂载 S3/R2/GCS/Azure Blob 云存储。读完本文,你将能基于该后端搭建可运行的沙箱 Agent(参考 examples/sandbox/extensions/daytona/daytona_runner.py),并理解其底层实现细节。
一、Daytona 沙箱扩展的定位
在 openai-agents-python 中,沙箱(sandbox)是 Agent 执行代码与文件操作的隔离环境。docs/sandbox/clients.md将沙箱客户端分为两类:本地客户端(UnixLocalSandboxClient、DockerSandboxClient)与托管客户端。Daytona 属于托管后端,适用于"需要生产级隔离、把工作区边界交给云端托管环境"的场景,安装方式为openai-agents[daytona]扩展。
agents.extensions.sandbox.daytona是官方扩展包,其__init__.py导出的核心符号包括:
DaytonaSandboxClient:管理沙箱生命周期(创建、恢复、删除)的客户端;DaytonaSandboxClientOptions:创建沙箱时的全部可调参数;DaytonaSandboxSession/DaytonaSandboxSessionState:沙箱会话实现与其可序列化状态;DaytonaSandboxResources/DaytonaSandboxTimeouts:资源与超时配置模型;DEFAULT_DAYTONA_WORKSPACE_ROOT:默认工作区根路径/home/daytona/workspace;DaytonaCloudBucketMountStrategy:rclone 云存储挂载策略(定义于mounts.py)。
从源码结构看,整个扩展是沙箱抽象层的 Daytona 实现:DaytonaSandboxClient继承BaseSandboxClient,DaytonaSandboxSession继承BaseSandboxSession,因此它可以无缝接入SandboxAgent+SandboxRunConfig的统一框架(见 src/agents/extensions/sandbox/daytona/sandbox.py 与 src/agents/extensions/sandbox/daytona/init.py)。
可选依赖与惰性导入
daytonaSDK 是可选依赖,源码采用了"惰性导入"策略:包级导出不强制导入该模块,模块内部的_import_daytona_sdk()等辅助函数会在真正使用时才加载AsyncDaytona、DaytonaConfig、CreateSandboxFromImageParams、CreateSandboxFromSnapshotParams等类;若未安装依赖,会抛出明确的ImportError:DaytonaSandboxClient requires the optional 'daytona' dependency。这意味着未安装该扩展的用户仍可正常导入整个agents包,只有实际使用 Daytona 客户端时才需要安装。对应的示例在导入失败时提示uv sync --extra daytona。
二、客户端与配置模型
2.1DaytonaSandboxClientOptions:创建沙箱的参数
DaytonaSandboxClientOptions继承BaseSandboxClientOptions,默认type="daytona",全部字段及默认值如下:
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
sandbox_snapshot_name | str \| None | None | 从命名快照创建沙箱(优先于image) |
image | str \| None | None | 从镜像创建沙箱 |
resources | DaytonaSandboxResources \| None | None | CPU/内存/磁盘资源规格 |
env_vars | dict[str, str] \| None | None | 注入沙箱的环境变量 |
pause_on_exit | bool | False | 退出时暂停(stop)而非删除沙箱 |
create_timeout | int | 60 | 创建沙箱的超时秒数 |
start_timeout | int | 60 | 启动沙箱的超时秒数 |
name | str \| None | None | 沙箱名称,缺省使用随机 session UUID |
auto_stop_interval | int | 0 | 自动停止间隔;0表示不自动停止 |
timeouts | DaytonaSandboxTimeouts \| dict \| None | None | 各操作超时,见 2.3 |
exposed_ports | tuple[int, ...] | () | 需要暴露的端口列表 |
exposed_port_url_ttl_s | int | 3600 | 签名预览 URL 的 TTL(秒),仅影响新连接建立 |
其中exposed_port_url_ttl_s的实现细节值得注意:源码注释明确指出,Daytona 仅在初始 HTTP 请求 / WebSocket 升级握手阶段校验签名预览 URL 的过期时间;已经建立的 WebSocket 连接在 URL 过期后仍保持连接,但重连或新握手必须使用重新解析的 URL。这解释了为何该 TTL 只作用于"新连接建立"。
2.2DaytonaSandboxResources:资源规格
class DaytonaSandboxResources(BaseModel): model_config = {"frozen": True} cpu: int | None = None memory: int | None = None disk: int | None = None三个字段均为可选的整数,None表示不指定、交给 Daytona 使用默认值。_build_create_params在构建CreateSandboxFromImageParams时,仅当三个字段任一非None时才会构造 SDK 的Resources对象。
2.3DaytonaSandboxTimeouts:操作超时集合
class DaytonaSandboxTimeouts(BaseModel): exec_timeout_unbounded_s: int = Field(default=24 * 60 * 60, ge=1) # 未指定 timeout 时的 exec 上限,默认 24 小时 keepalive_s: int = Field(default=10, ge=1) # 存活检测刷新超时 cleanup_s: int = Field(default=30, ge=1) # 会话清理/终止超时 fast_op_s: int = Field(default=30, ge=1) # 快速操作(如 mkdir、send_input)超时 file_upload_s: int = Field(default=1800, ge=1) # 文件上传超时(30 分钟) file_download_s: int = Field(default=1800, ge=1) # 文件下载超时(30 分钟) workspace_tar_s: int = Field(default=300, ge=1) # 工作区打包/解包超时(5 分钟)所有字段都有ge=1校验。当调用方未指定timeout时,_coerce_exec_timeout返回exec_timeout_unbounded_s(默认 24 小时);若显式传入timeout <= 0则会被钳制为0.001秒。
三、客户端生命周期:创建、恢复与关闭
DaytonaSandboxClient的构造签名:
DaytonaSandboxClient(*, api_key: str | None = None, api_url: str | None = None, instrumentation: Instrumentation | None = None, dependencies: Dependencies | None = None)构造函数通过DaytonaConfig(api_key=api_key, api_url=api_url)初始化AsyncDaytona客户端;当api_key与api_url均为None时使用 SDK 的默认配置(如DAYTONA_API_KEY环境变量)。从实现看,api_key与api_url通常由 Daytona 环境变量提供,示例代码中_require_env("DAYTONA_API_KEY")也印证了这一点。
3.1create:创建沙箱会话
async def create(self, *, snapshot: SnapshotSpec | SnapshotBase | None = None, manifest: Manifest | None = None, options: DaytonaSandboxClientOptions) -> SandboxSessioncreate的执行流程(src/agents/extensions/sandbox/daytona/sandbox.py):
manifest缺省时使用Manifest(root=DEFAULT_DAYTONA_WORKSPACE_ROOT),根目录为/home/daytona/workspace,并调用_validate_manifest_for_create校验;- 规范化
timeouts(None→ 默认实例,dict →model_validate); - 生成 session UUID,沙箱名缺省取 UUID 字符串;
_build_create_params按优先级选择创建方式:sandbox_snapshot_name>image> 默认快照参数;- 调用
await self._daytona.create(params, timeout=options.create_timeout)创建远程沙箱; - 组装
DaytonaSandboxSessionState并用DaytonaSandboxSession.from_state包装成会话,最后_wrap_session挂上Instrumentation监控。
_build_create_params的三分支逻辑:
- 有
sandbox_snapshot_name:CreateSandboxFromSnapshotParams(snapshot=..., env_vars=..., name=..., auto_stop_interval=...); - 有
image:CreateSandboxFromImageParams(image=..., env_vars=..., name=..., resources=..., auto_stop_interval=...),resources仅在至少指定一项资源时传入; - 两者皆无:退化为不带快照/镜像的
CreateSandboxFromSnapshotParams。
3.2resume:恢复已保存的会话
async def resume(self, state: SandboxSessionState) -> SandboxSessionresume只接受DaytonaSandboxSessionState(否则抛TypeError),先调用state.assert_path_grants_rebound()校验路径授权,然后:
- 通过
self._daytona.get(state.sandbox_id)查找已有沙箱;若沙箱状态不是SandboxState.STARTED,则调用daytona_sandbox.start(timeout=state.start_timeout)重新启动; - 若查找/启动失败(如沙箱已不存在),则用保存的
sandbox_snapshot_name/image/env_vars/resources/auto_stop_interval重建沙箱,并更新state.sandbox_id、重置workspace_root_ready; - 重新包装会话,并通过
_set_start_state_preserved(reconnected, system=reconnected)记录恢复方式。
这一"优先重连、失败重建"策略与沙箱会话可持久化设计相辅相成——DaytonaSandboxSessionState继承了SandboxSessionState,所有关键配置(sandbox_id、image、resources、timeouts、pause_on_exit等)都可序列化,因此会话状态可以跨进程保存后恢复。
3.3delete与close
delete(session)要求会话必须是DaytonaSandboxSession,调用inner.shutdown()后返回会话;close()关闭底层AsyncDaytonaHTTP 客户端;同时实现了__aenter__/__aexit__,支持async with用法。
3.4_shutdown_backend:退出策略
会话关闭时,_shutdown_backend依据pause_on_exit决定行为:
if self.state.pause_on_exit: await self._sandbox.stop() # 暂停:保留沙箱,便于下次恢复 else: await self._sandbox.delete() # 删除:彻底释放云端资源四、会话能力:命令执行、PTY、文件与工作区
DaytonaSandboxSession是核心会话类,所有操作都落在 Daytona SDK 的process与fs两个子对象上。
4.1 普通命令执行_exec_internal
非交互命令的完整链路如下:
- 用
shlex.join组装命令字符串; - 解析合并环境变量:
_resolved_envs()将base_env_vars与 manifest 的environment.resolve()结果合并; - 组装
cd <workspace_root> && env -- K=V ... <cmd>形式的 session 命令; - 生成
sandbox-<uuid 前 12 位>格式的 Daytona session ID; process.create_session→process.execute_session_command(SessionExecuteRequest(command=..., run_async=False)),两段调用都套上asyncio.wait_for以落实调用方超时;- 解析
exit_code、stdout、stderr返回ExecResult(字节流编码为 UTF-8,非法字节以replace容错);SDK 旧式结果只有output字段时,按退出码拆分为 stdout/stderr; finally中总是尝试process.delete_session清理,超时上限为cleanup_s。
超时与错误映射:asyncio.TimeoutError或 SDK 的DaytonaTimeoutError→ExecTimeoutError;其余异常 →_daytona_exec_transport_error构造的ExecTransportError(含backend="daytona"、HTTP 状态码、provider_error、provider_error_code、retryable等上下文)。
4.2 错误分类与重试语义
模块内置了 Daytona SDK 异常的分类表(_DAYTONA_HTTP_STATUS_RETRYABLE)与按类名导入异常的辅助函数:
- 可重试:
DaytonaRateLimitError、DaytonaTimeoutError、DaytonaConnectionError;HTTP 429/500/502/503/504; - 不可重试:
DaytonaNotFoundError、DaytonaAuthenticationError、DaytonaAuthorizationError、DaytonaValidationError、DaytonaConflictError;HTTP 400/401/403/404/409; - 消息包含
"is the sandbox started"或"no ip address found"时判定为sandbox_not_running(不可重试); asyncio.TimeoutError与 Daytona 超时异常在persist_workspace场景下被视为可重试(_retryable_persist_workspace_error_types)。
值得注意的是,Daytona 异常类型通过getattr(daytona_module, name)动态探测,SDK 版本缺少某类时自动降级为不匹配,体现了对 SDK 版本差异的兼容设计。persist_workspace中的 tar 命令执行还应用了retry_async装饰器,按上述可重试条件自动重试。
4.3 PTY 交互进程
supports_pty()返回True,说明 Daytona 后端支持交互式进程。PTY 实现分两条路径:
- TTY 模式(
tty=True):process.create_pty_session(id=..., on_data=..., cwd=..., envs=..., pty_size=PtySize(cols=80, rows=24))创建 PTY 会话,等待连接后send_input(cmd_str + "\n")发送命令,后台_run_pty_waiter等待进程退出并收集退出码; - 非 TTY 模式:
create_session+execute_session_command(SessionExecuteRequest(..., run_async=True))异步启动,再由_run_session_reader通过get_session_command_logs_async持续拉取 stdout/stderr 回调。
输出收集统一走collect_pty_output(chunks 队列 +output_notify事件 +yield_time_ms轮询),支持max_output_tokens截断,默认 yield 时间 10 秒(pty_exec_start)/ 250 毫秒(pty_write_stdin)。会话管理方面:
- 每个 PTY 进程分配整数
process_id(allocate_pty_process_id),受PTY_PROCESSES_MAX上限约束,达到上限时按last_used与输出关闭状态淘汰最久未用的进程(_prune_pty_sessions_if_needed); - 进程数达到
PTY_PROCESSES_WARNING阈值时打印告警日志; pty_write_stdin通过entry.pty_handle.send_input(chars)写入 stdin(仅 TTY 进程支持),随后采集输出;pty_terminate_all终止全部 PTY 会话并清空注册表。
4.4 文件读写
read(path):走fs.download_file(sandbox_path, file_download_s),返回io.BytesIO;SDK 抛DaytonaNotFoundError时映射为WorkspaceReadNotFoundError,其他异常映射为WorkspaceArchiveReadError;write(path, data):读取io.IOBase内容,str自动 UTF-8 编码,非bytes/bytearray抛WorkspaceWriteTypeError;随后fs.upload_file(payload, sandbox_path, timeout=file_upload_s),失败映射为WorkspaceArchiveWriteError;mkdir(path, parents, user):无user时先做路径授权校验(_validate_path_access),根路径/直接返回,随后fs.create_folder(path, "755");指定user时走_check_mkdir_with_exec的 exec 路径。
read/write指定user时同样走_check_read_with_exec/_check_write_with_exec的 exec 辅助路径,文件访问前都会通过_validate_remote_path_access做远程路径安全校验。
4.5 工作区持久化与恢复
这是 Daytona 后端最有价值的能力之一,允许把沙箱工作区打包保存、并在后续会话中恢复:
persist_workspace()将工作区打包为 tar 字节流返回:
- 在
/tmp/sandbox-persist-<session_id.hex>.tar生成 tar 文件,命令为tar <excludes> -C <root> -cf <tar_path> .,excludes来自shell_tar_exclude_args计算的需要跳过的相对路径; - 对 manifest 中所有临时挂载(ephemeral mounts)先执行
teardown_for_snapshot卸载,打包完成后再按逆序restore_after_snapshot重挂;任一步失败都会在最终错误上下文中记录earlier_unmount_error、snapshot_error_before_remount_corruption等细节,保证挂载状态可追溯; _run_persist_workspace_command带retry_async重试,成功后fs.download_file拉取 tar 字节;finally中总是rm -f清理远端临时文件(超时cleanup_s);- 返回
io.BytesIO(raw)。
hydrate_workspace(data)反向恢复工作区:
- 将字节写入
/tmp/sandbox-hydrate-<session_id.hex>.tar; - 安全性关键点:先调用
validate_tar_bytes(payload, allow_external_symlink_targets=False)校验 tar 内容,检测到不安全成员(如指向外部的符号链接)时抛WorkspaceArchiveWriteError(reason="unsafe_or_invalid_tar"),防止恶意 tar 包逃逸沙箱; mkdir(root, parents=True)确保根目录存在 →fs.upload_file上传 →tar -C <root> -xf <tar_path>解包;- 非零退出码映射为
WorkspaceArchiveWriteError(reason="tar_extract_failed"),最后同样清理远端临时文件。
五、云存储挂载:DaytonaCloudBucketMountStrategy
DaytonaCloudBucketMountStrategy(src/agents/extensions/sandbox/daytona/mounts.py)包装通用InContainerMountStrategy与RcloneMountPattern(mode="fuse"),为 Daytona 沙箱自动补齐 rclone 环境后挂载云存储。它支持S3、R2、GCS、Azure Blob(文档表格还列出 Box),无凭证场景走统一代码路径,带凭证的挂载要求对可信 manifest 做精确路径的运行时确认。
挂载激活前的准备步骤(activate/restore_after_snapshot):
validate_mount_activation_credential_boundary:校验凭证边界——若挂载辅助进程在模型可控的沙箱内执行且需要受保护权限,则要求 manifest 已显式确认;_assert_daytona_session:确认会话类型是DaytonaSandboxSession;_ensure_fuse_support:FUSE 模式先检查/dev/fuse字符设备与/proc/filesystems中的 fuse 内核模块(这两项无法安装,缺失直接报MountConfigError);缺少fusermount3/fusermount时通过apt-get安装fuse3并复查;_ensure_rclone:command -v rclone检测不到时自动安装 rclone(apt-get/apk二选一,最多重试 3 次,每次超时 180 秒且以 root 执行),全部失败抛带可操作提示的MountConfigError——这也意味着 Daytdaytona 镜像缺少包管理器时,需要在镜像中预装 rclone/fuse3。
使用示例(摘自daytona_runner.py与模块 docstring):
from agents.extensions.sandbox.daytona import DaytonaCloudBucketMountStrategy from agents.sandbox.entries import S3Mount mount = S3Mount( bucket="my-bucket", endpoint_url=endpoint_url, # 可选,S3 兼容端点(如 MinIO) prefix=key_prefix, # 可选键前缀 mount_path=Path("/mnt/bucket"), # 相对路径基于工作区根解析 read_only=False, mount_strategy=DaytonaCloudBucketMountStrategy(), )docs/sandbox/clients.md明确:Daytona 后端对S3Mount/R2Mount/GCSMount/AzureBlobMount/BoxMount的挂载支持均为 ✓。快照与持久化流程会把挂载路径当作临时条目处理(跳过或卸载,不把远端存储复制进保存的工作区),teardown_for_snapshot/restore_after_snapshot正是为此设计。
六、端到端示例:把 Daytona 接入SandboxAgent
以下是仓库内 examples/sandbox/extensions/daytona/daytona_runner.py 的核心流程(最小化示意):
import asyncio, os from pathlib import Path from agents import ModelSettings, Runner from agents.run import RunConfig from agents.sandbox import Manifest, SandboxAgent, SandboxRunConfig from agents.extensions.sandbox import ( DEFAULT_DAYTONA_WORKSPACE_ROOT, DaytonaSandboxClient, DaytonaSandboxClientOptions, ) def _require_env(name: str) -> None: if not os.environ.get(name): raise SystemExit(f"{name} must be set before running this example.") async def main() -> None: _require_env("OPENAI_API_KEY") _require_env("DAYTONA_API_KEY") manifest = Manifest( root=DEFAULT_DAYTONA_WORKSPACE_ROOT, entries={}, # 实际示例通过 text_manifest 填充 README.md / launch.md / tasks.md ) agent = SandboxAgent( name="Daytona Sandbox Assistant", model="gpt-5.6-sol", instructions="Inspect the workspace files before answering, stay concise.", default_manifest=manifest, capabilities=[WorkspaceShellCapability()], model_settings=ModelSettings(tool_choice="required"), ) client = DaytonaSandboxClient() run_config = RunConfig( sandbox=SandboxRunConfig( client=client, options=DaytonaSandboxClientOptions(pause_on_exit=False), ), workflow_name="Daytona sandbox example", ) try: result = await Runner.run(agent, "Summarize this cloud sandbox workspace in 2 sentences.", run_config=run_config) print(result.final_output) finally: await client.close() asyncio.run(main())运行前需要:
- 安装扩展:
uv sync --extra daytona(或对发布包执行pip install openai-agents[daytona]); - 设置环境变量
OPENAI_API_KEY与DAYTONA_API_KEY(必要时还有 Daytonaapi_url); - 可选参数:
--pause-on-exit(退出时暂停而非删除沙箱)、--stream(流式输出)、--cloud-bucket-name等(挂载公开 S3 桶做匿名挂载演示)。
示例同时演示了流式模式:Runner.run_streamed遍历stream_events(),对raw_response_event中的ResponseTextDeltaEvent增量打印assistant>输出;finally中总是await client.close()释放底层 HTTP 连接。
七、实现要点与适用前提总结
- 统一抽象:Daytona 后端完全基于
BaseSandboxClient/BaseSandboxSession抽象实现,切换后端只需改动SandboxRunConfig的client与options,SandboxAgent定义保持不变(docs/sandbox/clients.md的决策指南也强调这一点)。 - 状态可序列化:
DaytonaSandboxSessionState记录sandbox_id、镜像/快照、资源、超时、暴露端口、pause_on_exit等全部信息,配合resume实现跨进程会话恢复;恢复时优先重连既有沙箱,失败则按保存参数重建。 - 错误语义清晰:通过
_DAYTONA_HTTP_STATUS_RETRYABLE表与 SDK 异常类探测,将远端错误统一映射为ExecTimeoutError/ExecTransportError/WorkspaceArchiveReadError等框架内异常,并携带backend="daytona"与原始 HTTP 状态上下文,便于上层诊断。 - 安全边界:文件路径访问有
_validate_remote_path_access校验;tar 恢复前执行validate_tar_bytes拒绝外部符号链接;云存储挂载严格执行凭证边界确认。 - 适用前提:需要 Daytona 平台账号与 API Key;沙箱镜像需具备运行 rclone/FUSE 的条件(缺少时扩展会自动安装,但要求镜像存在
apt-get/apk包管理器);当前沙箱 Agent 相关能力整体仍处于 Beta(docs/sandbox/clients.md顶部明确标注),API 与默认值在正式发布前可能调整。
如需继续深入,可阅读 docs/sandbox/clients.md(后端选型与挂载总览)、src/agents/extensions/sandbox/daytona/sandbox.py(完整实现)、src/agents/extensions/sandbox/daytona/mounts.py(挂载策略)以及 examples/sandbox/extensions/daytona/daytona_runner.py(可运行示例)。
【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考