Implementar workers autoalojados

El inicio rápido ejecuta un worker de la CLI ant que sondea continuamente y ejecuta cada sesión en un solo proceso. Esta página cubre las otras formas de ejecutar un worker y cómo elegir entre ellas.

Elige un patrón de implementación

Al implementar workers, debes tomar dos decisiones: cómo el worker reclama trabajo y dónde se ejecuta cada sesión.

Cómo el worker reclama trabajo:

• Siempre activo: Un proceso de larga duración sondea la cola continuamente y solo necesita HTTPS saliente. Esta es la configuración más sencilla.
• Activado por webhook: Un controlador se activa con session.status_run_started y comienza a sondear. Esto evita un sondeador inactivo, pero requiere un endpoint de webhook al que Anthropic pueda acceder.

Dónde se ejecuta cada sesión:

• En proceso: El worker que reclama una sesión también ejecuta sus llamadas a herramientas, en un único directorio de trabajo compartido.
• Sandbox por sesión: Un sondeador lanza un sandbox nuevo para cada sesión reclamada. Elige esta opción para un aislamiento más fuerte: un sistema de archivos nuevo, límites de recursos o controles de red por sesión.

Los workers de la CLI y del SDK admiten diferentes combinaciones:

Consulta la referencia de workers autoalojados para ver cada flag de la CLI y cada opción del SDK. Para tener más control, llama directamente a los endpoints de Environments Work e implementa tu propio worker.

Ejecuta un worker siempre activo

Ambos workers se autentican con la clave de entorno del inicio rápido.

Con la CLI ant:

ant beta:worker poll --workdir /workspace

Con el SDK, EnvironmentWorker hace el mismo trabajo:

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())
        # Cancelar la tarea, en lugar de matar el proceso, permite al worker detener su
        # elemento de trabajo en curso y subir los archivos de memoria modificados antes de salir.
        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())

Activa workers desde webhooks

1. 1

   Suscríbete a los webhooks de sesión

   En la Console, define un endpoint de webhook que escuche eventos session.status_run_started. Consulta Webhooks para obtener más detalles.

2. 2

   Exporta la clave de firma del webhook

   Junto con el ID y la clave del entorno del inicio rápido, exporta la clave de firma del webhook en el host de tu controlador. El controlador la usa para verificar las cargas útiles entrantes.

   export ANTHROPIC_WEBHOOK_SIGNING_KEY="whsec_..."

   
3. 3

   Implementa el controlador del webhook

   Invoca el worker cuando se dispare session.status_run_started. El controlador vacía la cola y entrega cada elemento de trabajo reclamado a handle_item(), que descarga skills, ejecuta llamadas a herramientas, publica los resultados y retorna.

   PythonTypeScriptC#GoJavaPHPRuby

   

   Para verificar las firmas de webhook, instala el extra de 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,
   )
   # Cancelado por shutdown() para que un elemento de trabajo en curso pueda subir archivos de memoria modificados y
   # eliminar sus directorios de almacén antes de que el proceso termine.
   inflight: set[asyncio.Task[None]] = set()
   
   
   # Espera esto desde el hook de apagado del host, como un apagado de lifespan ASGI (el código después de
   # `yield` en un lifespan de FastAPI), que uvicorn ejecuta en SIGTERM. uvicorn deja que las solicitudes abiertas
   # terminen antes de ejecutar ese hook, así que define --timeout-graceful-shutdown para limitar la espera.
   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:
           # Protegido (shield): una entrega descartada o con timeout no debe cancelar el elemento; shutdown() sí lo hace.
           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,
               # El secreto por sesión es lo que permite al worker montar los almacenes de memoria de la sesión.
               work_secret=work.secret,
           )

   Como el controlador reclama el trabajo por sí mismo, debe reenviar el secreto del elemento de trabajo, como lo hace aquí el argumento work_secret.

Este controlador ejecuta cada elemento reclamado en un solo proceso en un solo host. Si tus sesiones adjuntan el mismo almacén de memoria, consulta Aislar sesiones que comparten un almacén.

