Menerapkan worker self-hosted

Quickstart menjalankan satu worker CLI ant yang melakukan polling secara terus-menerus dan menjalankan setiap sesi dalam satu proses. Halaman ini membahas cara lain untuk menjalankan worker dan cara memilih di antaranya.

Pilih pola deployment

Saat menerapkan worker, Anda perlu membuat dua pilihan: bagaimana worker mengklaim pekerjaan, dan di mana setiap sesi berjalan.

Bagaimana worker mengklaim pekerjaan:

• Always-on: Proses yang berjalan lama melakukan polling antrean secara terus-menerus dan hanya memerlukan HTTPS keluar. Ini adalah penyiapan paling sederhana.
• Dipicu webhook: Sebuah handler aktif pada session.status_run_started dan mulai melakukan polling. Ini menghindari poller yang menganggur, tetapi memerlukan endpoint webhook yang dapat dijangkau oleh Anthropic.

Di mana setiap sesi berjalan:

• Dalam proses: Worker yang mengklaim sebuah sesi juga menjalankan pemanggilan alatnya, dalam satu direktori kerja bersama.
• Sandbox per sesi: Sebuah poller meluncurkan sandbox baru untuk setiap sesi yang diklaim. Pilih ini untuk isolasi yang lebih kuat: sistem file baru, batas sumber daya, atau kontrol jaringan per sesi.

Worker CLI dan SDK mendukung kombinasi yang berbeda:

Lihat Referensi worker self-hosted untuk setiap flag CLI dan opsi SDK. Untuk kontrol lebih, panggil endpoint Environments Work secara langsung dan implementasikan worker Anda sendiri.

Menjalankan worker always-on

Kedua worker melakukan autentikasi dengan environment key dari quickstart.

Dengan CLI ant:

ant beta:worker poll --workdir /workspace

Dengan SDK, EnvironmentWorker melakukan pekerjaan yang sama:

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())
        # Membatalkan task, alih-alih mematikan proses, memungkinkan worker menghentikan
        # item pekerjaan yang sedang berjalan dan mengunggah file memori yang berubah sebelum keluar.
        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())

Memicu worker dari webhook

1. 1

   Berlangganan webhook sesi

   Di Console, tentukan endpoint webhook yang mendengarkan event session.status_run_started. Lihat Webhook untuk detailnya.

2. 2

   Ekspor kunci penandatanganan webhook

   Bersama dengan ID environment dan kunci dari quickstart, ekspor kunci penandatanganan webhook di host handler Anda. Handler menggunakannya untuk memverifikasi payload yang masuk.

   export ANTHROPIC_WEBHOOK_SIGNING_KEY="whsec_..."

   
3. 3

   Implementasikan handler webhook

   Panggil worker saat session.status_run_started terpicu. Handler menguras antrean dan menyerahkan setiap work item yang diklaim ke handle_item(), yang mengunduh skill, mengeksekusi pemanggilan alat, mengirimkan hasil kembali, lalu kembali.

   PythonTypeScriptC#GoJavaPHPRuby

   

   Untuk memverifikasi tanda tangan webhook, instal 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,
   )
   # Dibatalkan oleh shutdown() agar work item yang sedang berjalan dapat mengunggah file memori yang berubah dan
   # menghapus direktori store-nya sebelum proses berakhir.
   inflight: set[asyncio.Task[None]] = set()
   
   
   # Await ini dari hook shutdown milik host, misalnya shutdown lifespan ASGI (kode setelah
   # `yield` dalam lifespan FastAPI), yang dijalankan uvicorn saat SIGTERM. uvicorn membiarkan request terbuka
   # selesai sebelum hook itu berjalan, jadi atur --timeout-graceful-shutdown untuk membatasi waktu tunggu.
   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:
           # Dilindungi (shielded): pengiriman yang terputus atau timeout tidak boleh membatalkan item; shutdown() yang melakukannya.
           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,
               # Secret per sesi inilah yang memungkinkan worker me-mount memory store milik sesi tersebut.
               work_secret=work.secret,
           )

   Karena handler mengklaim pekerjaan sendiri, handler harus meneruskan secret work item, seperti yang dilakukan argumen work_secret di sini.

