Python SDK 總覽

NextPDF Python SDK 專為需要帶有 provenance（來源資訊）之 PDF 擷取的 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 只能用於遠端，而且並非串流介面。每個指令都會把整份 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。