セルフホストワーカーをデプロイする

クイックスタートでは、継続的にポーリングし、すべてのセッションを1つのプロセスで実行する ant CLIワーカーを1つ実行します。このページでは、ワーカーを実行するその他の方法と、それらの選び方について説明します。

デプロイパターンを選択する

ワーカーをデプロイする際には、ワーカーがワークを取得する方法と、各セッションを実行する場所の2つを選択する必要があります。

ワーカーがワークを取得する方法：

• 常時稼働： 長時間実行されるプロセスがキューを継続的にポーリングし、必要なのはアウトバウンドHTTPSのみです。これが最もシンプルな構成です。
• Webhookトリガー型： ハンドラーが session.status_run_started で起動し、ポーリングを開始します。これによりアイドル状態のポーラーを回避できますが、Anthropicから到達可能なWebhookエンドポイントが必要です。

各セッションを実行する場所：

• プロセス内： セッションを取得したワーカーが、そのツール呼び出しも1つの共有作業ディレクトリで実行します。
• セッションごとのサンドボックス： ポーラーが、取得したセッションごとに新しいサンドボックスを起動します。新しいファイルシステム、リソース制限、セッションごとのネットワーク制御など、より強力な分離が必要な場合はこちらを選択してください。

CLIワーカーとSDKワーカーは、それぞれ異なる組み合わせをサポートしています：

すべてのCLIフラグとSDKオプションについては、セルフホストワーカーリファレンスを参照してください。より細かく制御したい場合は、Environments Workエンドポイントを直接呼び出して独自のワーカーを実装してください。

常時稼働ワーカーを実行する

どちらのワーカーも、クイックスタートで取得した環境キーで認証します。

ant CLIの場合：

ant beta:worker poll --workdir /workspace

SDKの場合は、EnvironmentWorker が同じ処理を行います：

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())
        # プロセスを強制終了するのではなくタスクをキャンセルすることで、ワーカーは処理中の
        # 作業項目を停止し、変更されたメモリファイルを終了前にアップロードできます。
        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())

Webhookからワーカーをトリガーする

1. 1

   セッションWebhookをサブスクライブする

   Consoleで、session.status_run_started イベントをリッスンするWebhookエンドポイントを定義します。詳細についてはWebhookを参照してください。

2. 2

   Webhook署名キーをエクスポートする

   クイックスタートで取得した環境IDとキーに加えて、ハンドラーのホストでWebhook署名キーをエクスポートします。ハンドラーはこれを使用して受信ペイロードを検証します。

   export ANTHROPIC_WEBHOOK_SIGNING_KEY="whsec_..."

   
3. 3

   Webhookハンドラーを実装する

   session.status_run_started が発火したときにワーカーを呼び出します。ハンドラーはキューを処理し尽くし、取得した各ワークアイテムを handle_item() に渡します。これはスキルをダウンロードし、ツール呼び出しを実行し、結果を送り返してから戻ります。

   PythonTypeScriptC#GoJavaPHPRuby

   

   Webhook署名を検証するには、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,
   )
   # shutdown() によってキャンセルされ、実行中の作業項目が変更されたメモリファイルをアップロードし、
   # プロセス終了前にストアディレクトリを削除できるようにします。
   inflight: set[asyncio.Task[None]] = set()
   
   
   # ホストのシャットダウンフックから await してください。例: ASGI lifespan のシャットダウン（FastAPI lifespan の
   # `yield` 以降のコード）。uvicorn は SIGTERM 時にこれを実行します。uvicorn は未完了のリクエストを
   # フック実行前に完了させるため、--timeout-graceful-shutdown を設定して待機時間を制限してください。
   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:
           # シールド済み: 配信の切断やタイムアウトで項目をキャンセルしてはなりません。キャンセルは shutdown() が行います。
           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,
               # セッションごとのシークレットにより、ワーカーはセッションのメモリストアをマウントできます。
               work_secret=work.secret,
           )

   ハンドラーは自らワークを取得するため、ここでの work_secret 引数のように、ワークアイテムのシークレットを転送する必要があります。

