LightOn Python SDK

Seamlessly integrate state-of-the-art RAG directly into your software.

What is LightOn?

LightOn is a 🇪🇺 European AI lab building industrial-grade retrieval infrastructure: index your documents, then query them with grounded ask and search actions, and process documents on the fly with parse and extract for specific, standalone actions. This SDK wraps the LightOn API. Create an account and get an API key on console.lighton.ai 🚀

Note from the human maintainers:

This code-base is implemented with AI assistance to allow our team to keep up with the required development celerity, however be assured that all design-patterns, architectural decisions, code-reviews and QA cycles are fully human-backed to ensure that this SDK meet our standards of quality and that we maintainers keep full knowledge of its inner workings to better serve the developer community <3

Quick start

Install:

uv add lighton-sdk

Set your API key in your environment:

export LIGHTON_API_KEY="..."

Get your first result:

from lighton import LightOn, Workspace

with LightOn() as client:  # reads LIGHTON_API_KEY from the environment
    # Create a workspace and ingest a folder of PDFs (glob), blocking until searchable
    ws = Workspace(name="Docs").create(client)
    ws.ingest_many(["docs/**/*.pdf"], wait=True)

    # Search: retrieve the most relevant passages, scoped to that workspace
    chunks = client.search("Q4 revenues", workspaces=[ws])
    for r in chunks.results:
        print(r.score, r.source.filename, r.content)

    # Single-turn RAG for simple use-cases, using an LLM registered on your account
    answer = client.ask(
        "What were Q4 revenues?", workspaces=[ws], model="mistral-large-latest"
    )
    print(answer.answer)

Ingestion

Get documents in first: ask/search only see files ingested into a workspace. Upload one file with Workspace.ingest(), or many at once with ingest_many(). Uploading is the ingestion; a File carries a processing status you can poll.

from lighton import ExecMode, File, LightOn, Workspace

with LightOn() as client:
    ws = Workspace.get(client, 42)

    # One file, non-blocking (returns immediately, status "pending")
    f = ws.ingest(File(path="report.pdf"))
    ws.ingest(File(path="report.pdf"), wait=True)   # or block until embedded

ingest_many() takes paths, Files, and glob patterns (mixed). Every path is validated before any upload; it returns a BatchIngest with succeeded / failed:

batch = ws.ingest_many(
    ["contracts/*.pdf", "reports/**/*.docx", File(path="extra.pdf")],
    wait=True,             # wait for each to finish embedding
    ignore_errors=True,    # collect failures instead of raising on the first
)
print(len(batch.succeeded), "ok,", len(batch.failed), "failed")
for fail in batch.failed:
    print(fail.source, "→", fail.error)

The client paces every request (uploads and status polls) to stay under a per-minute cap and applies the 429 cooldown automatically. It defaults to 1000 requests/minute, the API's limit for most endpoints, so batches stay within bounds out of the box. Override it if your account differs (or pass None to disable pacing):

from lighton import LightOnConfiguration

# override the default cap (or pass None to disable pacing)
with LightOn(config=LightOnConfiguration(max_requests_per_minute=2000)) as client:
    Workspace.get(client, 42).ingest_many(["docs/**/*.pdf"])

Run it in the background with mode=ExecMode.ASYNC and poll the job's progress:

import time

from lighton import BatchIngest, BatchProgress

with LightOn() as client:
    ws = Workspace.get(client, 42)
    job = ws.ingest_many(["docs/**/*.pdf"], wait=True, mode=ExecMode.ASYNC)

    while not job.done:
        p: BatchProgress = job.poll()
        print(f"{p.uploaded}/{p.total} uploaded, {p.ingested} embedded, {p.failed} failed")
        time.sleep(2)

    result: BatchIngest = job.wait()   # once finished

More on file management (list, fetch, tags, delete) and polling in Files & ingestion and Async jobs & polling.

Primary verbs

Four actions live directly on the client. ask and search query your indexed documents, scope them with workspaces=, tags=, or files= (objects or bare ids), and narrow further with content_type=/attribute= (see Content types). parse and extract process a document on the fly, no indexing required, extract can also target a file you already ingested, with file=. Full reference at developers.lighton.ai. The per-verb snippets below assume a client opened with with LightOn() as client:.

