沙箱 SDK 用法

用 Python 或 TypeScript SDK 以编程方式创建和管理沙箱。

LangSmith SDK 提供了一个用于创建并与沙箱交互的编程接口。

安装

Python

# uv
uv add "langsmith[sandbox]"

# pip
pip install "langsmith[sandbox]"

TypeScript

npm install langsmith
# or
yarn add langsmith

Python 的 [sandbox] extra 会安装 websockets，它启用实时流式传输和 timeout=0。没有它，run() 会自动回退到 HTTP。对于 TypeScript，安装可选的 ws 包以进行 WebSocket 流式传输：

npm install ws

创建并运行一个沙箱

客户端从环境读取 LANGSMITH_API_KEY 和 LANGSMITH_ENDPOINT，因此在你创建沙箱之前导出两者：

export LANGSMITH_API_KEY="<LANGSMITH_API_KEY>"
export LANGSMITH_ENDPOINT="<LANGSMITH_ENDPOINT>"

LANGSMITH_ENDPOINT 默认为 https://api.smith.langchain.com（GCP US）。在 BYOC 上把它设为你的数据平面 URL，在自托管上设为你的实例 URL，或在其他 Cloud 区域上设为你所在区域的 API URL。

当你想从一个可复用的自定义文件系统镜像启动时，传入一个快照 ID 或名字；那个流程参阅快照。

Python

from langsmith.sandbox import SandboxClient

# Client uses LANGSMITH_ENDPOINT and LANGSMITH_API_KEY from environment
client = SandboxClient()

# Create a sandbox with the default runtime and run code
with client.sandbox() as sb:
    result = sb.run("python -c 'print(2 + 2)'")
    print(result.stdout)  # "4\n"
    print(result.success)  # True

TypeScript

import { SandboxClient } from "langsmith/sandbox";

// Client uses LANGSMITH_ENDPOINT and LANGSMITH_API_KEY from environment
const client = new SandboxClient();

// Create a sandbox with the default runtime and run code
const sandbox = await client.createSandbox();
const result = await sandbox.run("node -e 'console.log(2 + 2)'");
console.log(result.stdout); // "4\n"

// Don't forget to clean up
await sandbox.delete();

运行命令

每个 run() 调用都返回一个带 stdout、stderr、exit_code 和 success 的 ExecutionResult。

Python

with client.sandbox() as sb:
    result = sb.run("echo 'Hello, World!'")

    print(result.stdout)     # "Hello, World!\n"
    print(result.stderr)     # ""
    print(result.exit_code)  # 0
    print(result.success)    # True

    # Commands that fail return non-zero exit codes
    result = sb.run("exit 1")
    print(result.success)    # False
    print(result.exit_code)  # 1

TypeScript

const sandbox = await client.createSandbox();
try {
  const result = await sandbox.run("echo 'Hello, World!'");

  console.log(result.stdout);     // "Hello, World!\n"
  console.log(result.stderr);     // ""
  console.log(result.exit_code);  // 0

  // Pass environment variables and working directory
  const envResult = await sandbox.run("echo $MY_VAR", {
    env: { MY_VAR: "test-value" },
    cwd: "/tmp",
  });
} finally {
  await sandbox.delete();
}

流式传输输出

对于长时间运行的命令，用回调或一个 CommandHandle 实时流式传输输出。

用回调流式传输

Python

import sys

with client.sandbox() as sb:
    result = sb.run(
        "make build",
        timeout=600,
        on_stdout=lambda s: print(s, end=""),
        on_stderr=lambda s: print(s, end="", file=sys.stderr),
    )
    print(f"\nBuild {'succeeded' if result.success else 'failed'}")

TypeScript

const result = await sandbox.run("make build", {
  timeout: 600,
  onStdout: (data) => process.stdout.write(data),
  onStderr: (data) => process.stderr.write(data),
});
console.log(`Exit code: ${result.exit_code}`);

用 CommandHandle 流式传输

