Python SDK 開発者ガイド

概要

NextPDF Python Software Development Kit (SDK) は、NextPDF Connect エンドポイント向けの薄い型付きクライアントです。入力の検証、認証情報の取り扱い、並行処理ポリシーはアプリケーションが担い、リクエストの構築、トランスポート、レスポンスの型付けは SDK が担います。この境界を明確にしてください。PDF を安全に読み込み、クライアントを選択し、必要な ast メソッドを呼び出して、個別の障害を処理します。

このガイドは、SDK を中心に抽出サービス、asyncio バッチジョブ、AI エージェントツール、またはコマンドラインワークフローを構築する際に参照してください。概要とクイックスタートを読み終えており、Python 3.10 以降と NextPDF Connect エンドポイントを用意していることを前提とします。

アーキテクチャの境界

レイヤー	担当	責務	ここに置かないもの
入力ソース	アプリケーション	呼び出し元の認可、PDF ソースの検証、抽出ポリシーの選択。	エンドポイント URL や認証情報のリテラル。
クライアントの構築	アプリケーション	環境変数またはシークレットマネージャーからの `base_url` と `api_key` の読み取り。	ハードコードされたシークレット。
`NextPDF` / `AsyncNextPDF`	SDK	リクエストの構築、Connect の呼び出し、型付けされた Pydantic モデルの返却。	ドメインロジックやストレージポリシー。
`ast` メソッド名前空間	SDK	メソッド呼び出しの Connect エンドポイントへのマッピング、レスポンスの解析。	設定範囲を超えるリトライやバックオフのポリシー。
NextPDF Connect エンドポイント	デプロイ	抽出の実行、認証、クォータ、ライセンスの適用。	アプリケーションの認可。

SDK は光学文字認識 (OCR) を一切実行しません。スキャンされた PDF や画像のみの PDF には、抽出の前に OCR の手順が必要です。これはこの境界の外にある、アプリケーション側の関心事として扱ってください。

ランタイムのライフサイクル

ステージ	動作	開発者の対応
クライアントの構築	`base_url` と `api_key` の検証。空の値では `ValueError` が発生。	どちらも環境変数から読み取り、決してインライン化しないこと。
バックエンドの作成	リモートバックエンドによる Connect へのプール接続のオープン。	リクエストごとに構築せず、1 つのクライアントを複数の呼び出しで再利用。
メソッド呼び出し	`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 の検証は抽出から分離してください。抽出レイヤーは、認可済みでサイズチェック済みのバイト列を受け取り、さらに多層防御としてエンドポイントに依存する必要があります。

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)

同期クライアントのパターン

スクリプト、バッチジョブ、ノートブックでは、同期の 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

1 つの結果項目に期待される形状は次のとおりです。

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

非同期とバッチ処理のパターン

FastAPI などの asyncio ランタイム内では、非同期の AsyncNextPDF クライアントを使用します。1 つのクライアントを非同期コンテキストマネージャーとして構築し、並行する呼び出し全体で共有します。ドキュメントごとにクライアントを開かないでください。エンドポイントのクォータを守れるよう、セマフォで並行数を制限します。

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=...)`	テストでのカスタムバックエンドやローカルバックエンドの注入。	バックエンドは `PdfBackend` プロトコルを満たす必要あり。
`api_version` 引数	Connect API のバージョンの固定。	既定は `v1`。エンドポイントが対象のバージョンをサポートしている場合にのみ変更。
環境設定	CLI と MCP サーバーへの `NEXTPDF_BASE_URL` と `NEXTPDF_API_KEY` の受け渡し。	鍵はワークロードにスコープを限定したシークレットとして扱うこと。
MCP サーバー (`python -m nextpdf.mcp`)	抽出ツールの MCP 対応エージェントへの公開。	動作には `nextpdf[mcp]` エクストラと管理されたエンドポイントが必要。

開発ワークフロー

