Visão geral do SDK Python

Visão geral do SDK Python

O kit de desenvolvimento de software (SDK) Python do NextPDF é voltado a aplicações Python que precisam de extração do Portable Document Format (PDF) com proveniência. Ele retorna blocos estruturados com âncoras de citação, incluindo índice de página, confiança, caixas delimitadoras opcionais e identificadores de nós semânticos quando o PDF de origem expõe essa estrutura.

Use o SDK quando o pipeline precisar responder a perguntas como “De qual página veio este texto?”, “Qual tabela embasa este valor?” ou “O que mudou entre estes dois PDFs?” sem tratar a extração de PDF como texto simples sem origem identificável.

O que ele oferece

• Um cliente NextPDF síncrono para scripts, jobs em lote e notebooks.
• Um cliente AsyncNextPDF assíncrono para asyncio, FastAPI e outros runtimes assíncronos.
• Uma interface de linha de comando (CLI) nextpdf para extração em uma única execução a partir de um caminho de arquivo ou da entrada padrão, gravando na saída padrão ou em um arquivo.
• Um servidor opcional do Model Context Protocol (MCP) para que agentes de inteligência artificial (IA) possam chamar diretamente as ferramentas de extração de PDF.
• Um backend remoto para uso em produção com o NextPDF Connect.
• Um backend local, disponível apenas como biblioteca, para extração offline por meio do pypdf.

Escolhas de backend

O backend remoto envia os bytes do PDF para um servidor do NextPDF Connect. Esse é o caminho de produção recomendado porque centraliza o comportamento da extração, a autenticação, as cotas e os controles operacionais.

O backend local é executado dentro do processo Python e lê PDFs por meio do pypdf. Ele é útil para desenvolvimento offline e PDFs com tags, mas não fornece caixas delimitadoras precisas e usa extração heurística em nível de parágrafo para PDFs sem tags. O backend local está disponível apenas como biblioteca: injete um LocalBackend no AsyncNextPDF para usá-lo. A CLI nextpdf e o servidor MCP não podem usá-lo. Consulte Matriz de escolha de backend para a comparação completa.

Limitações

O SDK não realiza reconhecimento óptico de caracteres (OCR). PDFs digitalizados ou somente de imagem precisam de uma etapa de OCR antes que o NextPDF consiga extrair texto incorporado. Layouts complexos, texto sobreposto e produtores de PDF incomuns também podem reduzir a qualidade da extração.

A CLI nextpdf usa somente o backend remoto e não é uma interface de streaming. Cada comando lê todo o PDF na memória (a partir de um caminho de arquivo ou da entrada padrão), envia-o a um servidor do NextPDF Connect, monta o resultado completo na memória e o serializa em uma única gravação. Você pode redirecionar essa saída para um arquivo com --output (ou -o) ou para a saída padrão, mas o resultado é totalmente armazenado em buffer, não produzido de forma incremental. A CLI não pode usar o backend local pypdf.

Escolha de um cliente: síncrono x assíncrono

Ambos os clientes compartilham o mesmo namespace de métodos ast para operações de árvore de sintaxe abstrata (AST) e retornam os mesmos modelos Pydantic. Eles diferem apenas no modelo de concorrência.

Seu contexto	Use	Por que
Scripts e jobs em lote	NextPDF (síncrono)	Fluxo de controle linear; nenhum loop de eventos para gerenciar.
Notebooks Jupyter	NextPDF (síncrono)	run_sync detecta o loop de eventos em execução e despacha para uma thread de trabalho, de modo que chamadas bloqueantes funcionam dentro das células.
A nextpdf CLI	NextPDF (síncrono, interno)	A CLI monta um cliente síncrono para você.
asyncio (serviços)	AsyncNextPDF	Suporte a await nativo; sem delegação entre threads.
FastAPI, Starlette, Asynchronous Server Gateway Interface (ASGI)	AsyncNextPDF	Compartilha o loop de eventos da requisição e o mesmo pool de conexões.
Fan-out de alta concorrência	AsyncNextPDF	Execute muitas extrações simultaneamente com asyncio.gather sobre um único cliente em pool.

