Déployer des workers auto-hébergés

Le démarrage rapide exécute un worker CLI ant qui interroge la file en continu et exécute chaque session dans un seul processus. Cette page présente les autres façons d'exécuter un worker et comment choisir entre elles.

Choisir un modèle de déploiement

Lors du déploiement de workers, vous devez faire deux choix : comment le worker réclame le travail, et où chaque session s'exécute.

Comment le worker réclame le travail :

• Toujours actif : Un processus de longue durée interroge la file en continu et n'a besoin que de HTTPS sortant. C'est la configuration la plus simple.
• Déclenché par webhook : Un gestionnaire se réveille sur session.status_run_started et commence à interroger la file. Cela évite un « poller » (processus d'interrogation) inactif, mais nécessite un point de terminaison de webhook accessible par Anthropic.

Où chaque session s'exécute :

• Dans le processus : Le worker qui réclame une session exécute également ses appels d'outils, dans un répertoire de travail partagé unique.
• Une sandbox par session : Un poller lance une nouvelle sandbox pour chaque session réclamée. Choisissez cette option pour une isolation plus forte : un système de fichiers neuf, des limites de ressources ou des contrôles réseau par session.

Les workers CLI et SDK prennent en charge différentes combinaisons :

Consultez la référence des workers auto-hébergés pour chaque option de la CLI et du SDK. Pour plus de contrôle, appelez directement les points de terminaison Environments Work et implémentez votre propre worker.

Exécuter un worker toujours actif

Les deux workers s'authentifient avec la clé d'environnement du démarrage rapide.

Avec la CLI ant :

ant beta:worker poll --workdir /workspace

Avec le SDK, EnvironmentWorker effectue le même travail :

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())
        # Annuler la tâche, plutôt que de tuer le processus, permet au worker d'arrêter son
        # élément de travail en cours et de téléverser les fichiers mémoire modifiés avant de quitter.
        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())

Déclencher des workers à partir de webhooks

1. 1

   S'abonner aux webhooks de session

   Dans la Console, définissez un point de terminaison de webhook qui écoute les événements session.status_run_started. Consultez Webhooks pour plus de détails.

2. 2

   Exporter la clé de signature du webhook

   En plus de l'ID et de la clé d'environnement du démarrage rapide, exportez la clé de signature du webhook sur l'hôte de votre gestionnaire. Le gestionnaire l'utilise pour vérifier les charges utiles entrantes.

   export ANTHROPIC_WEBHOOK_SIGNING_KEY="whsec_..."

   
3. 3

   Implémenter le gestionnaire de webhook

   Invoquez le worker lorsque session.status_run_started se déclenche. Le gestionnaire vide la file et transmet chaque élément de travail réclamé à handle_item(), qui télécharge les skills, exécute les appels d'outils, renvoie les résultats et se termine.

   PythonTypeScriptC#GoJavaPHPRuby

   

   Pour vérifier les signatures des webhooks, installez 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,
   )
   # Annulé par shutdown() afin qu'un élément de travail en cours puisse téléverser les fichiers mémoire modifiés et
   # supprimer ses répertoires de store avant la fin du processus.
   inflight: set[asyncio.Task[None]] = set()
   
   
   # Attendez ceci depuis le hook d'arrêt de l'hôte, par ex. un arrêt de lifespan ASGI (le code après
   # `yield` dans un lifespan FastAPI), qu'uvicorn exécute sur SIGTERM. uvicorn laisse les requêtes ouvertes
   # se terminer avant l'exécution de ce hook ; définissez --timeout-graceful-shutdown pour borner l'attente.
   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:
           # Protégé : une livraison abandonnée ou expirée ne doit pas annuler l'élément ; shutdown() le fait.
           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,
               # Le secret par session est ce qui permet au worker de monter les stores mémoire de la session.
               work_secret=work.secret,
           )

   Comme le gestionnaire réclame lui-même le travail, il doit transmettre le secret de l'élément de travail, comme le fait ici l'argument work_secret.