Handler ini menjalankan setiap item yang diklaim dalam satu proses di satu host. Jika sesi Anda melampirkan memory store yang sama, lihat Mengisolasi sesi yang berbagi store.

Menjalankan satu sandbox per sesi

Sebuah poller di host mengklaim pekerjaan dan memanggil skrip Anda sekali per work item. Skrip tersebut meluncurkan sandbox untuk satu sesi itu.

1. 1

   Bangun image sandbox

   Instal ant dan tetapkan ant beta:worker run sebagai entrypoint. Saat sandbox dimulai, sandbox membaca detail sesi dari variabel lingkungan, menangani sesi tersebut, lalu keluar. Image dasar harus menyediakan /bin/bash; curl hanya digunakan saat 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

   Tulis skrip spawn

   Skrip meneruskan detail sesi ke sandbox baru. Skrip ini memerlukan jq di host poller.

   #!/bin/bash
   # spawn.sh: dipanggil sekali per item kerja yang diklaim
   # Item kerja yang diklaim diterima sebagai JSON melalui 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

   

   Poller menetapkan variabel ANTHROPIC_* yang diteruskan oleh skrip, kecuali secret. Lihat Variabel lingkungan.

   /host/outputs adalah direktori host yang Anda pilih. Memasangnya di /workspace memungkinkan Anda mengambil hasil kerja sesi setelah sandbox keluar. Mount tersebut juga menangkap pohon skills/ yang diunduh dan file perantara apa pun.

3. 3

   Mulai poller

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

   

Meneruskan secret work item

Setiap work item yang diklaim dapat membawa secret per sesi, yang diterbitkan oleh Anthropic. Worker yang menjalankan sesi memerlukannya untuk me-mount memory store.

Worker yang mengklaim dan menjalankan sesi dalam satu proses (ant beta:worker poll tanpa --on-work, atau EnvironmentWorker dengan run()) meneruskan secret itu sendiri. Ketika kode Anda sendiri berada di antara klaim dan worker, Anda yang meneruskannya:

Teruskan secret hanya ke sandbox yang melayani sesi tersebut, dan jangan pernah mencatatnya di log. Lihat Model keamanan untuk mengetahui hubungannya dengan environment key.

Meluncurkan sandbox dari poller SDK

Untuk mengklaim pekerjaan dari kode Anda sendiri alih-alih ant beta:worker poll --on-work, gunakan work.poller(). Fungsi ini melakukan polling antrean dan memberi Anda setiap sesi yang diklaim, lalu Anda meluncurkan 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}")
    # Ganti `docker run` dengan peluncur sandbox Anda sendiri. Teruskan kunci
    # environment (jangan pernah kunci API Anda) dan secret per sesi milik item kerja: worker
    # di dalamnya memerlukan secret tersebut untuk me-mount memory store sesi.
    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())

Menjalankan worker SDK di dalam sandbox

Ganti entrypoint ant beta:worker run dengan entrypoint SDK ketika sandbox harus melayani alat kustom atau menggunakan pengaturan sinkronisasi memori non-default. Entrypoint membuat EnvironmentWorker dan memanggil handle_item(), yang membaca variabel ANTHROPIC_* yang sama yang diteruskan oleh skrip spawn.

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")
        # Tanpa argumen, handle_item() membaca variabel ANTHROPIC_* yang diteruskan oleh
        # skrip spawn, termasuk ANTHROPIC_WORK_SECRET.
        task = asyncio.create_task(worker.handle_item())
        # Membatalkan task saat container dihentikan memungkinkan worker mengunggah
        # file memori yang berubah dan menghapus direktori store sebelum keluar.
        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())

Menyiapkan file untuk sesi

Anthropic tidak me-mount file atau repositori GitHub ke dalam sandbox self-hosted. Untuk menyediakan file khusus sesi:

1. Teruskan referensi file, seperti path S3 atau SHA commit, di field metadata sesi.
2. Di skrip spawn atau handler --on-work Anda, ambil sesi (GET /v1/sessions/{session_id}) dan baca metadata. Work item yang diklaim membawa ID sesi tetapi tidak membawa metadata.
3. Siapkan file ke dalam direktori kerja sebelum eksekusi alat dimulai.

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

Langkah selanjutnya