Развёртывание самостоятельно размещаемых воркеров

В кратком руководстве запускается один воркер CLI ant, который непрерывно опрашивает очередь и выполняет все сеансы в одном процессе. На этой странице описаны другие способы запуска воркера и то, как выбрать между ними.

Выбор шаблона развёртывания

При развёртывании воркеров необходимо сделать два выбора: как воркер получает работу и где выполняется каждый сеанс.

Как воркер получает работу:

• Постоянная работа («always-on»): долго работающий процесс непрерывно опрашивает очередь, и ему нужен только исходящий HTTPS. Это самая простая конфигурация.
• Запуск по вебхуку («webhook-triggered»): обработчик активируется по событию session.status_run_started и начинает опрос. Это позволяет избежать простаивающего опрашивающего процесса, но требует конечной точки вебхука, доступной для Anthropic.

Где выполняется каждый сеанс:

• В процессе: воркер, получивший сеанс, сам выполняет его вызовы инструментов в одном общем рабочем каталоге.
• Песочница на сеанс: опрашивающий процесс запускает новую песочницу для каждого полученного сеанса. Выбирайте этот вариант для более сильной изоляции: новая файловая система, ограничения ресурсов или сетевые ограничения для каждого сеанса.

Воркеры CLI и SDK поддерживают разные комбинации:

Все флаги CLI и параметры SDK описаны в справочнике по самостоятельно размещаемым воркерам. Для большего контроля вызывайте конечные точки Environments Work напрямую и реализуйте собственный воркер.

Запуск постоянно работающего воркера

Оба воркера проходят аутентификацию с помощью ключа среды из краткого руководства.

С помощью CLI ant:

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

   Экспортируйте ключ подписи вебхука

   Вместе с идентификатором среды и ключом из краткого руководства экспортируйте ключ подписи вебхука на хосте обработчика. Обработчик использует его для проверки входящих полезных данных.

   export ANTHROPIC_WEBHOOK_SIGNING_KEY="whsec_..."

   
3. 3

   Реализуйте обработчик вебхука

   Вызывайте воркер при срабатывании session.status_run_started. Обработчик опустошает очередь и передаёт каждый полученный элемент работы в handle_item(), который загружает навыки, выполняет вызовы инструментов, отправляет результаты обратно и завершает работу.

   PythonTypeScriptC#GoJavaPHPRuby

   

   Чтобы проверять подписи вебхуков, установите дополнительный пакет 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()
   
   
   # Ожидайте это из хука завершения хоста, например ASGI lifespan shutdown (код после
   # `yield` в lifespan FastAPI), который 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:
           # Защищено (shield): сброшенная или просроченная доставка не должна отменять элемент; это делает 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: вызывается один раз для каждого взятого в работу элемента
   # Взятый в работу элемент поступает в виде JSON через 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

   

   Опрашивающий процесс задаёт переменные ANTHROPIC_*, которые передаёт скрипт, за исключением секрета. См. раздел Переменные окружения.

   /host/outputs — это выбранный вами каталог на хосте. Его монтирование в /workspace позволяет получить результаты сеанса после завершения работы песочницы. В это монтирование также попадают загруженное дерево skills/ и любые промежуточные файлы.

3. 3

   Запустите опрашивающий процесс

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

   

Передача секрета элемента работы

Каждый полученный элемент работы может содержать secret для конкретного сеанса, который выдаёт Anthropic. Он нужен воркеру, выполняющему сеанс, для монтирования хранилищ памяти.

Воркер, который получает и выполняет сеансы в одном процессе (ant beta:worker poll без --on-work или EnvironmentWorker с run()), передаёт секрет сам. Если между получением работы и воркером находится ваш собственный код, секрет передаёте вы:

Передавайте секрет только в песочницу, обслуживающую этот сеанс, и никогда не записывайте его в журналы. О том, как он связан с ключом среды, см. в разделе Модель безопасности.

Запуск песочниц из опрашивающего процесса 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() читает переменные 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. Полученный элемент работы содержит идентификатор сеанса, но не метаданные.
3. Разместите файлы в рабочем каталоге до начала выполнения инструментов.

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

Следующие шаги