Distribuire worker self-hosted

La guida rapida esegue un worker della CLI ant che effettua il polling in modo continuo ed esegue ogni sessione in un unico processo. Questa pagina illustra gli altri modi per eseguire un worker e come scegliere tra di essi.

Scegli un modello di distribuzione

Quando distribuisci i worker, devi fare due scelte: come il worker acquisisce il lavoro e dove viene eseguita ciascuna sessione.

Come il worker acquisisce il lavoro:

• Sempre attivo: un processo di lunga durata effettua il polling della coda in modo continuo e richiede solo HTTPS in uscita. Questa è la configurazione più semplice.
• Attivato da webhook: un handler si attiva su session.status_run_started e avvia il polling. Questo evita un poller inattivo, ma richiede un endpoint webhook raggiungibile da Anthropic.

Dove viene eseguita ciascuna sessione:

• Nel processo: il worker che acquisisce una sessione esegue anche le sue chiamate agli strumenti, in un'unica directory di lavoro condivisa.
• Una sandbox per sessione: un poller avvia una nuova sandbox per ogni sessione acquisita. Scegli questa opzione per un isolamento più forte: un filesystem nuovo, limiti sulle risorse o controlli di rete per sessione.

I worker della CLI e dell'SDK supportano combinazioni diverse:

Consulta il Riferimento per i worker self-hosted per ogni flag della CLI e opzione dell'SDK. Per un maggiore controllo, chiama direttamente gli endpoint Work degli Environments e implementa il tuo worker.

Esegui un worker sempre attivo

Entrambi i worker si autenticano con la chiave dell'ambiente della guida rapida.

Con la CLI ant:

ant beta:worker poll --workdir /workspace

Con l'SDK, EnvironmentWorker svolge lo stesso lavoro:

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())
        # Annullare il task, anziché terminare il processo, consente al worker di fermare il suo
        # elemento di lavoro in corso e caricare i file di memoria modificati prima di uscire.
        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())

Attiva i worker tramite webhook

1. 1

   Iscriviti ai webhook delle sessioni

   Nella Console, definisci un endpoint webhook che ascolti gli eventi session.status_run_started. Consulta Webhook per i dettagli.

2. 2

   Esporta la chiave di firma del webhook

   Insieme all'ID e alla chiave dell'ambiente della guida rapida, esporta la chiave di firma del webhook sull'host del tuo handler. L'handler la usa per verificare i payload in arrivo.

   export ANTHROPIC_WEBHOOK_SIGNING_KEY="whsec_..."

   
3. 3

   Implementa l'handler del webhook

   Invoca il worker quando viene attivato session.status_run_started. L'handler svuota la coda e passa ogni elemento di lavoro acquisito a handle_item(), che scarica le skill, esegue le chiamate agli strumenti, invia i risultati e termina.

   PythonTypeScriptC#GoJavaPHPRuby

   

   Per verificare le firme dei webhook, installa l'extra 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,
   )
   # Annullato da shutdown() affinché un elemento di lavoro in corso possa caricare i file di memoria modificati e
   # rimuovere le proprie directory di store prima che il processo termini.
   inflight: set[asyncio.Task[None]] = set()
   
   
   # Attendi questo dall'hook di arresto dell'host, ad es. uno shutdown del lifespan ASGI (il codice dopo
   # `yield` in un lifespan FastAPI), che uvicorn esegue su SIGTERM. uvicorn lascia terminare le richieste aperte
   # prima che l'hook venga eseguito, quindi imposta --timeout-graceful-shutdown per limitare l'attesa.
   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:
           # Protetto (shield): una consegna persa o scaduta non deve annullare l'elemento; shutdown() lo fa.
           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,
               # Il segreto per sessione è ciò che consente al worker di montare gli store di memoria della sessione.
               work_secret=work.secret,
           )

   Poiché l'handler acquisisce il lavoro autonomamente, deve inoltrare il segreto dell'elemento di lavoro, come fa qui l'argomento work_secret.