このハンドラーは、取得したすべてのアイテムを1つのホスト上の1つのプロセスで実行します。セッションが同じメモリストアをアタッチする場合は、ストアを共有するセッションを分離するを参照してください。

セッションごとに1つのサンドボックスを実行する

ホスト上のポーラーがワークを取得し、ワークアイテムごとに1回スクリプトを呼び出します。スクリプトはその1つのセッション用のサンドボックスを起動します。

1. 1

   サンドボックスイメージをビルドする

   ant をインストールし、ant beta:worker run をエントリポイントとして設定します。サンドボックスが起動すると、環境変数からセッションの詳細を読み取り、そのセッションを処理して終了します。ベースイメージには /bin/bash が必要です。curl はビルド時にのみ使用されます。

   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

   スポーンスクリプトを作成する

   スクリプトはセッションの詳細を新しいサンドボックスに転送します。ポーラーのホストに jq が必要です。

   #!/bin/bash
   # spawn.sh: 取得した作業項目ごとに1回呼び出されます
   # 取得した作業項目は stdin から JSON として渡されます。
   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

   

   ポーラーは、スクリプトが転送する ANTHROPIC_* 変数を設定します（シークレットを除く）。環境変数を参照してください。

   /host/outputs は任意に選択するホストディレクトリです。これを /workspace にマウントすると、サンドボックスの終了後にセッションの成果物を取得できます。このマウントには、ダウンロードされた skills/ ツリーや中間ファイルも含まれます。

3. 3

   ポーラーを起動する

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

   

ワークアイテムのシークレットを転送する

取得された各ワークアイテムには、Anthropicが発行するセッションごとの secret が含まれる場合があります。セッションを実行するワーカーは、メモリストアをマウントするためにこれを必要とします。

セッションの取得と実行を1つのプロセスで行うワーカー（--on-work なしの ant beta:worker poll、または run() を使用する EnvironmentWorker）は、シークレットを自ら受け渡します。取得とワーカーの間に独自のコードが入る場合は、自分で転送します：

シークレットは、そのセッションを処理するサンドボックスにのみ渡し、決してログに記録しないでください。環境キーとの関係については、セキュリティモデルを参照してください。

SDKポーラーからサンドボックスを起動する

ant beta:worker poll --on-work の代わりに独自のコードからワークを取得するには、work.poller() を使用します。これはキューをポーリングして取得した各セッションを渡すので、お客様がサンドボックスを起動します：

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}")
    # `docker run` は独自のサンドボックス起動ツールに置き換えてください。環境
    # キー（APIキーは絶対に渡さないでください）と作業項目のセッションごとのシークレットを転送します。内部のワーカーは
    # セッションのメモリストアをマウントするためにこのシークレットを必要とします。
    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())

サンドボックス内でSDKワーカーを実行する

サンドボックスでカスタムツールを提供する必要がある場合や、デフォルト以外のメモリ同期設定を使用する必要がある場合は、ant beta:worker run エントリポイントをSDKエントリポイントに置き換えます。エントリポイントは EnvironmentWorker を構築して handle_item() を呼び出します。これはスポーンスクリプトが転送するのと同じ ANTHROPIC_* 変数を読み取ります。

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")
        # 引数なしの場合、handle_item() は spawn スクリプトが転送した ANTHROPIC_* 変数を読み取ります。
        # これには ANTHROPIC_WORK_SECRET も含まれます。
        task = asyncio.create_task(worker.handle_item())
        # コンテナ停止時にタスクをキャンセルすることで、ワーカーは終了前に
        # 変更されたメモリファイルをアップロードし、ストアディレクトリを削除できます。
        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())

セッション用のファイルをステージングする

Anthropicは、セルフホスト型サンドボックスにファイルやGitHubリポジトリをマウントしません。セッション固有のファイルを利用可能にするには：

1. S3パスやコミットSHAなどのファイル参照を、セッションの metadata フィールドで渡します。
2. スポーンスクリプトまたは --on-work ハンドラーで、セッションを取得（GET /v1/sessions/{session_id}）して metadata を読み取ります。取得したワークアイテムにはセッションIDは含まれますが、メタデータは含まれません。
3. ツールの実行が始まる前に、ファイルを作業ディレクトリにステージングします。

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

次のステップ