LocalParse — Python client

Official Python client for the LocalParse API. LocalParse gives you two products under one API:

1. Parse — turn any document into clean Markdown, JSON, or text. Drop-in LlamaParse-compatible, plus a deterministic accuracy layer: table-detection recovery (catches tables a layout model misses) and oracle-free financial-identity checks (flags Totals that don't reconcile).
2. Extract (Beta) — pull structured JSON out of parsed documents using your own JSON Schema. Server-side schema validation guarantees the returned object matches.

Install

pip install localparse

Quickstart — Parse

from localparse import LocalParse

client = LocalParse(api_key="lp-...")          # or set LOCALPARSE_API_KEY

result = client.parse("invoice.pdf", result_type="markdown")
print(result.markdown)

# Accuracy signal that plain OCR/LLM parsers don't give you:
print(result.identity_check)     # {tables_checked, violations, ...} or None
print(result.recovered_tables)   # tables recovered by detect_repair

Fetch JSON or the ingestion-ready structured contract instead:

result = client.parse("10k.pdf", result_type="json")
for page in result.pages:
    ...

structured = client.parse("10k.pdf", result_type="structured")

Quickstart — Extract (Beta)

Parse first to get clean markdown, then extract the fields you actually care about:

md = client.parse("invoice.pdf").markdown

invoice = client.extract(md, {
    "type": "object",
    "properties": {
        "vendor":     {"type": "string"},
        "invoice_no": {"type": "string"},
        "total":      {"type": "number"},
        "due_on":     {"type": "string", "description": "ISO 8601 date"},
    },
    "required": ["vendor", "total"],
})

print(invoice["vendor"], invoice["total"])

The server validates the model's output against your schema before returning: a non-conforming response surfaces as SchemaConformanceError rather than as silently broken data. Pre-flight safety failures (bad schema shape, depth/width caps, external $ref) surface as SchemaValidationError.

Webhooks (skip polling)

Pass webhook_url to either Parse or Extract and our worker POSTs a signed event the moment the job finishes — Stripe-shape HMAC-SHA256 so you can verify it came from us. One signing secret covers both products.

# Parse — fires parse.succeeded / parse.failed
client.upload(
    "report.pdf",
    webhook_url="https://yourapp.com/localparse/webhook",
)

# Extract — fires extract.succeeded / extract.failed
client.submit_extract(
    md,
    schema,
    webhook_url="https://yourapp.com/localparse/webhook",
)

Parse events are notifications (the full markdown / JSON can be many MB, so we don't inline it) — fetch the payload with client.get_result(job_id) once the event arrives. Extract events carry the schema-conformant data directly.

Grab your signing secret from Account → Webhooks in the dashboard (one per account, rotatable), then verify inbound requests with the bundled helper:

from localparse import verify_webhook, InvalidWebhookSignature

@app.post("/localparse/webhook")
def receive(request):
    try:
        event = verify_webhook(
            body             = request.body,                          # raw bytes
            signature_header = request.headers["X-LocalParse-Signature"],
            secret           = os.environ["LOCALPARSE_WEBHOOK_SECRET"],
        )
    except InvalidWebhookSignature:
        return 400

    if event["event"] == "extract.succeeded":
        ingest(event["data"])
    elif event["event"] == "parse.succeeded":
        # Notification only — fetch the result yourself:
        result = client.get_result(event["id"], "markdown")
        ingest_markdown(result.markdown)
    else:
        log_failure(event["id"], event["error_message"])
    return 200

verify_webhook rejects modified bodies, bad signatures, missing headers, and timestamps older than 5 minutes (replay protection). Webhook URLs must be https and must not resolve to private/loopback IPs (SSRF guard runs at submit-time).

Retry policy: first attempt is inline, then 10s → 60s → 300s → 1800s (5 total) before the delivery is marked failed in the Jobs dashboard.

Persist a whole folder (incremental)

Ingest a data room into a named case; re-runs only parse new/changed files (unchanged files are skipped by content hash):

results = client.parse_folder(
    "./data-room",
    case_id="acme",
    resume=True,
    on_progress=lambda path, res: print("parsed" if res else "skipped", path),
)

Full control (async jobs)

job = client.upload("big.pdf", result_type="json", case_id="acme")
job = client.wait(job.id)
if job.is_success:
    result = client.get_result(job.id, "json")

Configuration

Errors

AuthenticationError (401/403), QuotaExceededError (402), NotFoundError (404), RateLimitError (429, with .retry_after), APIError (other non-2xx), JobFailedError (job ended ERROR/CANCELED), JobTimeoutError, SchemaValidationError (400 — bad data_schema), SchemaConformanceError (422 — model output didn't match the schema) — all subclass LocalParseError.