자체 호스팅 워커 배포

빠른 시작에서는 지속적으로 폴링하고 모든 세션을 하나의 프로세스에서 실행하는 ant CLI 워커 하나를 실행합니다. 이 페이지에서는 워커를 실행하는 다른 방법과 그중에서 선택하는 방법을 다룹니다.

배포 패턴 선택

워커를 배포할 때는 두 가지를 선택해야 합니다. 워커가 작업을 가져오는 방식과 각 세션이 실행되는 위치입니다.

워커가 작업을 가져오는 방식:

• 상시 실행(Always-on): 장기 실행 프로세스가 큐를 지속적으로 폴링하며, 아웃바운드 HTTPS만 필요합니다. 가장 간단한 설정입니다.
• 웹훅 트리거(Webhook-triggered): 핸들러가 session.status_run_started에서 깨어나 폴링을 시작합니다. 유휴 폴러를 피할 수 있지만, Anthropic이 접근할 수 있는 웹훅 엔드포인트가 필요합니다.

각 세션이 실행되는 위치:

• 프로세스 내(In process): 세션을 가져온 워커가 하나의 공유 작업 디렉터리에서 해당 세션의 도구 호출도 실행합니다.
• 세션당 샌드박스(Sandbox per session): 폴러가 가져온 각 세션마다 새 샌드박스를 시작합니다. 새로운 파일 시스템, 리소스 제한 또는 세션별 네트워크 제어 등 더 강력한 격리가 필요할 때 이 방식을 선택하세요.

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())

웹훅으로 워커 트리거

1. 1

   세션 웹훅 구독

   Console에서 session.status_run_started 이벤트를 수신하는 웹훅 엔드포인트를 정의하세요. 자세한 내용은 웹훅을 참조하세요.

2. 2

   웹훅 서명 키 내보내기

   빠른 시작의 환경 ID 및 키와 함께, 핸들러 호스트에서 웹훅 서명 키를 내보내세요. 핸들러는 이 키를 사용하여 수신 페이로드를 검증합니다.

   export ANTHROPIC_WEBHOOK_SIGNING_KEY="whsec_..."

   
3. 3

   웹훅 핸들러 구현

   session.status_run_started가 발생하면 워커를 호출하세요. 핸들러는 큐를 비우면서 가져온 각 작업 항목을 handle_item()에 전달하며, 이 함수는 스킬을 다운로드하고, 도구 호출을 실행하고, 결과를 다시 게시한 후 반환합니다.

   PythonTypeScriptC#GoJavaPHPRuby

   

   웹훅 서명을 검증하려면 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,
   )
   # 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:
           # Shielded: 누락되거나 시간 초과된 전달이 항목을 취소해서는 안 되며, 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

   샌드박스 이미지 빌드

   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: 할당받은 작업 항목마다 한 번씩 호출됩니다
   # 할당받은 작업 항목은 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이 포함될 수 있습니다. 세션을 실행하는 워커는 메모리 스토어를 마운트하기 위해 이 시크릿이 필요합니다.

하나의 프로세스에서 세션을 가져오고 실행하는 워커(--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 키는 절대 아님)와 작업 항목의 세션별 시크릿을 전달하세요. 내부의 worker는
    # 세션의 메모리 저장소를 마운트하려면 이 시크릿이 필요합니다.
    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"},
)

다음 단계