部署自行託管的 worker

快速入門會執行一個 ant CLI worker，它會持續輪詢，並在單一程序中執行每個工作階段。本頁介紹執行 worker 的其他方式，以及如何在它們之間做出選擇。

選擇部署模式

部署 worker 時，您需要做出兩個選擇：worker 如何認領工作，以及每個工作階段在何處執行。

worker 如何認領工作：

• 常駐（Always-on）： 一個長時間執行的程序會持續輪詢佇列，且只需要對外的 HTTPS 連線。這是最簡單的設定。
• 由 webhook 觸發（Webhook-triggered）： 處理常式會在 session.status_run_started 時喚醒並開始輪詢。這可避免閒置的輪詢器，但需要一個 Anthropic 能夠連線到的 webhook 端點。

每個工作階段在何處執行：

• 在程序內（In process）： 認領工作階段的 worker 也會在一個共用的工作目錄中執行其工具呼叫。
• 每個工作階段一個沙箱（Sandbox per session）： 輪詢器會為每個認領的工作階段啟動一個全新的沙箱。若需要更強的隔離，請選擇此方式：全新的檔案系統、資源限制或每個工作階段的網路控制。

CLI 和 SDK worker 支援不同的組合：

請參閱自行託管 worker 參考以了解每個 CLI 旗標和 SDK 選項。若需要更多控制，請直接呼叫 Environments Work 端點並實作您自己的 worker。

執行常駐 worker

兩種 worker 都使用快速入門中的環境金鑰進行驗證。

使用 ant CLI：

ant beta:worker poll --workdir /workspace

使用 SDK 時，EnvironmentWorker 會執行相同的工作：

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())

從 webhook 觸發 worker

1. 1

   訂閱工作階段 webhook

   在 Console 中，定義一個監聽 session.status_run_started 事件的 webhook 端點。詳情請參閱 Webhooks。

2. 2

   匯出 webhook 簽署金鑰

   除了快速入門中的環境 ID 和金鑰之外，請在您的處理常式主機上匯出 webhook 簽署金鑰。處理常式會使用它來驗證傳入的酬載。

   export ANTHROPIC_WEBHOOK_SIGNING_KEY="whsec_..."

   
3. 3

   實作 webhook 處理常式

   在 session.status_run_started 觸發時呼叫 worker。處理常式會清空佇列，並將每個認領的工作項目交給 handle_item()，它會下載技能、執行工具呼叫、回傳結果，然後返回。

   PythonTypeScriptC#GoJavaPHPRuby

   

   若要驗證 webhook 簽章，請安裝 webhooks 額外套件：pip install "anthropic[webhooks]"。

   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,
           )

   由於處理常式會自行認領工作，因此它必須轉送工作項目的密鑰，如此處的 work_secret 引數所示。

此處理常式會在一台主機上的單一程序中執行每個認領的項目。如果您的工作階段附加了相同的記憶體儲存區，請參閱隔離共用儲存區的工作階段。

每個工作階段執行一個沙箱

主機上的輪詢器會認領工作，並針對每個工作項目呼叫一次您的指令碼。該指令碼會為該工作階段啟動一個沙箱。

1. 1

   建置沙箱映像檔

   安裝 ant 並將 ant beta:worker run 設為進入點。沙箱啟動時，會從環境變數讀取工作階段詳細資訊、處理該工作階段，然後結束。基礎映像檔必須提供 /bin/bash；curl 僅在建置時使用。

   FROM your-base-image
   ARG ANT_VERSION=1.39.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"]

   
2. 2

   撰寫啟動指令碼

   該指令碼會將工作階段詳細資訊轉送到一個全新的沙箱中。它需要輪詢器主機上安裝 jq。

   #!/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_* 變數，但密鑰除外。請參閱環境變數。

   /host/outputs 是您選擇的主機目錄。將其掛載到 /workspace 可讓您在沙箱結束後取回工作階段的產出物。此掛載也會包含下載的 skills/ 目錄樹以及任何中間檔案。

3. 3

   啟動輪詢器

   ant beta:worker poll --on-work ./spawn.sh

   

轉送工作項目的密鑰

每個認領的工作項目都可以攜帶一個由 Anthropic 核發的每工作階段 secret。執行該工作階段的 worker 需要它來掛載記憶體儲存區。

在單一程序中認領並執行工作階段的 worker（不帶 --on-work 的 ant beta:worker poll，或使用 run() 的 EnvironmentWorker）會自行傳遞密鑰。當您自己的程式碼位於認領與 worker 之間時，您需要自行轉送它：

請僅將密鑰傳入服務該工作階段的沙箱，且切勿將其記錄到日誌中。請參閱安全模型以了解它與環境金鑰的關係。

從 SDK 輪詢器啟動沙箱

若要從您自己的程式碼認領工作，而非使用 ant beta:worker poll --on-work，請使用 work.poller()。它會輪詢佇列並將每個認領的工作階段交給您，由您啟動沙箱：

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())

在沙箱內執行 SDK worker

當沙箱必須提供自訂工具或使用非預設的記憶同步設定時，請將 ant beta:worker run 進入點替換為 SDK 進入點。該進入點會建構 EnvironmentWorker 並呼叫 handle_item()，它會讀取啟動指令碼所轉送的相同 ANTHROPIC_* 變數。

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())

為工作階段準備檔案

Anthropic 不會將檔案或 GitHub 儲存庫掛載到自行託管的沙箱中。若要提供工作階段專屬的檔案：

1. 在工作階段的 metadata 欄位中傳遞檔案參照，例如 S3 路徑或 commit SHA。
2. 在您的啟動指令碼或 --on-work 處理常式中，擷取工作階段（GET /v1/sessions/{session_id}）並讀取 metadata。認領的工作項目攜帶工作階段 ID，但不包含 metadata。
3. 在工具執行開始之前，將檔案準備到工作目錄中。

session = client.beta.sessions.create(
    agent=agent.id,
    environment_id=environment.id,
    metadata={"input_file": "s3://my-bucket/data.csv"},
)

後續步驟