同步/异步调用、异常重试、资源防泄漏完整可商用代码模板。ConnectionConfig 初始化、asyncio.gather 并发调度、SandboxManager 兜底回收，最小实现直接复制跑通。

OpenSandbox Python SDK开发实战｜代码全自动操控沙箱，无缝嵌入AI项目

系列第 9 篇：SDK 开发实战。把沙箱创建、执行、销毁写进代码，AI 项目才能自动用沙箱，而不是人肉敲 CLI。

1 导语：CLI 无法嵌入程序，AI 项目要代码级操控沙箱

你的 Agent 要在程序里临时跑一段代码，CLI 只能人肉敲，程序没法自己开终端。用几行 Python 把沙箱创建、执行、销毁托管起来，才是把沙箱能力嵌进 AI 项目的第一步。

CLI 适合人机交互，进了自动化流程就失效。

Agent 需要「代码里动态建沙箱 → 跑命令 → 取输出 → 回收」的完整闭环。

批量与并发场景下，手动操作必然资源堆积：

• 建 10 个沙箱就开 10 次终端，跑完忘了杀
• 残留沙箱全挂在服务端，CPU 与内存持续被占
• 账单一直涨，回收全靠人的记忆

代码化之后，创建、执行、回收写死在流程里，不再依赖人的记忆。这正是本文要交付的东西。

读完这篇，你会拿到：

• SDK 接入：ConnectionConfig 初始化，Sandbox / SandboxSync 双模式选型
• 异步调度：asyncio.gather 并发建沙箱，共享 transport 控制并发上限
• 异常处理：SandboxException 与 httpx 超时、连接错误双层捕获
• 资源防泄漏：async with 自动关闭本地资源，SandboxManager 兜底回收
• 商用模板：完整工程源码，最小实现一个文件跑通

2 前置准备：Python 3.10+、服务运行与 SDK 安装

依赖极简，pip 一行安装。前置条件是 OpenSandbox 服务可用（本地或远端），SDK 本质是它的 HTTP 客户端：

pip install opensandbox
# 或 uv add opensandbox

版本以 PyPI 为准，落笔时最新 0.1.15（requires-python >=3.10）。服务与 CLI 的部署见系列前篇，本文默认你已有一个可连实例。连不上先查服务地址与端口，再怀疑 SDK。

3 原理铺垫：HTTP 封装、双模式对象与上下文生命周期

SDK 是 HTTP 接口的面向对象封装。核心类 Sandbox（异步，默认）与 SandboxSync（同步），API 一致，按调用方环境选型。ConnectionConfig 承载连接参数，生命周期由上下文管理器托管：

选型就一句话：调用方是 async 代码就用 Sandbox，是同步脚本就用 SandboxSync。两个类的方法签名一致，切模式只改 import 和配置类，业务代码不用动。

• Sandbox：异步，默认路径。适合并发批量，一次 gather 全出去
• SandboxSync：同步。适合脚本与简单任务，不用事件循环

生命周期方法分三组：

• 状态：get_info 查询、is_healthy 探活、renew 续期、pause 暂停、resume 恢复
• 终止：kill 立即终止远程实例，close 关闭本地资源
• 托管：async with / with 进入自动建连，退出自动 close

from datetime import timedelta
from opensandbox.sandbox import Sandbox
from opensandbox.config import ConnectionConfig

config = ConnectionConfig(domain="api.opensandbox.io", api_key="your-api-key")
sandbox = await Sandbox.create("ubuntu", connection_config=config)

async with sandbox:                # 进入自动建连，退出自动 close() 本地资源
    execution = await sandbox.commands.run("echo 'Hello Sandbox!'")
    print(execution.logs.stdout[0].text)

await sandbox.kill()               # 立即终止远程实例
await sandbox.renew(timedelta(minutes=30))   # 续期：过期时间重置为 now+duration
await sandbox.pause()              # 暂停（挂起所有进程）
sandbox = await Sandbox.resume(sandbox_id=sandbox.id, connection_config=config)  # 恢复
healthy = await sandbox.is_healthy()         # 健康检查

async with 只关本地资源，远程实例靠 kill 或 timeout 自动过期。资源回收没有 GC 魔法，靠的是上下文管理器加兜底清单，这个原则贯穿全文。

timeout 传 None 表示手动清理模式，沙箱不自动过期，回收完全交给代码。默认 10 分钟自动终止，日常任务用默认值就够了。

4 分步实战：从同步调用到异步批量调度

4-① SDK 安装与客户端初始化（ConnectionConfig）

