flaxflow 1.0.2

Official Python SDK for the FlaxFlow Engine API — AI document processing.

flaxflow — Python SDK

SDK oficial em Python para a FlaxFlow Engine API — extração de dados de documentos (PDF, imagens) com IA, classificação automática, processamento em lote e webhooks.

pip install flaxflow

Requisitos: Python 3.10+ · dependências: httpx, pydantic v2

---

Índice

1. Conceitos da API
2. Configuração
3. Guia 1 — Extração rápida com prompt
4. Guia 2 — Document class a partir de amostra
5. Guia 3 — Document class manual (campos definidos)
6. Guia 4 — Classifier (roteamento automático)
7. Guia 5 — Processamento em lote
8. Guia 6 — Jobs: histórico, analytics e download
9. Guia 7 — Webhooks
10. Guia 8 — API keys (administração)
11. Referência da API
12. Arquivos de entrada
13. Erros
14. Tipos e modelos

---

Conceitos da API

Antes de codar, entenda os blocos que a Engine expõe:

Conceito	O que é	Quando usar
Document class	Template de extração: campos + prompt que a IA segue	Quando você processa o mesmo tipo de documento repetidamente (NF, contrato, RG…)
Processing job	Uma execução de extração sobre um arquivo	Retorna result (JSON extraído), status, timestamps, etc.
Classifier	Roteador: olha o arquivo e escolhe qual document class aplicar	Quando você recebe vários tipos de documento no mesmo fluxo (ex.: NF + boleto + contrato)
Batch	Grupo de jobs criados juntos via process_many	Processar dezenas/centenas de arquivos de uma vez
Webhook	HTTP POST que a FlaxFlow manda quando um job termina ou falha	Integrações assíncronas (filas, ERP, Slack…) sem polling

Fluxo típico de produção:

1. Criar document classes (ou classifier)
2. process_document / process_many
3. Ler job.result  OU  receber webhook job.completed

---

Configuração

Passo 1 — Obter API key

No app FlaxFlow: Settings → API keys → Create. A chave começa com sk_....

Passo 2 — Instanciar o client

from flaxflow import FlaxFlow

# Opção A: passar a chave direto
fx = FlaxFlow(api_key="sk_...")

# Opção B: variável de ambiente (recomendado em produção)
# export FLAXFLOW_API_KEY=sk_...
fx = FlaxFlow()

# Opção C: context manager (fecha conexão HTTP ao sair)
with FlaxFlow() as fx:
    ...

Opção	Default	Variável de ambiente
api_key	obrigatório	FLAXFLOW_API_KEY
base_url	https://api.flaxia.com.br	FLAXFLOW_BASE_URL
timeout	60 segundos	—

Passo 3 — Escolher o modo de extração

Modo	Método	Precisa de document class?
Prompt livre	processing.process_with_prompt	Não
Document class fixa	processing.process_document(document_id=...)	Sim
Classifier	processing.process_document(classifier_id=...)	Sim (várias classes ligadas ao classifier)

---

Guia 1 — Extração rápida com prompt

Use quando quiser testar a API sem criar nada antes. Ideal para protótipos.

Passo 1 — Processar

from flaxflow import FlaxFlow

fx = FlaxFlow(api_key="sk_...")

job = fx.processing.process_with_prompt(
    "nota-fiscal.pdf",
    prompt=(
        "Extraia: número da nota, data de emissão, CNPJ do emitente, "
        "valor total e lista de itens com descrição e quantidade."
    ),
)

print(job.status)       # ProcessingJobStatus.completed
print(job.result)       # dict com os campos extraídos
print(job.pages_count)  # páginas processadas

Passo 2 — Inspecionar o resultado

result = job.result or {}

numero = result.get("numero_nota") or result.get("invoice_number")
total = result.get("valor_total") or result.get("total_amount")

print(f"NF {numero} — R$ {total}")

Passo 3 — Tratar falha

from flaxflow import ProcessingJobStatus  # via generated models se quiser tipar

if job.status.value == "failed":
    print("Extração falhou — veja logs no portal ou reprocesse")
elif job.status.value == "waiting_validation":
    print("Aguardando validação humana no portal")

Limitação: prompt livre não persiste configuração. Para produção com o mesmo layout, crie uma document class (Guias 2 ou 3).

---

Guia 2 — Document class a partir de amostra

A IA analisa um arquivo exemplo e infere nome, descrição e campos. Você só envia a amostra uma vez; depois reutiliza o document_id.

