Python SDK 开发者指南

概述

NextPDF Python 软件开发工具包（SDK）是一个轻量且带类型的客户端，封装在 NextPDF Connect endpoint（端点）之上。你的应用程序负责输入验证、凭证处理和并发策略；SDK 负责构建请求、传输以及将响应类型化。请明确这条边界：安全读取 PDF、选择客户端、调用你需要的 ast 方法，并处理相应的明确失败。

当你要围绕 SDK 构建提取服务、asyncio 批处理任务、AI Agent（代理）工具或命令行工作流时，请使用本指南。本指南假设你已读过概述和快速上手，并已准备好 Python 3.10 或更新版本，以及一个 NextPDF Connect endpoint。

架构边界

层级	负责方	职责	不应放在此处
输入来源	应用程序	授权调用方、验证 PDF 来源、选择提取策略。	endpoint URL 或凭证字面值。
客户端构建	应用程序	从环境或秘密管理工具读取 `base_url` 和 `api_key`。	硬编码的秘密值。
`NextPDF` / `AsyncNextPDF`	SDK	构建请求、调用 Connect、返回带类型的 Pydantic 模型。	领域逻辑或持久化策略。
`ast` 方法命名空间	SDK	将一次方法调用映射到 Connect endpoint，并解析响应。	超出你所设置范围的重试或退避策略。
NextPDF Connect endpoint	部署	运行提取，并执行认证、配额和授权。	应用程序层的授权。

SDK 从不运行光学字符识别（OCR）。扫描或纯图像的 PDF 在提取前需要先经过一道 OCR 步骤；请将其视为这条边界之外的应用程序层事务。

运行时生命周期

阶段	行为	开发者动作
客户端构建	`base_url` 和 `api_key` 会被验证；空值会引发 `ValueError`。	两者都从环境读取；绝不可硬编码到程序中。
backend（后端）创建	远程 backend 会为 Connect 打开一条池化（pooled）连接。	在多次调用之间复用同一个客户端，而不是每次请求都重新构建。
方法调用	对应的 `ast` 方法会序列化请求、发送 PDF 字节，并将响应解析为一个 Pydantic 模型。	传入已验证过的 `bytes`。
错误映射	不成功的 HTTP 状态会被映射到一个明确的异常子类别。	先拦截最明确的类别。
关闭	`AsyncNextPDF.close()` 会释放连接池；异步上下文管理器会替你调用它。	使用 `async with`，或在 `finally` 块中调用 `close()`。

建议的应用程序结构

路径	用途
`app/pdf/clients.py`	构建并缓存一个配置好的 `NextPDF` 或 `AsyncNextPDF`。
`app/pdf/extraction.py`	应用程序对 `ast` 方法调用的包装层。
`app/pdf/validation.py`	PDF 来源验证、大小限制和内容检查。
`tests/pdf/`	提取、失败模式和异步批量处理的测试。

让 PDF 验证与提取彼此分离。提取层应该接收已授权且已检查大小的字节，同时仍将 endpoint 作为纵深防御的一环。

import os

from nextpdf import NextPDF


def build_client() -> NextPDF:
    """Construct a synchronous client from environment configuration.

    Raises:
        KeyError: When a required environment variable is missing.
    """
    base_url = os.environ["NEXTPDF_BASE_URL"]
    api_key = os.environ["NEXTPDF_API_KEY"]
    return NextPDF(base_url=base_url, api_key=api_key)

同步客户端模式

在脚本、批量任务和 notebook 中，使用同步的 NextPDF 客户端。在调用 SDK 之前先验证输入，并处理该次调用可能引发的明确失败。

from pathlib import Path

from nextpdf import (
    NextPDF,
    CitedTextBlock,
    NextPDFAPIError,
    NextPDFError,
    QuotaExceededError,
)

MAX_PDF_BYTES = 100 * 1024 * 1024  # Reject documents above 100 MiB for the in-memory path.


def read_pdf(path: Path) -> bytes:
    """Read and validate a PDF from disk.

    Raises:
        ValueError: When the file is missing, empty, oversized, or not a PDF.
    """
    if not path.is_file():
        raise ValueError(f"Not a file: {path}")
    data = path.read_bytes()
    if not data:
        raise ValueError("PDF is empty")
    if len(data) > MAX_PDF_BYTES:
        raise ValueError("PDF exceeds the configured size limit; use the CLI streaming path")
    if not data.startswith(b"%PDF-"):
        raise ValueError("File does not look like a PDF")
    return data


def extract_text(client: NextPDF, path: Path) -> list[CitedTextBlock]:
    """Extract cited text blocks, handling the most specific failures first."""
    pdf_bytes = read_pdf(path)
    try:
        return client.ast.extract_cited_text(pdf_bytes)
    except QuotaExceededError as error:
        raise RuntimeError(f"Quota exceeded; retry after {error.retry_after}s") from error
    except NextPDFAPIError as error:
        raise RuntimeError(f"API error {error.status_code}: {error}") from error
    except NextPDFError as error:
        raise RuntimeError(f"SDK error: {error}") from error

单个结果项的预期结构如下：

block = blocks[0]
print(block.text)                  # the extracted text
print(block.citation.page_index)   # 0-based page index
print(block.citation.confidence)   # 0.0 - 1.0

异步与批量处理模式

在 asyncio 运行环境（例如 FastAPI）中，使用异步的 AsyncNextPDF 客户端。以异步上下文管理器的形式构建一个客户端，并在多个并发调用之间共用它；不要为每份文档各开一个客户端。用 semaphore 限制并发量，以遵守 endpoint 的配额。

import asyncio
import os

from nextpdf import (
    AsyncNextPDF,
    ExtractCitedTablesResponse,
    NextPDFError,
    QuotaExceededError,
)


