Deploy self-hosted workers

The quickstart runs one ant CLI worker that polls continuously and runs every session in one process. This page covers the other ways to run a worker and how to choose between them.

Choose a deployment pattern

When deploying workers, you need to make two choices: how the worker claims work, and where each session runs.

How the worker claims work:

• Always-on: A long-running process polls the queue continuously and needs only outbound HTTPS. This is the simplest setup.
• Webhook-triggered: A handler wakes on session.status_run_started and starts polling. This avoids an idle poller, but requires a webhook endpoint that Anthropic can reach.

Where each session runs:

• In process: The worker that claims a session also runs its tool calls, in one shared working directory.
• Sandbox per session: A poller launches a fresh sandbox for each claimed session. Choose this for stronger isolation: a fresh filesystem, resource limits, or per-session network controls.

The CLI and SDK workers support different combinations:

See Self-hosted worker reference for every CLI flag and SDK option. For more control, call the Environments Work endpoints directly and implement your own worker.

Run an always-on worker

Both workers authenticate with the environment key from the quickstart.

With the ant CLI:

ant beta:worker poll --workdir /workspace

With the SDK, EnvironmentWorker does the same work:

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())
        # Cancelling the task, rather than killing the process, lets the worker stop its
        # in-flight work item and upload changed memory files before it exits.
        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())

Trigger workers from webhooks

1. 1

   Subscribe to session webhooks

   In the Console, define a webhook endpoint that listens for session.status_run_started events. See Webhooks for details.

2. 2

   Export the webhook signing key

   Along with the environment ID and key from the quickstart, export the webhook signing key on your handler host. The handler uses it to verify incoming payloads.

   export ANTHROPIC_WEBHOOK_SIGNING_KEY="whsec_..."

   
3. 3

   Implement the webhook handler

   Invoke the worker when session.status_run_started fires. The handler drains the queue and hands each claimed work item to handle_item(), which downloads skills, executes tool calls, posts results back, and returns.

   PythonTypeScriptC#GoJavaPHPRuby

   

   To verify webhook signatures, install the webhooks extra: 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,
   )
   # Cancelled by shutdown() so an in-flight work item can upload changed memory files and
   # remove its store directories before the process exits.
   inflight: set[asyncio.Task[None]] = set()
   
   
   # Await this from the host's shutdown hook, such as an ASGI lifespan shutdown (the code after
   # `yield` in a FastAPI lifespan), which uvicorn runs on SIGTERM. uvicorn lets open requests
   # finish before that hook runs, so set --timeout-graceful-shutdown to bound the wait.
   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:
           # Shielded: a dropped or timed-out delivery must not cancel the item; shutdown() does.
           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,
               # The per-session secret is what lets the worker mount the session's memory stores.
               work_secret=work.secret,
           )

   Because the handler claims work itself, it must forward the work item's secret, as the work_secret argument does here.

This handler runs every claimed item in one process on one host. If your sessions attach the same memory store, see Isolate sessions that share a store.

Run one sandbox per session

A poller on the host claims work and calls your script once per work item. The script launches a sandbox for that one session.

1. 1

   Build the sandbox image

   Install ant and set ant beta:worker run as the entrypoint. When a sandbox starts, it reads session details from environment variables, handles that session, and exits. The base image must provide /bin/bash; curl is only used at build time.

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

   Write the spawn script

   The script forwards the session details into a fresh sandbox. It requires jq on the poller host.

   #!/bin/bash
   # spawn.sh: called once per claimed work item
   # The claimed work item arrives as JSON on 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

   

   The poller sets the ANTHROPIC_* variables that the script forwards, except the secret. See Environment variables.

   /host/outputs is a host directory you choose. Mounting it at /workspace lets you retrieve the session's deliverables after the sandbox exits. The mount also picks up the downloaded skills/ tree and any intermediate files.

3. 3

   Start the poller

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

   

Forward the work item's secret

Each claimed work item can carry a per-session secret, which Anthropic issues. The worker that runs the session needs it to mount memory stores.

A worker that claims and runs sessions in one process (ant beta:worker poll without --on-work, or EnvironmentWorker with run()) passes the secret along itself. When your own code sits between the claim and the worker, you forward it:

Pass the secret only into the sandbox that serves that session, and never log it. See Security model for how it relates to the environment key.

Launch sandboxes from the SDK poller

To claim work from your own code instead of ant beta:worker poll --on-work, use work.poller(). It polls the queue and gives you each claimed session, and you launch the sandbox:

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}")
    # Replace `docker run` with your own sandbox launcher. Forward the environment
    # key (never your API key) and the work item's per-session secret: the worker
    # inside needs the secret to mount the session's memory stores.
    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())

Run the SDK worker inside the sandbox

Replace the ant beta:worker run entrypoint with an SDK entrypoint when the sandbox must serve custom tools or use non-default memory sync settings. The entrypoint constructs EnvironmentWorker and calls handle_item(), which reads the same ANTHROPIC_* variables that the spawn script forwards.

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")
        # With no arguments, handle_item() reads the ANTHROPIC_* variables the spawn
        # script forwarded, including ANTHROPIC_WORK_SECRET.
        task = asyncio.create_task(worker.handle_item())
        # Cancelling the task when the container is stopped lets the worker upload
        # changed memory files and remove the store directories before it exits.
        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())

Stage files for a session

Anthropic doesn't mount files or GitHub repositories into self-hosted sandboxes. To make session-specific files available:

1. Pass file references, such as an S3 path or commit SHA, in the session's metadata field.
2. In your spawn script or --on-work handler, retrieve the session (GET /v1/sessions/{session_id}) and read metadata. The claimed work item carries the session ID but not the metadata.
3. Stage the files into the working directory before tool execution begins.

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

Next steps