Python SDK 概览

NextPDF Python SDK 面向需要从 PDF 中提取内容并保留 provenance（来源信息）的 Python 应用程序。当来源 PDF 提供相应结构时，它会返回带引用锚点的结构化块，例如页面 Index（索引）、置信度、可选的 bounding box，以及语义节点标识符。

当你的流程需要回答「这段文本来自哪一页？」、「哪个表格佐证了这个数值？」或「这两份 PDF 之间有什么变动？」这类问题时，应使用这个 SDK，而不是把 PDF 提取结果当作没有来源的纯文本处理。

它提供什么

一个同步的 NextPDF client，适用于脚本、批量作业与 notebook。
一个异步的 AsyncNextPDF client，适用于 asyncio、FastAPI 及其他异步运行环境。
一个 nextpdf 命令行接口（CLI），可从文件路径或标准输入执行一次性提取，并输出到标准输出或文件。
一个可选的 Model Context Protocol（MCP）server，让 AI agent 能够直接调用 PDF 提取工具。
一个搭配 NextPDF Connect、供生产环境使用的远程 backend。
一个通过 pypdf 在离线场景中执行纯库提取的本地 backend。

Backend 选择

远程 backend 会把 PDF 字节数据发送到 NextPDF Connect server。这是推荐的生产环境路径，因为它会集中管理提取行为、authentication、配额与运维控制。

本地 backend 在 Python 进程内运行，并通过 pypdf 读取 PDF。它适合离线开发与 tagged PDF，不过无法提供精确的 bounding box，而且对于 untagged PDF 会采用启发式的 segment（段落）层级提取。本地 backend 是纯库模式：你需要将 LocalBackend 注入 AsyncNextPDF 才能使用它。nextpdf CLI 与 MCP server 都无法使用它。完整比较请参阅 Backend choice matrix。

限制

此 SDK 不会运行光学字符识别（OCR）。扫描件或纯图像 PDF 需要先经过 OCR，NextPDF 才能提取嵌入文本。复杂版面、重叠文本以及不常见的 PDF 生成器，也都可能降低提取质量。

这个 nextpdf CLI 只能使用远程 backend，而且不是流式接口。每个命令都会把整份 PDF 读入内存（从文件路径或标准输入），发送到 NextPDF Connect server，在内存中创建完整结果，再一次性序列化写出。你可以用 --output（或 -o）将该输出重定向到文件或标准输出，但结果是完整缓冲后生成的，并非逐步生成。CLI 无法使用本地 pypdf backend。

选择 client：同步 vs 异步

两种 client 共享同一个 ast 方法命名空间，并返回相同的 Pydantic 模型。差别在于并发模型。

你的情境	使用	原因
脚本与批量作业	`NextPDF`（同步）	线性控制流程；不需要管理 event loop。
Jupyter notebook（笔记本）	`NextPDF`（同步）	`run_sync` 会检测正在运行的 event loop，并把工作调度到 worker 线程，因此阻塞式调用能在 cell 内运行。
`nextpdf` CLI	`NextPDF`（同步，内部）	CLI 会为你创建一个同步 client。
`asyncio` 服务	`AsyncNextPDF`	原生 `await`；没有线程交接。
FastAPI、Starlette、ASGI 等异步运行环境	`AsyncNextPDF`	共享请求的 event loop 与同一个连接池。
高并发扇出	`AsyncNextPDF`	用 `asyncio.gather` 在单个连接池 client 上并行运行多个提取。

NextPDF 会封装一个内部的 AsyncNextPDF，并让每次调用都经过 run_sync。在运行中的 event loop 内（例如 notebook），run_sync 会把协程调度到一个带有自身 loop 的单个 worker 线程，因此你不会遇到嵌套 asyncio.run 的错误。在 asyncio 或 ASGI 服务中，请直接调用 AsyncNextPDF，避免每次调用都付出线程交接的成本。

异步 client 持有一个用于连接池的 httpx.AsyncClient，因此请复用同一个 AsyncNextPDF 实例，并只关闭它一次。同步的 NextPDF client 不会公开 close() 方法。对于长期运行的异步工作负载，请优先使用 AsyncNextPDF 并明确管理其生命周期（请参阅 Production operational model）。

Backend 选择矩阵

backend 会实现 PdfBackend 协议。远程 backend（RemoteBackend）会在你传入 base_url 与 api_key 时自动选用。本地 backend（LocalBackend）必须通过 backend= 参数明确注入到 AsyncNextPDF；它不会从顶层 nextpdf 包导出，也无法从 CLI 或 MCP server 使用。