Passo 1 — Enviar amostra

from flaxflow import FlaxFlow

fx = FlaxFlow(api_key="sk_...")

# A IA infere campos a partir deste PDF/imagem
doc = fx.documents.create_from_sample("exemplos/nota-fiscal-modelo.pdf")

print(doc.document_id)    # UUID — guarde isso
print(doc.display_name)   # nome sugerido pela IA
print(doc.description)
print(doc.fields)         # lista de FieldModel inferidos

Passo 2 — Conferir campos inferidos

for field in doc.fields or []:
    print(field.field_name, field.display_name, field.field_type)

Passo 3 — Processar documentos reais

job = fx.processing.process_document(
    "entrada/nf-cliente-001.pdf",
    document_id=str(doc.document_id),
)

print(job.result)

Passo 4 — Processar vários arquivos com a mesma class

paths = ["nf-001.pdf", "nf-002.pdf", "nf-003.pdf"]

for path in paths:
    job = fx.processing.process_document(path, document_id=str(doc.document_id))
    print(path, job.result)

Passo 5 — Listar e remover classes antigas

for d in fx.documents.list():
    print(d.document_id, d.display_name, d.document_type)

# fx.documents.delete(str(doc.document_id))

---

Guia 3 — Document class manual (campos definidos)

Use quando você já sabe exatamente quais campos quer, sem depender da inferência da IA.

Passo 1 — Definir campos

from flaxflow import FlaxFlow
from flaxflow.resources.documents import PROCESS_METHOD_FIELDS

fx = FlaxFlow(api_key="sk_...")

doc = fx.documents.create_manual(
    document_type="pdf",
    display_name="Nota Fiscal de Serviço",
    description="NFSe com prestador, tomador e valores",
    process_method=PROCESS_METHOD_FIELDS,  # extração por campos (default)
    fields=[
        {
            "fieldName": "numero_nfse",
            "displayName": "Número da NFSe",
            "fieldType": "string",
            "fieldPrompt": "Número da nota fiscal de serviço no topo do documento",
        },
        {
            "fieldName": "valor_liquido",
            "displayName": "Valor Líquido",
            "fieldType": "number",
            "fieldPrompt": "Valor líquido total a pagar",
        },
        {
            "fieldName": "data_emissao",
            "displayName": "Data de Emissão",
            "fieldType": "date",
        },
    ],
)

Passo 2 — Processar

job = fx.processing.process_document(
    "nfse.pdf",
    document_id=str(doc.document_id),
)
print(job.result)

Passo 3 — Document class baseada em prompt (sem lista de campos)

from flaxflow.resources.documents import PROCESS_METHOD_PROMPT

doc = fx.documents.create_manual(
    document_type="pdf",
    display_name="Contrato genérico",
    description="Extrai partes, objeto e vigência",
    process_method=PROCESS_METHOD_PROMPT,
    prompt=(
        "Identifique contratante, contratado, objeto do contrato, "
        "data de início e data de término."
    ),
)

Passo 4 — Atualizar uma class existente

updated = fx.documents.update(
    str(doc.document_id),
    document_type="pdf",
    display_name="NFSe — versão 2",
    description="Campos revisados",
    fields=[...],  # nova lista de campos
)

---

Guia 4 — Classifier (roteamento automático)

Um classifier escolhe automaticamente qual document class usar. Cenário clássico: caixa de entrada com NF, boleto e contrato misturados.

Passo 1 — Criar document classes (uma por tipo)

nf = fx.documents.create_from_sample("exemplos/nf.pdf")
boleto = fx.documents.create_from_sample("exemplos/boleto.pdf")
contrato = fx.documents.create_from_sample("exemplos/contrato.pdf")

Passo 2 — Criar o classifier

classifier_id = fx.classifiers.create("Caixa de entrada — financeiro")
print(classifier_id)

Passo 3 — Vincular document classes ao classifier

fx.classifiers.set_models(
    classifier_id,
    document_ids=[
        str(nf.document_id),
        str(boleto.document_id),
        str(contrato.document_id),
    ],
)

Passo 4 — Processar sem saber o tipo antecipadamente

job = fx.processing.process_document(
    "documento-desconhecido.pdf",
    classifier_id=classifier_id,
)

print(job.document_id)   # qual class foi escolhida
print(job.result)

Passo 5 — Gerenciar classifiers

# Listar
for c in fx.classifiers.list():
    print(c.classifier_id, c.name)

# Detalhe (com documentos vinculados)
detail = fx.classifiers.get(classifier_id)
print(detail)

