Implantar workers auto-hospedados

O início rápido executa um worker da CLI ant que faz polling continuamente e executa todas as sessões em um único processo. Esta página aborda as outras formas de executar um worker e como escolher entre elas.

Escolher um padrão de implantação

Ao implantar workers, você precisa fazer duas escolhas: como o worker reivindica trabalho e onde cada sessão é executada.

Como o worker reivindica trabalho:

• Sempre ativo: Um processo de longa duração faz polling da fila continuamente e precisa apenas de HTTPS de saída. Esta é a configuração mais simples.
• Acionado por webhook: Um handler é ativado em session.status_run_started e começa a fazer polling. Isso evita um poller ocioso, mas requer um endpoint de webhook que a Anthropic consiga alcançar.

Onde cada sessão é executada:

• No próprio processo: O worker que reivindica uma sessão também executa suas chamadas de ferramentas, em um diretório de trabalho compartilhado.
• Sandbox por sessão: Um poller inicia um sandbox novo para cada sessão reivindicada. Escolha esta opção para um isolamento mais forte: um sistema de arquivos novo, limites de recursos ou controles de rede por sessão.

Os workers da CLI e do SDK suportam combinações diferentes:

Consulte a Referência de workers auto-hospedados para ver todas as flags da CLI e opções do SDK. Para ter mais controle, chame diretamente os endpoints Environments Work e implemente seu próprio worker.

Executar um worker sempre ativo

Ambos os workers se autenticam com a chave de ambiente do início rápido.

Com a CLI ant:

ant beta:worker poll --workdir /workspace

Com o SDK, EnvironmentWorker faz o mesmo trabalho:

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())
        # Cancelar a tarefa, em vez de matar o processo, permite que o worker pare seu
        # item de trabalho em andamento e envie os arquivos de memória alterados antes de sair.
        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())

Acionar workers a partir de webhooks

1. 1

   Inscreva-se nos webhooks de sessão

   No Console, defina um endpoint de webhook que escute eventos session.status_run_started. Consulte Webhooks para mais detalhes.

2. 2

   Exporte a chave de assinatura do webhook

   Junto com o ID e a chave do ambiente do início rápido, exporte a chave de assinatura do webhook no host do seu handler. O handler a usa para verificar os payloads recebidos.

   export ANTHROPIC_WEBHOOK_SIGNING_KEY="whsec_..."

   
3. 3

   Implemente o handler do webhook

   Invoque o worker quando session.status_run_started for disparado. O handler esvazia a fila e entrega cada item de trabalho reivindicado para handle_item(), que baixa skills, executa chamadas de ferramentas, envia os resultados de volta e retorna.

   PythonTypeScriptC#GoJavaPHPRuby

   

   Para verificar assinaturas de webhook, instale o extra de 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,
   )
   # Cancelado por shutdown() para que um item de trabalho em andamento possa enviar arquivos de memória alterados e
   # remover seus diretórios de store antes que o processo termine.
   inflight: set[asyncio.Task[None]] = set()
   
   
   # Aguarde isto no hook de encerramento do host, como um shutdown de lifespan ASGI (o código após
   # `yield` em um lifespan do FastAPI), que o uvicorn executa no SIGTERM. O uvicorn deixa requisições abertas
   # terminarem antes desse hook rodar, então defina --timeout-graceful-shutdown para limitar a espera.
   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:
           # Protegido (shield): uma entrega descartada ou expirada não deve cancelar o item; shutdown() faz isso.
           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,
               # O segredo por sessão é o que permite ao worker montar os memory stores da sessão.
               work_secret=work.secret,
           )

   Como o próprio handler reivindica o trabalho, ele deve encaminhar o segredo do item de trabalho, como o argumento work_secret faz aqui.

Este handler executa todos os itens reivindicados em um único processo em um único host. Se suas sessões anexam o mesmo memory store, consulte Isolar sessões que compartilham um store.

Executar um sandbox por sessão

Um poller no host reivindica trabalho e chama seu script uma vez por item de trabalho. O script inicia um sandbox para essa única sessão.

1. 1

   Construa a imagem do sandbox

   Instale o ant e defina ant beta:worker run como entrypoint. Quando um sandbox é iniciado, ele lê os detalhes da sessão a partir de variáveis de ambiente, processa essa sessão e encerra. A imagem base deve fornecer /bin/bash; o curl é usado apenas no momento do 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

   Escreva o script de spawn

   O script encaminha os detalhes da sessão para um sandbox novo. Ele requer jq no host do poller.

   #!/bin/bash
   # spawn.sh: chamado uma vez por item de trabalho reivindicado
   # O item de trabalho reivindicado chega como JSON via 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

   

   O poller define as variáveis ANTHROPIC_* que o script encaminha, exceto o segredo. Consulte Variáveis de ambiente.

   /host/outputs é um diretório do host que você escolhe. Montá-lo em /workspace permite recuperar os entregáveis da sessão depois que o sandbox encerra. A montagem também captura a árvore skills/ baixada e quaisquer arquivos intermediários.

3. 3

   Inicie o poller

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

   

Encaminhar o segredo do item de trabalho

Cada item de trabalho reivindicado pode carregar um secret por sessão, emitido pela Anthropic. O worker que executa a sessão precisa dele para montar memory stores.

Um worker que reivindica e executa sessões em um único processo (ant beta:worker poll sem --on-work, ou EnvironmentWorker com run()) repassa o segredo por conta própria. Quando seu próprio código fica entre a reivindicação e o worker, você o encaminha:

Passe o segredo apenas para o sandbox que atende essa sessão e nunca o registre em logs. Consulte Modelo de segurança para ver como ele se relaciona com a chave de ambiente.

Iniciar sandboxes a partir do poller do SDK

Para reivindicar trabalho a partir do seu próprio código em vez de ant beta:worker poll --on-work, use work.poller(). Ele faz polling da fila e entrega cada sessão reivindicada, e você inicia o 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}")
    # Substitua `docker run` pelo seu próprio inicializador de sandbox. Repasse a chave do
    # ambiente (nunca sua chave de API) e o segredo por sessão do item de trabalho: o worker
    # interno precisa do segredo para montar os armazenamentos de memória da sessão.
    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())

Executar o worker do SDK dentro do sandbox

Substitua o entrypoint ant beta:worker run por um entrypoint do SDK quando o sandbox precisar servir ferramentas personalizadas ou usar configurações de sincronização de memória não padrão. O entrypoint constrói EnvironmentWorker e chama handle_item(), que lê as mesmas variáveis ANTHROPIC_* que o script de spawn encaminha.

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")
        # Sem argumentos, handle_item() lê as variáveis ANTHROPIC_* que o script de spawn
        # encaminhou, incluindo ANTHROPIC_WORK_SECRET.
        task = asyncio.create_task(worker.handle_item())
        # Cancelar a tarefa quando o contêiner é parado permite que o worker envie
        # os arquivos de memória alterados e remova os diretórios de armazenamento antes de sair.
        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())

Preparar arquivos para uma sessão

A Anthropic não monta arquivos nem repositórios do GitHub em sandboxes auto-hospedados. Para disponibilizar arquivos específicos da sessão:

1. Passe referências de arquivos, como um caminho do S3 ou um SHA de commit, no campo metadata da sessão.
2. No seu script de spawn ou handler --on-work, recupere a sessão (GET /v1/sessions/{session_id}) e leia metadata. O item de trabalho reivindicado carrega o ID da sessão, mas não os metadados.
3. Prepare os arquivos no diretório de trabalho antes que a execução das ferramentas comece.

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

Próximos passos