`ask`: single-turn RAG

Retrieval-augmented generation: retrieves the most relevant chunks and has an LLM answer your question grounded in them, returning the answer plus the sources it used. Reach for it when you want a direct answer over a corpus. Choose the answering model with model= (any LLM registered on your account).

resp = client.ask(
    "What were Q4 revenues?",
    workspaces=[42],
    max_results=5,
    model="mistral-large-latest",
)
print(resp.answer)
for r in resp.results:          # the chunks used as grounding
    print(r.source.filename, r.score)

Pass schema= to constrain the answer to structured output, same inputs as extract (a pydantic model or a JSON-Schema dict, describing an object). The answer comes back as JSON text in .answer, so parse it yourself:

from pydantic import BaseModel, Field


class Revenue(BaseModel):
    amount: float = Field(description="Revenue figure, in millions.")
    currency: str = Field(description="ISO 4217 code, e.g. 'EUR'.")
    quarter: str | None = Field(None, description="Fiscal quarter, or null.")


resp = client.ask("What were Q4 revenues?", workspaces=[42], schema=Revenue)
revenue = Revenue.model_validate_json(resp.answer)
print(revenue.amount, revenue.currency)
print(resp.results)             # sources still come back alongside

Pass stream=True for anything user-facing: you get Server-Sent Events instead of one blocking call, so the answer appears as it generates rather than after it.

from lighton import DoneEvent, SourcesEvent, TokenEvent

for event in client.ask("What were Q4 revenues?", workspaces=[42], stream=True):
    if isinstance(event, SourcesEvent):
        print("grounded in", len(event.results), "chunks")   # arrives first
    elif isinstance(event, TokenEvent):
        print(event.text, end="", flush=True)                # the answer, as it lands
    elif isinstance(event, DoneEvent):
        print()

SourcesEvent.results holds exactly what non-streaming ask returns, so you can show sources immediately and let the text fill in. Switching on event.type ("sources", "token", "done") works too, if that reads better than isinstance.

Streaming composes with schema=: the tokens spell out the JSON, so join them and parse at the end.

text = "".join(
    e.text
    for e in client.ask("What were Q4 revenues?", workspaces=[42], schema=Revenue, stream=True)
    if isinstance(e, TokenEvent)
)
revenue = Revenue.model_validate_json(text)

Two things to know. It's a generator, so nothing is sent until you start iterating, and a bad request surfaces on the first step rather than at the call; iterate it fully or .close() it so the connection is released. And if generation fails partway, you get a StreamError rather than a silently short answer, because a truncated answer that looks finished is the worse failure.

`search`: retrieval only, no generation

Hybrid semantic + lexical retrieval that returns ranked chunks with scores, source metadata, and (optionally) page images, but no generated answer. Use it to feed context into your own pipeline/LLM, build custom ranking, or surface sources to users.

from lighton import RelevanceScoring, SearchMode

resp = client.search(
    "termination clause",
    tags=[7],
    mode=SearchMode.text,        # .text (hybrid) or .vision (page-image)
    include_image=True,          # attach a base64 page image per chunk
)
for r in resp.results:
    print(r.score, r.content)

Filter on the taxonomy with content_type= and attribute= (both apply to ask too):

resp = client.search(
    "termination clause",
    content_type=["legal:contract"],          # or a ContentType object
    attribute=["fiscal_year:2024|2025", "status:active"],
)

content_type paths are OR-matched and exact-or-subtree (legal also matches legal:contract), with wildcards (legal:contract*, *nda*). attribute entries are ANDed, | ORs within one entry; also name (has any value), name:>value, name:prefix*, name:*text*.

relevance_scoring tunes the scoring step (applies to ask too):

.scoring_and_filtering (default): score, drop chunks below the quality threshold
.scoring_only: score every candidate, return them all
.none: skip scoring; lowest latency, r.scores.relevance is None

`parse`: document → Markdown

One-off conversion of a PDF, Office file, or image into structured per-page Markdown, without storing it in your index. Ideal for feeding documents into another tool. Pass a local path or a public url (exactly one).

