部署自托管 worker

快速入门运行一个 ant CLI worker，它持续轮询并在一个进程中运行每个会话。本页介绍运行 worker 的其他方式以及如何在它们之间进行选择。

选择部署模式

部署 worker 时，您需要做出两个选择：worker 如何认领工作，以及每个会话在何处运行。

worker 如何认领工作：

• 常驻运行： 一个长时间运行的进程持续轮询队列，只需要出站 HTTPS。这是最简单的设置。
• Webhook 触发： 处理程序在 session.status_run_started 时被唤醒并开始轮询。这避免了空闲的轮询器，但需要一个 Anthropic 可以访问的 webhook 端点。

每个会话在何处运行：

• 进程内： 认领会话的 worker 同时在一个共享工作目录中运行其工具调用。
• 每个会话一个沙箱： 轮询器为每个认领的会话启动一个全新的沙箱。如需更强的隔离，请选择此方式：全新的文件系统、资源限制或按会话的网络控制。

CLI 和 SDK worker 支持不同的组合：

有关每个 CLI 标志和 SDK 选项，请参阅自托管 worker 参考。如需更多控制，请直接调用 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() 取消，以便正在进行的工作项能够上传已更改的内存文件，
   # 并在进程退出前移除其存储目录。
   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,
               # 每会话密钥使 worker 能够挂载该会话的内存存储。
               work_secret=work.secret,
           )

   由于处理程序自行认领工作，它必须转发工作项的 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_* 变量，但 secret 除外。请参阅环境变量。

   /host/outputs 是您选择的主机目录。将其挂载到 /workspace 后，您可以在沙箱退出后取回会话的交付成果。该挂载还会包含下载的 skills/ 目录树以及所有中间文件。

3. 3

   启动轮询器

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

   

转发工作项的 secret

每个认领的工作项都可以携带一个由 Anthropic 签发的按会话 secret。运行该会话的 worker 需要它来挂载记忆存储。

在一个进程中认领并运行会话的 worker（不带 --on-work 的 ant beta:worker poll，或使用 run() 的 EnvironmentWorker）会自行传递该 secret。当您自己的代码位于认领和 worker 之间时，需要由您来转发它：

仅将 secret 传入为该会话服务的沙箱，并且切勿记录它。有关它与环境密钥的关系，请参阅安全模型。

从 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())
        # 在容器停止时取消任务，可让工作进程在退出前上传
        # 已更改的内存文件并移除存储目录。
        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 路径或提交 SHA。
2. 在您的启动脚本或 --on-work 处理程序中，获取会话（GET /v1/sessions/{session_id}）并读取 metadata。认领的工作项携带会话 ID，但不携带元数据。
3. 在工具执行开始之前，将文件放入工作目录。

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

后续步骤