设 wait=False 以获得一个 CommandHandle，对输出流拥有完全控制。

Python

with client.sandbox() as sb:
    handle = sb.run("make build", timeout=600, wait=False)

    print(f"Command ID: {handle.command_id}")

    for chunk in handle:
        prefix = "OUT" if chunk.stream == "stdout" else "ERR"
        print(f"[{prefix}] {chunk.data}", end="")

    result = handle.result
    print(f"\nExit code: {result.exit_code}")

TypeScript

const handle = await sandbox.run("python train.py", {
  wait: false,
  timeout: 600,
});

console.log(`Command ID: ${handle.commandId}`);
console.log(`PID: ${handle.pid}`);

for await (const chunk of handle) {
  if (chunk.stream === "stdout") {
    process.stdout.write(chunk.data);
  } else {
    process.stderr.write(chunk.data);
  }
}

const result = await handle.result;
console.log(`Exit code: ${result.exit_code}`);

发送 stdin 和终止命令

Python

with client.sandbox() as sb:
    handle = sb.run(
        "python -c 'name = input(\"Name: \"); print(f\"Hello {name}\")'",
        timeout=30,
        wait=False,
    )

    for chunk in handle:
        if "Name:" in chunk.data:
            handle.send_input("World\n")
        print(chunk.data, end="")

    result = handle.result

TypeScript

const handle = await sandbox.run("python -i", { wait: false });

// Send input to stdin
handle.sendInput("print(2 + 2)\n");
handle.sendInput("exit()\n");

for await (const chunk of handle) {
  process.stdout.write(chunk.data);
}

终止一个正在运行的命令：

Python

with client.sandbox() as sb:
    handle = sb.run("python server.py", timeout=0, wait=False)

    for chunk in handle:
        print(chunk.data, end="")
        if "Ready" in chunk.data:
            break

    handle.kill()

TypeScript

const handle = await sandbox.run("sleep 300", { wait: false });
handle.kill();

const result = await handle.result;
console.log(result.exit_code); // non-zero

重新连接到一个正在运行的命令

如果一个客户端断开，用命令 ID 重新连接：

Python

with client.sandbox() as sb:
    handle = sb.run("make build", timeout=600, wait=False)
    command_id = handle.command_id

    # Later, possibly in a different process
    handle = sb.reconnect(command_id)
    for chunk in handle:
        print(chunk.data, end="")
    result = handle.result

TypeScript

const handle = await sandbox.run("long-task", { wait: false });
const commandId = handle.commandId;

// Later, or from a different client
const newHandle = await sandbox.reconnect(commandId);
for await (const chunk of newHandle) {
  process.stdout.write(chunk.data);
}

文件操作

在沙箱中读写文件：

Python

with client.sandbox() as sb:
    # Write a file
    sb.write("/app/script.py", "print('Hello from file!')")

    # Run the script
    result = sb.run("python /app/script.py")
    print(result.stdout)  # "Hello from file!\n"

    # Read a file (returns bytes)
    content = sb.read("/app/script.py")
    print(content.decode())  # "print('Hello from file!')"

    # Write binary files
    sb.write("/app/data.bin", b"\x00\x01\x02\x03")

TypeScript

const sandbox = await client.createSandbox();
try {
  // Write a file (string content)
  await sandbox.write("/app/script.py", "print('Hello from file!')");

  // Run the script
  const result = await sandbox.run("python /app/script.py");
  console.log(result.stdout);  // "Hello from file!\n"

  // Read a file (returns Uint8Array)
  const content = await sandbox.read("/app/script.py");
  console.log(new TextDecoder().decode(content));

  // Write binary files
  await sandbox.write("/app/data.bin", new Uint8Array([0x00, 0x01, 0x02, 0x03]));
} finally {
  await sandbox.delete();
}

把一个文件作为链接分享

要把一个文件交给那些无法发送 API 密钥的东西，例如一个浏览器标签、一个 <a href>、或一个 webhook 消费者，请签发一个下载链接而非自己读取字节：

