Python SDK 개요

Python SDK 개요

NextPDF Python SDK는 출처 정보가 포함된 PDF 추출이 필요한 Python 애플리케이션을 위한 것입니다. 소스 PDF가 해당 구조를 노출하면 페이지 인덱스, 신뢰도, 선택적 경계 상자, 의미론적 노드 식별자 같은 인용 앵커가 포함된 구조화된 블록을 반환합니다.

파이프라인이 “이 텍스트는 어느 페이지에서 왔는가?”, “어느 표가 이 값을 뒷받침하는가?” 또는 “이 두 PDF 사이에 무엇이 바뀌었는가?”와 같은 질문에 답해야 하고, PDF 추출을 익명의 일반 텍스트로 취급하지 않아야 할 때 SDK를 사용하십시오.

제공 기능

• 스크립트, 배치 작업, 노트북을 위한 동기 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 백엔드를 사용할 수 없습니다.

클라이언트 선택: 동기 vs 비동기

두 클라이언트는 동일한 ast 메서드 네임스페이스를 공유하며 동일한 Pydantic 모델을 반환합니다. 차이는 동시성 모델입니다.

사용 환경	사용	이유
스크립트 및 배치 작업	NextPDF (동기)	선형 제어 흐름. 관리할 이벤트 루프가 없습니다.
Jupyter 노트북	NextPDF (동기)	run_sync는 실행 중인 이벤트 루프를 감지하여 작업자 스레드로 디스패치하므로 셀 안에서도 블로킹 호출이 작동합니다.
이 nextpdf CLI	NextPDF (동기, 내부)	CLI가 동기 클라이언트를 자동으로 생성합니다.
asyncio 서비스	AsyncNextPDF	네이티브 await. 스레드 핸드오프가 없습니다.
FastAPI, Starlette, ASGI 환경	AsyncNextPDF	요청 이벤트 루프와 동일한 연결 풀을 공유합니다.
고동시성 팬아웃	AsyncNextPDF	하나의 풀링된 클라이언트에서 asyncio.gather를 사용하여 여러 추출을 동시에 실행합니다.

NextPDF는 내부 AsyncNextPDF를 래핑하고 각 호출을 run_sync를 통해 실행합니다. 실행 중인 이벤트 루프 내부(예: 노트북)에서 run_sync는 코루틴을 자체 루프가 있는 단일 작업자 스레드로 디스패치하므로, 중첩 asyncio.run 오류가 발생하지 않습니다. asyncio 또는 ASGI 서비스에서는 매 호출마다 이런 스레드 핸드오프 비용을 감수하지 말고 AsyncNextPDF를 직접 호출하십시오.

비동기 클라이언트는 연결 풀링을 위해 httpx.AsyncClient를 보유하므로, 하나의 AsyncNextPDF 인스턴스를 재사용하고 한 번만 닫으십시오. 동기 NextPDF 클라이언트는 close() 메서드를 노출하지 않습니다. 수명이 긴 비동기 워크로드의 경우, AsyncNextPDF를 사용하고 그 수명 주기를 명시적으로 관리하는 것이 좋습니다(프로덕션 운영 모델 참조).

백엔드 선택 매트릭스

백엔드는 PdfBackend 프로토콜을 구현합니다. 원격 백엔드(RemoteBackend)는 base_url과 api_key를 전달하면 자동으로 선택됩니다. 로컬 백엔드(LocalBackend)는 backend= 매개변수를 통해 AsyncNextPDF에 명시적으로 주입해야 합니다. 이는 최상위 nextpdf 패키지에서 내보내지지 않으며 CLI나 MCP 서버에서 접근할 수 없습니다.

기능	원격 (RemoteBackend)	로컬 (LocalBackend)
선택 방법	base_url + api_key	AsyncNextPDF(backend=LocalBackend(...))
네트워크	HyperText Transfer Protocol Secure(HTTPS)를 통한 NextPDF Connect	없음. 프로세스 내에서 실행됩니다
인증, 할당량, 미터링	서버에서 중앙 집중식으로 관리	없음
관찰 가능성 및 운영 제어	서버 측	없음
태그가 있는 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 하나를 유지합니다. 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를 읽고, 인용 텍스트를 추출하며, 구조화된 출력을 작성합니다. 풀링된 클라이언트 하나를 재사용하고, 세마포어로 동시성을 제한하며, 위의 재시도 헬퍼를 적용하십시오.

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

웹 서비스

FastAPI 서비스는 애플리케이션 수명 동안 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 참조를 참조하십시오.