# Renomear
fx.classifiers.update(classifier_id, name="Financeiro v2")

# Remover
# fx.classifiers.delete(classifier_id)

---

Guia 5 — Processamento em lote

Envie vários arquivos numa única requisição. A API cria um batch_id e um job por arquivo.

Passo 1 — Disparar o batch

batch = fx.processing.process_many(
    ["lote/a.pdf", "lote/b.pdf", "lote/c.png"],
    document_id=str(doc.document_id),
    # ou: classifier_id=classifier_id
)

print(batch.batch_id)
print(len(batch.jobs))
for job in batch.jobs:
    print(job.process_job_id, job.status, job.result)

Passo 2 — Tipos diferentes por arquivo (opcional)

batch = fx.processing.process_many(
    [open("scan.png", "rb"), "doc.pdf"],
    classifier_id=classifier_id,
    document_types=["png", "pdf"],       # um tipo por arquivo
    file_names=["scan-001.png", "doc.pdf"],
)

Passo 3 — Consultar o batch depois

batch = fx.jobs.get_batch(batch.batch_id)

for job in batch.jobs:
    if job.status.value == "completed":
        print(job.document_file_name, job.result)

---

Guia 6 — Jobs: histórico, analytics e download

Todo processamento gera um ProcessingJob. Use fx.jobs para auditar e recuperar dados.

Passo 1 — Listar jobs recentes

page = fx.jobs.list(page=0, page_size=20)

print(page.total, "jobs no total")
for job in page.items:
    print(
        job.process_job_id,
        job.status.value,
        job.document_file_name,
        job.started_at,
    )

Passo 2 — Filtrar por status

# status: "processing" | "completed" | "failed" | "waiting_validation" | "rejected"
failed = fx.jobs.list(page=0, page_size=50, status="failed")

for job in failed.items:
    print(job.process_job_id, job.document_file_name)

Passo 3 — Buscar um job específico

job = fx.jobs.get("123e4567-e89b-12d3-a456-426614174000")

print(job.status)
print(job.result)
print(job.is_validating, job.document_validated)
print(job.validated_at)

Passo 4 — Analytics agregados

stats = fx.jobs.analytics()

print(stats.total_jobs)
print(stats.completed_jobs, stats.failed_jobs)
print(f"Taxa de sucesso: {stats.success_rate:.1%}")
print(f"Páginas processadas: {stats.total_pages_processed}")
print(stats.jobs_by_status)   # dict por status

Passo 5 — Baixar o arquivo original do job

raw_bytes = fx.jobs.download_document(job.process_job_id)

with open("copia-original.pdf", "wb") as f:
    f.write(raw_bytes)

---

Guia 7 — Webhooks

Receba job.completed e job.failed via HTTP POST no seu servidor.

Passo 1 — Registrar o webhook

hook = fx.webhooks.create(
    "https://api.seusistema.com.br/hooks/flaxflow",
    event_types=["job.completed", "job.failed"],
)

# GUARDE O SECRET — só aparece nesta resposta
print(hook.webhook_id)
print(hook.secret)

Passo 2 — Validar assinatura no seu servidor (conceito)

A FlaxFlow assina o body com HMAC-SHA256 usando hook.secret. No handler HTTP do seu backend, compare o header de assinatura com o hash do payload bruto.

Passo 3 — Atualizar ou desativar

fx.webhooks.update(
    str(hook.webhook_id),
    url="https://api.seusistema.com.br/hooks/flaxflow-v2",
    event_types=["job.completed", "job.failed"],
    is_active=True,
)

Passo 4 — Debugar entregas

for log in fx.webhooks.delivery_logs():
    print(log.webhook_id, log.status_code, log.error_message, log.created_at)

Passo 5 — Listar e remover

for w in fx.webhooks.list():
    print(w.webhook_id, w.url, w.is_active)

# fx.webhooks.delete(str(hook.webhook_id))

---

Guia 8 — API keys (administração)

Nota: fx.api_keys.create exige autenticação JWT (login no app), não funciona só com outra API key. Use para scripts admin ou automação interna.

created = fx.api_keys.create("Integração ERP — produção")
print(created.api_key)   # sk_... completa — só nesta resposta

for key in fx.api_keys.list():
    print(key.api_key_id, key.name, key.created_at)

# fx.api_keys.delete(str(key.api_key_id))

---

Referência da API

fx.documents — Document classes