Python

link = sb.generate_download_url("/app/report.csv", expires_in_seconds=3600)
print(link.download_url)

TypeScript

const link = await sandbox.generateDownloadURL("/app/report.csv", {
  expiresInSeconds: 3600,
});
console.log(link.download_url);

省略过期时间以获得一个永不失效的链接。之后不要修改该文件：一个链接绑定到一个路径而非内容的一个快照，因此之后的写入可能反映、也可能不反映在该链接所提供的内容中。选项和安全考量参阅沙箱下载链接。

挂载一个 Context Hub 仓库

挂载一个 Context Hub 仓库，以给你的沙箱代码对 agent 和技能的 filesystem 访问。仓库的最新 commit 树被镜像进挂载路径，并在沙箱的整个生命周期内保持同步，因此一个新 commit 会出现在运行中的沙箱里而无需重启。

WARNING

Context Hub 挂载是只读的。同步是单向的，从仓库到沙箱：agent 在挂载路径下写的文件永不被推回仓库，下一次同步会覆盖它们。把沙箱输出写到挂载之外的路径，如果它属于仓库，则用 SDK 推送它。

通过 mount_config 传入该挂载。创建沙箱的 API 密钥必须能访问该仓库，否则创建会失败并返回一个 403。

Python

from langsmith.sandbox import AsyncSandboxClient, context_hub_mount, mount_config

async def main():
    async with AsyncSandboxClient() as client:
        async with await client.sandbox(
            name="context-hub-mount-sandbox",
            mount_config=mount_config(
                mounts=[
                    context_hub_mount(
                        id="memories",
                        mount_path="/memories",
                        repo="-/my-agent",
                    )
                ],
            ),
        ) as sb:
            result = await sb.run("ls /memories")
            print(result.stdout)

TypeScript

import {
  SandboxClient,
  contextHubMount,
  mountConfig,
} from "langsmith/sandbox";

const client = new SandboxClient();

const sandbox = await client.createSandbox({
  name: "context-hub-mount-sandbox",
  mountConfig: mountConfig({
    mounts: [
      contextHubMount({
        id: "memories",
        mountPath: "/memories",
        repo: "-/my-agent",
      }),
    ],
  }),
});

try {
  const result = await sandbox.run("ls /memories");
  console.log(result.stdout);
} finally {
  await sandbox.delete();
}

repo 是仓库句柄，可选地限定为 owner/repo，其中 - 是当前工作区。mount_path 必须是一个绝对、干净、位于系统目录之外的路径，且不同于 bucket 和 Git 挂载，它不局限于 /mnt/mounts。

有关同步行为、同步限制，以及与仓库一起挂载 bucket 和 Git 仓库，参阅沙箱挂载。

为沙箱定规格

传入 vcpus、mem_bytes 和 fs_capacity_bytes（TypeScript 中为 vCpus、memBytes、fsCapacityBytes）以在创建时为沙箱定规格。省略它们，沙箱会获得下面的默认值。

Python

sb = client.create_sandbox(
    name="big-vm",
    vcpus=4,
    mem_bytes=16 * 1024**3,        # 16 GiB
    fs_capacity_bytes=32 * 1024**3, # 32 GiB
)

TypeScript

const sb = await client.createSandbox({
  name: "big-vm",
  vCpus: 4,
  memBytes: 17_179_869_184,        // 16 GiB
  fsCapacityBytes: 34_359_738_368, // 32 GiB
});

资源	默认	范围
CPU	0.5 vCPU	0.05 到 16 vCPU。用 cpu_millicores 做子核请求（500 是 0.5 vCPU）；它优先于 vcpus。
内存	每 vCPU 4 GiB	最多 64 GiB。必须保持在每 vCPU 目标的 50% 以内，因此一个 1 vCPU 沙箱接受 2 到 6 GiB。只设内存不设 CPU 时，CPU 会从同一比例派生。
文件系统	快照容量	最多 64 GiB，且从不小于它启动自的快照。