api_key 与 domain 可走环境变量，CI 里不写死。protocol（http/https，默认 http）、debug（HTTP 调试日志）、headers（自定义请求头）、transport（共享 httpx 连接池）是生产化关键参数：

from opensandbox.config import ConnectionConfig
from datetime import timedelta

config = ConnectionConfig(
    api_key="your-api-key",            # 或用环境变量 OPEN_SANDBOX_API_KEY
    domain="api.opensandbox.io",       # 或用环境变量 OPEN_SANDBOX_DOMAIN
    request_timeout=timedelta(seconds=30),   # 默认 30 秒
)

环境变量由 SDK 原生支持：OPEN_SANDBOX_API_KEY 与 OPEN_SANDBOX_DOMAIN 填好后，config 里可以留空。密钥走环境变量后，代码能在本地、CI、生产三套环境间复用，改的只有变量值，不碰代码。

4-② 同步调用：SandboxSync 创建→执行→输出→销毁

同步路径适合脚本与简单任务。SandboxSync.create 建沙箱，with 进入上下文，commands.run 执行，取 stdout 首行，kill 销毁：

from datetime import timedelta
import httpx
from opensandbox import SandboxSync
from opensandbox.config import ConnectionConfigSync

config = ConnectionConfigSync(
    domain="api.opensandbox.io",
    api_key="your-api-key",
    request_timeout=timedelta(seconds=30),
    transport=httpx.HTTPTransport(limits=httpx.Limits(max_connections=20)),
)
sandbox = SandboxSync.create("ubuntu", connection_config=config)
with sandbox:
    execution = sandbox.commands.run("echo 'Hello Sandbox!'")
    print(execution.logs.stdout[0].text)
    sandbox.kill()

execution.logs.stdout[0].text 是命令输出的第一行。多行输出遍历 stdout 列表即可。注意 with 块内主动 kill，脚本结束不留远程实例。同步路径的代价是阻塞，任务密集时性能吃亏，这正是下一节异步批量要解决的。

4-③ 异步批量调度：asyncio.gather + 共享 transport 连接池

异步是默认路径，同步是特例。并发控制靠 httpx transport 的 limits（max_connections / max_keepalive_connections），SDK 没有自带并发参数。大量实例共享一个 transport 才省资源，自定义 transport 需自行 aclose()：

import asyncio
import httpx
from datetime import timedelta
from opensandbox.sandbox import Sandbox
from opensandbox.config import ConnectionConfig

config = ConnectionConfig(
    api_key="your-key",
    domain="api.opensandbox.io",
    transport=httpx.AsyncHTTPTransport(
        limits=httpx.Limits(
            max_connections=100,          # 总连接上限（并发控制核心）
            max_keepalive_connections=50, # 保活连接上限
            keepalive_expiry=30.0,
        )
    ),
)

async def run_task(name: str):
    sandbox = await Sandbox.create("python:3.11", connection_config=config,
                                   timeout=timedelta(minutes=10))
    async with sandbox:
        r = await sandbox.commands.run(f"echo 'task {name}'")
        return name, r.logs.stdout[0].text

async def main():
    tasks = [run_task(f"job-{i}") for i in range(10)]
    results = await asyncio.gather(*tasks)   # 并发创建 10 个沙箱
    for name, out in results:
        print(name, out)
    await config.transport.aclose()          # 自定义 transport 需自行关闭

并发上限写在 transport 里，不写在代码逻辑里。max_connections=100 就是最多同时 100 个连接，超出的请求排队等连接释放。想收敛并发就调这个数，别自己写信号量。

• gather 返回顺序与任务顺序一致，按 name 配对输出不会乱
• 每个沙箱都带 timeout=10 分钟，个别任务卡死也会自动到期销毁

4-④ 异常捕获：SandboxException 与 httpx 超时/连接错误

两层异常都要 catch。服务端业务错误抛 SandboxException，e.error.code 与 e.error.message 是错误码与消息。底层连接、超时抛 httpx 异常。实测注意：连接失败在 create 阶段会被 SDK 包成 SandboxException，消息里带 Network connectivity error，用消息内容判断，别只信错误码：

from opensandbox.exceptions import SandboxException

try:
    sandbox = await Sandbox.create("ubuntu", connection_config=config)
except SandboxException as e:
    print(f"沙箱错误: [{e.error.code}] {e.error.message}")
except httpx.TimeoutException:
    print("请求超时，检查 request_timeout 或服务端负载")
except httpx.ConnectError:
    print("连接失败，检查 domain/protocol 是否填错")

