Python SDK の概要

Python SDK の概要

NextPDF Python SDK は、来歴付きの PDF 抽出を必要とする Python アプリケーションを対象としています。ソース PDF が構造情報を公開している場合、ページインデックス、信頼度、オプションのバウンディングボックス、意味的なノード識別子などの引用アンカーを含む構造化ブロックを返します。

PDF 抽出を匿名のプレーンテキストとして扱うのではなく、「このテキストはどのページに由来するのか」「この値の根拠になっているのはどの表か」「これら 2 つの PDF の間で何が変わったのか」といった問いにパイプライン内で答える必要がある場合に、SDK を使用します。SDK は、PDF 抽出を匿名のプレーンテキストとして扱いません。

提供される機能

• スクリプト、バッチジョブ、ノートブック向けの同期 NextPDF クライアント。
• 非同期 AsyncNextPDF クライアント（asyncio、FastAPI、その他の非同期ランタイム向け）。
• ファイルパスまたは標準入力からワンショット抽出を実行し、標準出力またはファイルへ出力する nextpdf コマンドラインインターフェイス（CLI）。
• AI エージェントが PDF 抽出ツールを直接呼び出せるようにする、任意の Model Context Protocol（MCP）サーバー。
• NextPDF Connect を使用した本番利用向けのリモートバックエンド。
• pypdf を介した、オフラインかつライブラリのみの抽出用ローカルバックエンド。

バックエンドの選択

リモートバックエンドは、PDF のバイト列を NextPDF Connect サーバーに送信します。抽出の動作、認証、クォータ、運用上の制御を一元化できるため、本番環境ではこの経路が推奨されます。

ローカルバックエンドは Python プロセス内で実行され、pypdf を介して PDF を読み取ります。オフライン開発やタグ付き PDF の処理に役立ちますが、正確なバウンディングボックスは提供できず、タグなし PDF にはヒューリスティックな段落単位の抽出を使用します。ローカルバックエンドはライブラリ専用で、LocalBackend を AsyncNextPDF に注入することで利用します。nextpdf CLI と MCP サーバーでは使用できません。詳細な比較は、バックエンド選択マトリックスを参照してください。

制限事項

SDK は光学文字認識（OCR）を実行しません。スキャンされた PDF や画像のみの PDF では、NextPDF が埋め込みテキストを抽出する前に OCR 処理が必要です。複雑なレイアウト、重なり合うテキスト、特殊な PDF 生成元によっても、抽出品質が低下することがあります。

この nextpdf CLI はリモート専用であり、ストリーミングインターフェイスではありません。各コマンドは PDF 全体を（ファイルパスまたは標準入力から）メモリに読み込み、それを NextPDF Connect サーバーへ送信し、完全な結果をメモリ内に構築してから、単一の書き込みでシリアライズします。出力は --output（または -o）でファイルにリダイレクトすることも、標準出力へリダイレクトすることもできますが、結果は完全にバッファリングされ、段階的には生成されません。CLI はローカルの pypdf バックエンドを使用できません。

クライアントの選択: 同期と非同期

両方のクライアントは 1 つの ast メソッド名前空間を共有し、同じ Pydantic モデルを返します。違いは並行処理モデルです。

コンテキスト	使用するもの	理由
スクリプトとバッチジョブ	NextPDF（同期）	線形な制御フロー。管理すべきイベントループなし。
Jupyter ノートブック	NextPDF（同期）	run_sync が実行中のイベントループを検出してワーカースレッドへディスパッチするため、セル内でもブロッキング呼び出しが機能。
この nextpdf CLI	NextPDF（同期、内部）	CLI が同期クライアントを構築。
asyncio サービス	AsyncNextPDF	ネイティブな await。スレッドの受け渡しなし。
FastAPI、Starlette、ASGI 向け	AsyncNextPDF	リクエストのイベントループと同じコネクションプールを共有。
高並行のファンアウト	AsyncNextPDF	1 つのプール済みクライアント上で asyncio.gather を使い、多数の抽出を並行実行。

NextPDF は内部の AsyncNextPDF をラップし、各呼び出しを run_sync 経由で実行します。実行中のイベントループ内（たとえばノートブック）では、run_sync がコルーチンを独自のループを持つシングルワーカースレッドへディスパッチするため、ネストした asyncio.run のエラーは発生しません。asyncio または ASGI サービスでは、呼び出しごとにスレッド間の受け渡しコストを負うのではなく、AsyncNextPDF を直接呼び出してください。

非同期クライアントはコネクションプーリングのために httpx.AsyncClient を保持するため、1 つの AsyncNextPDF インスタンスを再利用し、一度だけクローズしてください。同期 NextPDF クライアントは close() メソッドを公開していません。長時間実行する非同期ワークロードでは、AsyncNextPDF を優先し、そのライフサイクルを明示的に管理してください（本番運用モデルを参照）。

バックエンド選択マトリックス

バックエンドは PdfBackend プロトコルを実装します。リモートバックエンド（RemoteBackend）は、base_url と api_key を渡すと自動的に選択されます。ローカルバックエンド（LocalBackend）は、backend= パラメーター経由で AsyncNextPDF に明示的に注入する必要があります。LocalBackend はトップレベルの nextpdf パッケージからエクスポートされておらず、CLI や MCP サーバーからは到達できません。

