python-sandbox-sdk

Python Sandbox 官方 Python SDK — 在隔离的 Docker 容器中执行 Python 代码、Shell 命令、管理 pip 包、读写文件。

对应后端：python-sandbox（Spring Boot 3 + Java 17）。

✨ 特性

• 🚀 简洁 API：10 个方法覆盖会话、代码执行、Shell、pip、文件读写、文件上传下载、健康检查
• 🔌 上下文管理器：自动管理 HTTP 连接与会话生命周期
• 🛡️ 异常体系：独立的 ApiRequestError / SandboxError 类型
• 🧵 线程安全：requests.Session 复用，支持 ThreadPoolExecutor 并发
• ⚡ 异步支持：提供 AsyncSandboxClient，基于 aiohttp 实现完整异步 API
• 🐍 Python ≥ 3.8：同步客户端仅依赖 requests>=2.31.0

📦 安装

# 方式 A：从 PyPI 安装（推荐）
pip install python-sandbox-sdk

# 方式 B：从源码安装
git clone https://github.com/zjwan461/python-sandbox.git
cd python-sandbox/sdk/python-sandbox-sdk
pip install -e .

# 安装异步支持（可选）
pip install python-sandbox-sdk[async]

🚀 快速开始

0. 启动后端服务

参考 python-sandbox/README.md 启动：

cd python-sandbox
cp .env.example .env
docker-compose up -d --build
curl http://localhost:8080/health

1. 最简示例

from python_sandbox_sdk import SandboxClient

with SandboxClient("http://localhost:8080", "sandbox-secret-key") as client:
    if not client.is_health():
        raise SystemExit("Sandbox not ready")

    session_id = client.create_session()
    try:
        result = client.exec_python(session_id, "print('Hello from sandbox!')")
        print(result.stdout)              # Hello from sandbox!
        print(result.exit_code)           # 0
        print(result.success)             # True
    finally:
        client.delete_session(session_id)

2. 安装并使用第三方包

client.pip_install(session_id, "requests==2.31.0")
result = client.exec_python(session_id, """
import requests
print(requests.get('https://httpbin.org/get', timeout=5).status_code)
""")

3. 文件读写

# 文本
client.write_file(session_id, "/tmp/config.json", '{"key": "value"}')
content = client.read_file(session_id, "/tmp/config.json")

# 二进制
client.upload_file(session_id, "/tmp/image.png", b"\\x89PNG...")
data = client.download_file(session_id, "/tmp/image.png")

📚 使用样例（覆盖全部 API 与典型场景）

完整可运行样例见 usage/ 目录：

运行样例：

cd usage/
export SANDBOX_API_KEY="sandbox-secret-key"

python 01_hello_world.py
for f in 0[1-9]_*.py 1[0-1]_*.py; do python "$f"; done

🔧 API 速查

CommandResult 字段：exit_code, stdout, stderr, success（property）, combined_output（property）。

💡 异步版本：AsyncSandboxClient 提供完全相同的 API，所有方法均为 async，支持 await 调用和 async with 上下文管理器。

⚙️ 环境变量

为避免硬编码，样例通过环境变量注入连接信息：

🛡️ 异常处理

from python_sandbox_sdk import SandboxClient, ApiRequestError, SandboxError

try:
    client.exec_python(session_id, "print(1)")
except ApiRequestError as e:
    # 后端返回的 4xx / 5xx（鉴权失败、会话不存在、黑名单、容器上限等）
    print(f"API Error: HTTP {e.status_code} - {e}")
except SandboxError as e:
    # SDK 抛出的通用错误
    print(f"SDK Error: {e}")
except OSError as e:
    # 网络错误（连接失败、超时等）
    print(f"Network Error: {e}")

🧵 并发使用

同步并发（ThreadPoolExecutor）

from concurrent.futures import ThreadPoolExecutor
# 每个任务使用独立的 client + session（线程安全）
with ThreadPoolExecutor(max_workers=5) as pool:
    futures = [pool.submit(run_one_task, i) for i in range(5)]
    for f in futures:
        print(f.result())

异步并发（AsyncSandboxClient）

import asyncio
from python_sandbox_sdk import AsyncSandboxClient

async def run_task(task_id):
    async with AsyncSandboxClient("http://localhost:8080", "sandbox-secret-key") as client:
        session_id = await client.create_session()
        try:
            result = await client.exec_python(
                session_id, f"print('Task {task_id} done')"
            )
            return result.stdout
        finally:
            await client.delete_session(session_id)

async def main():
    # 并发执行 5 个任务
    tasks = [run_task(i) for i in range(5)]
    results = await asyncio.gather(*tasks)
    for r in results:
        print(r)

asyncio.run(main())

⚠️ 并发数受后端 sandbox.max-containers（默认 10）约束。

📦 打包发布

# 安装构建工具
pip install build twine

# 构建
cd sdk/python-sandbox-sdk
python -m build

# 上传到 TestPyPI
twine upload --repository testpypi dist/*

# 上传到 PyPI
twine upload dist/*

发布前请修改 pyproject.toml 中的 version 与 urls 字段。

🔗 相关项目

• 后端服务：python-sandbox/
• Java SDK：sdk/java-sandbox-sdk/
• SDK 总览：sdk/README.md

📄 许可证

MIT License