当宿主有余量时，沙箱会burst到其请求 CPU 的两倍。用 update_sandbox / updateSandbox 调整一个现有沙箱会在它下次启动时生效，且一次调整只强制 64 GiB 上限而非每 vCPU 比例。

NOTE

自由形式的 labels（每个沙箱最多 128 个，每键 256 字节、每值 4096 字节）可在创建时通过 REST API 设置。沙箱继承其快照的标签，除非被覆盖。langsmith.sandbox 客户端尚未暴露这个字段。

沙箱生命周期与保留

沙箱受一个以 idle activity 与 stopped 状态为锚的两级保留模型管辖。

字段	它控制什么	何时触发
idle_ttl_seconds	启动器在这些秒数的不活动后停止沙箱。任何命令执行或文件 I/O 都重置计时器。0 禁用空闲停止。	省略时默认 600（10 分钟）。
delete_after_stop_seconds	一旦沙箱进入 stopped 状态，此计时器开始。它过完之后，沙箱行 + filesystem 克隆被一个服务端扫描永久删除。0 禁用以停止为锚的删除（需手动清理）。	省略时服务端应用它配置的默认（通常 14 天）。

两个值都必须是 60 的倍数（分钟精度），且 delete_after_stop_seconds 上限为 2592000（30 天）。完整的生命周期是：

running ──(idle for idle_ttl_seconds)──▶ stopped ──(delete_after_stop_seconds)──▶ deleted

你也可以显式调用 stop_sandbox / stopSandbox，在空闲超时触发之前释放资源；那也会填入 stopped_at 并启动删除计时器。之后你无需再次启动它：一个已停止的沙箱会在下次命令、文件操作或 service-URL 请求时唤醒。

Python

# Default retention (server defaults: 10-min idle stop, 14-day delete)
with client.sandbox() as sb:
    sb.run("echo hello")

# Aggressive: stop after 5 min idle, delete 1 hour after stop
sb = client.create_sandbox(
    idle_ttl_seconds=300,
    delete_after_stop_seconds=3600,
)

# Long-running: never auto-stop, delete 7 days after manual stop
sb = client.create_sandbox(
    idle_ttl_seconds=0,
    delete_after_stop_seconds=604800,
)

# Update retention on an existing sandbox
sb = client.update_sandbox(
    sb.name,
    idle_ttl_seconds=1800,
    delete_after_stop_seconds=2592000,  # 30 days
)

TypeScript

// Default retention (server defaults applied)
const sandbox = await client.createSandbox();

// Aggressive: stop after 5 min idle, delete 1 hour after stop
const sb = await client.createSandbox({
  idleTtlSeconds: 300,
  deleteAfterStopSeconds: 3600,
});

// Long-running: never auto-stop, delete 7 days after manual stop
const longRunning = await client.createSandbox({
  idleTtlSeconds: 0,
  deleteAfterStopSeconds: 604800,
});

// Update retention on an existing sandbox
await client.updateSandbox(sb.name, {
  idleTtlSeconds: 1800,
  deleteAfterStopSeconds: 2592000, // 30 days
});

命令生命周期与 TTL

沙箱守护进程用两种超时机制管理命令会话生命周期：

• 会话 TTL（已完成的命令）：一个命令完成后，它的会话在一个 TTL 期内（默认：5 分钟）保留在内存中。在这个窗口内你可以重新连接以取回输出。TTL 过完之后，会话被清理。设 ttl_seconds 为 -1 以无限期保留会话。
• 空闲超时（运行中的命令）：没有已连接客户端的运行中命令，在一个空闲超时（默认：1 小时）后被终止。每当一个客户端连接，空闲计时器就重置。设 idle_timeout 为 -1 以无空闲超时。

组合生命周期选项

Python

