hobots

SDK Python da Hobots para robôs (RPAs) e integrações. Tudo é uma task: você registra o que o robô faz e a plataforma passa a receber logs, itens processados, progresso, prints, erros e o "estou vivo" do robô.

• hobots.register(task, handler) — sob demanda: o robô fica online aguardando solicitações criadas no app (formulários e agendamentos) e executa o handler a cada uma. Módulo Solicitações.
• hobots.start(task, handler) — sempre ligado: o robô abre a própria execução ao iniciar e roda o handler (o seu loop) dentro dela; a plataforma só recebe logs, itens e progresso. Sem formulário, sem solicitante.

Cliente HTTP puro, sem dependências de runtime (só a stdlib). Requer Python 3.9+. A API é síncrona: handlers são funções normais (def) e o SDK administra as threads de fundo — combina com Selenium, Playwright síncrono, pyautogui e código legado.

Todas as durações são em milissegundos, como no SDK Node: retry_delay=5000 são 5 segundos, hobots.sleep(2000) dorme 2 segundos. Os nomes não levam sufixo _ms.

Instalação

pip install hobots

import hobots  # o próprio módulo é o objeto global

Em 30 segundos

export HOBOTS_CLIENT_SECRET=hb_<64 hex>   # Configurações → Agentes
export HOBOTS_INSTANCE_ID=vm-cliente-01   # opcional — default: hostname da máquina
export HOBOTS_ENVIRONMENT=staging         # opcional — default: production

import hobots


# sob demanda: uma execução por solicitação do app (o task_slug casa com o formulário)
def processar_nfe(params, ctx):
    hobots.log.info(f"lote {params['lote']}")
    hobots.progress.total(len(params["notas"]))
    for nota in params["notas"]:
        baixar(nota)
        hobots.items.succeeded({"numero_nota": nota}, id=nota)
        hobots.progress.advance()


# sempre ligado: o handler é o loop do robô; cada iteração reporta o que fez
def monitor_pasta(params, ctx):
    while True:
        hobots.sleep(60_000)  # cancelável
        for arquivo in novos_arquivos():
            hobots.items.succeeded({"arquivo": arquivo})
        hobots.log.info("ciclo ok")


hobots.register("processar-nfe", processar_nfe)
hobots.start("monitor-pasta", monitor_pasta)
# as duas chamadas voltam na hora; o processo fica vivo (heartbeat) até hobots.close()
# ou Ctrl-C — encerramento gracioso em "Encerramento", abaixo

O módulo hobots é o objeto global: hobots.log, hobots.items, hobots.progress, hobots.transaction… resolvem a execução atual de qualquer módulo do programa — sem passar contexto de mão em mão, mesmo com várias execuções em paralelo.

Credenciais e identidade

A opção passada em register()/start() tem precedência sobre a env — útil para copiar e colar o snippet do app. O secret nunca aparece em repr() do SDK.

Os dois modos

As duas chamadas voltam na hora: o polling e o handler do modo start rodam em threads de fundo. Os dois modos podem coexistir no mesmo processo: um monitor sempre ligado + tasks sob demanda, um heartbeat, uma instância. Cada slug só pode ser registrado uma vez.

O handler é sempre handler(params, ctx) — dois argumentos posicionais, também no modo start (onde params é {}). O retorno é ignorado.

Modo start e o limite de logs: o servidor guarda no máximo 1000 linhas de log por execução. Num loop que roda por dias, logue o que importa (ciclos com novidade, avisos, erros) e registre o trabalho em hobots.items — itens não têm esse teto tão baixo (10.000 por execução).

O módulo hobots

O handler também recebe o mesmo contexto como 2º argumento (ctx.log, ctx.items, ctx.progress, ctx.transaction, ctx.signal, ctx.params, ctx.sleep, ctx.attach, ctx.retry, ctx.raise_if_canceled), para quem prefere explícito. Fora de qualquer execução, os membros de contexto caem em fallbacks seguros e avisam uma vez — hobots.log continua imprimindo no terminal.

Evite from hobots import log (ou items, progress…) no topo de um módulo: o valor capturado no import é o de fora de execução. Importe o módulo e chame hobots.log.info(...) — cada acesso resolve a execução atual.

