Selbst gehostete Worker bereitstellen

Der Schnellstart führt einen ant-CLI-Worker aus, der kontinuierlich pollt und jede Sitzung in einem Prozess ausführt. Diese Seite behandelt die anderen Möglichkeiten, einen Worker auszuführen, und wie du zwischen ihnen wählst.

Ein Bereitstellungsmuster wählen

Beim Bereitstellen von Workern musst du zwei Entscheidungen treffen: wie der Worker Arbeit beansprucht und wo jede Sitzung läuft.

Wie der Worker Arbeit beansprucht:

• Dauerhaft aktiv: Ein langlebiger Prozess pollt die Warteschlange kontinuierlich und benötigt nur ausgehendes HTTPS. Dies ist die einfachste Einrichtung.
• Per Webhook ausgelöst: Ein Handler wird bei session.status_run_started aktiviert und beginnt mit dem Polling. Dadurch wird ein untätiger Poller vermieden, es ist jedoch ein Webhook-Endpunkt erforderlich, den Anthropic erreichen kann.

Wo jede Sitzung läuft:

• Im Prozess: Der Worker, der eine Sitzung beansprucht, führt auch deren Tool-Aufrufe aus, in einem gemeinsamen Arbeitsverzeichnis.
• Sandbox pro Sitzung: Ein Poller startet für jede beanspruchte Sitzung eine neue Sandbox. Wähle dies für stärkere Isolation: ein frisches Dateisystem, Ressourcenlimits oder Netzwerkkontrollen pro Sitzung.

Die CLI- und SDK-Worker unterstützen unterschiedliche Kombinationen:

Siehe Referenz für selbst gehostete Worker für jedes CLI-Flag und jede SDK-Option. Für mehr Kontrolle rufe die Environments-Work-Endpunkte direkt auf und implementiere deinen eigenen Worker.

Einen dauerhaft aktiven Worker ausführen

Beide Worker authentifizieren sich mit dem Umgebungsschlüssel aus dem Schnellstart.

Mit der ant CLI:

ant beta:worker poll --workdir /workspace

Mit dem SDK erledigt EnvironmentWorker dieselbe Arbeit:

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())
        # Das Abbrechen des Tasks statt des Beendens des Prozesses erlaubt dem Worker, sein
        # laufendes Work-Item zu stoppen und geänderte Memory-Dateien vor dem Beenden hochzuladen.
        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())

Worker über Webhooks auslösen

1. 1

   Sitzungs-Webhooks abonnieren

   Definiere in der Console einen Webhook-Endpunkt, der auf session.status_run_started-Events lauscht. Details findest du unter Webhooks.

2. 2

   Den Webhook-Signaturschlüssel exportieren

   Exportiere zusammen mit der Umgebungs-ID und dem Schlüssel aus dem Schnellstart den Webhook-Signaturschlüssel auf deinem Handler-Host. Der Handler verwendet ihn, um eingehende Payloads zu verifizieren.

   export ANTHROPIC_WEBHOOK_SIGNING_KEY="whsec_..."

   
3. 3

   Den Webhook-Handler implementieren

   Rufe den Worker auf, wenn session.status_run_started ausgelöst wird. Der Handler leert die Warteschlange und übergibt jedes beanspruchte Arbeitselement an handle_item(), das Skills herunterlädt, Tool-Aufrufe ausführt, Ergebnisse zurücksendet und zurückkehrt.

   PythonTypeScriptC#GoJavaPHPRuby

   

   Um Webhook-Signaturen zu verifizieren, installiere das 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,
   )
   # Wird von shutdown() abgebrochen, damit ein laufendes Work-Item geänderte Memory-Dateien hochladen und
   # seine Store-Verzeichnisse entfernen kann, bevor der Prozess endet.
   inflight: set[asyncio.Task[None]] = set()
   
   
   # Awaite dies im Shutdown-Hook des Hosts, z. B. einem ASGI-Lifespan-Shutdown (der Code nach
   # `yield` in einem FastAPI-Lifespan), den uvicorn bei SIGTERM ausführt. uvicorn lässt offene Requests
   # vor diesem Hook abschließen; setze daher --timeout-graceful-shutdown, um die Wartezeit zu begrenzen.
   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:
           # Geschützt (shielded): Eine verworfene oder abgelaufene Zustellung darf das Item nicht abbrechen; shutdown() tut es.
           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,
               # Das sitzungsspezifische Secret ermöglicht dem Worker, die Memory-Stores der Sitzung zu mounten.
               work_secret=work.secret,
           )

   Da der Handler die Arbeit selbst beansprucht, muss er das Secret des Arbeitselements weiterleiten, wie es hier das Argument work_secret tut.