with client.sandbox() as sb:
    # Long-running task: 30-min idle timeout, 1-hour session TTL
    handle = sb.run(
        "python train.py",
        timeout=0,              # No command timeout
        idle_timeout=1800,      # Kill after 30min with no clients
        ttl_seconds=3600,       # Keep session for 1 hour after exit
        wait=False,
    )

    # Fire-and-forget: no idle timeout, infinite TTL
    handle = sb.run(
        "python background_job.py",
        timeout=0,
        idle_timeout=-1,        # Never kill due to idle
        ttl_seconds=-1,         # Keep session forever
        wait=False,
    )

TypeScript

const sandbox = await client.createSandbox();
try {
  // Long-running task: 30-min idle timeout, 1-hour session TTL
  const handle = await sandbox.run("python train.py", {
    timeout: 0,              // No command timeout
    idleTimeout: 1800,       // Kill after 30min with no clients
    ttlSeconds: 3600,        // Keep session for 1 hour after exit
    wait: false,
  });

  // Fire-and-forget: no idle timeout, infinite TTL
  const bg = await sandbox.run("python background_job.py", {
    timeout: 0,
    idleTimeout: -1,         // Never kill due to idle
    ttlSeconds: -1,          // Keep session forever
    wait: false,
  });
} finally {
  await sandbox.delete();
}

设 kill_on_disconnect=True（Python）或 killOnDisconnect: true（TypeScript），以在最后一个客户端断开时立即终止命令，而非等待空闲超时。

服务 URL（Python）

通过一个已认证的 URL 访问沙箱内运行的一个 HTTP 服务。你可以在浏览器中打开它、从代码中调用它，或与队友分享它。

with client.sandbox() as sb:
    sb.run("python -m http.server 8000", timeout=0, wait=False)

    svc = sb.service(port=8000)

    # Open in a browser
    print(svc.browser_url)

    # Or make requests with built-in helpers (auth is injected automatically)
    resp = svc.get("/api/data")
    resp = svc.post("/api/data", json={"key": "value"})

更多详情，包括用例、REST API 访问和一个完整的 FastAPI 示例，参阅服务 URL。

TCP 隧道（Python）

像访问本地一样访问沙箱内运行的任何 TCP 服务。该隧道打开一个本地 TCP 端口，并通过一个 WebSocket 把连接转发到沙箱内的目标端口。

import psycopg2

# Snapshot built from the official postgres:16 image
sb = client.create_sandbox(snapshot_id=postgres_snapshot_id)
pg_handle = sb.run(
    "POSTGRES_HOST_AUTH_METHOD=trust docker-entrypoint.sh postgres",
    timeout=0,
    wait=False,
)
import time; time.sleep(6)  # Wait for Postgres to start

try:
    with sb.tunnel(remote_port=5432, local_port=25432) as t:
        conn = psycopg2.connect(
            host="127.0.0.1",
            port=t.local_port,
            user="postgres",
        )
        cursor = conn.cursor()
        cursor.execute("SELECT version()")
        print(cursor.fetchone())
        conn.close()
finally:
    pg_handle.kill()
    client.delete_sandbox(sb.name)

隧道适用于任何 TCP 服务（Redis、HTTP 服务器等），你可以同时打开多个隧道：

with sb.tunnel(remote_port=5432, local_port=25432) as t1, \
     sb.tunnel(remote_port=6379, local_port=26379) as t2:
    # Use both Postgres and Redis simultaneously
    pass

异步支持 (Python)

Python SDK 提供一个完整的异步客户端：

from langsmith.sandbox import AsyncSandboxClient

async def main():
    async with AsyncSandboxClient() as client:
        async with await client.sandbox() as sb:
            result = await sb.run("python -c 'print(1 + 1)'")
            print(result.stdout)  # "2\n"

            await sb.write("/app/test.txt", "async content")
            content = await sb.read("/app/test.txt")
            print(content.decode())

            # Async streaming
            handle = await sb.run("make build", timeout=600, wait=False)
            async for chunk in handle:
                print(chunk.data, end="")
            result = await handle.result

            # Async service URLs
            svc = await sb.service(port=8000)
            resp = await svc.get("/api/data")
            url = await svc.get_service_url()
            token = await svc.get_token()

