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

Capability	Como habilita	Para quê	Ciclo de vida
Telemetria	sempre ativa no init()	O que o programa emite sozinho: erros, transactions/etapas, heartbeat.	Threads daemon — não seguram o processo. Chame close() no fim.
Tasks	tasks=True (ou opções) no init()	Execução remota sob demanda via polling.	start() segura o processo vivo até stop()/close().

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

Opção	Tipo	Default	Descrição
client_secret	str	obrigatório	o Client Secret do agente (hb_<64 hex>)
instance_id	str	obrigatório	identidade única da instância — heartbeat e poll de tasks convergem nela
environment	str	'production'	ambiente lógico
release	str	—	versão do programa, ex. robo-nfe@1.4.2
tags	dict[str, str]	—	tags aplicadas a todos os eventos
sample_rate	float	1.0	amostragem de erros (0..1)
traces_sample_rate	float	1.0	amostragem de transactions (0..1)
max_breadcrumbs	int	100	tamanho do buffer de breadcrumbs
capture_unhandled	bool	True	handlers globais de erro (sys.excepthook, threading.excepthook, atexit)
heartbeat	bool | HeartbeatOptions | dict	False	heartbeat periódico
tasks	bool | TasksOptions | dict	False	habilita a execução sob demanda
before_send	(event) -> event | None	—	edita/descarta evento (retorne None para descartar)
debug	bool	False	loga atividade do SDK
flush_interval	int (ms)	2000	intervalo do flush automático do transport
max_batch_size	int	10	itens por envelope antes do flush imediato

Captura de erros e mensagens

try:
    processar_nota(nota)
except Exception as error:
    hobots.capture_exception(
        error,
        {
            "type": "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 type, 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.

EventType = 'fatal' | 'error' | 'warning' | 'success'. Não há 'info'/'debug': narração não abre issue — para isso existem o log da execução e as breadcrumbs. 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).

Opção de tasks	Tipo	Default	Descrição
poll_interval	int (ms)	5000 (mín. 1000)	intervalo entre polls sem trabalho; com trabalho na fila o próximo poll é imediato
concurrency	int	1	runs simultâneas (clamp 1..5), cada uma na sua thread
task_timeout	int (ms)	sem timeout	teto client-side por run — não mata o handler
log_flush_interval	int (ms)	2000	intervalo do envio incremental de logs (ou na hora, ao acumular 20)
cancel_check_interval	int (ms)	5000 (clamp 1000..15000)	detecta cancelamento e mantém o sinal de vida da run
install_exit_handlers	bool	True	handlers de SIGINT/SIGTERM que reportam as runs ativas como AgentTerminated

O ctx (RunContext)

ctx.run_id  # id da execução
ctx.task  # nome da task
ctx.log  # .debug/.info/.warn/.error/.success(mensagem)
ctx.items  # .succeeded/.failed/.occurrence — os itens que a run processou (a entrega do bot)
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.

Itens de execução (ctx.items)

Registre os itens que a run processou/gerou — cada um com status succeeded, failed (falha técnica) ou occurrence (ocorrência de negócio, ex.: "nota nº 10 não encontrada") e um payload de atributos livres. Eles aparecem na aba itens da solicitação, com totalizadores; as colunas exibidas são configuradas por formulário no app.

@hobots.task("baixar-notas")
def handler(params, ctx):
    for nota in listar_notas(params):
        try:
            baixar(nota)
            ctx.items.succeeded({"numero_nota": nota.numero, "cnpj_prestador": nota.cnpj}, id=nota.numero)
        except Exception as erro:
            ctx.items.failed(str(erro), {"numero_nota": nota.numero}, id=nota.numero)

Chamadas síncronas e bufferizadas (lotes de até 500; dreno antes do reporte final). O id= opcional é a identidade do item na run — repetido, é ignorado pelo servidor (registro único, retry idempotente). Limites: 10.000 itens/run, payload 16 KB, mensagem 2.000 caracteres. Guia: Itens de execução.

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

Documento	Conteúdo
Getting started	instalação, primeiro robô, encerramento
Conceitos	evento, tag, breadcrumb, run, transaction, fila…
Telemetria	erros, scope, performance, heartbeat
Tasks	polling, handlers, cancelamento, concorrência, anexos
Referência de API	todas as assinaturas, opções e prazos
Protocolo & transporte	endpoints, wire format, retries
Exemplos	11 scripts executáveis, um por funcionalidade

Desenvolvimento

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

Publicação

Para publicar uma nova versão no PyPI:

1. Ajuste a versão em src/hobots/__about__.py:

   __version__ = "1.2.0"  # incremente conforme 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:

   git add src/hobots/__about__.py
   git commit -m "chore: bump SDK Python version to 1.2.0"
   

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

Licença

MIT