Cancelamento automático

Uma execução cancelada no app não mata o handler — o SDK dispara hobots.signal e o próximo ponto de parada levanta RunCanceled:

• hobots.items.*, hobots.progress.total/advance/set, hobots.transaction.start_child() (e step.start_child()), hobots.sleep() e cada tentativa de hobots.retry().
• Nunca hobots.log.* — um log dentro de um finally de limpeza não pode explodir.

Num loop de itens, isso costuma bastar: registrar o item processado já é o ponto de parada. Para controle fino, hobots.raise_if_canceled() e hobots.signal.is_set() continuam disponíveis. Quando o cancelamento veio do app, o RunCanceled é suprimido — a execução já está cancelada no servidor e nada é reportado por cima. Limpeza (fechar navegador, desfazer estado) vai em try/finally.

Atenção: não use if hobots.signal.is_set(): return. Retornar normalmente reporta a execução como succeeded. É preciso levantar — e é o que os pontos de parada fazem. Se capturar o RunCanceled para limpar, re-levante.

Retry

Quando o handler levanta, o SDK tenta de novo na mesma execução: aviso no log (tentativa 1/2 falhou: … — nova tentativa em 5000ms), espera cancelável, re-invocação com os mesmos params. Default: 1 tentativa extra com 5 s. Itens e progresso já enviados permanecem — escreva handlers idempotentes (use o id de negócio nos itens: repetido, o servidor ignora).

hobots.register("processar-nfe", handler, retries=3, retry_delay=lambda tentativa: tentativa * 10_000)

Nunca há nova tentativa em cancelamento, timeout (TaskTimeout é terminal) ou execução já cancelada no app. Para uma chamada instável dentro do handler, use hobots.retry().

Opções

hobots.register(task, handler, *, client_secret=None, ..., concurrency=None, poll_interval=None)
hobots.start(task, handler, *, client_secret=None, ...)

Todas as opções são keyword-only.

As opções de processo (client_secret, instance_id, environment, release, tags, debug, machine_metrics, capture_unhandled, install_exit_handlers, e poll_interval entre os register()) valem as da primeira chamada; chamadas seguintes com valor diferente avisam e são ignoradas.

Encerramento

O heartbeat mantém o processo vivo por design — é a única thread não daemon do SDK, o que faz o script continuar rodando depois da última linha. hobots.close(timeout=30_000) para o polling, aguarda as execuções em voo, para o heartbeat (libera o processo) e faz o flush final; devolve False se algo não terminou a tempo.

Para um encerramento gracioso no Ctrl-C/SIGTERM (esperar a execução em andamento terminar), instale o seu handler de sinal e segure a main thread até ele disparar:

import signal
import threading

import hobots

parar = threading.Event()
signal.signal(signal.SIGINT, lambda signum, frame: parar.set())
signal.signal(signal.SIGTERM, lambda signum, frame: parar.set())

hobots.register("processar-nfe", processar_nfe)

while not parar.wait(1):  # o timeout mantém o Ctrl-C responsivo também no Windows
    pass
hobots.close(15_000)  # espera a execução em voo, para o heartbeat e faz o flush

Com handler próprio de sinal, o SDK não interfere. Sem ele, um SIGINT/SIGTERM com execuções ativas reporta cada uma como failed/AgentTerminated, faz o flush da telemetria e o processo sai com 130/143 — o app mostra a falha na hora. Uma exceção não tratada na main thread também encerra o processo (exit 1), mesmo com o heartbeat vivo.

Não chame sys.exit() de dentro do handler de sinal quando o script principal já terminou: o SystemExit cai dentro do threading._shutdown do interpretador ("Exception ignored in…"). O padrão acima — sinal só seta um Event, a main thread chama close() — evita isso.

Dados sensíveis: logs, itens, mensagens de erro e prints são enviados ao servidor e ficam visíveis no app. O SDK redige padrões óbvios de segredo (hb_…, Bearer, sk-…, credenciais em URL), mas não borra prints nem filtra dados pessoais. Capture só o necessário.

Migrando da 2.x

