Python SDK 개발자 가이드

한눈에 보기

NextPDF Python 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는 추출 전에 OCR 단계가 필요합니다. 이는 이 경계 밖에 있는 애플리케이션의 관심사로 다루십시오.

런타임 수명 주기

단계	동작	개발자 조치
클라이언트 구성	`base_url`과 `api_key`가 검증되며, 빈 값은 `ValueError`를 발생시킵니다.	둘 다 환경에서 읽으십시오. 절대로 인라인으로 작성하지 마십시오.
백엔드 생성	원격 백엔드가 Connect에 대한 풀링된 연결을 엽니다.	요청마다 생성하지 말고 하나의 클라이언트를 여러 호출에 재사용하십시오.
메서드 호출	이 단계에서 `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

결과 항목 하나는 다음과 같은 형태입니다:

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 클라이언트를 사용하십시오. 하나의 클라이언트를 비동기 컨텍스트 관리자로 구성하고 동시 호출 전체에서 공유하십시오. 문서마다 클라이언트를 열지 마십시오. 세마포어로 동시성을 제한하여 엔드포인트의 할당량을 준수하십시오.

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- 매직 바이트)를 검증합니다.
프로세스당 하나의 클라이언트를 생성하여 재사용합니다. asyncio의 경우 async with으로 열린 상태를 유지합니다.
작업에 맞는 가장 좁은 ast 메서드를 호출합니다. 산문에는 extract_cited_text(), 표에는 extract_cited_tables(), 전체 트리가 필요할 때만 get_document_ast()를 사용합니다.
대응할 수 있는 가장 구체적인 예외를 캐치한 다음, NextPDFError로 폴백합니다.
100MiB를 초과하는 문서의 경우, 모든 블록을 메모리에 구체화하지 말고 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 검증	활성화됨. 안전하지 않은 스위치는 노출되지 않습니다.	프로덕션 트래픽에서는 절대로 비활성화하지 마십시오.
자격 증명	환경에서 읽습니다. 절대로 인라인으로 작성하지 않습니다.	프로덕션에서는 시크릿 관리자를 사용하십시오.
인메모리 크기 제한	클라이언트 경로에서는 100MiB를 초과하는 PDF를 거부합니다.	멀티 테넌트 서비스의 경우 더 낮추십시오. 더 큰 파일에는 CLI를 사용하십시오.
동시성	비동기 배치에서 세마포어로 제한됩니다.	호스트의 코어 수가 아니라 엔드포인트의 할당량에 맞춰 조정하십시오.
로깅	파일 이름, 크기, 상태, 소요 시간을 로깅합니다.	절대로 PDF 바이트나 API 키를 로깅하지 마십시오.

테스트 체크리스트

구성 테스트는 base_url 또는 api_key가 비어 있으면 ValueError가 발생하는지 단언합니다.
유효성 검사 테스트는 누락, 빈 값, 크기 초과, PDF가 아닌 입력을 다룹니다.
추출 테스트는 반환된 모델 타입과 각 블록에 CitationAnchor가 있는지 단언합니다.
실패 모드 테스트는 AstNoStructTreeError, AstBuildTimeoutError, NextPDFLicenseError, QuotaExceededError, NextPDFAPIError를 다룹니다.
비동기 테스트는 클라이언트가 async with 컨텍스트 관리자로 사용되고 동시성이 세마포어 제한 내에 머무는지 단언합니다.
수명 주기 테스트는 close()가 전송을 해제하고 멱등적인지 단언합니다.
테스트에서 AsyncNextPDF(backend=...)로 가짜 백엔드를 주입하여 라이브 엔드포인트 없이 실행되도록 합니다.