NextPDF encapsula um AsyncNextPDF interno e executa cada chamada por meio do run_sync. Dentro de um loop de eventos em execução, como em um notebook, o run_sync despacha a corrotina para uma única thread de trabalho com seu próprio loop, de modo que você não encontra o erro de asyncio.run aninhado. Em um serviço asyncio ou ASGI, chame o AsyncNextPDF diretamente em vez de pagar o custo dessa delegação entre threads a cada chamada.

O cliente assíncrono mantém um httpx.AsyncClient para pool de conexões, portanto reutilize uma instância de AsyncNextPDF e feche-a uma única vez. O cliente síncrono NextPDF não expõe um método close(). Para cargas assíncronas de longa duração, prefira o AsyncNextPDF e gerencie explicitamente o ciclo de vida dele (consulte Modelo operacional de produção).

Matriz de escolha de backend

Um backend implementa o protocolo PdfBackend. O backend remoto (RemoteBackend) é selecionado automaticamente quando você passa base_url e api_key. Você deve injetar o backend local (LocalBackend) explicitamente por meio do parâmetro backend= do AsyncNextPDF; ele não é exportado pelo pacote nextpdf de nível superior e não é acessível pela CLI nem pelo servidor MCP.

Capacidade	Remoto (RemoteBackend)	Local (LocalBackend)
Selecionado por	base_url + api_key	AsyncNextPDF(backend=LocalBackend(...))
Rede	NextPDF Connect por Hypertext Transfer Protocol Secure (HTTPS)	Nenhuma; roda no processo
Autenticação, cotas, medição	Centralizadas no servidor	Nenhuma
Observabilidade e controles operacionais	No lado do servidor	Nenhum
Extração de PDF com tags (StructTree)	Sim	Sim
Extração de PDF sem tags	Engine do servidor	Divisão heurística de parágrafos, confiança 0.5
Caixas delimitadoras	Sim (quando o servidor as fornece)	Não (bbox é None)
Extração de tabelas em PDFs sem tags	Engine do servidor	Não retorna tabelas
Acessível pela CLI / servidor MCP	Sim	Não (apenas como biblioteca)
Recomendado para	Produção	Desenvolvimento offline, testes com PDFs com tags

Use o backend remoto em produção porque ele é o único caminho com autenticação centralizada, aplicação de cotas, medição e observabilidade. Use o backend local para desenvolvimento offline e testes com PDFs com tags, aceitando resultados heurísticos, ausência de caixas delimitadoras e nenhuma tabela em entradas sem tags.

"""Inject the local backend for offline, library-only extraction."""

from nextpdf import AsyncNextPDF
from nextpdf.backends.local import LocalBackend


async def extract_offline(pdf_bytes: bytes) -> None:
    """Extract cited text without a NextPDF Connect server."""
    async with AsyncNextPDF(backend=LocalBackend()) as client:
        blocks = await client.ast.extract_cited_text(pdf_bytes)
        for block in blocks:
            # Heuristic blocks on untagged PDFs report confidence 0.5.
            print(block.citation.confidence, block.text)

Modelo operacional de produção

Em produção, execute o backend remoto com o NextPDF Connect. Os padrões a seguir cobrem reutilização de cliente, tratamento de erros, novas tentativas, tratamento de cotas e timeouts. Cada símbolo usado aqui existe no SDK. O SDK não faz novas tentativas por você, portanto o loop de novas tentativas é responsabilidade sua.

Reutilize o cliente e faça pool de conexões

RemoteBackend mantém um httpx.AsyncClient persistente para pool de conexões. Crie o AsyncNextPDF uma vez, compartilhe-o entre as requisições e feche-o no encerramento. Não crie um cliente por requisição.

"""Reuse one pooled async client for the lifetime of the process."""

import asyncio
import os
from pathlib import Path

from nextpdf import AsyncNextPDF


async def main() -> None:
    """Run several extractions over a single pooled client."""
    base_url = os.environ["NEXTPDF_BASE_URL"]
    # Treat the API key as a secret; read it from the environment, never hard-code it.
    api_key = os.environ["NEXTPDF_API_KEY"]

    async with AsyncNextPDF(base_url=base_url, api_key=api_key) as client:
        pdf_paths = (Path("a.pdf"), Path("b.pdf"), Path("c.pdf"))
        tasks = [
            client.ast.get_document_ast(path.read_bytes())
            for path in pdf_paths
        ]
        documents = await asyncio.gather(*tasks)
        for document in documents:
            print(document.page_count, document.estimated_tokens)


