Agentic Document Extraction Python Library

Docs · Playground · LandingAI

The official Python library for the LandingAI Agentic Document Extraction (ADE) API. Parse PDFs and images into structured, grounded Markdown, then extract typed fields with a JSON Schema or Pydantic model.

• Fully typed requests and Pydantic response models
• Sync and async clients with identical surfaces
• Async jobs with a built-in wait() helper for large documents
• Automatic retries with exponential backoff
• Optional save_to parameter to write responses to disk

Installation

pip install landingai-ade

Set Your API Key

Generate an API key, then export it as an environment variable. The client reads it automatically.

export VISION_AGENT_API_KEY=<your-api-key>

You can also pass the key directly with LandingAIADE(apikey=...). To keep keys out of source control, use a tool like python-dotenv.

Quickstart

Parse a document, then extract structured data from it:

from pathlib import Path
from pydantic import BaseModel, Field
from landingai_ade import LandingAIADE


class Invoice(BaseModel):
    invoice_number: str = Field(description="The invoice number")
    total: str = Field(description="Invoice grand total")


client = LandingAIADE()  # reads VISION_AGENT_API_KEY

# 1. Parse: convert the document to structured Markdown
parsed = client.v2.parse(document=Path("invoice.pdf"))
print(parsed.markdown)

# 2. Extract: pull typed fields out of the Markdown
result = client.v2.extract(schema=Invoice, markdown=parsed.markdown)
print(result.extraction)

Use client.v2 for new projects. It is the current API, powered by the DPT-3 model family. The earlier v1 methods (client.parse, client.extract, client.split, and others) remain fully supported; see v1 API.

The full method reference for both APIs is in api.md; usage guides are at docs.landing.ai.

Parse

Use client.v2.parse to convert a document into Markdown plus a structure tree and grounding (pixel-coordinate bounding boxes for every element). Provide exactly one of document (a local file) or document_url.

from pathlib import Path
from landingai_ade import LandingAIADE

client = LandingAIADE()

# Parse a local file
parsed = client.v2.parse(
    document=Path("path/to/file.pdf"),
    model="dpt-3-pro-latest",  # optional; defaults to the latest DPT-3 Pro model
    save_to="./output",  # optional; saves as {input_file}_parse_output.json
)

# Or parse a file at a URL
parsed = client.v2.parse(document_url="https://example.com/file.pdf")

print(parsed.markdown)  # full document as Markdown
print(parsed.metadata.page_count)  # pages processed

The response is a V2ParseResponse:

If some pages cannot be parsed, the request still succeeds (HTTP 206) and metadata.failed_pages lists the pages that failed. If a synchronous parse times out, the client raises V2SyncTimeoutError; use jobs instead.

Encrypted PDFs

Password-protected PDFs parse directly — pass password and skip decrypting the file yourself. The document is decrypted once at the start of processing, and the password is not retained with the result.

import os
from pathlib import Path

parsed = client.v2.parse(
    document=Path("locked.pdf"),
    password=os.environ["PDF_PASSWORD"],
)

password applies to PDFs only, and the server rejects the three mistakes with a 422, each naming its case: password_unsupported_content_type (a password sent with an image or an Office document), encrypted_pdf_wrong_password (the password does not open the PDF), and encrypted_pdf_password_required (a locked PDF submitted without one).

password is shorthand for the contract field options["password"], which is where the SDK puts it on the wire — and the only place it puts it. Both forms work; if you supply both, the explicit options["password"] wins:

# sends options["password"] = "from-options"
client.v2.parse(
    document=Path("locked.pdf"),
    options={"password": "from-options"},
    password="ignored",
)

That applies to an explicit None too: options={"password": None} means "no password" and silences the password argument behind it. options itself accepts a mapping or a JSON string that decodes to an object. Malformed JSON raises json.JSONDecodeError; a value that decodes to a non-object raises TypeError. Both are raised before the request is sent.

ade-typescript aligns on the same precedence rule in #121.

Extract

Use client.v2.extract to pull structured fields out of Markdown (typically from a parse response) using a schema. The schema parameter accepts a Pydantic BaseModel subclass, a dict, or a JSON string. Provide exactly one Markdown source: markdown or markdown_url.

from pathlib import Path
from pydantic import BaseModel, Field
from landingai_ade import LandingAIADE


class Person(BaseModel):
    name: str = Field(description="Person's full name")
    age: int = Field(description="Person's age")


client = LandingAIADE()
parsed = client.v2.parse(document=Path("path/to/file.pdf"))

result = client.v2.extract(
    schema=Person,  # Pydantic model, dict, or JSON string
    markdown=parsed.markdown,  # or markdown_url="https://example.com/doc.md"
    save_to="./output",  # optional
)