追踪沙箱活动

通过在 run() 上传入 LangSmith 追踪环境变量的 env 参数，从沙箱内运行的代码发送追踪。在进程退出前调用 flush() 以确保所有追踪都被投递。

Python

from langsmith.sandbox import SandboxClient

client = SandboxClient()

tracing_env = {
    "LANGSMITH_API_KEY": "lsv2_pt_...",
    "LANGSMITH_ENDPOINT": "https://api.smith.langchain.com",
    "LANGSMITH_TRACING": "true",
    "LANGSMITH_PROJECT": "my-sandbox-traces",
}

with client.sandbox() as sandbox:
    sandbox.run("pip install langsmith", timeout=120, env=tracing_env)
    result = sandbox.run("python3 my_agent.py", env=tracing_env)
    print(result.stdout)

TypeScript

import { SandboxClient } from "langsmith/sandbox";

const client = new SandboxClient();

const tracingEnv = {
  LANGSMITH_API_KEY: "lsv2_pt_...",
  LANGSMITH_ENDPOINT: "https://api.smith.langchain.com",
  LANGSMITH_TRACING: "true",
  LANGSMITH_PROJECT: "my-sandbox-traces",
};

const sandbox = await client.createSandbox();
try {
  await sandbox.run("pip install langsmith", { timeout: 120, env: tracingEnv });
  const result = await sandbox.run("python3 my_agent.py", { env: tracingEnv });
  console.log(result.stdout);
} finally {
  await sandbox.delete();
}

在沙箱内，任何带 LangSmith 插桩的代码（@traceable、LangChain、LangGraph）都会自动从注入的环境变量中拾取追踪配置。

WARNING

务必在沙箱进程退出前调用 flush()——Python 中的 langsmith.Client().flush()，或 TypeScript 中的 await new Client().flush()。没有它，追踪可能会丢失，因为容器在命令完成时被销毁。

错误处理

两个 SDK 都为特定错误处理提供类型化的异常：

Python

from langsmith.sandbox import (
    SandboxClientError,       # Base exception
    ResourceCreationError,    # Provisioning failed
    ResourceNotFoundError,    # Resource doesn't exist
    ResourceTimeoutError,     # Operation timed out
    SandboxNotReadyError,     # Sandbox not ready yet
    SandboxConnectionError,   # Network/WebSocket error
    CommandTimeoutError,      # Command exceeded timeout
    QuotaExceededError,       # Quota limit reached
)

try:
    with client.sandbox() as sb:
        result = sb.run("sleep 999", timeout=10)
except CommandTimeoutError as e:
    print(f"Command timed out: {e}")
except ResourceNotFoundError as e:
    print(f"{e.resource_type} not found: {e}")
except SandboxClientError as e:
    print(f"Error: {e}")

TypeScript

import {
  LangSmithSandboxError,
  LangSmithResourceNotFoundError,
  LangSmithResourceTimeoutError,
  LangSmithSandboxConnectionError,
  LangSmithCommandTimeoutError,
  LangSmithQuotaExceededError,
} from "langsmith/sandbox";

try {
  const sandbox = await client.createSandbox("not-a-real-snapshot");
  await sandbox.delete();
} catch (e) {
  if (e instanceof LangSmithResourceNotFoundError) {
    console.log(`${e.resourceType} not found: ${e.message}`);
  } else if (e instanceof LangSmithResourceTimeoutError) {
    console.log(`Timeout waiting for ${e.resourceType}: ${e.message}`);
  } else if (e instanceof LangSmithSandboxError) {
    console.log(`Error: ${e.message}`);
  }
}

NOTE

更多详情，参阅 GitHub 上针对 Python 或 TypeScript 的沙箱 SDK 参考。

---