機能	リモート（RemoteBackend）	ローカル（LocalBackend）
選択方法	base_url + api_key	AsyncNextPDF(backend=LocalBackend(...))
ネットワーク	NextPDF Connect（HyperText Transfer Protocol Secure（HTTPS）を経由）	なし。プロセス内実行
認証、クォータ、計測	サーバー側で一元管理	なし
可観測性と運用上の制御	サーバー側	なし
タグ付き PDF（StructTree）の抽出	対応	対応
タグなし PDF の抽出	サーバーエンジン	ヒューリスティックな段落分割、信頼度 0.5
バウンディングボックス	対応（サーバーが提供する場合）	非対応（bbox は None）
タグなし PDF での表抽出	サーバーエンジン	表を返しません
CLI / MCP サーバーから到達可能	対応	非対応（ライブラリのみ）
推奨される用途	本番環境	オフライン開発、タグ付き PDF のテスト

本番環境ではリモートバックエンドを使用してください。一元化された認証、クォータの適用、計測、可観測性を提供する唯一の経路です。オフライン開発やタグ付き PDF に対するテストにはローカルバックエンドを使用し、タグなし入力ではヒューリスティックな結果、バウンディングボックスなし、表なしであることを受け入れてください。

"""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 に接続するリモートバックエンドを使用します。以下のパターンは、クライアントの再利用、エラー処理、再試行、クォータ処理、タイムアウトを扱います。ここで使用するすべてのシンボルは SDK に存在します。SDK は自動では再試行しないため、再試行ループはご自身で実装してください。

クライアントを再利用してコネクションをプールする

RemoteBackend は、コネクションプーリングのために永続的な httpx.AsyncClient を 1 つ保持します。AsyncNextPDF を一度だけ構築し、リクエスト間で共有し、シャットダウン時にクローズしてください。リクエストごとにクライアントを作成しないでください。

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

非同期コンテキストマネージャーは終了時に close() を呼び出し、基盤となるトランスポートをクローズします。コンテキストマネージャーを使用しない場合は、ご自身で await client.close() を呼び出してください。

例外階層でエラーを処理する

SDK は型付けされた例外階層を送出します。すべてのエラーは NextPDFError から派生します。HTTP レベルの失敗は NextPDFAPIError から派生し、status_code を保持します。処理できる具体的な型をキャッチし、基底型にフォールバックしてください。

例外	送出される条件	主な属性
NextPDFError	すべての SDK エラーの基底型	status_code
NextPDFAPIError	サーバーからの任意の HTTP エラー	status_code、error_code
NextPDFLicenseError	HTTP 402。機能にはより上位のサーバーティアが必要	status_code（402）
QuotaExceededError	HTTP 429。レート制限またはクォータ超過	retry_after
AstNoStructTreeError	HTTP 422。ヒューリスティックモードがオフのタグなし 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 失敗を再試行し、サーバーの 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")

クォータ、レート制限、タイムアウトを管理する

クォータとレート制限はサーバー上で適用されます。HTTP 429 では、SDK は QuotaExceededError を送出し、Retry-After ヘッダーを retry_after に解析します。リモートバックエンドはレンダリングのレスポンスで X-RateLimit-* ヘッダーも公開するため、ハードリミットに達する前に事前にスロットリングできます。

リクエストのタイムアウトには、合計 60 秒で接続タイムアウトが 10 秒という固定のデフォルト値を使用します（httpx.Timeout(60.0, connect=10.0)）。長い AST の構築を制限するには、タイムアウトだけに頼るのではなく、page_range_start、page_range_end、または token_budget で作業範囲を絞り込むことを推奨します。構築が長すぎる場合は AstBuildTimeoutError（HTTP 504）を返します。

アーキテクチャの例

バッチジョブ

バッチワーカーは PDF を読み取り、引用付きテキストを抽出し、構造化された出力を書き込みます。1 つのプール済みクライアントを再利用し、セマフォで並行数を制限し、前述の再試行ヘルパーを適用してください。

"""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 サービスは、アプリケーションのライフスパン全体で 1 つの AsyncNextPDF をリクエスト間で共有するため、すべてのリクエストがコネクションプールを再利用します。認証情報は環境から読み取り、API キーはシークレットとして扱ってください。

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

エージェントツール

AI エージェント向けには、MCP サーバーを実行します。これは PDF ツール（たとえば nextpdf_extract_text、nextpdf_extract_tables、nextpdf_get_ast、nextpdf_info、nextpdf_search、nextpdf_get_outline、nextpdf_diff、nextpdf_health）を標準 input/output 経由で公開します。サーバーは環境から NEXTPDF_BASE_URL と NEXTPDF_API_KEY を読み取るため、リモートバックエンドに依存します。CLI と同様に、ローカルバックエンドは使用できません。オプションの追加パッケージをインストールし、モジュールを実行してください。

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

エージェント統合のチュートリアルについては Python MCP サーバーを、ターミナルでの使用方法については Python CLIを、クライアント、モデル、例外の全体については Python API リファレンスを参照してください。