只 catch SandboxException 会漏掉连接层错误，只 catch httpx 会漏掉业务错误。双层捕获是生产代码的底线，重试逻辑挂在哪层看错误类型：超时可重试，业务错误重试也没用。

4-⑤ 资源防泄漏：async with 自动关闭 + SandboxManager 兜底回收

三层防线。async with 自动 close() 本地资源。SandboxManager.list_sandbox_infos 查残留。kill_sandbox 兜底回收。没有 GC 自动回收魔法，程序崩溃后的残留只能靠清单清理：

from opensandbox.manager import SandboxManager
from opensandbox.models.sandboxes import SandboxFilter

async def cleanup_orphans(config):
    async with await SandboxManager.create(connection_config=config) as manager:
        infos = await manager.list_sandbox_infos(
            SandboxFilter(states=["RUNNING"], page_size=100)
        )
        for info in infos.sandbox_infos:
            await manager.kill_sandbox(info.id)
            print(f"回收残留沙箱: {info.id}")

SandboxManager 是管理面，跟单个沙箱解耦。崩溃、超时、忘了 kill，最后都靠它兜底。生产脚本收尾调一次 cleanup_orphans，比事后逐个排查省事得多。

5 踩坑：地址填错、并发限流与崩溃残留

三个高频坑，症状、根因、解法一次说清：

• 连接失败：本地服务地址是 127.0.0.1:8080，远端服务域名不同，protocol 默认 http，连 https 服务不显式配置必挂。解法：is_healthy() 先探活，创建时传 health_check 自定义覆盖，把地址错误挡在创建之前
• 并发限流：批量任务从 5 个加到 50 个，服务端开始返回限流错误。解法：transport limits 的 max_connections 从 100 调到 20，重跑即恢复。并发不是越多越好，超过服务承载就是给自己找事
• 崩溃残留：脚本半夜崩了，第二天发现服务端挂着一排 RUNNING。解法：timeout 自动过期 + SandboxManager 兜底回收双保险，清理动作写进收尾，别等第二天人工排查

# 服务连通校验：服务端 /health 返回 {"status": "healthy"}
healthy = await sandbox.is_healthy()   # SDK 侧探活
# 创建时也可传自定义 health_check，覆盖默认 ping 检查

timeout 默认 10 分钟自动终止，None 表示手动清理。并发量上去了，把 timeout 当强制保险写进每个 create。崩了不心疼，timeout 到点自动销毁，残留清单再补一刀。

6 生产优化：连接池、配置抽离与日志埋点

生产化三件套。连接池封装（共享 transport，控 max_connections 与 max_keepalive_connections）。配置抽离（env 变量或配置类，CI 不写死密钥）。日志埋点（debug=True 开 HTTP 调试日志）。

连接池不是优化，是并发场景的前提。每个沙箱自建连接，10 个任务就是 10 份握手。连接池把上限和保活统一收口：

import os
import httpx
from datetime import timedelta
from opensandbox.config import ConnectionConfig

class SDKConfig:
    @staticmethod
    def build() -> ConnectionConfig:
        transport = httpx.AsyncHTTPTransport(
            limits=httpx.Limits(max_connections=20, max_keepalive_connections=10),
        )
        return ConnectionConfig(
            api_key=os.environ.get("OPEN_SANDBOX_API_KEY", ""),
            domain=os.environ.get("OPEN_SANDBOX_DOMAIN", "api.opensandbox.io"),
            request_timeout=timedelta(seconds=30),
            debug=os.environ.get("OPEN_SANDBOX_DEBUG") == "1",
            transport=transport,
        )

配置类把散落的参数收进一个 build()。换环境只改环境变量，不改代码。连接池参数按任务规模定：

• 小任务 20 连接够用，大批量再调大 max_connections
• max_keepalive_connections 设成 max_connections 一半，保活连接不占满上限
• debug=True 联调期打开，HTTP 请求响应全打日志，上线前关掉，排查时再临时开

7 资源：完整 Python 工程源码与 SDK 配置模板（最小实现）

最小实现一个文件跑通全流程：初始化 → 并发 3 个沙箱执行 echo → 双层异常 → 兜底回收。文件 minimal_impl.py，依赖 opensandbox + httpx：

sdk-demo/
├── minimal_impl.py   # 本文完整源码，复制即用
└── .env              # OPEN_SANDBOX_API_KEY / OPEN_SANDBOX_DOMAIN

单文件适合快速验证。生产项目按职责拆四个文件，逻辑与本文一致：config.py 管配置装配，client.py 管同步与异步执行，cleanup.py 管兜底回收，main.py 管调度入口。