print(result.extraction)  # {"name": "...", "age": ...}
print(result.extraction_metadata)  # per-field source ranges in the Markdown

The response is a V2ExtractResult:

By default, unsupported schema fields are skipped and extraction continues. Pass strict=True to reject such schemas with an error (HTTP 422) instead.

Pass grounding=False to skip the grounding stage: extraction runs as usual and finishes faster, but every extraction_metadata field comes back with ranges: None instead of pointing into the Markdown. Preview — with grounding off the extraction itself can differ slightly from a grounded run. extract_jobs.create takes both options too.

Process Large Documents Asynchronously (Jobs)

For documents that take longer than a synchronous request allows, create a job and wait for it. client.v2.parse_jobs and client.v2.extract_jobs share the same shape: create, get, list, and wait.

from pathlib import Path
from landingai_ade import LandingAIADE
from landingai_ade.lib.v2_errors import JobFailedError, JobWaitTimeoutError

client = LandingAIADE()

job = client.v2.parse_jobs.create(
    document=Path("path/to/large_file.pdf"),
    service_tier="standard",  # "standard" (default, lower cost) or "priority" (faster)
)
print(job.job_id, job.status)

# Block until the job finishes (polls with backoff)
try:
    done = client.v2.parse_jobs.wait(job.job_id, timeout=600, raise_on_failure=True)
    if done.result is not None:  # a cancelled job can be terminal with no result
        print(done.result.markdown[:200])
except JobWaitTimeoutError:
    print("Job did not finish in time; it is still running server-side.")
except JobFailedError as e:
    print(f"Job failed: {e}")

The create, get, and wait methods return a normalized Job with job_id, status (pending, processing, completed, failed, or cancelled), progress, result, error, is_terminal, and raw (the unmodified API envelope, for any field not surfaced on the typed model). The list method returns a JobList, a list of Job items that also carries has_more. Parse job lists also populate org_id, and extract job lists also populate page and page_size; fields an endpoint doesn't populate are None.

# Poll manually instead of blocking
job = client.v2.parse_jobs.get(job.job_id)

# List jobs, with optional filtering
jobs = client.v2.parse_jobs.list(status="completed", page=0, page_size=10)
for job in jobs:
    print(job.job_id, job.status)
print(jobs.has_more)

Extract jobs work the same way. The create method takes the same schema and Markdown arguments as client.v2.extract, plus service_tier; it does not accept save_to.

Async Client

Import AsyncLandingAIADE and await each call. The async client mirrors the entire sync surface, including client.v2.

import asyncio
from pathlib import Path
from landingai_ade import AsyncLandingAIADE


async def main() -> None:
    async with AsyncLandingAIADE() as client:
        parsed = await client.v2.parse(document=Path("path/to/file.pdf"))
        print(parsed.markdown)


asyncio.run(main())

For higher concurrency, you can use aiohttp as the HTTP backend instead of the default httpx:

pip install landingai-ade[aiohttp]

import asyncio
from landingai_ade import AsyncLandingAIADE, DefaultAioHttpClient


async def main() -> None:
    async with AsyncLandingAIADE(http_client=DefaultAioHttpClient()) as client:
        ...  # same usage as the example above


asyncio.run(main())

Environments

The environment argument selects the region. Set it in code or with the LANDINGAI_ADE_ENVIRONMENT environment variable.

from landingai_ade import LandingAIADE

client = LandingAIADE(environment="eu")  # "production" (default) or "eu"

API keys are per-environment: an EU key works only with environment="eu".

v1 API

The v1 methods sit directly on the client.

import json
from pathlib import Path
from landingai_ade import LandingAIADE

client = LandingAIADE()

# Split a combined file into sub-documents
parsed = client.parse(document=Path("statements.pdf"), model="dpt-2-latest")
split = client.split(
    split_class=json.dumps(
        [
            {"name": "Bank Statement", "description": "Summarizes account activity over a period."},
            {"name": "Pay Stub", "description": "Details an employee's earnings for a pay period."},
        ]
    ),
    markdown=parsed.markdown,
    model="split-latest",
)
for s in split.splits:
    print(s.classification, s.pages)

Handling Errors

All HTTP errors inherit from landingai_ade.APIError.

• Connection problems raise a subclass of landingai_ade.APIConnectionError.
• Non-success HTTP status codes (4xx, 5xx) raise a subclass of landingai_ade.APIStatusError with status_code and response properties.

The v2 helper exceptions are separate: V2SyncTimeoutError, JobWaitTimeoutError, and JobFailedError (importable from landingai_ade.lib.v2_errors) inherit from LandingAiadeError rather than APIError, so catch them explicitly as shown in the jobs example.

