Extracting structured data

Extract takes a document and a JSON Schema, and returns the fields described by the schema as a structured object. You define what you want in the schema, the API finds it in the document: invoices, forms, ID documents, contracts, anything. By default the call is synchronous: you send the request, the API blocks until the extraction is done, and the response comes back with the full result. For larger documents, opt into async mode with options.async = true: you get back a job ID, and you poll until done.

Full request/response schema for POST /api/v3/extract and GET /api/v3/extract/{job_id} lives in the API reference.

When to use Extract

	Input	Output	Best for
Search	Query string	Ranked text chunks	Finding passages across many documents
Parse	Document file	Markdown text	Converting a document to clean text
Extract	Document + JSON Schema	Structured object	Pulling typed fields from forms, invoices, contracts

Use Extract when you need machine-readable values out of a document, mapped to fields you’ve named.

Extract applies your schema independently to every page of the document: result.data comes back with one object per page. That makes it ideal for mechanical, repetitive document processing, where each page is a self-contained record (a stack of invoices, a batch of forms, a multi-page table) and you want the same fields pulled from every one.If instead you need a single structured JSON for the whole document (synthesizing information spread across pages into one consolidated object), Extract is the wrong tool. Use the POST /api/v3/search endpoint inside your own agentic loop to retrieve the relevant passages, then have your model generate the structured output from them.

Sync extraction (small documents)

Sync mode handles documents up to 20 MB, 15 pages and returns the full result in one response.

from lighton import LightOn
from pydantic import BaseModel, Field


class Invoice(BaseModel):
    invoice_number: str = Field(description="The invoice reference number")
    total: float = Field(description="The total amount due")
    due_date: str | None = Field(None, description="Due date in ISO format")


with LightOn() as client:  # reads LIGHTON_API_KEY from the environment
    # A pydantic model is converted to the guided-generation schema for you.
    # Field descriptions steer the model, so write them as instructions.
    response = client.extract(
        schema=Invoice,
        url="https://example.com/invoices/inv-2025-004.pdf",
    )
    print(response.status)
    print(response.result.data)  # one object per page

import os
import requests

headers = {"Authorization": f"Bearer {os.environ['LIGHTON_API_KEY']}"}

response = requests.post(
    "https://api.lighton.ai/api/v3/extract",
    headers=headers,
    json={
        "document": "https://example.com/invoices/inv-2025-004.pdf",
        "schema": {
            "type": "object",
            "properties": {
                "invoice_number": {"type": "string", "description": "The invoice reference number"},
                "total":          {"type": "number", "description": "The total amount due"},
                "due_date":       {"type": "string", "description": "Due date in ISO format"},
            },
        },
    },
)

result = response.json()
print(result["status"])              # → completed
print(result["result"]["data"])      # one object per page

result.data holds one object per page, for example [{"invoice_number": "INV-2025-004", "total": 4750.0, "due_date": "2025-12-01"}, ...]. You can also send a file via multipart/form-data instead of a URL:

from lighton import LightOn

with LightOn() as client:
    # `schema` also takes a raw JSON Schema dict, sent as-is after validation
    response = client.extract(
        schema={"type": "object", "properties": {"invoice_number": {"type": "string"}}},
        path="invoice.pdf",
    )
    print(response.result.data)

import os
import requests

headers = {"Authorization": f"Bearer {os.environ['LIGHTON_API_KEY']}"}

with open("invoice.pdf", "rb") as f:
    response = requests.post(
        "https://api.lighton.ai/api/v3/extract",
        headers=headers,
        files={"file": f},
        data={"schema": '{"type": "object", "properties": {"invoice_number": {"type": "string"}}}'},
    )
print(response.json()["result"]["data"])

In multipart requests, schema arrives as a JSON-encoded string and is decoded server-side.

Extracting from a document you already ingested

If the document is already in the platform, you don’t need to re-upload it. Pass file_id instead of document or a file part (file in the SDK, which also takes a File object), and the API extracts from the stored copy. Get the id from GET /api/v3/files, or from the response of the POST /api/v3/files call that ingested it.

from lighton import LightOn

with LightOn() as client:
    response = client.extract(
        schema={
            "type": "object",
            "properties": {
                "invoice_number": {"type": "string"},
                "total": {"type": "number"},
            },
        },
        file=42,  # replace with your file ID, or a File object
    )
    print(response.result.data)

import os
import requests