Dieser Handler führt jedes beanspruchte Arbeitselement in einem Prozess auf einem Host aus. Wenn deine Sitzungen denselben Memory Store anhängen, siehe Sitzungen isolieren, die sich einen Store teilen.

Eine Sandbox pro Sitzung ausführen

Ein Poller auf dem Host beansprucht Arbeit und ruft dein Skript einmal pro Arbeitselement auf. Das Skript startet eine Sandbox für genau diese eine Sitzung.

1. 1

   Das Sandbox-Image erstellen

   Installiere ant und lege ant beta:worker run als Entrypoint fest. Wenn eine Sandbox startet, liest sie die Sitzungsdetails aus Umgebungsvariablen, bearbeitet diese Sitzung und beendet sich. Das Basis-Image muss /bin/bash bereitstellen; curl wird nur zur Build-Zeit verwendet.

   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

   Das Spawn-Skript schreiben

   Das Skript leitet die Sitzungsdetails in eine neue Sandbox weiter. Es benötigt jq auf dem Poller-Host.

   #!/bin/bash
   # spawn.sh: wird einmal pro beanspruchtem Arbeitselement aufgerufen
   # Das beanspruchte Arbeitselement kommt als JSON über stdin an.
   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

   

   Der Poller setzt die ANTHROPIC_*-Variablen, die das Skript weiterleitet, mit Ausnahme des Secrets. Siehe Umgebungsvariablen.

   /host/outputs ist ein Host-Verzeichnis deiner Wahl. Wenn du es unter /workspace einbindest, kannst du die Ergebnisse der Sitzung abrufen, nachdem die Sandbox beendet wurde. Der Mount erfasst außerdem den heruntergeladenen skills/-Baum und alle Zwischendateien.

3. 3

   Den Poller starten

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

   

Das Secret des Arbeitselements weiterleiten

Jedes beanspruchte Arbeitselement kann ein sitzungsspezifisches secret enthalten, das Anthropic ausstellt. Der Worker, der die Sitzung ausführt, benötigt es, um Memory Stores einzubinden.

Ein Worker, der Sitzungen in einem Prozess beansprucht und ausführt (ant beta:worker poll ohne --on-work oder EnvironmentWorker mit run()), gibt das Secret selbst weiter. Wenn dein eigener Code zwischen dem Beanspruchen und dem Worker steht, leitest du es weiter:

Übergib das Secret nur an die Sandbox, die diese Sitzung bedient, und protokolliere es niemals. Unter Sicherheitsmodell erfährst du, wie es sich zum Umgebungsschlüssel verhält.

Sandboxes aus dem SDK-Poller starten

Um Arbeit aus deinem eigenen Code statt mit ant beta:worker poll --on-work zu beanspruchen, verwende work.poller(). Es pollt die Warteschlange und gibt dir jede beanspruchte Sitzung, und du startest die 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}")
    # Ersetze `docker run` durch deinen eigenen Sandbox-Launcher. Leite den Environment-
    # Key (niemals deinen API-Key) und das sitzungsspezifische Secret des Work Items weiter: Der Worker
    # darin braucht das Secret, um die Memory Stores der Sitzung einzubinden.
    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())

Den SDK-Worker innerhalb der Sandbox ausführen

Ersetze den Entrypoint ant beta:worker run durch einen SDK-Entrypoint, wenn die Sandbox benutzerdefinierte Tools bereitstellen oder nicht standardmäßige Memory-Synchronisierungseinstellungen verwenden muss. Der Entrypoint erstellt EnvironmentWorker und ruft handle_item() auf, das dieselben ANTHROPIC_*-Variablen liest, die das Spawn-Skript weiterleitet.

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")
        # Ohne Argumente liest handle_item() die ANTHROPIC_*-Variablen, die das Spawn-
        # Skript weitergereicht hat, einschließlich ANTHROPIC_WORK_SECRET.
        task = asyncio.create_task(worker.handle_item())
        # Das Abbrechen des Tasks beim Stoppen des Containers ermöglicht dem Worker, geänderte
        # Memory-Dateien hochzuladen und die Store-Verzeichnisse vor dem Beenden zu entfernen.
        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())

Dateien für eine Sitzung bereitstellen

Anthropic bindet keine Dateien oder GitHub-Repositories in selbst gehostete Sandboxes ein. So machst du sitzungsspezifische Dateien verfügbar:

1. Übergib Dateiverweise, etwa einen S3-Pfad oder einen Commit-SHA, im Feld metadata der Sitzung.
2. Rufe in deinem Spawn-Skript oder --on-work-Handler die Sitzung ab (GET /v1/sessions/{session_id}) und lies metadata. Das beanspruchte Arbeitselement enthält die Sitzungs-ID, aber nicht die Metadaten.
3. Lege die Dateien im Arbeitsverzeichnis ab, bevor die Tool-Ausführung beginnt.

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

Nächste Schritte