Skip to content

Going to production

What an integration needs once the quickstart works: results delivered to you, runs you can stop, calls with known limits, and work sent in bulk. Every error class is on the errors page; every endpoint is in the API reference.

Contents

Webhooks #

A deep check takes about 90 seconds and a review a few minutes, so in production let Lenz call you instead of waiting. Pass webhook_url (webhookUrl in TypeScript) when you submit; the call returns at once with its id, and Lenz POSTs the result to that URL when the run ends. Each delivery is signed with HMAC-SHA256 over the raw body in X-Lenz-Signature; LenzWebhooks checks the signature, refuses a stale delivery and returns a typed event. Set the signing secret, and a default URL, at /api-credentials, where Send test sends a signed webhook.test to your endpoint.

1. Submit, from the job or script that has the work. It returns at once with the run’s id:

from lenz_io import Lenz

client = Lenz()
task = client.verify(
    claim="The Eiffel Tower is in Berlin.",
    webhook_url="https://example.com/lenz-webhook",
)
print(task.task_id)

2. Receive, in its own module, which submits nothing: a build or a cold start imports it.

# receiver.py: a FastAPI app at https://example.com
import os

from fastapi import FastAPI, Request
from lenz_io import (
    LenzWebhooks,
    VerificationCancelled,
    VerificationCompleted,
    VerificationFailed,
)

webhooks = LenzWebhooks(secret=os.environ["LENZ_WEBHOOK_SECRET"])
app = FastAPI()


@app.post("/lenz-webhook")
async def lenz_webhook(request: Request) -> dict[str, str]:
    # the raw bytes: the signature covers them exactly
    event = webhooks.parse(
        raw_body=await request.body(),
        headers=request.headers,
    )
    if isinstance(event, VerificationCompleted) and event.verification:
        v = event.verification.result
        if v:
            print(v.verdict, v.lenz_score, v.key_finding)
    elif isinstance(event, VerificationFailed) and event.failure:
        print("failed:", event.failure.code, event.failure.retryable)
    elif isinstance(event, VerificationCancelled):
        print("stopped elsewhere; nothing to retry")
    return {"received": "ok"}
  • Edge runtimes. In the Node SDK, await webhooks.unwrap(request) takes a standard Request, reads the raw body once and verifies it with WebCrypto, with no Node built-in. It is tested on Cloudflare Workers, Deno and Bun; Vercel Edge and Next.js route handlers are supported but not tested. Don’t read the body yourself first. await webhooks.parseAsync(rawBody, headers) does the same for a framework that hands you the body and the headers.
  • Node servers. On Express, read the raw body with express.raw({ type: "application/json" }) and call webhooks.parse(req.body, req.headers).
  • Events. verification.completed, .failed, .needs_input and .cancelled; review.* and citecheck.* carry the whole review or check. Deduplicate on event_id (eventId), which stays the same across retries of one delivery, and ignore an event you don’t recognise. Work submitted with SDK 3.x sends *.cancelled when a run is stopped, not *.failed.
  • A rejected signature raises LenzWebhookSignatureError; its causes are on the errors page.

Polling and resuming #

Without a webhook, submit and keep the id. get_status (getStatus) is one non-blocking read; wait polls until the run ends. Both work from another process or after a restart, so a worker can submit and another can collect. A wait that reaches its own deadline raises LenzTimeoutError and the run keeps going: read it again by id, never submit it again. 

# one read: processing | needs_input | completed | failed | cancelled
status = client.get_status(task_id)
if status.status == "completed" and status.result:
    print(status.result.verdict)

# or block until it ends (timeout: how long to wait, in seconds)
v = client.wait(task_id, timeout=300)

A loop of your own must stop on cancelled as well as on completed and failed. A needs_input run found several claims and waits for you to pick: its options are on the status, and select resumes it. Reviews and citation checks read the same way, with get_review / getReview and get_citecheck / getCitecheck.

Stopping a run #

POST /verify/{task_id}/cancel stops a verification that has not finished. A cancelled run is not charged and saves no verification. The call is safe to repeat and answers 200 in every state: cancelled: true with status: "cancelled", or cancelled: false with "completed" or "failed" when the run had already ended, and then nothing changes. The status reports cancelled from then on, and a webhook_url receives verification.cancelled. A /verify/batch has one task per item, so cancel each task. A task that select already resolved answers cancelled: false with status: "needs_input", and the tasks select started keep running: cancel each task id select returned. To stop a review or a citation check, use POST /reviews/{review_id}/cancel or POST /citechecks/{citecheck_id}/cancel: what was already delivered stays charged, the rest is refunded. A review’s deep checks are stopped through the review; cancelling one alone answers 409 use_review_cancel.

The SDKs (3.0 and later) have one method for each:

from lenz_io import Lenz

client = Lenz()