headers = {"Authorization": f"Bearer {os.environ['LIGHTON_API_KEY']}"}

file_id = 42  # replace with your file ID, from GET /api/v3/files

response = requests.post(
    "https://api.lighton.ai/api/v3/extract",
    headers=headers,
    json={
        "file_id": file_id,
        "schema": {
            "type": "object",
            "properties": {
                "invoice_number": {"type": "string"},
                "total": {"type": "number"},
            },
        },
    },
)
print(response.json()["result"]["data"])

file_id, document, and a multipart file part are mutually exclusive: send exactly one. Sending none returns a 400, and a file_id your API key can’t read returns a 404.

Async extraction (large documents)

For documents up to 100 MB, 1000 pages, set options.async = true. The API returns a 202 Accepted immediately with a job ID. Poll GET /api/v3/extract/{job_id} until status is completed or failed. Recommended cadence: 1 s for the first 10 s, then 5 s, capped at 30 s. The SDK does this for you when you pass wait=True; drop it and call job.poll() yourself if you want to report progress as it runs.

from lighton import ExecMode, LightOn

with LightOn() as client:
    # wait=True polls for you and returns once the job is terminal
    job = client.extract(
        schema={"type": "object", "properties": {"title": {"type": "string"}}},
        url="https://example.com/large-report.pdf",
        mode=ExecMode.ASYNC,
        wait=True,
    )

    print(job.result.data)

import os
import time
import requests

headers = {"Authorization": f"Bearer {os.environ['LIGHTON_API_KEY']}"}

# Returns 202 Accepted immediately with a job ID instead of blocking
response = requests.post(
    "https://api.lighton.ai/api/v3/extract",
    headers=headers,
    json={
        "document": "https://example.com/large-report.pdf",
        "schema": {"type": "object", "properties": {"title": {"type": "string"}}},
        "options": {"async": True},
    },
)
job_id = response.json()["id"]  # e.g. "ext_0196e4b2a3c14d5e8f7a9b2c1d0e3f4a"

for _ in range(120):  # give up after ~4 minutes
    response = requests.get(
        f"https://api.lighton.ai/api/v3/extract/{job_id}",
        headers=headers,
    )
    data = response.json()
    if data["status"] in ("completed", "failed"):
        break
    time.sleep(2)

print(data["result"]["data"])

Reading the response

Sync and async responses share the same shape:

{
  "id": "ext_0196e4b2a3c14d5e8f7a9b2c1d0e3f4a",
  "status": "completed",
  "created_at": "2026-03-31T10:00:00+00:00",
  "completed_at": "2026-03-31T10:00:04+00:00",
  "processing_time_ms": 3200,
  "document": {
    "filename": "invoice.pdf",
    "page_count": 3,
    "file_size_bytes": 245120,
    "mime_type": "application/pdf"
  },
  "result": {
    "data": [
      {"invoice_number": "INV-2026-001", "total": null},
      {"invoice_number": null,           "total": 1250.00}
    ],
    "pagination": {
      "page": 1,
      "page_size": 15,
      "total_items": 3,
      "total_pages": 1,
      "has_next": false,
      "has_prev": false
    }
  },
  "usage": {
    "pages_processed": 3
  }
}

result.data is one entry per page, each shaped like your schema. Fields that weren’t found on a given page are null. When the document has more than 15 pages, result.data is paginated: request additional pages with ?page=N on the GET /api/v3/extract/{job_id} endpoint.

Common errors

Status	Cause
`400`	No document, file, or `file_id` supplied; unsupported format; or page limit exceeded
`401`	Missing or invalid API key
`404`	Extract job not found, or `file_id` is unknown, unauthorized, or its stored file is missing
`413`	File exceeds the size limit (20 MB sync, 100 MB async)
`422`	JSON Schema is malformed, uses unsupported features, or exceeds limits
`429`	Rate limit exceeded (6 requests/second per tenant)
`503`	Parsing backend is overloaded. Retry later

Overview

Build a searchable knowledge base

Classify and organise documents

Process documents on the fly

When to use Extract

Sync extraction (small documents)

Extracting from a document you already ingested

Async extraction (large documents)

Reading the response

Common errors

​When to use Extract

​Sync extraction (small documents)

​Extracting from a document you already ingested

​Async extraction (large documents)

​Reading the response

​Common errors

When to use Extract

Sync extraction (small documents)

Extracting from a document you already ingested

Async extraction (large documents)

Reading the response

Common errors