async def extract_tables_batch(
    pdfs: list[bytes],
    *,
    max_concurrency: int = 4,
) -> list[ExtractCitedTablesResponse | None]:
    """Extract tables from many PDFs concurrently with one shared client.

    Returns one response per input PDF, or None where extraction failed.
    """
    base_url = os.environ["NEXTPDF_BASE_URL"]
    api_key = os.environ["NEXTPDF_API_KEY"]
    semaphore = asyncio.Semaphore(max_concurrency)

    async with AsyncNextPDF(base_url=base_url, api_key=api_key) as client:

        async def one(pdf_bytes: bytes) -> ExtractCitedTablesResponse | None:
            async with semaphore:
                try:
                    return await client.ast.extract_cited_tables(pdf_bytes)
                except QuotaExceededError as error:
                    # Surface the backpressure signal; do not silently drop it.
                    raise RuntimeError(f"Quota exceeded; retry after {error.retry_after}s") from error
                except NextPDFError:
                    return None

        return await asyncio.gather(*(one(pdf) for pdf in pdfs))

绝不要编写空的 except。要么针对该失败采取动作、将其转换为明确定义的结果，要么重新抛出它。

扩展点

扩展点	用途	限制
`AsyncNextPDF(backend=...)`	在测试中注入自定义或本地 backend。	该 backend 必须满足 `PdfBackend` 协议。
`api_version` 参数	固定一个 Connect API 版本。	默认为 `v1`；只有在 endpoint 支持目标版本时才更改。
环境配置	将 `NEXTPDF_BASE_URL` 和 `NEXTPDF_API_KEY` 提供给 CLI 和 MCP 服务器。	将密钥视为仅限该工作负载范围内使用的秘密。
MCP 服务器（`python -m nextpdf.mcp`）	将提取工具暴露给支持 MCP 的代理。	需要 `nextpdf[mcp]` 额外包和一个受控的 endpoint。

开发工作流程

用 pip install nextpdf 安装 SDK，或用 pip install nextpdf[mcp] 安装代理服务器。
从环境读取 NEXTPDF_BASE_URL 和 NEXTPDF_API_KEY，这样就不会有任何秘密进入源代码版本控制。
在调用 SDK 之前，验证每个 PDF 来源——是否存在、大小，以及 %PDF- 魔术字节。
每个进程构建一个客户端并复用；在 asyncio 中，以 async with 让它保持打开。
针对该任务调用最精确的 ast 方法：正文文本用 extract_cited_text()、表格用 extract_cited_tables()，只有在你需要完整树结构时才用 get_document_ast()。
先拦截你能据以处置的最明确异常，再回退到 NextPDFError。
对于超过 100 MiB 的文档，请改用 CLI 流式路径，而不是把每个区块都物化到内存中。
以 mypy 的严格模式运行类型检查，并为你处理的每个异常加上一个失败模式测试。

失败处理

失败情况	例外	建议的回应方式
未加标签的 PDF，且启发式判断已关闭	`AstNoStructTreeError`（HTTP 422）	在 endpoint 上启用启发式模式，或提供一份已加标签的 PDF。
服务器端构建超时	`AstBuildTimeoutError`（HTTP 504）	缩小页面范围后重试。
需要相应的授权级别	`NextPDFLicenseError`（HTTP 402）	升级服务器授权，或回退到允许使用的功能。
速率限制或配额	`QuotaExceededError`（HTTP 429）	等待 `retry_after` 秒后，再以退避策略重试。
其他 HTTP 错误	`NextPDFAPIError`	检查 `status_code` 和 `error_code`；记录并对外呈现一个明确定义的错误。
任何 SDK 错误	`NextPDFError`	最后的兜底；绝不要让它以未处理异常的形式逸出。

endpoint 使用与 RFC 9110 对齐的 HTTP 状态语义报告失败，并以与 RFC 9457 对齐的机器可读错误主体呈现。每个异常都会保留原始的 status_code。将这些状态映射到你自己的错误响应，而不是把传输层细节泄漏给调用方。

安全默认值

考虑项	默认	何时应覆盖
API 版本	`v1`。	只有在 endpoint 支持更新版本时才更改。
TLS 验证	已启用；不对外提供任何不安全开关。	绝不要为正式环境流量停用它。
凭证	从环境读取；绝不硬编码到程序中。	在正式环境使用秘密管理工具。
内存内大小限制	在客户端路径上拒绝超过 100 MiB 的 PDF。	对多租户服务调得更低；较大的文件请改用 CLI。
并发	在异步批量中由 semaphore 限制。	根据 endpoint 的配额调整，而不是根据主机的内核数。
日志	记录文件名、大小、状态和耗时。	绝不要记录 PDF 字节或 API 密钥。

测试检查清单

构建测试要断言：空的 base_url 或 api_key 会引发 ValueError。
验证测试要涵盖缺失、空、过大和非 PDF 的输入。
提取测试要断言返回的模型类型，以及每个区块上都有一个 CitationAnchor。
失败模式测试要涵盖 AstNoStructTreeError、AstBuildTimeoutError、NextPDFLicenseError、QuotaExceededError 与 NextPDFAPIError。
异步测试要断言：客户端是以 async with 上下文管理器的形式使用，且并发量保持在 semaphore 的上限内。
生命周期测试要断言 close() 会释放传输层资源，且该操作是幂等的。
用 AsyncNextPDF(backend=...) 注入一个假的 backend，使测试无需实际运行的 endpoint 也能运行。

另请参阅

Python API 参考——涵盖每个客户端方法、模型和异常。
Python CLI——针对大型文档的流式提取。
Python MCP 服务器——将提取工具暴露给 AI 代理。
Python SDK 概述——backend 选项和限制。