Questo handler esegue ogni elemento acquisito in un unico processo su un unico host. Se le tue sessioni collegano lo stesso memory store, consulta Isolare le sessioni che condividono uno store.

Esegui una sandbox per sessione

Un poller sull'host acquisisce il lavoro e chiama il tuo script una volta per ogni elemento di lavoro. Lo script avvia una sandbox per quella singola sessione.

1. 1

   Crea l'immagine della sandbox

   Installa ant e imposta ant beta:worker run come entrypoint. Quando una sandbox si avvia, legge i dettagli della sessione dalle variabili d'ambiente, gestisce quella sessione e termina. L'immagine di base deve fornire /bin/bash; curl viene usato solo in fase di build.

   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

   Scrivi lo script di avvio

   Lo script inoltra i dettagli della sessione in una nuova sandbox. Richiede jq sull'host del poller.

   #!/bin/bash
   # spawn.sh: chiamato una volta per ogni elemento di lavoro acquisito
   # L'elemento di lavoro acquisito arriva come JSON su 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

   

   Il poller imposta le variabili ANTHROPIC_* che lo script inoltra, tranne il segreto. Consulta Variabili d'ambiente.

   /host/outputs è una directory dell'host a tua scelta. Montarla su /workspace ti consente di recuperare i risultati della sessione dopo che la sandbox è terminata. Il mount include anche l'albero skills/ scaricato e gli eventuali file intermedi.

3. 3

   Avvia il poller

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

   

Inoltra il segreto dell'elemento di lavoro

Ogni elemento di lavoro acquisito può contenere un secret per sessione, emesso da Anthropic. Il worker che esegue la sessione ne ha bisogno per montare i memory store.

Un worker che acquisisce ed esegue le sessioni in un unico processo (ant beta:worker poll senza --on-work, oppure EnvironmentWorker con run()) trasmette il segreto autonomamente. Quando il tuo codice si trova tra l'acquisizione e il worker, sei tu a inoltrarlo:

Passa il segreto solo alla sandbox che serve quella sessione e non registrarlo mai nei log. Consulta Modello di sicurezza per capire come si relaziona alla chiave dell'ambiente.

Avvia le sandbox dal poller dell'SDK

Per acquisire il lavoro dal tuo codice invece che con ant beta:worker poll --on-work, usa work.poller(). Effettua il polling della coda e ti fornisce ogni sessione acquisita, mentre tu avvii la 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}")
    # Sostituisci `docker run` con il tuo launcher di sandbox. Inoltra la chiave
    # dell'ambiente (mai la tua chiave API) e il segreto per sessione del work item: il worker
    # all'interno ha bisogno del segreto per montare i memory store della sessione.
    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())

Esegui il worker dell'SDK all'interno della sandbox

Sostituisci l'entrypoint ant beta:worker run con un entrypoint dell'SDK quando la sandbox deve servire strumenti personalizzati o usare impostazioni di sincronizzazione della memoria non predefinite. L'entrypoint costruisce EnvironmentWorker e chiama handle_item(), che legge le stesse variabili ANTHROPIC_* inoltrate dallo script di avvio.

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")
        # Senza argomenti, handle_item() legge le variabili ANTHROPIC_* che lo script di spawn
        # ha inoltrato, inclusa ANTHROPIC_WORK_SECRET.
        task = asyncio.create_task(worker.handle_item())
        # Annullare il task quando il container viene arrestato consente al worker di caricare
        # i file di memoria modificati e rimuovere le directory dello store prima di uscire.
        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 i file per una sessione

Anthropic non monta file o repository GitHub nelle sandbox self-hosted. Per rendere disponibili i file specifici di una sessione:

1. Passa i riferimenti ai file, come un percorso S3 o lo SHA di un commit, nel campo metadata della sessione.
2. Nel tuo script di avvio o nell'handler --on-work, recupera la sessione (GET /v1/sessions/{session_id}) e leggi metadata. L'elemento di lavoro acquisito contiene l'ID della sessione ma non i metadati.
3. Prepara i file nella directory di lavoro prima che inizi l'esecuzione degli strumenti.

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

Passaggi successivi