Método	Descrição
create_from_sample(file, document_type=None)	IA infere campos a partir de amostra
create_manual(...)	Cria class com campos/prompt definidos manualmente
update(document_id, ...)	Atualiza class existente
list()	Lista todas as classes do usuário
delete(document_id)	Remove uma class

Constantes úteis:

from flaxflow.resources.documents import PROCESS_METHOD_FIELDS, PROCESS_METHOD_PROMPT

fx.processing — Extração

Método	Descrição
process_with_prompt(file, prompt, ...)	Extração ad-hoc, sem document class
process_document(file, document_id=..., classifier_id=...)	Extração com class fixa ou classifier
process_many(files, document_id=..., classifier_id=...)	Batch de arquivos

fx.jobs — Histórico

Método	Descrição
list(page=0, page_size=10, status=None)	Lista paginada
get(processing_job_id)	Detalhe de um job
get_batch(batch_id)	Jobs de um batch
analytics()	Métricas agregadas
download_document(processing_job_id)	Bytes do arquivo original

fx.classifiers — Roteamento

Método	Descrição
create(name) → str	Cria classifier, retorna id
list()	Lista classifiers
get(classifier_id)	Detalhe + documentos vinculados
update(classifier_id, name=...)	Renomeia
set_models(classifier_id, document_ids=[...])	Define classes permitidas
delete(classifier_id)	Remove

fx.webhooks — Eventos HTTP

Método	Descrição
create(url, event_types=[...])	Registra webhook
list()	Lista webhooks
update(webhook_id, url, event_types, is_active)	Atualiza
delete(webhook_id)	Remove
delivery_logs()	Logs de entrega

fx.api_keys — Chaves de API

Método	Descrição
create(name)	Cria chave (retorna sk_... completa)
list()	Lista chaves (ofuscadas)
delete(api_key_id)	Revoga

---

Arquivos de entrada

O SDK aceita três formatos para o parâmetro file:

# 1. Caminho — document_type inferido pela extensão
fx.processing.process_with_prompt("nota.pdf", prompt="...")

# 2. Bytes
with open("nota.pdf", "rb") as f:
    data = f.read()
fx.processing.process_with_prompt(data, document_type="pdf", prompt="...")

# 3. File-like (objeto com .read())
with open("nota.pdf", "rb") as f:
    fx.processing.process_with_prompt(f, document_type="pdf", prompt="...")

Tipos suportados (SUPPORTED_DOCUMENT_TYPES):

from flaxflow import SUPPORTED_DOCUMENT_TYPES
# ('pdf', 'png', 'jpeg', 'webp', 'gif')

Extensões mapeadas automaticamente: .pdf, .png, .jpg/.jpeg, .webp, .gif.

---

Erros

Toda falha levanta uma subclasse de FlaxFlowError:

Exceção	HTTP	Quando
AuthenticationError	401	API key inválida ou ausente
PermissionDeniedError	403	Sem permissão
NotFoundError	404	Job/document/class não existe
ConflictError	409	Conflito de estado
UnprocessableEntityError	422	Payload inválido
RateLimitError	429	Muitas requisições — faça backoff
ServerError	5xx	Erro no servidor FlaxFlow
APIConnectionError	—	Timeout, DNS, rede

from flaxflow import (
    FlaxFlow,
    AuthenticationError,
    NotFoundError,
    RateLimitError,
    UnprocessableEntityError,
)
import time

fx = FlaxFlow()

def process_with_retry(path: str, document_id: str, max_retries: int = 3):
    for attempt in range(max_retries):
        try:
            return fx.processing.process_document(path, document_id=document_id)
        except RateLimitError:
            time.sleep(2 ** attempt)
        except UnprocessableEntityError as e:
            print("Payload inválido:", e.message, e.body)
            raise
        except NotFoundError:
            print("Document class não encontrada")
            raise
    raise RuntimeError("Rate limit persistente")

---

Tipos e modelos

Respostas são modelos Pydantic v2 gerados do OpenAPI. Nomes em Python são snake_case; o JSON na wire usa camelCase.

from flaxflow._generated.models import (
    Document,
    ProcessingJob,
    ProcessingJobStatus,
    ProcessingJobsPage,
    ProcessingJobAnalytics,
    ClassifierDetail,
    Webhook,
    FieldModel,
)

job: ProcessingJob = fx.jobs.get("...")
print(job.process_job_id)
print(job.status == ProcessingJobStatus.completed)
print(job.result)  # dict ou None

Importe tipos pelo módulo gerado ou use autocompletion no retorno dos métodos.