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_toparameter to write responses to disk
pip install landingai-adeGenerate 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.
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.
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 processedThe response is a V2ParseResponse:
| Field | Description |
|---|---|
markdown |
The full document as one Markdown string, in reading order. |
structure |
A typed tree (document → pages → elements); each node carries its spatial grounding (page, character range into markdown, and a normalized-coordinate bounding box). |
grounding |
Legacy tree mirroring structure, present only on older gateway responses — newer responses carry grounding inline on structure nodes. |
metadata |
Processing details: page_count, failed_pages, duration_ms, and billing (credits used). |
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.
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.
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 MarkdownThe response is a V2ExtractResult:
| Field | Description |
|---|---|
extraction |
The extracted values, matching your schema. |
extraction_metadata |
Mirrors extraction; each field carries the character ranges in the Markdown that the value came from. |
markdown |
The Markdown the extraction ran against, echoed back. |
metadata |
Processing details, including credits used. |
By default, unsupported schema fields are skipped and extraction continues. Pass strict=True to reject such schemas with an error (HTTP 422) instead.
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.
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())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".
The v1 methods sit directly on the client.
| Method | What it does |
|---|---|
client.parse(...) |
Parse a document with the DPT-2 model family. |
client.extract(...) |
Extract fields from Markdown. |
client.split(...) |
Split a multi-document file into sub-documents by classification. |
client.classify(...) |
Classify each page of a document. |
client.section(...) |
Generate a hierarchical table of contents. |
client.extract_build_schema(...) |
Generate an extraction schema from sample documents. |
client.parse_jobs, client.extract_jobs |
Async jobs (create, get, list). |
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)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.APIStatusErrorwithstatus_codeandresponseproperties.
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)| Status Code | Error Type |
|---|---|
| 400 | BadRequestError |
| 401 | AuthenticationError |
| 403 | PermissionDeniedError |
| 404 | NotFoundError |
| 422 | UnprocessableEntityError |
| 429 | RateLimitError |
| >=500 | InternalServerError |
| N/A | APIConnectionError |
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-requestRequests 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-requestOn 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.
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 returnedUse .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.
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-AgentNested 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.
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.')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.
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=...).
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 exitSet the LANDINGAI_ADE_LOG environment variable to info (or debug for more detail):
export LANDINGAI_ADE_LOG=infoThis package generally follows SemVer conventions, though certain backwards-incompatible changes may be released as minor versions:
- Changes that only affect static types, without breaking runtime behavior.
- 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.)
- 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__)Python 3.9 or higher.
See the contributing documentation. We welcome issues with questions, bugs, or suggestions.