功能	远程（`RemoteBackend`）	本地（`LocalBackend`）
选用方式	`base_url` + `api_key`	`AsyncNextPDF(backend=LocalBackend(...))`
网络	通过 HyperText Transfer Protocol Secure（HTTPS）连接到 NextPDF Connect	无；在进程内运行
Authentication、配额、计量	集中由 server 管理	无
可观测性与运维控制	服务器端	无
Tagged PDF（StructTree）提取	是	是
Untagged PDF 提取	Server 引擎	启发式段落切分，置信度 `0.5`
边界框（Bounding box）	是（当 server 提供时）	否（`bbox` 为 `None`）
Untagged PDF 上的表格提取	Server 引擎	不返回任何表格
可从 CLI / MCP server 使用	是	否（纯库）
建议用于	生产环境	离线开发、tagged PDF 测试

生产环境请使用远程 backend：它是唯一能让你获得集中式 authentication、配额强制执行、计量与可观测性的路径。离线开发以及针对 tagged PDF 的测试请使用本地 backend，同时接受启发式结果、没有 bounding box，且对 untagged 输入没有表格。

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

生产环境运维模型

在生产环境中，你会搭配 NextPDF Connect 运行远程 backend。以下模式涵盖 client 复用、错误处理、重试、配额处理与超时。这里用到的每个符号在 SDK 中都存在；SDK 不会替你重试，因此重试循环由你负责。

复用 client 并共享连接池

RemoteBackend 会维护一个持久的 httpx.AsyncClient 用于连接池。请只创建一次 AsyncNextPDF，在多个请求之间共享它，并在关闭时关闭它。不要每个请求都创建一个 client。

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

异步 context manager 会在退出时调用 close()，进而关闭底层传输层。如果没有使用 context manager，请自行调用 await client.close()。

用异常层次结构处理错误

此 SDK 会抛出一个带类型的层次结构。所有错误都派生自 NextPDFError；HTTP 层级的失败则派生自 NextPDFAPIError，并带有 status_code。请捕获你能处理的具体类型，并用基类兜底。

例外	何时抛出	主要属性
`NextPDFError`	每个 SDK 错误的基类	`status_code`
`NextPDFAPIError`	来自 server 的任何 HTTP 错误	`status_code`, `error_code`
`NextPDFLicenseError`	HTTP 402；此功能需要更高的 server 方案等级	`status_code`（402）
`QuotaExceededError`	HTTP 429；超过速率限制或配额	`retry_after`
`AstNoStructTreeError`	HTTP 422；untagged PDF 且关闭启发式模式	`status_code`（422）
`AstBuildTimeoutError`	HTTP 504；AST 构建超时	`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)

使用退避策略重试临时性失败

此 SDK 不会自动重试。请将调用包进你自己的循环，在临时性 HTTP 失败时重试，并遵守 server 的 Retry-After 值；QuotaExceededError 会将这个值公开为 retry_after（以秒为单位的整数，或 None）。对于其他临时性状态，请使用指数退避，并且不要重试 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")

管理配额、速率限制与超时

配额与速率限制都在 server 上强制执行。HTTP 429 时，SDK 会抛出 QuotaExceededError，并将 Retry-After 标头 resolve（解析）为 retry_after。远程 backend 也会在 render 响应上提供 X-RateLimit-* 标头，因此你可以在触及硬性上限之前主动节流。

请求超时采用固定的默认值：总计 60 秒，连接超时 10 秒（httpx.Timeout(60.0, connect=10.0)）。若要限制过长的 AST 构建，请优先用 page_range_start、page_range_end 或 token_budget 缩小工作范围，而不要只依赖超时；过长的构建会返回 AstBuildTimeoutError（HTTP 504）。

架构示例

批量作业

批量 worker 会读取 PDF、提取带引用来源的文本，并写出结构化输出。请复用一个连接池 client，用 semaphore 限制并发量，并应用上面的重试辅助函数。

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

Web 服务

FastAPI 服务会在应用程序的 lifespan 期间在多个请求之间共享一个 AsyncNextPDF，因此每个请求都会复用该连接池。请从环境变量读取凭证，并把 API key 当作 secret 处理。

"""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]}

Agent 工具

对于 AI agent，请运行 MCP server。它会通过标准 input/output 公开多个 PDF 工具（例如 nextpdf_extract_text、nextpdf_extract_tables、nextpdf_get_ast、nextpdf_info、nextpdf_search、nextpdf_get_outline、nextpdf_diff，以及 nextpdf_health）。此 server 会从环境变量读取 NEXTPDF_BASE_URL 与 NEXTPDF_API_KEY，因此由远程 backend 支持；与 CLI 一样，它也无法使用本地 backend。请安装这个可选附加组件并运行该模块。

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

agent 集成的逐步教程请参阅 Python MCP server，终端用法请参阅 Python CLI，完整的 client、模型与异常接口请参阅 Python API reference。