doc = client.parse(path="report.pdf")
# doc = client.parse(url="https://example.com/report.pdf")
for page in doc.result.pages:
    print(page.index, page.markdown)

Large documents can time out synchronously, run them async and poll (see Async jobs & polling): client.parse(path="big.pdf", mode=ExecMode.ASYNC).

`extract`: schema-guided structured data

Pull specific, typed fields out of a document for a custom pipeline: you describe the shape (a pydantic model or a raw JSON Schema) and get back data matching it, one object per page. See Extract below for the full schema guide.

resp = client.extract(schema=InvoiceModel, path="invoice.pdf")
# or, on a file already in your index: client.extract(schema=InvoiceModel, file=f)
print(resp.result.data)

Workspaces

Workspaces are the containers your documents live in, retrieval scopes to them. They're active-record objects: an instance manages its own lifecycle.

from lighton import LightOn, Workspace

with LightOn() as client:
    # Create
    ws = Workspace(name="Legal", description="Contracts & NDAs").create(client)

    # Edit, then persist
    ws.name = "Legal EU"
    ws.save()

    # Re-fetch from the API
    ws.refresh()

    # List (follows pagination) and retrieve by id
    for w in Workspace.list(client):
        print(w.id, w.name)

    ws = Workspace.get(client, ws.id)

    # Delete
    ws.delete()

What a listing tells you

Workspace.list() carries a few read-only fields worth knowing about, so you can survey an account without a request per workspace:

for w in Workspace.list(client):
    print(w.name, w.files_count, w.user_role)     # your role: owner / editor / viewer

    if w.taxonomy:                                 # classification coverage
        print(f"  {w.taxonomy.classified_files_rate:.0%} classified")
        for root in w.taxonomy.root_content_types:
            print("  ", root.label, root.count)

    if w.sync:                                     # connected datasource, if any
        print("  synced from", w.sync.datasource_type, w.sync.last_status)

taxonomy is the cheapest way to see how much of a workspace is classified (see Content types); it's None until something is. sync is None for workspaces you upload to directly, and user_role is None when you hold no role on the workspace.

One wrinkle: only list() returns taxonomy. The detail endpoint leaves the field out, so get() and refresh() neither populate it nor clear an already-loaded value; re-list when you want fresh coverage numbers.

Files & ingestion

Uploading a file into a workspace is the ingestion, there's no separate job to track. The returned File carries a processing status; poll it with refresh(), or wait() to block until it's embedded. Ingestion is non-blocking by default. Once it's through, the document keeps both its original bytes and the text the platform parsed, reachable with download() and pages().

from lighton import LightOn, Workspace, File, wait_all

with LightOn() as client:
    ws = Workspace.get(client, 42)

    # Upload, returns immediately, f.status == "pending"
    f = ws.ingest(File(path="report.pdf"))

    f.refresh()          # poll status whenever you like
    print(f.status)      # pending → parsing → embedding → embedded

    # Or block until ready (opt-in)
    ws.ingest(File(path="report.pdf"), wait=True)

    # Bulk upload, then wait on all concurrently (threads, the SDK is sync)
    files = [ws.ingest(File(path=p)) for p in ("a.pdf", "b.pdf", "c.pdf")]
    wait_all(files)

    # Manage existing files (active-record, like Workspace/ApiKey)
    for doc in File.list(client, workspace_id=42):
        print(doc.id, doc.filename, doc.status)

    doc = File.get(client, f.id)

    # Or fetch by name within a workspace — matches the title, so pass the name you
    # uploaded (the server uniquifies the stored filename). The extension is optional.
    # Returns every match (titles aren't unique), empty if there are none.
    # workspace takes a Workspace or an id.
    docs = File.get_by_name(client, "report.pdf", workspace=42)
    doc = docs[0]
    doc.title = "Q4 Report"
    doc.save()

    # Read back the text parsed at ingestion, no re-upload (see Getting the bytes back)
    for page in doc.pages():
        print(page.index, page.markdown)

    # Assign / remove tags, by Tag object, id, or name (see Tags below)
    doc.tag([7, "contracts"])
    doc.untag([12])

    # Or replace the whole set in one save (see External metadata below)
    doc.save(tags=["contracts", 7])

    doc.delete()

    # Deleting many? One request, not one per file.
    File.delete_many(client, File.list(client, workspace_id=42))