Ejecuta un sandbox por sesión

Un sondeador en el host reclama trabajo y llama a tu script una vez por cada elemento de trabajo. El script lanza un sandbox para esa sesión.

1. 1

   Construye la imagen del sandbox

   Instala ant y establece ant beta:worker run como punto de entrada. Cuando un sandbox se inicia, lee los detalles de la sesión desde variables de entorno, gestiona esa sesión y termina. La imagen base debe proporcionar /bin/bash; curl solo se usa en tiempo de construcción.

   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

   Escribe el script de lanzamiento

   El script reenvía los detalles de la sesión a un sandbox nuevo. Requiere jq en el host del sondeador.

   #!/bin/bash
   # spawn.sh: se llama una vez por cada elemento de trabajo reclamado
   # El elemento de trabajo reclamado llega como JSON por 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

   

   El sondeador establece las variables ANTHROPIC_* que el script reenvía, excepto el secreto. Consulta Variables de entorno.

   /host/outputs es un directorio del host que tú eliges. Montarlo en /workspace te permite recuperar los entregables de la sesión después de que el sandbox termine. El montaje también recoge el árbol skills/ descargado y cualquier archivo intermedio.

3. 3

   Inicia el sondeador

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

   

Reenvía el secreto del elemento de trabajo

Cada elemento de trabajo reclamado puede llevar un secret por sesión, emitido por Anthropic. El worker que ejecuta la sesión lo necesita para montar almacenes de memoria.

Un worker que reclama y ejecuta sesiones en un solo proceso (ant beta:worker poll sin --on-work, o EnvironmentWorker con run()) transmite el secreto por sí mismo. Cuando tu propio código se sitúa entre la reclamación y el worker, tú lo reenvías:

Pasa el secreto solo al sandbox que atiende esa sesión y nunca lo registres en logs. Consulta Modelo de seguridad para ver cómo se relaciona con la clave de entorno.

Lanza sandboxes desde el sondeador del SDK

Para reclamar trabajo desde tu propio código en lugar de ant beta:worker poll --on-work, usa work.poller(). Sondea la cola y te entrega cada sesión reclamada, y tú lanzas el 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}")
    # Reemplaza `docker run` con tu propio lanzador de sandbox. Reenvía la clave del
    # entorno (nunca tu clave de API) y el secreto por sesión del elemento de trabajo: el worker
    # interno necesita el secreto para montar los almacenes de memoria de la sesión.
    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())

Ejecuta el worker del SDK dentro del sandbox

Reemplaza el punto de entrada ant beta:worker run por un punto de entrada del SDK cuando el sandbox deba servir herramientas personalizadas o usar configuraciones de sincronización de memoria no predeterminadas. El punto de entrada construye EnvironmentWorker y llama a handle_item(), que lee las mismas variables ANTHROPIC_* que reenvía el script de lanzamiento.

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")
        # Sin argumentos, handle_item() lee las variables ANTHROPIC_* que el script de
        # lanzamiento reenvió, incluida ANTHROPIC_WORK_SECRET.
        task = asyncio.create_task(worker.handle_item())
        # Cancelar la tarea cuando se detiene el contenedor permite al worker subir
        # los archivos de memoria modificados y eliminar los directorios del almacén antes de salir.
        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())

Prepara archivos para una sesión

Anthropic no monta archivos ni repositorios de GitHub en sandboxes autoalojados. Para que los archivos específicos de la sesión estén disponibles:

1. Pasa referencias de archivos, como una ruta de S3 o un SHA de commit, en el campo metadata de la sesión.
2. En tu script de lanzamiento o controlador --on-work, recupera la sesión (GET /v1/sessions/{session_id}) y lee metadata. El elemento de trabajo reclamado lleva el ID de la sesión, pero no los metadatos.
3. Coloca los archivos en el directorio de trabajo antes de que comience la ejecución de herramientas.

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

Próximos pasos