A 3.0 remove o init() e a telemetria "solta": tudo passa a acontecer dentro de uma task.

Antes (2.x):

import hobots

hobots.init(client_secret="hb_…", instance_id="vm-01", heartbeat=True, tasks={"concurrency": 2})


@hobots.task("processar-nfe")
def processar(params, ctx):
    for nota in params["notas"]:
        ctx.raise_if_cancelled()
        with ctx.tx.start_child("http", f"GET /notas/{nota}"):
            baixar(nota)
        ctx.items.succeeded({"numero_nota": nota}, id=nota)


hobots.run_forever()

Depois (3.0):

import hobots  # HOBOTS_CLIENT_SECRET e HOBOTS_INSTANCE_ID no ambiente


def processar(params, ctx):
    for nota in params["notas"]:
        with hobots.transaction.start_child("http", f"GET /notas/{nota}"):  # ponto de parada
            baixar(nota)
        hobots.items.succeeded({"numero_nota": nota}, id=nota)  # ponto de parada


hobots.register("processar-nfe", processar, concurrency=2)

Diferenças em relação ao SDK Node

Mesmo protocolo, mesmos defaults, nomes em snake_case. O que muda por causa da linguagem:

• Boot síncrono: o primeiro register()/start() configura o processo na hora (no Node, num microtask). Tasks registradas depois entram no próximo poll, antecipado na hora.
• Sem globalThis: import hobots já é o objeto global — os membros de contexto são resolvidos por __getattr__ do módulo, com contextvars no lugar do AsyncLocalStorage.
• attach devolve concurrent.futures.Future em vez de Promise; hobots.signal é um threading.Event em vez de AbortSignal.
• Threads próprias do handler não herdam o contexto da execução. Com uma execução só ativa, hobots.* ainda a encontra; com concurrency > 1, passe o ctx ou rode a thread com contextvars.copy_context().run(...).
• Requisições em voo não são interrompíveis (urllib não tem AbortSignal): um close() durante um poll espera a resposta e devolve à plataforma, como AgentTerminated, qualquer run que ele ainda entregue.

Changelog resumido

• 3.1.0 — hobots.log.* aceita vários argumentos (como print); eco de transaction/steps no terminal só com debug=True; None dentro de extra/payload/params/metadata viaja como null; heartbeat com jitter de ±10% e interval_ms clampado à faixa do servidor; fila da telemetria nunca descarta heartbeat; prazos do reporter (flush, progresso, heartbeat de cancelamento) valem mesmo com o handler num loop apertado; logs emitidos depois do fim da execução seguem como log tardio; close() nunca deixa uma run reivindicada sem dono; PATCH terminal nunca passa de 100 logs; progress.set(inf)/advance concorrente corrigidos; tipos Environment/EnvironmentInput e todos os tipos do protocolo exportados.

Documentação

• Getting started — instalação, envs, primeira task nos dois modos, encerramento.
• Tasks — handler, hobots.*, itens, progresso, anexos, cancelamento, retry, concorrência, idempotência.
• Conceitos — o vocabulário: task, execução, evento, transaction, step, heartbeat, instância, claim…
• Referência de API — assinaturas, tabelas de opções e tempos.
• Protocolo & transporte — endpoints, envelope, retries, claim.
• Exemplos — scripts executáveis, um por funcionalidade.

Desenvolvimento

pip install -e ".[dev]"
pytest            # sem rede (servidor HTTP local nas fixtures)
ruff check . && mypy src
coverage run -m pytest && coverage report   # reprova abaixo de 97% (linhas + branches)

Publicação

0. Rode a suíte — não publique com ela vermelha:

   pytest && ruff check . && mypy src && coverage run -m pytest && coverage report
   

1. Ajuste a versão em src/hobots/__about__.py (semver).
2. Construa o pacote:

   python3 -m pip install build twine
   python3 -m build
   

3. Publique no PyPI (requer credenciais em ~/.pypirc):

   python3 -m twine upload dist/*
   

4. Confirme a publicação em pypi.org/project/hobots.
5. Crie um commit com a nova versão.

Os artifacts em dist/ podem ser descartados após a publicação.

Licença

MIT © Hobots