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 端點	部署	執行擷取，並強制執行認證、配額與授權。	應用程式層的授權。

SDK 從不執行光學字元辨識（OCR）。掃描或純影像的 PDF 在擷取前需要先經過 OCR 步驟；請把它視為這條界線之外的應用程式層事務。

執行期生命週期

階段	行為	開發者動作
用戶端建構	`base_url` 與 `api_key` 會被驗證；空值會引發 `ValueError`。	兩者都從環境讀取；絕不可寫死在程式內。
backend（後端）建立	遠端 backend 會與 Connect 建立一條集區化（pooled）連線。	在多次呼叫之間重複使用同一個用戶端，而非每次請求都重新建構。
方法呼叫	`ast` 方法會序列化請求、送出 PDF 位元組，並將回應剖析成 Pydantic 模型。	傳入已驗證過的 `bytes`。
錯誤對映	非成功 HTTP 狀態會被對映到明確的例外子類別。	先攔截最明確的類別。
關閉	`AsyncNextPDF.close()` 會釋放連線集區；非同步情境管理器會代為呼叫它。	使用 `async with`，或呼叫 `close()`（置於 `finally` 區塊中）。

建議的應用程式結構

路徑	用途
`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 選項與限制。