delete_many() takes File objects or bare ids (mixed), and is all-or-nothing: if any id is unknown or isn't yours the API rejects the whole call and deletes nothing, which surfaces as a NotFoundError. There is no partial-success report because there is no partial success. An empty list is a local no-op.

Getting the bytes back

Retrieval gives you passages; sometimes you need the document itself, to show it in a UI or hand it to another tool. download() returns the stored bytes:

from pathlib import Path
from lighton import DownloadPurpose

Path("report.pdf").write_bytes(doc.download())                       # as uploaded
Path("render.pdf").write_bytes(doc.download(DownloadPurpose.rendered_pdf))

Already-ingested documents keep the text the platform parsed at ingestion, so you can read it back instead of re-uploading and re-parsing:

for page in doc.pages():
    print(page.index, page.markdown)

That's the same {index, markdown} shape parse returns, the same Page model, so code can move between parsing a local file and reading an ingested one without reshaping anything. It costs an extra request (the text can be large), so it isn't part of refresh().

purpose= picks which stored version you get (original, rendered_pdf, transcript); the server falls back to original when a document has no such rendition, so it won't 404 on you for that.

Thumbnails (256x256 WebP) are generated asynchronously and independently of ingestion, so an embedded file may still have none. Check before you fetch:

from lighton import ThumbnailStatus

doc.refresh()
if doc.thumbnail and doc.thumbnail.status is ThumbnailStatus.READY:
    Path("thumb.webp").write_bytes(doc.download_thumbnail())

Fetching one that isn't READY raises NotFoundError.

External metadata

Documents ingested from a third-party system keep their origin with ExternalMetadata, so a later sync can match a platform document back to the record it came from. Set it on upload, or with save():

from lighton import ExternalMetadata, File

doc = ws.ingest(File(
    path="incident.pdf",
    external_metadata=ExternalMetadata(
        external_id="JIRA-123",
        doc_type="incident",
        additional_metadata={"url": "https://jira/INC-123", "version": 3},
    ),
))
print(doc.external_metadata.external_id)   # round-trips, including after refresh()

To change it, pass it to save(). Every update merges, all the way into additional_metadata, so a partial edit leaves the other keys alone. There is no replace or overwrite mode: merging is the only update the API offers.

doc.save(external_metadata=ExternalMetadata(doc_type="ticket"))
print(doc.external_metadata)   # external_id and additional_metadata are still there

Because everything merges, you remove a value by writing an empty one, and not everything can be removed:

to remove	how
`doc_type`	`ExternalMetadata(doc_type="")`
one key of `additional_metadata`	`ExternalMetadata(additional_metadata={"version": None})`
`external_id`	not possible, it can only be overwritten
the whole record	not possible

doc.save(external_metadata=ExternalMetadata(doc_type=""))

Treat external_id as permanent. The API rejects it both blank (422 may not be blank) and null (422 may not be null), and there's no way to drop the record as a whole. Set it deliberately on upload; you can overwrite it later, never clear it.

save() absorbs the response, so doc.external_metadata always shows what the server actually kept, not the partial value you sent.

What `save()` writes

It writes the title field, plus whatever you hand it explicitly. tags and external_metadata are arguments rather than fields, because neither is a plain set server-side and the keyword says which one you mean: tags replaces the whole set, external_metadata merges. Omit either and that part is left alone:

doc.title = "Q4 Report"
doc.save()                          # writes only the title
doc.save(tags=["contracts", 7])     # replaces every tag (objects, ids, or names)
doc.save(tags=[])                   # removes them all

Use tag() / untag() to add or remove a few without replacing the rest. filename is immutable server-side and never sent.

Replacing a document's content

replace() swaps the content of an existing document and re-ingests it. The document keeps its id, tags, and content-type classifications, so links and stored references survive what used to need a delete plus a re-upload. The new file can be of a different type.

doc.replace("report_v2.pdf", wait=True)   # same id, new content

Because titles and filenames aren't unique, replace() is addressed by id: it's an instance method on a document you already resolved. Resolve it first with File.get(client, id), or check what File.get_by_name() returned:

docs = File.get_by_name(client, "report.pdf", workspace=42)
if len(docs) != 1:
    raise SystemExit(f"{len(docs)} documents named report.pdf, pick one by id")
docs[0].replace("report_v2.pdf", wait=True)

Two things to expect. filename follows the new file while title is preserved, so get_by_name() still finds the document under the name it was uploaded with. And the response still describes the previous content, because the old version stays served until re-ingestion actually starts; the queued work shows up as pending_reprocess:

doc.replace("report_v2.pdf")
print(doc.pending_reprocess, doc.status)   # "update" embedded  <- embedded is STALE
doc.wait()                                 # blocks while a reprocess is queued
print(doc.pending_reprocess, doc.status)   # None embedded      <- the new run

wait() treats a queued reprocess as not-terminal, so wait=True and a hand-written doc.replace(...); doc.wait() are both safe. Reading status yourself right after a replace is not: it reports the previous run until processing starts. wait=True accounts for that; if you poll yourself, don't trust the first status you read back.

Once a file reaches embedded, it's retrievable by ask/search. You can also run extract straight on it, client.extract(schema=Invoice, file=doc), instead of uploading the document a second time (see Extract).

Async jobs & polling

Two things in the SDK are asynchronous and polled: ingestion (a File's status, via refresh() / wait() shown above) and parse / extract run in async mode, which return a job handle you poll. Same idea in both, kick off the work, poll until it reaches a terminal state.

parse and extract take mode= (an ExecMode, default ExecMode.SYNC). Pass ExecMode.ASYNC to queue the job, the call returns a ParseJob / ExtractJob handle instead of blocking. Call job.poll() to refresh it in place; job.succeeded is the one success state and job.done means terminal (finished either way). Handy for large documents that would otherwise time out.

import time

# queue the job, returns right away, job.status == "pending"
job = client.extract(schema=Letter, path="big-scan.pdf", mode=ExecMode.ASYNC)

while not job.poll().succeeded:
    if job.done:                                # terminal but not completed → failure
        raise RuntimeError(f"extract job {job.id} ended as {job.status!r}")
    if job.progress:                            # optional live progress
        print(f"{job.progress.percentage}% ({job.progress.pages_processed} pages)")
    time.sleep(2)

for row in job.result.data:
    print(row)

poll() mutates the job and returns it, so while not job.poll().succeeded: reads naturally; raising once job.done (terminal but not successful) means a stuck or failed job surfaces instead of looping forever.

If you don't need live progress, don't write the loop: pass wait=True to block until the job is terminal (same wait= / timeout= pair as ingest), or call job.wait() yourself. Both return the finished job, raise TimeoutError past timeout (default 300s), and raise LightOnError if the job ends in failure, so the result is there when the call returns.

# async endpoint (no sync timeout to hit), but blocking, no polling code
job = client.extract(schema=Letter, path="big-scan.pdf", mode=ExecMode.ASYNC, wait=True)
for row in job.result.data:
    print(row)

# equivalent, and how to tune the poll interval
job = client.parse(path="big.pdf", mode=ExecMode.ASYNC).wait(timeout=1800, poll=5)

wait=True only makes sense with ExecMode.ASYNC (sync already blocks); passing it without is a ValueError.

parse is the same shape, on failure a ParseJob carries an error block you can raise with directly:

import time

job = client.parse(path="big.pdf", mode=ExecMode.ASYNC)

while not job.poll().succeeded:
    if job.error is not None:                   # terminal failure
        raise RuntimeError(f"parse job {job.id} failed: {job.error.message}")
    time.sleep(2)

for page in job.result.pages:
    print(page.index, page.markdown)

Extract

extract(schema, *, path | url | file) pulls structured data from a document. Pass exactly one source:

path=: a local file, uploaded multipart
url=: a publicly accessible URL the server fetches
file=: a file already ingested into your index (a File or a bare id), no re-upload, the cheap option when the document is already there

The schema drives guided generation and can be a pydantic model or a raw JSON-Schema dict, use whichever you have.

A pydantic model is the easy path: nested models, list[...], and X | None fields all convert to a valid vLLM response_format schema for you.

Give every field a meaningful Field(description=...), the descriptions are carried into the schema and steer the model, so they materially improve extraction quality. Treat them as instructions, not documentation.

from lighton import LightOn
from pydantic import BaseModel, Field


class Person(BaseModel):
    last_name: str = Field(description="Family name, as written in the document.")
    first_name: str | None = Field(
        None, description="Given name; null if not stated."
    )
    role: str | None = Field(
        None, description="Title or role if given, e.g. 'sender', 'recipient'."
    )


class Letter(BaseModel):
    people: list[Person] = Field(
        description="Every person or entity named in the letter."
    )
    subject: str | None = Field(
        None, description="The letter's stated subject line, or null if absent."
    )


with LightOn() as client:
    resp = client.extract(schema=Letter, path="letter.pdf")
    # or from a public URL: client.extract(schema=Letter, url="https://example.com/letter.pdf")
    for row in resp.result.data:          # one object per page
        print(row)

Or pass the schema dict directly, it's validated against the JSON-Schema meta-schema (raises jsonschema.SchemaError if malformed), then normalized the same way a model is: the endpoint rejects $ref, so $defs/$ref are inlined whether the schema came from a model class or from your own Model.model_json_schema() call:

with LightOn() as client:
    resp = client.extract(
        url="https://example.com/invoice.pdf",
        schema={
            "$schema": "https://json-schema.org/draft/2020-12/schema",
            "type": "object",
            "properties": {
                "total": {"type": "number"},
                "currency": {"type": ["string", "null"]},
            },
            "required": ["total"],
        },
    )
    print(resp.result.data)

Need the converted schema without calling the API (to inspect or cache it)?

from lighton.utils import convert_pydantic_to_response_format_json

schema = convert_pydantic_to_response_format_json(Letter)

Content types

Semantic search is good at what does this say and bad at which documents count. Ask for termination clauses across 50,000 documents and you get the most similar-looking passages in the whole corpus, supplier invoices and last year's HR handbook included.

Content types fix that by giving documents a category and typed fields, so you can narrow the corpus before the search runs, and answer questions that aren't semantic at all ("every contract signed in France in 2024"). Three words, one SDK class each:

Content type (ContentType): a node in your company-wide taxonomy tree, such as legal:contract:nda. It says what kind of document this is.
Attribute (Attribute): a typed field on a content type, such as jurisdiction (select) or signed_on (date). It says what you can record.
Facet (Facet): one content type assigned to one file, together with that file's attribute values. It says what this document actually is.

The payoff lands in retrieval, where content_type= and attribute= filter both ask and search:

# 1. describe the kind of document, once, company-wide
nda = ContentType.define(client, "nda", "NDA", parent="legal:contract")
ContentType.define_attribute(
    client, nda, "jurisdiction", AttributeType.select, choices=["FR", "US", "UK"]
)

# 2. classify a document and record its values
doc.classify(nda)
doc.set_attribute(nda, "jurisdiction", "FR")

# 3. retrieve against the narrowed corpus
client.search(
    "termination clause",
    content_type=["legal:contract:nda"],
    attribute=["jurisdiction:FR"],
)

Matching is exact-or-subtree, so classifying precisely still answers broad questions: a document filed under legal:contract:nda also comes back for content_type=["legal"]. That is the difference from tags, which are flat labels with no values. Reach for tags for cross-cutting marks like confidential, and content types for what a document is.

Browsing the taxonomy

ContentType.list() returns the tree (each node has children, and attributes when include_attributes=True):

from lighton import ContentType

for ct in ContentType.list(client, include_attributes=True):
    print(ct.path, ct.label)
    for attr in ct.attributes:
        print("  ", attr.name, attr.type, attr.choices)

Classifying a file

Assign a content type and record its attribute values. classify, unclassify, set_attribute, and clear_attribute all take a ContentType object or a plain path string, so you can pass whichever you have to hand:

from lighton import File

doc = File.get_by_name(client, "nda-2026.pdf", workspace=42)[0]

doc.classify("legal:contract:nda")
doc.set_attribute("legal:contract:nda", "jurisdiction", "FR")
doc.set_attribute("legal:contract:nda", "signed_on", "2026-07-01")  # date → "YYYY-MM-DD"

# Inspect what's assigned (a Facet per content type, with attribute values)
for facet in doc.facets():
    print(facet.path, {a.name: a.value for a in facet.attributes})

doc.clear_attribute("legal:contract:nda", "jurisdiction")
doc.unclassify("legal:contract:nda")

Building the taxonomy

Starting from nothing? Adopt a starter tree from the catalog:

for tpl in ContentType.templates(client):
    print(tpl.path, tpl.label)          # legal, healthcare, finance, tech, ...

ContentType.adopt(client, ["legal", "finance"])

Or define your own. Nodes and attributes are both idempotent: defining an existing one updates it, so define() is also how you rename:

from lighton import AttributeType

compliance = ContentType.define(client, "compliance", "Compliance")
audit = ContentType.define(client, "audit-report", "Audit Report", parent=compliance)

ContentType.define_attribute(client, audit, "fiscal_year", AttributeType.number)
ContentType.define_attribute(
    client, audit, "jurisdiction", AttributeType.select, choices=["FR", "US", "UK"]
)

code is lowercase alphanumeric with hyphens (audit-report), attribute name is snake_case, and choices is required for select and multi-select (passing neither raises before the request goes out).

Removing cascades, so undefining a node takes its whole subtree with it:

ContentType.undefine_attribute(client, audit, "fiscal_year")
ContentType.undefine(client, "compliance")   # also removes compliance:audit-report

Building a tree is several calls, so batch() sends them in one request. Each entry is the body a single method would send, and you get one result per action, in order:

results = ContentType.batch(client, [
    {"action": "adopt", "content_types": ["legal"]},
    {"action": "define_attribute", "content_type_path": "legal",
     "name": "jurisdiction", "attribute_type": "select", "choices": ["FR", "US"]},
])
print([r["status"] for r in results])   # [201, 201]

API keys

Same active-record style. The plaintext secret is available only right after create().

from lighton import LightOn, ApiKey, ApiKeyScope, Role

with LightOn() as client:
    key = ApiKey(
        name="ci-pipeline",
        scopes=[ApiKeyScope(workspace_id=42, role=Role.viewer)],  # omit for an unscoped key
    ).create(client)

    print(key.key.get_secret_value())  # plaintext secret (SecretStr), shown once, store it now

    # Manage existing keys
    for k in ApiKey.list(client):
        print(k.id, k.name, k.prefix)

    key = ApiKey.get(client, key.id)
    key.name = "ci-pipeline-v2"
    key.save()
    key.delete()

Client configuration

LightOn() with no arguments reads LIGHTON_API_KEY from the environment and talks to https://api.lighton.ai. To point it somewhere else — a self-hosted deployment, a staging environment, a local instance — pass a base_url:

from lighton import LightOn, LightOnConfiguration

with LightOn(
    api_key="sk-...",  # or omit and let it read LIGHTON_API_KEY
    config=LightOnConfiguration(base_url="https://lighton.internal.acme.com"),
) as client:
    print(client.search("onboarding policy").results)

base_url is the host only: the SDK appends the /api/v3/... paths itself, and a trailing slash is stripped, so https://host/ and https://host behave the same.

Everything on LightOnConfiguration is optional and independent, override only what you need:

field	default	what it controls
`base_url`	`https://api.lighton.ai`	API root; point at another deployment
`timeout`	5 s connect, 120 s read	`httpx.Timeout`; raise the read timeout for slow parses
`retries`	`3`	connection-level retries with backoff (httpx transport), not HTTP errors
`max_requests_per_minute`	`1000`	paces every request under the API cap; `None` disables pacing
`rate_limit_retries`	`3`	retries on HTTP 429, waiting `Retry-After` when present; `0` disables
`transport`	`None`	a custom `httpx.BaseTransport`, for a proxy, or `MockTransport` in tests

The API key stays a direct LightOn() argument rather than a config field, so a config object can be shared or logged without carrying a secret.

A local instance over plain HTTP, with a longer read timeout and no pacing:

import httpx

config = LightOnConfiguration(
    base_url="http://localhost:8000",
    timeout=httpx.Timeout(600.0, connect=5.0),
    max_requests_per_minute=None,
)
with LightOn(config=config) as client:
    ...

Agent Frameworks

LightOn drops into any agent framework as a retrieval tool: wrap a client.search() call that returns text the LLM can read, and hand it to your agent. (Swap search for ask if you'd rather the tool return a grounded answer than raw chunks.)

The snippets assume a client (see Quick start), in a long-running agent, open it once for the process lifetime. They share this helper, which searches a workspace and formats the hits into a string:

def lighton_search(query: str) -> str:
    """Search the company's document corpus for passages relevant to the query."""
    resp = client.search(query, workspaces=[42], max_results=5)
    return "\n\n".join(f"[{r.source.filename}] {r.content}" for r in resp.results)

LangChain

# pip install langchain-core
from langchain_core.tools import tool

lighton_tool = tool(lighton_search)          # name + description come from the function
# bind it: llm.bind_tools([lighton_tool]), or pass to create_react_agent(...)

LangGraph

Reuses the LangChain lighton_tool above, pass it to a prebuilt ReAct agent:

# pip install langgraph
from langgraph.prebuilt import create_react_agent

agent = create_react_agent(model="openai:gpt-5", tools=[lighton_tool])
result = agent.invoke({"messages": [{"role": "user", "content": "What were Q4 revenues?"}]})

LlamaIndex

# pip install llama-index-core
from llama_index.core.tools import FunctionTool

lighton_tool = FunctionTool.from_defaults(fn=lighton_search)
# agent = ReActAgent.from_tools([lighton_tool], llm=...)

OpenAI Agents SDK

# pip install openai-agents
from agents import Agent, function_tool

agent = Agent(name="Search", tools=[function_tool(lighton_search)])

CrewAI

# pip install crewai
from crewai.tools import tool

lighton_tool = tool("lighton_search")(lighton_search)
# pass tools=[lighton_tool] to your crewai Agent

Name		Name	Last commit message	Last commit date
.github/workflows		.github/workflows
docs/images		docs/images
lighton		lighton
tests		tests
.gitignore		.gitignore
.pre-commit-config.yaml		.pre-commit-config.yaml
AGENTS.md		AGENTS.md
CLAUDE.md		CLAUDE.md
CONTRIBUTING.md		CONTRIBUTING.md
LICENSE		LICENSE
Makefile		Makefile
README.md		README.md
cliff.toml		cliff.toml
pyproject.toml		pyproject.toml
uv.lock		uv.lock

Navigation Menu

Repository files navigation

LightOn Python SDK

What is LightOn?

Contents

Quick start

Ingestion

Primary verbs

`ask`: single-turn RAG

`search`: retrieval only, no generation

`parse`: document → Markdown

`extract`: schema-guided structured data

Workspaces

What a listing tells you

Files & ingestion

Getting the bytes back

External metadata

What `save()` writes

Replacing a document's content

Async jobs & polling

Extract

Tags

Content types

Browsing the taxonomy

Classifying a file

Building the taxonomy

API keys

Client configuration

Agent Frameworks

LangChain

LangGraph

LlamaIndex

OpenAI Agents SDK

CrewAI

About

Releases

Contributors

Languages

Latest commit

History

Folders and files

Repository files navigation

LightOn Python SDK

What is LightOn?

Contents

Quick start

Ingestion

Primary verbs

ask: single-turn RAG

search: retrieval only, no generation

parse: document → Markdown

extract: schema-guided structured data

Workspaces

What a listing tells you

Files & ingestion

Getting the bytes back

External metadata

What save() writes

Replacing a document's content

Async jobs & polling

Extract

Tags

Content types

Browsing the taxonomy

Classifying a file

Building the taxonomy

API keys

Client configuration

Agent Frameworks

LangChain

LangGraph

LlamaIndex

OpenAI Agents SDK

CrewAI

About

Topics

Resources

Contributing

Stars

Watchers

Forks

Releases

Contributors

Languages

`ask`: single-turn RAG

`search`: retrieval only, no generation

`parse`: document → Markdown

`extract`: schema-guided structured data

What `save()` writes