import asyncio
import os
from datetime import timedelta

import httpx
from opensandbox import Sandbox
from opensandbox.config import ConnectionConfig
from opensandbox.exceptions import SandboxException
from opensandbox.manager import SandboxManager
from opensandbox.models.sandboxes import SandboxFilter

# 共享 transport = 连接池：并发上限在这里控制，不是 SDK 自带参数
SHARED_TRANSPORT = httpx.AsyncHTTPTransport(
    limits=httpx.Limits(
        max_connections=20,            # 总连接上限（并发控制核心）
        max_keepalive_connections=10,  # 保活连接上限
        keepalive_expiry=30.0,
    )
)


def build_config() -> ConnectionConfig:
    """从环境变量装配连接配置，CI 里不写死密钥。"""
    return ConnectionConfig(
        api_key=os.environ.get("OPEN_SANDBOX_API_KEY", ""),
        domain=os.environ.get("OPEN_SANDBOX_DOMAIN", "localhost:8080"),
        request_timeout=timedelta(seconds=30),  # 默认 30 秒
        debug=os.environ.get("OPEN_SANDBOX_DEBUG") == "1",
        transport=SHARED_TRANSPORT,             # 所有沙箱共享一个连接池
    )


async def run_task(config: ConnectionConfig, name: str) -> str:
    """创建一个沙箱跑 echo，返回输出文本。"""
    try:
        sandbox = await Sandbox.create(
            "python:3.11",
            connection_config=config,
            timeout=timedelta(minutes=10),  # 沙箱 TTL，到期自动终止
        )
        async with sandbox:                 # 退出自动 close() 本地资源
            execution = await sandbox.commands.run(f"echo 'task {name}'")
            return execution.logs.stdout[0].text
    except SandboxException as e:
        # 连接失败会被 SDK 包装成 SandboxException，message 里带 Network connectivity error
        if "Network connectivity error" in str(e):
            return "[ERR] 连接失败，检查 domain/protocol 是否填错"
        # 服务端业务错误：错误码 + 消息
        return f"[ERR] 沙箱错误: [{e.error.code}] {e.error.message}"
    except httpx.TimeoutException:
        return "[ERR] 请求超时，检查 request_timeout 或服务端负载"
    except httpx.ConnectError:
        return "[ERR] 连接失败，检查 domain/protocol 是否填错"


async def cleanup_orphans(config: ConnectionConfig) -> None:
    """兜底回收：列出 RUNNING 残留并逐个 kill。"""
    async with await SandboxManager.create(connection_config=config) as manager:
        infos = await manager.list_sandbox_infos(
            SandboxFilter(states=["RUNNING"], page_size=100)
        )
        for info in infos.sandbox_infos:
            await manager.kill_sandbox(info.id)
            print(f"回收残留沙箱: {info.id}")


async def main() -> None:
    config = build_config()
    try:
        tasks = [run_task(config, f"job-{i}") for i in range(3)]
        results = await asyncio.gather(*tasks)   # 并发创建 3 个沙箱
        for name, out in zip([f"job-{i}" for i in range(3)], results):
            print(name, out.strip())
    finally:
        # 兜底回收与连接池关闭都做容错：连接失败时不让主流程崩掉
        try:
            await cleanup_orphans(config)
        except Exception as e:
            print(f"[WARN] 残留回收失败: {e}")
        # 自定义 transport 需自行关闭；SDK 默认 transport 会随 close() 关闭
        await config.transport.aclose()


if __name__ == "__main__":
    asyncio.run(main())

连接正常时，预期输出三行 echo 结果，退出码 0：

job-0 task job-0
job-1 task job-1
job-2 task job-2

domain 填错时，每任务打印 [ERR] 连接失败，兜底回收给出 WARN，脚本仍正常退出：

job-0 [ERR] 连接失败，检查 domain/protocol 是否填错
job-1 [ERR] 连接失败，检查 domain/protocol 是否填错
job-2 [ERR] 连接失败，检查 domain/protocol 是否填错
[WARN] 残留回收失败: Network connectivity error: All connection attempts failed

运行方式：export OPEN_SANDBOX_API_KEY 与 OPEN_SANDBOX_DOMAIN 后 python minimal_impl.py。跑完用 SandboxManager.list_sandbox_infos 确认无 RUNNING 残留，这就是完整闭环。

引用链接

• Python SDK | OpenSandbox^[1] — open-sandbox.ai — official
• Python SDK - OpenSandbox^[2] — mintlify — official
• Python SDK Reference^[3] — mintlify — official
• opensandbox on PyPI^[4] — pypi.org — official