if __name__ == "__main__":
    asyncio.run(main())

O gerenciador de contexto assíncrono chama close() ao sair, o que fecha o transporte subjacente. Sem um gerenciador de contexto, chame await client.close() você mesmo.

Trate erros com a hierarquia de exceções

O SDK levanta uma hierarquia tipada de exceções. Todos os erros derivam de NextPDFError; falhas em nível de Hypertext Transfer Protocol (HTTP) derivam de NextPDFAPIError e carregam um status_code. Capture os tipos específicos sobre os quais você pode agir e use o tipo base como fallback.

Exceção	Levantada quando	Atributos principais
NextPDFError	Tipo base para todos os erros do SDK	status_code
NextPDFAPIError	Qualquer erro HTTP do servidor	status_code, error_code
NextPDFLicenseError	HTTP 402; o recurso precisa de um nível de servidor superior	status_code (402)
QuotaExceededError	HTTP 429; limite de taxa ou cota excedida	retry_after
AstNoStructTreeError	HTTP 422; PDF sem tags com o modo heurístico desativado	status_code (422)
AstBuildTimeoutError	HTTP 504; o build da AST excedeu o tempo limite	status_code (504)

"""Map SDK exceptions to caller-facing outcomes."""

from nextpdf import (
    AstBuildTimeoutError,
    AstNoStructTreeError,
    AsyncNextPDF,
    NextPDFAPIError,
    NextPDFError,
    NextPDFLicenseError,
    QuotaExceededError,
)


async def safe_extract(client: AsyncNextPDF, pdf_bytes: bytes) -> str:
    """Extract text, translating known failures into a stable status string."""
    try:
        blocks = await client.ast.extract_cited_text(pdf_bytes)
    except QuotaExceededError as exc:
        # exc.retry_after holds the server Retry-After value in seconds, or None.
        return f"rate-limited; retry after {exc.retry_after}s"
    except NextPDFLicenseError:
        return "feature requires a higher server tier"
    except AstNoStructTreeError:
        return "untagged PDF; enable heuristic mode or use a tagged PDF"
    except AstBuildTimeoutError:
        return "build timed out; reduce the page range"
    except NextPDFAPIError as exc:
        return f"server error (status {exc.status_code})"
    except NextPDFError:
        return "extraction failed"
    return "\n".join(block.text for block in blocks)

Tente novamente falhas transitórias com backoff

O SDK não faz novas tentativas automaticamente. Envolva as chamadas em seu próprio loop de novas tentativas para falhas HTTP transitórias e respeite o valor Retry-After do servidor, que o QuotaExceededError expõe como retry_after (um número inteiro de segundos ou None). Use backoff exponencial para outros status transitórios e não tente novamente o NextPDFLicenseError.

"""Retry transient failures with exponential backoff and Retry-After support."""

import asyncio
from collections.abc import Awaitable, Callable
from typing import TypeVar

from nextpdf import NextPDFAPIError, QuotaExceededError

_RETRYABLE_STATUS = frozenset({500, 502, 503, 504})

_T = TypeVar("_T")


async def with_retry(
    coro_factory: Callable[[], Awaitable[_T]],
    *,
    max_attempts: int = 4,
) -> _T:
    """Call coro_factory() with bounded retries on transient server errors.

    Args:
        coro_factory: A zero-argument callable returning a fresh awaitable.
        max_attempts: Maximum number of attempts before giving up.

    Returns:
        The awaited result of the first successful attempt.

    Raises:
        NextPDFAPIError: When all attempts fail or the error is not retryable.
    """
    delay = 1.0
    for attempt in range(1, max_attempts + 1):
        try:
            return await coro_factory()
        except QuotaExceededError as exc:
            if attempt == max_attempts:
                raise
            await asyncio.sleep(exc.retry_after if exc.retry_after is not None else delay)
            delay *= 2.0
        except NextPDFAPIError as exc:
            if attempt == max_attempts or exc.status_code not in _RETRYABLE_STATUS:
                raise
            await asyncio.sleep(delay)
            delay *= 2.0
    raise RuntimeError("unreachable")

