--- title: 部署自行託管的 worker url: https://platform.claude.com/docs/zh-TW/managed-agents/self-hosted-sandboxes-workers description: 選擇自行託管的沙箱 worker 如何認領工作以及工作階段在何處執行：常駐或由 webhook 觸發，在單一程序中執行或每個工作階段一個沙箱。 featureMetadata: topic: title: Managed Agents url: https://platform.claude.com/docs/en/managed-agents/overview status: beta betaHeader: managed-agents-2026-04-01 --- [快速入門](https://platform.claude.com/docs/zh-TW/managed-agents/self-hosted-sandboxes#quickstart)會執行一個 `ant` CLI worker，它會持續輪詢，並在單一程序中執行每個工作階段。本頁介紹執行 worker 的其他方式，以及如何在它們之間做出選擇。 ## 選擇部署模式 部署 worker 時，您需要做出兩個選擇：worker 如何認領工作，以及每個工作階段在何處執行。 **worker 如何認領工作：** * **常駐（Always-on）：** 一個長時間執行的程序會持續輪詢佇列，且只需要對外的 HTTPS 連線。這是最簡單的設定。 * **由 webhook 觸發（Webhook-triggered）：** 處理常式會在 `session.status_run_started` 時喚醒並開始輪詢。這可避免閒置的輪詢器，但需要一個 Anthropic 能夠連線到的 [webhook](https://platform.claude.com/docs/zh-TW/managed-agents/webhooks) 端點。 **每個工作階段在何處執行：** * **在程序內（In process）：** 認領工作階段的 worker 也會在一個共用的工作目錄中執行其工具呼叫。 * **每個工作階段一個沙箱（Sandbox per session）：** 輪詢器會為每個認領的工作階段啟動一個全新的沙箱。若需要更強的隔離，請選擇此方式：全新的檔案系統、資源限制或每個工作階段的網路控制。 CLI 和 SDK worker 支援不同的組合： | 功能 | `ant` CLI | SDK（Python、TypeScript、Go） | | ------------------------------------------------------------------------------------------------ | ---------- | ------------------------- | | 常駐輪詢 | 是 | 是 | | 由 webhook 觸發 | 否 | 是 | | 每個工作階段一個沙箱 | 是 | 是 | | [記憶體儲存區](https://platform.claude.com/docs/zh-TW/managed-agents/self-hosted-sandboxes-memory) | 是，使用預設同步設定 | 是，可設定同步 | | [自訂工具](https://platform.claude.com/docs/zh-TW/managed-agents/self-hosted-sandboxes-custom-tools) | 否 | 是 | 請參閱[自行託管 worker 參考](https://platform.claude.com/docs/zh-TW/managed-agents/self-hosted-sandboxes-reference)以了解每個 CLI 旗標和 SDK 選項。若需要更多控制，請直接呼叫 [Environments Work 端點](https://platform.claude.com/docs/zh-TW/api/beta/environments/work)並實作您自己的 worker。 ## 執行常駐 worker 啟動一個持續輪詢佇列的 worker。它會使用快速入門中的[環境金鑰](https://platform.claude.com/docs/zh-TW/managed-agents/self-hosted-sandboxes#run-your-first-session)進行驗證： ```bash CLI ant beta:worker poll --workdir /workspace ``` ```python Python import asyncio import contextlib import os import signal from anthropic import AsyncAnthropic from anthropic.lib.environments import EnvironmentWorker async def main() -> None: environment_key = os.environ["ANTHROPIC_ENVIRONMENT_KEY"] environment_id = os.environ["ANTHROPIC_ENVIRONMENT_ID"] async with AsyncAnthropic(auth_token=environment_key) as client: worker = EnvironmentWorker( client, environment_id=environment_id, environment_key=environment_key, workdir="/workspace", ) task = asyncio.create_task(worker.run()) # 取消任務（而非終止程序）可讓 worker 停止其 # 進行中的工作項目，並在結束前上傳已變更的記憶檔案。 loop = asyncio.get_running_loop() for signum in (signal.SIGINT, signal.SIGTERM): loop.add_signal_handler(signum, task.cancel) with contextlib.suppress(asyncio.CancelledError): await task asyncio.run(main()) ``` ```typescript TypeScript import Anthropic from "@anthropic-ai/sdk"; import { EnvironmentWorker } from "@anthropic-ai/sdk/helpers/beta/environments"; const environmentKey = process.env.ANTHROPIC_ENVIRONMENT_KEY!; const environmentId = process.env.ANTHROPIC_ENVIRONMENT_ID!; const client = new Anthropic({ authToken: environmentKey }); const controller = new AbortController(); // 在任一訊號下中止，可讓 worker 在程序結束前上傳已變更的記憶檔案， // 並移除其儲存目錄。 process.once("SIGINT", () => controller.abort()); process.once("SIGTERM", () => controller.abort()); await new EnvironmentWorker({ client, environmentId, environmentKey, workdir: "/workspace", signal: controller.signal }).run(); ``` C# SDK 沒有 `EnvironmentWorker`。請改為執行 `ant` CLI worker：`ant beta:worker poll --workdir /workspace`。 ```go Go package main import ( "context" "log" "os" "os/signal" "syscall" "github.com/anthropics/anthropic-sdk-go" "github.com/anthropics/anthropic-sdk-go/lib/environments" "github.com/anthropics/anthropic-sdk-go/option" ) func main() { environmentKey := os.Getenv("ANTHROPIC_ENVIRONMENT_KEY") environmentID := os.Getenv("ANTHROPIC_ENVIRONMENT_ID") ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() client := anthropic.NewClient(option.WithAuthToken(environmentKey)) worker := environments.NewEnvironmentWorker(client, environments.EnvironmentWorkerOptions{ EnvironmentID: environmentID, EnvironmentKey: environmentKey, Workdir: "/workspace", }) if err := worker.Run(ctx); err != nil { log.Fatalf("worker: %v", err) } } ``` Java SDK 沒有 `EnvironmentWorker`。請改為執行 `ant` CLI worker：`ant beta:worker poll --workdir /workspace`。 PHP SDK 沒有 `EnvironmentWorker`。請改為執行 `ant` CLI worker：`ant beta:worker poll --workdir /workspace`。 Ruby SDK 沒有 `EnvironmentWorker`。請改為執行 `ant` CLI worker：`ant beta:worker poll --workdir /workspace`。 ## 從 webhook 觸發 worker 在 [Console](https://platform.claude.com/settings/workspaces/default/webhooks) 中，定義一個監聽 `session.status_run_started` 事件的 webhook 端點。詳情請參閱 [Webhooks](https://platform.claude.com/docs/zh-TW/managed-agents/webhooks)。 除了[快速入門](https://platform.claude.com/docs/zh-TW/managed-agents/self-hosted-sandboxes#run-your-first-session)中的環境 ID 和金鑰之外，請在您的處理常式主機上匯出 webhook 簽署金鑰。處理常式會使用它來驗證傳入的酬載。 ```bash export ANTHROPIC_WEBHOOK_SIGNING_KEY="whsec_..." ``` 在 `session.status_run_started` 觸發時呼叫 worker。處理常式會清空佇列，並將每個認領的工作項目交給 `handle_item()` (typescript: `handleItem()`; go: `HandleItem()`)，它會下載技能、執行工具呼叫、回傳結果，然後返回。 若要驗證 webhook 簽章，請安裝 webhooks 額外套件：`pip install "anthropic[webhooks]"`。 ```python Python import asyncio import os import anthropic import standardwebhooks # installed by the anthropic[webhooks] extra environment_key = os.environ["ANTHROPIC_ENVIRONMENT_KEY"] environment_id = os.environ["ANTHROPIC_ENVIRONMENT_ID"] client = anthropic.AsyncAnthropic( auth_token=environment_key, ) # 由 shutdown() 取消，讓進行中的工作項目能在程序結束前上傳已變更的記憶檔案 # 並移除其 store 目錄。 inflight: set[asyncio.Task[None]] = set() # 請在主機的關閉掛鉤中 await 此函式，例如 ASGI lifespan 關閉階段（FastAPI lifespan 中 # `yield` 之後的程式碼），uvicorn 會在收到 SIGTERM 時執行。uvicorn 會先讓未完成的請求 # 結束後才執行該掛鉤，因此請設定 --timeout-graceful-shutdown 以限制等待時間。 async def shutdown() -> None: for task in inflight: task.cancel() await asyncio.gather(*inflight, return_exceptions=True) async def handle(raw: bytes, headers: dict[str, str]) -> tuple[dict[str, str], int]: try: event = client.beta.webhooks.unwrap(raw.decode(), headers=headers) except standardwebhooks.WebhookVerificationError: return {"error": "signature verification failed"}, 401 if event.data.type != "session.status_run_started": return {"status": "ignored"}, 200 task = asyncio.create_task(run_queued_work()) inflight.add(task) task.add_done_callback(inflight.discard) try: # 已加 shield 保護：遺失或逾時的傳遞不得取消該項目；由 shutdown() 負責取消。 await asyncio.shield(task) except asyncio.CancelledError: return {"status": "shutting down"}, 503 return {"status": "ok"}, 200 async def run_queued_work() -> None: async for work in client.beta.environments.work.poller( environment_id=environment_id, environment_key=environment_key, block_ms=None, reclaim_older_than_ms=2000, drain=True, auto_stop=False, ): await client.beta.environments.work.worker(workdir="/workspace").handle_item( work_id=work.id, environment_id=environment_id, session_id=work.data.id, environment_key=environment_key, # 每個工作階段專屬的 secret 讓 worker 得以掛載該工作階段的記憶 store。 work_secret=work.secret, ) ``` ```typescript TypeScript import Anthropic from "@anthropic-ai/sdk"; const environmentKey = process.env.ANTHROPIC_ENVIRONMENT_KEY!; const environmentId = process.env.ANTHROPIC_ENVIRONMENT_ID!; const client = new Anthropic({ authToken: environmentKey }); // 請在主機的 SIGTERM/SIGINT 處理程式中呼叫 shutdown.abort()，同時關閉伺服器， // 然後在結束前等待進行中的 handle() 呼叫完成：abort 可讓執行中的工作項目 // 先上傳已變更的記憶體檔案並移除其 store 目錄。 export const shutdown = new AbortController(); export async function handle(req: Request): Promise { // 切勿確認不會在此執行其工作的遞送；回傳 503 可讓傳送端重試。 if (shutdown.signal.aborted) { return Response.json({ status: "shutting down" }, { status: 503 }); } const body = await req.text(); let event; try { event = client.beta.webhooks.unwrap(body, { headers: Object.fromEntries(req.headers) }); } catch { return new Response("signature verification failed", { status: 401 }); } if (event.data.type !== "session.status_run_started") { return Response.json({ status: "ignored" }); } for await (const work of client.beta.environments.work.poller({ environmentId, environmentKey, blockMs: null, reclaimOlderThanMs: 2000, drain: true, autoStop: false, signal: shutdown.signal })) { await client.beta.environments.work.worker({ workdir: "/workspace" }).handleItem({ workId: work.id, environmentId, sessionId: work.data.id, environmentKey, // 每個 session 專屬的 secret 讓 worker 得以掛載該 session 的記憶體 store。 workSecret: work.secret ?? undefined, signal: shutdown.signal }); } // poller 與 handleItem 在 abort 時會靜默返回，因此被中斷的 drain 會落到這裡。 if (shutdown.signal.aborted) { return Response.json({ status: "shutting down" }, { status: 503 }); } return Response.json({ status: "ok" }); } ``` C# SDK 沒有 `EnvironmentWorker`。若要從您自己的程式碼處理工作項目，請呼叫 [Environments Work 端點](https://platform.claude.com/docs/zh-TW/api/beta/environments/work)。 ```go Go package main import ( "context" "encoding/json" "errors" "io" "log/slog" "net/http" "os" "os/signal" "syscall" "github.com/anthropics/anthropic-sdk-go" "github.com/anthropics/anthropic-sdk-go/lib/environments" "github.com/anthropics/anthropic-sdk-go/option" "github.com/anthropics/anthropic-sdk-go/packages/param" ) var ( environmentKey = os.Getenv("ANTHROPIC_ENVIRONMENT_KEY") environmentID = os.Getenv("ANTHROPIC_ENVIRONMENT_ID") client = anthropic.NewClient( option.WithAuthToken(environmentKey), option.WithWebhookKey(os.Getenv("ANTHROPIC_WEBHOOK_SIGNING_KEY")), ) worker = environments.NewEnvironmentWorker(client, environments.EnvironmentWorkerOptions{ Workdir: "/workspace", }) // 於 SIGINT 或 SIGTERM 時取消（在 main 中設定），讓進行中的工作項目能 // 在結束前上傳已變更的記憶檔案並移除其 store 目錄。 shutdown context.Context ) func handle(w http.ResponseWriter, r *http.Request) { body, err := io.ReadAll(r.Body) if err != nil { http.Error(w, "bad request", http.StatusBadRequest) return } event, err := client.Beta.Webhooks.Unwrap(body, r.Header) if err != nil { http.Error(w, "signature verification failed", http.StatusUnauthorized) return } if event.Data.Type != "session.status_run_started" { json.NewEncoder(w).Encode(map[string]string{"status": "ignored"}) return } // Go SDK 未提供 RunOne 便利方法：請以 WorkPoller 取出待處理項目， // 並以 HandleItem 逐一執行。 // 與 r.Context() 分離：session 的存續時間可能超過 webhook 傳遞逾時。 // 整個行程的關閉 context 仍會在 SIGTERM 時乾淨地結束該項目。 ctx := shutdown poller := environments.NewWorkPoller(ctx, client, environments.WorkPollerOptions{ EnvironmentID: environmentID, EnvironmentKey: environmentKey, BlockMs: param.Null[int64](), ReclaimOlderThanMs: param.NewOpt[int64](2000), Drain: true, AutoStop: param.NewOpt(false), }) defer poller.Close() for poller.Next() { item := poller.Current() if err := worker.HandleItem(ctx, environments.HandleItemOptions{ WorkID: item.ID, EnvironmentID: item.EnvironmentID, SessionID: item.Data.ID, EnvironmentKey: environmentKey, // 每個 session 專屬的 secret 讓 worker 得以掛載該 session 的記憶 store。 WorkSecret: item.Secret, }); err != nil { slog.Error("handle work item", "work_id", item.ID, "err", err) http.Error(w, "internal error", http.StatusInternalServerError) return } } if err := poller.Err(); err != nil { slog.Error("poll work queue", "err", err) http.Error(w, "internal error", http.StatusInternalServerError) return } json.NewEncoder(w).Encode(map[string]string{"status": "ok"}) } func main() { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() shutdown = ctx server := &http.Server{Addr: ":8080"} http.HandleFunc("POST /webhook", handle) go func() { if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { slog.Error("http server", "err", err) os.Exit(1) } }() // 收到訊號時，停止接受傳遞，並僅在進行中的 // 處理程式（及其工作項目的記憶拆除作業）完成後才返回。 <-ctx.Done() if err := server.Shutdown(context.Background()); err != nil { slog.Error("http shutdown", "err", err) } } ``` Java SDK 沒有 `EnvironmentWorker`。若要從您自己的程式碼處理工作項目，請呼叫 [Environments Work 端點](https://platform.claude.com/docs/zh-TW/api/beta/environments/work)。 PHP SDK 沒有 `EnvironmentWorker`。若要從您自己的程式碼處理工作項目，請呼叫 [Environments Work 端點](https://platform.claude.com/docs/zh-TW/api/beta/environments/work)。 Ruby SDK 沒有 `EnvironmentWorker`。若要從您自己的程式碼處理工作項目，請呼叫 [Environments Work 端點](https://platform.claude.com/docs/zh-TW/api/beta/environments/work)。 由於處理常式會自行認領工作，因此它必須[轉送工作項目的密鑰](https://platform.claude.com/docs/zh-TW/managed-agents/self-hosted-sandboxes-workers#forward-the-work-items-secret)，如此處的 `work_secret` (typescript: `workSecret`; go: `WorkSecret`) 引數所示。 此處理常式會在一台主機上的單一程序中執行每個認領的項目。如果您的工作階段附加了相同的記憶體儲存區，請參閱[隔離共用儲存區的工作階段](https://platform.claude.com/docs/zh-TW/managed-agents/self-hosted-sandboxes-memory#isolate-sessions-that-share-a-store)。 ## 每個工作階段執行一個沙箱 主機上的輪詢器會認領工作，並針對每個工作項目呼叫一次您的指令碼。該指令碼會為該工作階段啟動一個沙箱。 安裝 `ant` 並將 `ant beta:worker run` 設為進入點。沙箱啟動時，會從環境變數讀取工作階段詳細資訊、處理該工作階段，然後結束。基礎映像檔必須提供 `/bin/bash`；`curl` 僅在建置時使用。 ```dockerfile FROM your-base-image ARG ANT_VERSION=1.42.0 ARG TARGETARCH RUN ARCH=$([ "$TARGETARCH" = "arm64" ] && echo arm64 || echo amd64) && \ curl -fsSL "https://github.com/anthropics/anthropic-cli/releases/download/v${ANT_VERSION}/ant_${ANT_VERSION}_linux_${ARCH}.tar.gz" \ | tar -xz -C /usr/local/bin ant WORKDIR /workspace VOLUME /workspace ENTRYPOINT ["ant", "beta:worker", "run"] ``` 該指令碼會將工作階段詳細資訊轉送到一個全新的沙箱中。它需要輪詢器主機上安裝 `jq`。 ```bash #!/bin/bash # spawn.sh：每個已認領的工作項目呼叫一次 # 已認領的工作項目會以 JSON 格式從 stdin 傳入。 ANTHROPIC_WORK_SECRET="$(jq -r '.secret // empty')" export ANTHROPIC_WORK_SECRET mkdir -p "/host/outputs/$ANTHROPIC_SESSION_ID" exec docker run --rm \ -e ANTHROPIC_SESSION_ID -e ANTHROPIC_ENVIRONMENT_KEY \ -e ANTHROPIC_WORK_ID -e ANTHROPIC_ENVIRONMENT_ID -e ANTHROPIC_BASE_URL \ -e ANTHROPIC_WORK_SECRET \ -v "/host/outputs/$ANTHROPIC_SESSION_ID":/workspace \ your-image ``` 輪詢器會設定指令碼所轉送的 `ANTHROPIC_*` 變數，但密鑰除外。請參閱[環境變數](https://platform.claude.com/docs/zh-TW/managed-agents/self-hosted-sandboxes-reference#environment-variables)。 `/host/outputs` 是您選擇的主機目錄。將其掛載到 `/workspace` 可讓您在沙箱結束後取回工作階段的產出物。此掛載也會包含下載的 `skills/` 目錄樹以及任何中間檔案。 ```bash ant beta:worker poll --on-work ./spawn.sh ``` ### 轉送工作項目的密鑰 每個已認領的工作項目都可以攜帶一個由 Anthropic 核發的每工作階段 `secret` (csharp, go: `Secret`)。執行該工作階段的 worker 需要它來掛載[記憶體儲存區](https://platform.claude.com/docs/zh-TW/managed-agents/self-hosted-sandboxes-memory)。 在單一程序中認領並執行工作階段的 worker（不帶 `--on-work` 的 `ant beta:worker poll`，或使用 `run()` (go: `Run()`) 的 `EnvironmentWorker`）會自行傳遞密鑰。當您自己的程式碼位於認領與 worker 之間時，您需要自行轉送它： | 您認領工作的方式 | 密鑰的傳入形式 | 傳遞給 worker 的方式 | | ---------------------------------------------------------- | ------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `ant beta:worker poll --on-work` | 您指令碼標準輸入上工作項目 JSON 的 `secret` 欄位 | 沙箱環境中的 `ANTHROPIC_WORK_SECRET` | | SDK 的 `work.poller()` (go: `environments.NewWorkPoller()`) | 每個已認領工作項目的 `secret` (go: `Secret`) 欄位 | 沙箱環境中的 `ANTHROPIC_WORK_SECRET`，或傳給 `handle_item()` (typescript: `handleItem()`; go: `HandleItem()`) 的 `work_secret` (typescript: `workSecret`; go: `WorkSecret`) 引數 | 請僅將密鑰傳入服務該工作階段的沙箱，且切勿將其記錄到日誌中。請參閱[安全模型](https://platform.claude.com/docs/zh-TW/managed-agents/self-hosted-sandboxes-security)以了解它與環境金鑰的關係。 ### 從 SDK 輪詢器啟動沙箱 若要從您自己的程式碼認領工作，而非使用 `ant beta:worker poll --on-work`，請使用 `work.poller()` (go: `environments.NewWorkPoller()`)。它會輪詢佇列並將每個認領的工作階段交給您，由您啟動沙箱： 在迴圈中認領工作並為每個工作階段啟動沙箱，所需的邏輯超出單一 shell 指令所能處理的範圍。從 shell 中，請搭配啟動指令碼執行 `ant beta:worker poll --on-work`，如[每個工作階段執行一個沙箱](https://platform.claude.com/docs/zh-TW/managed-agents/self-hosted-sandboxes-workers#run-one-sandbox-per-session)所示。 若要從 shell 認領工作並為每個工作階段啟動沙箱，請搭配啟動指令碼執行 `ant beta:worker poll --on-work`，如[每個工作階段執行一個沙箱](https://platform.claude.com/docs/zh-TW/managed-agents/self-hosted-sandboxes-workers#run-one-sandbox-per-session)所示。 ```python Python import asyncio import os from anthropic import AsyncAnthropic from anthropic.types.beta.environments import BetaSelfHostedWork SANDBOX_ENV = ( "ANTHROPIC_ENVIRONMENT_ID", "ANTHROPIC_ENVIRONMENT_KEY", "ANTHROPIC_WORK_ID", "ANTHROPIC_SESSION_ID", "ANTHROPIC_WORK_SECRET", "ANTHROPIC_BASE_URL", # forwarded only when set on this host ) async def launch_container(work: BetaSelfHostedWork) -> None: print(f"claimed session {work.data.id}") # 請將 `docker run` 替換為您自己的沙箱啟動器。請轉送環境 # 金鑰（切勿轉送您的 API 金鑰）以及工作項目的每個工作階段密鑰：內部的 worker # 需要該密鑰才能掛載該工作階段的記憶體儲存區。 env = os.environ | { "ANTHROPIC_WORK_ID": work.id, "ANTHROPIC_SESSION_ID": work.data.id, "ANTHROPIC_WORK_SECRET": work.secret or "", } forward = [arg for name in SANDBOX_ENV for arg in ("-e", name)] launcher = await asyncio.create_subprocess_exec( "docker", "run", "--rm", "--detach", *forward, "your-image", env=env ) await launcher.wait() async def main() -> None: environment_key = os.environ["ANTHROPIC_ENVIRONMENT_KEY"] environment_id = os.environ["ANTHROPIC_ENVIRONMENT_ID"] async with AsyncAnthropic(auth_token=environment_key) as client: async for work in client.beta.environments.work.poller( environment_id=environment_id, environment_key=environment_key, auto_stop=False, # the launched sandbox owns the stop call ): await launch_container(work) asyncio.run(main()) ``` ```typescript TypeScript import { spawn } from "node:child_process"; import { once } from "node:events"; import Anthropic from "@anthropic-ai/sdk"; import { WorkPoller } from "@anthropic-ai/sdk/helpers/beta/environments"; import type { BetaSelfHostedWork } from "@anthropic-ai/sdk/resources/beta/environments"; const SANDBOX_ENV = [ "ANTHROPIC_ENVIRONMENT_ID", "ANTHROPIC_ENVIRONMENT_KEY", "ANTHROPIC_WORK_ID", "ANTHROPIC_SESSION_ID", "ANTHROPIC_WORK_SECRET", "ANTHROPIC_BASE_URL" // forwarded only when set on this host ]; const environmentKey = process.env.ANTHROPIC_ENVIRONMENT_KEY!; const environmentId = process.env.ANTHROPIC_ENVIRONMENT_ID!; const client = new Anthropic({ authToken: environmentKey }); async function launchContainer(work: BetaSelfHostedWork): Promise { console.log(`claimed session ${work.data.id}`); // 請將 `docker run` 替換為您自己的沙箱啟動器。請轉送環境 // 金鑰（切勿轉送您的 API 金鑰）以及工作項目的每個工作階段密鑰：內部的 worker // 需要該密鑰才能掛載該工作階段的記憶體儲存區。 const env = { ...process.env, ANTHROPIC_WORK_ID: work.id, ANTHROPIC_SESSION_ID: work.data.id, ANTHROPIC_WORK_SECRET: work.secret ?? "" }; const forward = SANDBOX_ENV.flatMap((name) => ["-e", name]); const launcher = spawn("docker", ["run", "--rm", "--detach", ...forward, "your-image"], { env, stdio: "inherit" }); await once(launcher, "close"); } const poller = new WorkPoller({ client, environmentId, environmentKey, autoStop: false // the launched sandbox owns the stop call }); for await (const work of poller) { await launchContainer(work); } ``` C# SDK 沒有輪詢工作的輔助工具。若要從您自己的程式碼認領工作，請呼叫 [Environments Work 端點](https://platform.claude.com/docs/zh-TW/api/beta/environments/work)。 ```go Go package main import ( "context" "fmt" "log" "os" "os/exec" "github.com/anthropics/anthropic-sdk-go" "github.com/anthropics/anthropic-sdk-go/lib/environments" "github.com/anthropics/anthropic-sdk-go/option" "github.com/anthropics/anthropic-sdk-go/packages/param" ) var sandboxEnv = []string{ "ANTHROPIC_ENVIRONMENT_ID", "ANTHROPIC_ENVIRONMENT_KEY", "ANTHROPIC_WORK_ID", "ANTHROPIC_SESSION_ID", "ANTHROPIC_WORK_SECRET", "ANTHROPIC_BASE_URL", // forwarded only when set on this host } func launchContainer(ctx context.Context, work *anthropic.BetaSelfHostedWork) error { fmt.Printf("claimed session %s\n", work.Data.ID) // 請將 `docker run` 替換為您自己的沙箱啟動器。轉送環境 // 金鑰（切勿轉送您的 API 金鑰）以及工作項目的每個工作階段密鑰：其中的 worker // 需要此密鑰才能掛載該工作階段的記憶體儲存區。 args := []string{"run", "--rm", "--detach"} for _, name := range sandboxEnv { args = append(args, "-e", name) } launcher := exec.CommandContext(ctx, "docker", append(args, "your-image")...) launcher.Env = append(os.Environ(), "ANTHROPIC_WORK_ID="+work.ID, "ANTHROPIC_SESSION_ID="+work.Data.ID, "ANTHROPIC_WORK_SECRET="+work.Secret, ) launcher.Stdout, launcher.Stderr = os.Stdout, os.Stderr return launcher.Run() } func main() { environmentID := os.Getenv("ANTHROPIC_ENVIRONMENT_ID") environmentKey := os.Getenv("ANTHROPIC_ENVIRONMENT_KEY") client := anthropic.NewClient(option.WithAuthToken(environmentKey)) ctx := context.Background() poller := environments.NewWorkPoller(ctx, client, environments.WorkPollerOptions{ EnvironmentID: environmentID, EnvironmentKey: environmentKey, AutoStop: param.NewOpt(false), // the launched sandbox owns the stop call }) defer poller.Close() for work, err := range poller.All() { if err != nil { log.Fatal(err) } if err := launchContainer(ctx, work); err != nil { log.Fatal(err) } } } ``` Java SDK 沒有輪詢工作的輔助工具。若要從您自己的程式碼認領工作，請呼叫 [Environments Work 端點](https://platform.claude.com/docs/zh-TW/api/beta/environments/work)。 PHP SDK 沒有輪詢工作的輔助工具。若要從您自己的程式碼認領工作，請呼叫 [Environments Work 端點](https://platform.claude.com/docs/zh-TW/api/beta/environments/work)。 Ruby SDK 沒有輪詢工作的輔助工具。若要從您自己的程式碼認領工作，請呼叫 [Environments Work 端點](https://platform.claude.com/docs/zh-TW/api/beta/environments/work)。 ### 在沙箱內執行 SDK worker 當沙箱必須[提供自訂工具](https://platform.claude.com/docs/zh-TW/managed-agents/self-hosted-sandboxes-custom-tools)或使用[非預設的記憶同步設定](https://platform.claude.com/docs/zh-TW/managed-agents/self-hosted-sandboxes-memory#configure-sync)時，請將 `ant beta:worker run` 進入點替換為 SDK 進入點。該進入點會建構 `EnvironmentWorker` 並呼叫 `handle_item()` (typescript: `handleItem()`; go: `HandleItem()`)，它會讀取啟動指令碼所轉送的相同 `ANTHROPIC_*` 變數。 ```python Python import asyncio import contextlib import os import signal from anthropic import AsyncAnthropic from anthropic.lib.environments import EnvironmentWorker async def main() -> None: async with AsyncAnthropic( auth_token=os.environ["ANTHROPIC_ENVIRONMENT_KEY"] ) as client: worker = EnvironmentWorker(client, workdir="/workspace") # 不帶引數時，handle_item() 會讀取 spawn 指令碼所轉發的 # ANTHROPIC_* 變數，包括 ANTHROPIC_WORK_SECRET。 task = asyncio.create_task(worker.handle_item()) # 在容器停止時取消任務，可讓 worker 在結束前上傳 # 已變更的記憶檔案並移除儲存目錄。 loop = asyncio.get_running_loop() for signum in (signal.SIGINT, signal.SIGTERM): loop.add_signal_handler(signum, task.cancel) with contextlib.suppress(asyncio.CancelledError): await task asyncio.run(main()) ``` ```typescript TypeScript import Anthropic from "@anthropic-ai/sdk"; import { EnvironmentWorker } from "@anthropic-ai/sdk/helpers/beta/environments"; const client = new Anthropic({ authToken: process.env.ANTHROPIC_ENVIRONMENT_KEY }); const controller = new AbortController(); // 在容器停止時中止，可讓 worker 在結束前上傳已變更的記憶體 // 檔案並移除儲存目錄。 process.once("SIGTERM", () => controller.abort()); process.once("SIGINT", () => controller.abort()); // 不帶引數時，handleItem() 會讀取 spawn 指令碼所轉發的 ANTHROPIC_* 變數， // 包括 ANTHROPIC_WORK_SECRET。 await new EnvironmentWorker({ client, workdir: "/workspace", signal: controller.signal }).handleItem(); ``` C# SDK 沒有 `EnvironmentWorker`。 ```go Go package main import ( "context" "log" "os" "os/signal" "syscall" "github.com/anthropics/anthropic-sdk-go" "github.com/anthropics/anthropic-sdk-go/lib/environments" "github.com/anthropics/anthropic-sdk-go/option" ) func main() { // 在容器停止時取消 context，可讓 worker 在結束前上傳 // 已變更的記憶檔案並移除儲存目錄。 ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() client := anthropic.NewClient(option.WithAuthToken(os.Getenv("ANTHROPIC_ENVIRONMENT_KEY"))) worker := environments.NewEnvironmentWorker(client, environments.EnvironmentWorkerOptions{ Workdir: "/workspace", }) // 使用零值選項時，HandleItem 會讀取 spawn 腳本所轉發的 // ANTHROPIC_* 變數，包括 ANTHROPIC_WORK_SECRET。 if err := worker.HandleItem(ctx, environments.HandleItemOptions{}); err != nil { log.Fatalf("worker: %v", err) } } ``` Java SDK 沒有 `EnvironmentWorker`。 PHP SDK 沒有 `EnvironmentWorker`。 Ruby SDK 沒有 `EnvironmentWorker`。 ## 為工作階段準備檔案 Anthropic 不會將檔案或 GitHub 儲存庫掛載到自行託管的沙箱中。若要提供工作階段專屬的檔案： 1. 在工作階段的 `metadata` 欄位中傳遞檔案參照，例如 S3 路徑或 commit SHA。 2. 在您的啟動指令碼或 `--on-work` 處理常式中，擷取工作階段（`GET /v1/sessions/{session_id}`）並讀取 `metadata`。認領的工作項目攜帶工作階段 ID，但不包含 metadata。 3. 在工具執行開始之前，將檔案準備到工作目錄中。 ```bash cURL curl -sS --fail-with-body https://api.anthropic.com/v1/sessions \ -H "x-api-key: $ANTHROPIC_API_KEY" \ -H "anthropic-version: 2023-06-01" \ -H "anthropic-beta: managed-agents-2026-04-01" \ -H "content-type: application/json" \ -d '{ "agent": "'"$AGENT_ID"'", "environment_id": "'"$ANTHROPIC_ENVIRONMENT_ID"'", "metadata": {"input_file": "s3://my-bucket/data.csv"} }' ``` ```bash CLI ant beta:sessions create \ --agent "$AGENT_ID" \ --environment-id "$ANTHROPIC_ENVIRONMENT_ID" \ --metadata '{"input_file": "s3://my-bucket/data.csv"}' ``` ```python Python session = client.beta.sessions.create( agent=agent.id, environment_id=environment.id, metadata={"input_file": "s3://my-bucket/data.csv"}, ) ``` ```typescript TypeScript const session = await client.beta.sessions.create({ agent: agent.id, environment_id: environment.id, metadata: { input_file: "s3://my-bucket/data.csv" } }); ``` ```csharp C# var session = await client.Beta.Sessions.Create(new() { Agent = agent.ID, EnvironmentID = environment.ID, Metadata = new Dictionary { ["input_file"] = "s3://my-bucket/data.csv" }, }); ``` ```go Go session, err := client.Beta.Sessions.New(ctx, anthropic.BetaSessionNewParams{ Agent: anthropic.BetaSessionNewParamsAgentUnion{OfString: anthropic.String(agent.ID)}, EnvironmentID: environment.ID, Metadata: map[string]string{ "input_file": "s3://my-bucket/data.csv", }, }) if err != nil { panic(err) } ``` ```java Java var session = client.beta().sessions().create(SessionCreateParams.builder() .agent(agent.id()) .environmentId(environment.id()) .metadata(SessionCreateParams.Metadata.builder() .putAdditionalProperty("input_file", JsonValue.from("s3://my-bucket/data.csv")) .build()) .build()); ``` ```php PHP $session = $client->beta->sessions->create( agent: $agent->id, environmentID: $environment->id, metadata: ['input_file' => 's3://my-bucket/data.csv'], ); ``` ```ruby Ruby session = client.beta.sessions.create( agent: agent.id, environment_id: environment.id, metadata: {input_file: "s3://my-bucket/data.csv"} ) ``` ## 後續步驟 讀取佇列深度、乾淨地停止工作階段和 worker，並修正常見的失敗。 自行託管沙箱環境的共同責任模型。