hobots

SDK Python da Hobots para instrumentar agentes (RPAs) e integrações.

• Telemetria — o que o programa emite sozinho: erros com stack trace (módulo Issues do app), performance (transactions/etapas) e heartbeats de uptime (módulo Agentes).
• Tasks — execução sob demanda: o agente faz polling no app da Hobots, executa handlers registrados e reporta status e logs (módulo Solicitações).

Um único import e um único init() cobrem as duas capabilities. 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 cuida das threads de fundo. É o que combina com Selenium, Playwright síncrono, pyautogui e código legado.

Todas as durações são em milissegundos, como no SDK Node: poll_interval=5000 são 5 segundos, ctx.sleep(2000) dorme 2 segundos.

Instalação

pip install hobots

import hobots  # um import só — telemetria e tasks

Um SDK, duas capabilities, um init()

As duas capabilities são autenticadas pelo mesmo Client Secret do agente e configuradas em um único init():

Você usa uma, outra ou as duas:

• Só saber se o robô quebrou, quanto demora e se está vivo? → init(client_secret=..., instance_id=..., heartbeat=True).
• Disparar o robô sob demanda a partir de um formulário no app? → init(..., tasks=True) + handlers + run_forever().
• Ambos? Passe heartbeat e tasks no mesmo init() — o instance_id é a identidade única, e os dois canais convergem em um agente no painel. Com tasks=True, o heartbeat é opcional: o próprio poll de tasks já marca a presença do agente.

import hobots

hobots.init(
    client_secret="hb_<64 hex>",
    instance_id="vm-cliente-01",
    heartbeat=True,  # telemetria: "estou vivo" a cada 30s
    tasks={"poll_interval": 5_000},  # execução sob demanda
)


@hobots.task("processar-nfe")
def processar_nfe(params, ctx):
    ctx.log.info(f"processando o lote {params['lote']}…")
    ctx.log.success("12 notas processadas")  # o retorno é ignorado — reporte por log


hobots.run_forever()  # inicia o polling e bloqueia até Ctrl-C/SIGTERM

O Client Secret

Cada agente tem um único Client Secret — a mesma credencial autentica a telemetria e as tasks:

hb_<64 hex>

A credencial é secreta e viaja sempre no header Authorization: Bearer — nunca na URL, e nunca no upload de anexo para o S3. Copie o Client Secret no app, em Configurações → Agentes.

O SDK fala com a API oficial https://api2.hobots.app automaticamente; em desenvolvimento, aponte para outro host com a env HOBOTS_API_URL:

HOBOTS_API_URL=http://localhost:3001 python meu_robo.py

O secret nunca aparece em repr() nem em serialização do Client, do InitOptions ou do ResolvedConfig — só no acesso explícito (get_client().config.key).

Telemetria

Todas as funções abaixo são no-op seguras se hobots.init() não foi chamado.

init(**options)

Inicialize uma vez, o mais cedo possível no processo. Cria o cliente, instala handlers globais de erro (a menos que capture_unhandled=False) e inicia o heartbeat se configurado.

hobots.init(
    client_secret="hb_<64 hex>",
    instance_id="vm-cliente-01",
    environment="production",
    release="robo-nfe@1.4.2",
    tags={"squad": "automacoes"},
    heartbeat=True,
)

Captura de erros e mensagens

try:
    processar_nota(nota)
except Exception as error:
    hobots.capture_exception(
        error,
        {
            "level": "error",
            "tags": {"cliente": "acme"},
            "extra": {"numero": nota.numero},
            "fingerprint": ["sefaz-timeout"],  # agrupa manualmente
        },
    )
    raise

hobots.capture_message("Lote processado com sucesso", "success")

As duas devolvem o event_id (ou None se o SDK não foi inicializado). O hint aceita level, tags, extra e fingerprint.

Erros não tratados são capturados e flushados automaticamente antes do processo sair, preservando o comportamento padrão do Python (traceback + exit 1). Exceções em threads também são capturadas (threading.excepthook). KeyboardInterrupt e SystemExit não são capturados — são intenção do operador, não defeito. Desligue tudo com capture_unhandled=False.

Severity = 'fatal' | 'error' | 'warning' | 'info' | 'debug' | 'success'. Use 'success' para reportar itens concluídos com êxito — contam como eventos, mas não viram issue nem disparam alertas.

Breadcrumbs

A trilha do que aconteceu antes do erro. Não são enviados sozinhos: viajam anexados ao próximo evento.

hobots.add_breadcrumb(message="abriu o portal", category="nav")
hobots.add_breadcrumb({"message": "login efetuado", "level": "info"})

Scope

Dados que acompanham os próximos eventos:

hobots.set_tag("cliente", "acme")
hobots.set_tags({"regiao": "sudeste", "turno": "noite"})
hobots.set_user({"id": "42", "username": "operador"})
hobots.set_context("portal", {"url": "https://portal.exemplo"})
hobots.set_extra("tentativas", 3)

Precedência de tags: init.tags → scope → evento → hint. set_context(nome, None) apaga o grupo; set_user(None) remove o usuário.

Para isolar alterações num bloco, use o context manager:

with hobots.with_scope() as scope:
    scope.set_tag("lote", "2026-08")
    hobots.capture_message("processando")
# a tag `lote` não existe mais aqui

O isolamento é por thread (contextvars): um with_scope() numa thread não afeta as outras.

Transactions (performance)

with hobots.start_transaction("sincronizar-notas", "job") as tx:
    with tx.start_child("browser", "login no portal") as login:
        login.set_data("usuario", "operador")
    with tx.start_child("http", "GET /notas") as download:
        download.set_data("quantidade", 12)