task = client.verify(claim="The Eiffel Tower is in Berlin.")
result = client.cancel(task.task_id)  # a CancelResult
if result.cancelled:
    print("stopped:", result.status)  # "cancelled": not charged
else:
    print("already ended:", result.status)  # "completed" or "failed"

# A review: everything in it stops; what it delivered stays charged
review = client.review(text="The Eiffel Tower is in Berlin. It opened in 1899.")
stopped = client.cancel_review(review.review_id)  # as get_review reads it
print(stopped.status, stopped.credits.charged)

cancel_citecheck (cancelCitecheck) stops a citation check the same way. A wait on a cancelled run raises the failed-run error with failure_class cancelled.

Timeouts, retries and idempotency #

The SDKs send 5xx, 429 and dropped connections again with backoff, honouring Retry-After. Every call that runs or charges for work sends an Idempotency-Key and reuses it across its own retries, so a retried call is answered from the first one, not run twice. Every error carries retryable: whether sending the same request again can succeed.

Two timeouts read alike and mean opposite things. LenzRequestTimeoutError is one request that got no answer, after the retries: the work may have started, so resend only with the key the error carries (idempotency_key, idempotencyKey), never as a plain new call, which can run and charge it twice. LenzTimeoutError is a wait that reached its own deadline: the work is running, so read it by id and don’t resend. 

from lenz_io import LenzRequestTimeoutError, LenzTimeoutError

try:
    v = client.verify_and_wait(claim=claim, timeout=300)
except LenzTimeoutError as exc:
    status = client.get_status(exc.task_id)  # still running: read it later
except LenzRequestTimeoutError as exc:
    v = client.verify_and_wait(claim=claim, idempotency_key=exc.idempotency_key)

More on each error class: connection failures and timeouts, retrying safely.

Per-call options #

Every method takes request options for that one call: in Python the keyword arguments timeout (seconds, one HTTP attempt), max_retries and extra_headers; in TypeScript an options argument with timeoutMs, maxRetries, headers and signal. On the wait helpers the timeout is how long to wait, and the retry count applies to the submit; wait takes none. with_options / withOptions returns a copy of the client with other defaults that shares its connections; the original does not change. The SDK’s own headers (X-Lenz-API-Version, Idempotency-Key, Authorization, …) can’t be set this way. 

# one call: a 10 s attempt, an extra header
task = client.verify(
    claim="The Eiffel Tower is in Berlin.",
    timeout=10,
    extra_headers={"X-Trace-Id": trace_id},
)
usage = client.usage(max_retries=0)

# on a wait, timeout is how long to wait (seconds)
v = client.wait(task.task_id, timeout=180)

# a copy with other defaults, sharing the connection pool
fast = client.with_options(timeout=10, max_retries=0)
fast.get_status(task.task_id)

extract and assess wait at least 150 s and 100 s when the timeout comes from the client or a copy. A timeout passed to the call is used as given, even below that, and can end a call the server is still running: send it again with the same idempotency key to get its answer.

In TypeScript, a signal that fires stops the call and throws LenzAbortError, but cancels nothing on the server; the error carries the run’s id once it was accepted, so you can cancel it (LenzAbortError).

Calling Lenz from async Python #

The Python client is synchronous, and a call takes from seconds (assess) to minutes (a review). Called directly inside an async def, it blocks the event loop for that long, and every other request on that worker waits. Run it in a worker thread, or use a plain def route, which FastAPI runs in its thread pool:

import asyncio
from lenz_io import Lenz

client = Lenz()  # one client for the whole app; safe to share across threads

async def quick_check(claim: str) -> str:
    out = await asyncio.to_thread(client.assess, claim=claim)
    return out.claims[0].verdict

Cancelling the asyncio task does not stop the call: the thread runs until the call returns. Bound each call with timeout=, and stop paid work on the server with the cancel methods. For a long check behind a web request, a webhook ties up no thread. More in the Python SDK README.

Batch and depth #

To deep-check many claims, send up to 20 in one verify_batch call rather than one call per claim: each item is its own task, and verify_batch_and_wait returns one result per claim, in input order, without raising because one claim failed. A quick check takes a list too: assess(claims=[...]), up to 20, one row per claim. 

results = client.verify_batch_and_wait(
    claims=[
        {"claim": "The Eiffel Tower is in Berlin."},
        {"claim": "Water boils at 100 °C at sea level."},
    ],
    depth="low",
)
for r in results:
    if r.status == "completed" and r.verification:
        print(r.claim, r.verification.verdict)
    else:
        print(r.claim, r.status)  # needs_input | failed | timeout

depth="low" is a shallower deep check: fewer sources, a shorter debate (opening arguments, no rebuttals), back sooner, the same models, for 5 credits instead of 10. You are charged for the depth you asked for. An answer served from the last hour’s cache is free, unless it issues you a new warranty certificate, which costs the depth you asked for.