Gerencie cotas, limites de taxa e timeouts

A aplicação de cotas e de limites de taxa ocorre no servidor. No HTTP 429, o SDK levanta QuotaExceededError e converte o cabeçalho Retry-After em retry_after. O backend remoto também expõe cabeçalhos X-RateLimit-* nas respostas de renderização, para que você possa limitar a taxa proativamente antes de atingir um limite rígido.

Os timeouts de requisição usam um padrão fixo de 60 segundos no total, com um timeout de conexão de 10 segundos (httpx.Timeout(60.0, connect=10.0)). Para limitar builds longos de AST, restrinja o trabalho com page_range_start, page_range_end ou token_budget em vez de depender apenas do timeout; um build longo demais retorna AstBuildTimeoutError (HTTP 504).

Arquiteturas de exemplo

Job em lote

Um worker em lote lê PDFs, extrai texto citado e grava uma saída estruturada. Reutilize um único cliente em pool, limite a concorrência com um semáforo e aplique o helper de novas tentativas acima.

"""Batch-extract a directory of PDFs over one pooled async client."""

import asyncio
import os
from pathlib import Path

from nextpdf import AsyncNextPDF


async def run_batch(input_dir: Path, concurrency: int = 8) -> None:
    """Extract cited text for every PDF in input_dir, bounded by concurrency."""
    semaphore = asyncio.Semaphore(concurrency)

    async def worker(client: AsyncNextPDF, path: Path) -> None:
        async with semaphore:
            blocks = await client.ast.extract_cited_text(path.read_bytes())
            out = path.with_suffix(".txt")
            out.write_text("\n".join(b.text for b in blocks), encoding="utf-8")

    async with AsyncNextPDF(
        base_url=os.environ["NEXTPDF_BASE_URL"],
        api_key=os.environ["NEXTPDF_API_KEY"],
    ) as client:
        await asyncio.gather(*(worker(client, p) for p in input_dir.glob("*.pdf")))

Serviço web

Um serviço FastAPI compartilha um único AsyncNextPDF entre as requisições durante o ciclo de vida da aplicação, de modo que cada requisição reutiliza o pool de conexões. Leia as credenciais do ambiente e trate a chave de API como um segredo.

"""FastAPI service that shares one pooled NextPDF client across requests."""

import os
from contextlib import asynccontextmanager

from fastapi import FastAPI, UploadFile

from nextpdf import AsyncNextPDF


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Create the pooled client on startup and close it on shutdown."""
    app.state.client = AsyncNextPDF(
        base_url=os.environ["NEXTPDF_BASE_URL"],
        api_key=os.environ["NEXTPDF_API_KEY"],
    )
    try:
        yield
    finally:
        await app.state.client.close()


app = FastAPI(lifespan=lifespan)


@app.post("/extract")
async def extract(file: UploadFile) -> dict[str, list[str]]:
    """Return cited text blocks for an uploaded PDF."""
    pdf_bytes = await file.read()
    blocks = await app.state.client.ast.extract_cited_text(pdf_bytes)
    return {"text": [block.text for block in blocks]}

Ferramenta de agente

Para agentes de IA, execute o servidor MCP. Ele expõe ferramentas de PDF (por exemplo nextpdf_extract_text, nextpdf_extract_tables, nextpdf_get_ast, nextpdf_info, nextpdf_search, nextpdf_get_outline, nextpdf_diff e nextpdf_health) pela entrada e saída padrão. O servidor lê NEXTPDF_BASE_URL e NEXTPDF_API_KEY do ambiente, portanto usa o backend remoto; como a CLI, ele não pode usar o backend local. Instale o extra opcional e execute o módulo.

pip install "nextpdf[mcp]"
python -m nextpdf.mcp

Consulte Servidor MCP do Python para o passo a passo de integração com agentes, CLI do Python para uso no terminal e Referência da API do Python para a superfície completa de cliente, modelo e exceções.