Também funciona na forma explícita, com .finish():

tx = hobots.start_transaction("sincronizar-notas", "job")
step = tx.start_child("http", "GET /notas")
step.finish()
tx.finish()  # nada é enviado antes daqui

Steps aninham em profundidade arbitrária (step.start_child(...)) e o finish() da transaction fecha as que ficaram abertas. StepStatus = 'ok' | 'warning' | 'error' | 'cancelled'.

Heartbeat

hobots.init(
    client_secret="hb_<64 hex>",
    instance_id="vm-cliente-01",
    heartbeat={"interval": 30_000, "name": "VM 01", "metadata": {"regiao": "sudeste"}},
)

Bate imediatamente no init() e depois a cada interval ms. O valor vai no payload e define quanto silêncio o servidor tolera antes de marcar a instância offline.

Encerramento

hobots.flush(5_000)  # empurra a fila; o SDK continua utilizável
hobots.close(5_000)  # encerra tudo: tasks → heartbeat → flush final

As threads da telemetria são daemon e não seguram o processo — sempre chame close() antes de sair, ou você perde o que estava em buffer.

Tasks (execução sob demanda)

Habilite com tasks=True (ou opções) no init(), registre handlers e chame run_forever().

hobots.init(
    client_secret="hb_<64 hex>",
    instance_id="vm-cliente-01",
    tasks={"poll_interval": 5_000, "concurrency": 2},
)


@hobots.task("processar-nfe")  # o task_slug do formulário no app
def processar_nfe(params, ctx):
    ctx.log.info(f"lote {params['lote']}")


hobots.run_forever()

O decorator é equivalente a hobots.register("processar-nfe", processar_nfe).

O ctx (RunContext)

ctx.run_id  # id da execução
ctx.task  # nome da task
ctx.log  # .debug/.info/.warn/.error/.success(mensagem)
ctx.tx  # a transaction desta run — NÃO chame finish() nela
ctx.cancel_event  # threading.Event que dispara no cancelamento/timeout/shutdown
ctx.is_cancelled()  # -> bool
ctx.raise_if_cancelled()  # levanta RunCanceled
ctx.sleep(ms)  # espera cancelável (levanta RunCanceled)
ctx.attach(imagem, ...)  # anexa um print, bloqueando até subir -> AttachResult
ctx.attach_nowait(imagem, ...)  # idem, sem esperar o upload

O retorno do handler é ignorado — o protocolo não transporta resultado (dados de negócio podem ser sensíveis e não são coletados). Reporte o que importa com ctx.log.

Cancelamento é cooperativo

O SDK não consegue interromper código Python de fora. Ele dispara o cancel_event; o handler precisa parar:

@hobots.task("lote-longo")
def lote_longo(params, ctx):
    for item in itens:
        ctx.raise_if_cancelled()  # ponto de parada de uma linha
        processar(item)
        ctx.sleep(1_000)  # espera cancelável, no lugar de time.sleep

⚠️ Nunca capture o RunCanceled para retornar normalmente — uma run que retorna sem erro é reportada como succeeded. Se precisar fazer limpeza, capture, limpe e re-levante.

Anexos (screenshots)

@hobots.task("capturar-tela")
def capturar(params, ctx):
    ctx.attach(page.screenshot(), caption="portal após o login")

Aceita bytes, bytearray, memoryview ou um caminho de arquivo (str/os.PathLike). A linha de log entra antes de qualquer I/O, então a ordem dos logs fica intacta. attach() bloqueia até o upload terminar; attach_nowait() volta na hora e o SDK drena os uploads pendentes (até 15s) antes de fechar a run.

Nunca levanta: toda falha vira AttachResult(ok=False, reason=...) e um aviso no log da run. reason é um de 'unsupported', 'too-large', 'quota', 'read-failed', 'unsupported-format', 'upload-failed', 'canceled'.

Limites (aplicados no SDK e no servidor): PNG/JPEG/WEBP, 5 MB por imagem, 20 por execução, 4 por linha de log.

Encerramento

hobots.run_forever(timeout=15_000)  # bloqueia até SIGINT/SIGTERM e encerra

Se precisar da main thread para outra coisa, use hobots.start() (volta na hora) e encerre você mesmo:

hobots.start()
...
hobots.close(15_000)  # para as tasks (aguardando as runs) + heartbeat + flush
hobots.stop(30_000)  # para SÓ as tasks; a telemetria continua

⚠️ O timeout do close() é gasto duas vezes, em sequência — uma esperando as runs, outra no flush final. close(15_000) pode levar até ~30s no pior caso. Dimensione pelo prazo do seu SIGTERM (Kubernetes, systemd).

Segurança

• O Client Secret viaja só no header Authorization: Bearer, nunca na URL, e nunca no PUT do anexo para o S3.
• Mensagens de log, captions, mensagens de erro e stacks passam por scrub_secrets() antes de sair do processo — client secrets hb_…, tokens Bearer, chaves sk-… e credenciais embutidas em URL são redigidos.
• O secret nunca aparece em repr()/serialização do Client, InitOptions ou ResolvedConfig.
• Logs, erros e prints são visíveis no app. Não coloque segredos ou dados sensíveis neles.

Documentação

Desenvolvimento

pip install -e ".[dev]"
pytest            # 123 testes, sem rede (servidor HTTP local nas fixtures)
ruff check . && ruff format --check .
mypy src

Publicação: ajuste __version__ em src/hobots/__about__.py, rode python -m build (ou uv build) e twine upload dist/*.

Licença

MIT