Ce gestionnaire exécute chaque élément réclamé dans un seul processus sur un seul hôte. Si vos sessions attachent le même magasin de mémoire, consultez Isoler les sessions qui partagent un magasin.

Exécuter une sandbox par session

Un poller sur l'hôte réclame le travail et appelle votre script une fois par élément de travail. Le script lance une sandbox pour cette session unique.

1. 1

   Construire l'image de la sandbox

   Installez ant et définissez ant beta:worker run comme point d'entrée. Lorsqu'une sandbox démarre, elle lit les détails de la session à partir des variables d'environnement, traite cette session, puis se termine. L'image de base doit fournir /bin/bash ; curl n'est utilisé qu'au moment de la construction.

   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

   Écrire le script de lancement

   Le script transmet les détails de la session à une nouvelle sandbox. Il nécessite jq sur l'hôte du poller.

   #!/bin/bash
   # spawn.sh : appelé une fois par élément de travail réclamé.
   # L'élément de travail réclamé arrive au format JSON sur 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

   

   Le poller définit les variables ANTHROPIC_* que le script transmet, à l'exception du secret. Consultez Variables d'environnement.

   /host/outputs est un répertoire de l'hôte que vous choisissez. Le monter sur /workspace vous permet de récupérer les livrables de la session après la fin de la sandbox. Le montage récupère également l'arborescence skills/ téléchargée et tous les fichiers intermédiaires.

3. 3

   Démarrer le poller

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

   

Transmettre le secret de l'élément de travail

Chaque élément de travail réclamé peut contenir un secret propre à la session, émis par Anthropic. Le worker qui exécute la session en a besoin pour monter les magasins de mémoire.

Un worker qui réclame et exécute les sessions dans un seul processus (ant beta:worker poll sans --on-work, ou EnvironmentWorker avec run()) transmet lui-même le secret. Lorsque votre propre code se situe entre la réclamation et le worker, c'est à vous de le transmettre :

Transmettez le secret uniquement à la sandbox qui sert cette session, et ne le journalisez jamais. Consultez Modèle de sécurité pour comprendre son lien avec la clé d'environnement.

Lancer des sandboxes depuis le poller du SDK

Pour réclamer le travail depuis votre propre code au lieu de ant beta:worker poll --on-work, utilisez work.poller(). Il interroge la file et vous fournit chaque session réclamée, et c'est vous qui lancez 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}")
    # Remplacez `docker run` par votre propre lanceur de sandbox. Transmettez la clé
    # d'environnement (jamais votre clé API) et le secret par session de l'élément de travail : le worker
    # à l'intérieur a besoin du secret pour monter les magasins de mémoire de la session.
    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())

Exécuter le worker du SDK dans la sandbox

Remplacez le point d'entrée ant beta:worker run par un point d'entrée SDK lorsque la sandbox doit servir des outils personnalisés ou utiliser des paramètres de synchronisation de mémoire non par défaut. Le point d'entrée construit EnvironmentWorker et appelle handle_item(), qui lit les mêmes variables ANTHROPIC_* que celles transmises par le script de lancement.

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")
        # Sans arguments, handle_item() lit les variables ANTHROPIC_* que le script de lancement
        # a transmises, y compris ANTHROPIC_WORK_SECRET.
        task = asyncio.create_task(worker.handle_item())
        # Annuler la tâche lorsque le conteneur est arrêté permet au worker de téléverser
        # les fichiers mémoire modifiés et de supprimer les répertoires du store avant de quitter.
        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())

Préparer des fichiers pour une session

Anthropic ne monte pas de fichiers ni de dépôts GitHub dans les sandboxes auto-hébergées. Pour rendre disponibles des fichiers propres à une session :

1. Transmettez des références de fichiers, comme un chemin S3 ou un SHA de commit, dans le champ metadata de la session.
2. Dans votre script de lancement ou votre gestionnaire --on-work, récupérez la session (GET /v1/sessions/{session_id}) et lisez metadata. L'élément de travail réclamé contient l'ID de session, mais pas les métadonnées.
3. Placez les fichiers dans le répertoire de travail avant le début de l'exécution des outils.

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

Étapes suivantes