SDK は pip install nextpdf でインストールします。エージェントサーバーが必要な場合は pip install nextpdf[mcp] を使用します。
シークレットがソース管理に入らないよう、NEXTPDF_BASE_URL と NEXTPDF_API_KEY は環境変数から読み取ります。
SDK を呼び出す前に、すべての PDF ソースについて、存在、サイズ、%PDF- のマジックバイトを検証します。
プロセスごとに 1 つのクライアントを構築し、再利用します。asyncio の場合は、async with で開いたまま保持します。
タスクに対して最も範囲の狭い ast メソッドを呼び出します。本文には extract_cited_text()、テーブルには extract_cited_tables()、ツリー全体が必要な場合にのみ get_document_ast() を使用します。
対処できる最も具体的な例外をキャッチし、その後 NextPDFError にフォールバックします。
100 MiB を超えるドキュメントでは、すべてのブロックをメモリ上に展開するのではなく、CLI のストリーミングパスを使用します。
mypy の strict モードで型チェックを実行し、処理する各例外に対して障害モードのテストを追加します。

障害処理

障害	例外	推奨される対応
タグなし PDF、ヒューリスティックが無効	`AstNoStructTreeError` (HTTP 422)	エンドポイントでヒューリスティックモードを有効化するか、タグ付き PDF を受け渡し。
サーバー側のビルドタイムアウト	`AstBuildTimeoutError` (HTTP 504)	ページ範囲を狭めて再試行。
必要なライセンス階層	`NextPDFLicenseError` (HTTP 402)	サーバーのライセンスのアップグレード、または許可されている機能へのフォールバック。
レート制限またはクォータ	`QuotaExceededError` (HTTP 429)	待機時間として `retry_after` 秒待ってから、バックオフを伴って再試行。
その他の HTTP エラー	`NextPDFAPIError`	応答の `status_code` と `error_code` の確認、定義されたエラーのログ記録と提示。
あらゆる SDK エラー	`NextPDFError`	最終的なフォールバック。未処理例外として外部に漏れないようにすること。

エンドポイントは、RFC 9110 に準拠した HTTP ステータスのセマンティクスと、RFC 9457 に準拠した機械可読のエラーボディで障害を報告します。各例外は、発生元の status_code を保持します。トランスポートの詳細を呼び出し元に漏らすのではなく、これらを独自のエラーレスポンスにマッピングします。

安全な既定値

関心事	既定値	オーバーライドするタイミング
API バージョン	`v1`。	エンドポイントがより新しいバージョンをサポートしている場合にのみ変更。
TLS の検証	有効。安全でない切り替えスイッチは非公開。	本番トラフィックでは決して無効にしないこと。
認証情報	環境変数から読み取り、決してインライン化しない。	本番ではシークレットマネージャーを使用。
インメモリのサイズ制限	クライアントパスでは 100 MiB を超える PDF を拒否。	マルチテナントサービスでは引き下げ、より大きなファイルには CLI を使用。
並行処理	非同期バッチではセマフォで制限。	ホストのコア数ではなく、エンドポイントのクォータに合わせて調整。
ロギング	ファイル名、サイズ、ステータス、処理時間をログに記録。	PDF のバイト列や API キーは決してログに記録しないこと。

テストチェックリスト

構築テストでは、空の base_url または api_key で ValueError が発生することを検証します。
検証テストでは、存在しない入力、空の入力、サイズ超過の入力、PDF 以外の入力を網羅します。
抽出テストでは、返されるモデルの型と、各ブロックに CitationAnchor が存在することを検証します。
障害モードテストでは、AstNoStructTreeError、AstBuildTimeoutError、NextPDFLicenseError、QuotaExceededError、NextPDFAPIError を網羅します。
非同期テストでは、クライアントが async with コンテキストマネージャーとして使用されること、および並行数がセマフォの上限内に収まることを検証します。
ライフサイクルテストでは、close() がトランスポートを解放し、冪等であることを検証します。
フェイクバックエンドは AsyncNextPDF(backend=...) で注入し、稼働中のエンドポイントがなくてもテストを実行できるようにします。