import landingai_ade
from landingai_ade import LandingAIADE
from landingai_ade.lib.v2_errors import V2SyncTimeoutError

client = LandingAIADE()

try:
    client.v2.parse(document_url="https://example.com/file.pdf")
except landingai_ade.APIConnectionError as e:
    print("The server could not be reached")
    print(e.__cause__)
except V2SyncTimeoutError:
    print("The synchronous request timed out; use parse_jobs for this document.")
except landingai_ade.RateLimitError:
    print("A 429 status code was received; back off and retry.")
except landingai_ade.APIStatusError as e:
    print(e.status_code)
    print(e.response)

Retries

Connection errors, 408, 409, 429, and 5xx responses are retried twice by default with exponential backoff. Configure with max_retries:

client = LandingAIADE(max_retries=0)  # default is 2
client.with_options(max_retries=5).v2.parse(...)  # per-request

Timeouts

Requests time out after 8 minutes by default. Configure with timeout (a float or an httpx.Timeout):

client = LandingAIADE(timeout=20.0)  # seconds
client.with_options(timeout=5.0).v2.parse(...)  # per-request

On a client-side transport timeout, an APITimeoutError is raised, and the request is retried twice by default. The v2 synchronous endpoints also have a server-side wait window: exceeding it returns HTTP 504 and raises V2SyncTimeoutError instead; switch to jobs for those documents.

Advanced Usage

Accessing raw response data (e.g. headers)

Prefix any v1 method call with .with_raw_response. to get the raw HTTP response (these wrappers cover the v1 methods only, not client.v2):

response = client.with_raw_response.parse(document=Path("file.pdf"), model="dpt-2-latest")
print(response.headers.get("X-My-Header"))
parsed = response.parse()  # the object the method would have returned

Use .with_streaming_response instead (also v1 methods only) to stream the body rather than reading it eagerly; it requires a context manager and reads the body only when you call .read(), .text(), .json(), .iter_bytes(), .iter_text(), .iter_lines(), or .parse(). These return APIResponse (or AsyncAPIResponse) objects.

Client identity headers

Every request carries two headers that let the platform attribute traffic to this SDK:

• X-Source: sdk
• a structured User-Agent, e.g. ade-python/1.15.0 (Darwin arm64) python/3.12 httpx/0.28.1

You can override either one by passing the same header (using its canonical name) via default_headers on the client or extra_headers on a call — a caller-supplied value replaces the SDK default. Leave them untouched to keep accurate attribution.

client = LandingAIADE(default_headers={"User-Agent": "my-app/1.0"})  # replaces the SDK User-Agent

Nested params and file uploads

Nested request parameters are TypedDicts; responses are Pydantic models with helpers such as model.to_json() and model.to_dict(). File upload parameters accept bytes, a PathLike instance, or a (filename, contents, media type) tuple; the async client reads PathLike files asynchronously.

How to tell whether None means null or missing

if response.my_field is None:
    if "my_field" not in response.model_fields_set:
        print('The "my_field" key was absent from the response.')
    else:
        print('The response contained "my_field": null.')

Making custom or undocumented requests

Use client.get / client.post for undocumented endpoints (client options such as retries still apply), and extra_query, extra_body, or extra_headers for undocumented parameters. Undocumented response properties are available via response.unknown_prop or response.model_extra.

Configuring the HTTP client

Override the httpx client for proxies, custom transports, or other advanced behavior:

import httpx
from landingai_ade import LandingAIADE, DefaultHttpxClient

client = LandingAIADE(
    http_client=DefaultHttpxClient(
        proxy="http://my.test.proxy.example.com",
        transport=httpx.HTTPTransport(local_address="0.0.0.0"),
    ),
)

You can also change it per-request with client.with_options(http_client=...).

Managing HTTP resources

The client closes HTTP connections when garbage collected. Close it explicitly with .close(), or use a context manager:

with LandingAIADE() as client:
    ...  # connections close on exit

Logging

Set the LANDINGAI_ADE_LOG environment variable to info (or debug for more detail):

export LANDINGAI_ADE_LOG=info

Versioning

This package generally follows SemVer conventions, though certain backwards-incompatible changes may be released as minor versions:

1. Changes that only affect static types, without breaking runtime behavior.
2. Changes to library internals which are technically public but not intended or documented for external use. (Please open a GitHub issue to let us know if you are relying on such internals.)
3. Changes that we do not expect to impact the vast majority of users in practice.

We take backwards-compatibility seriously and work hard to ensure you can rely on a smooth upgrade experience.

To check the version in use at runtime:

import landingai_ade

print(landingai_ade.__version__)

Requirements

Python 3.9 or higher.

Contributing

See the contributing documentation. We welcome issues with questions, bugs, or suggestions.