needlesearchai 2.1.0

Install

Not on PyPI yet. Install from the repository:

pip install "git+https://github.com/Legal-Analytics-Inc/AI-LegalSearch#subdirectory=sdk/python"

Once released the command becomes pip install needlesearchai. Note the name: plain needlesearch on PyPI is an abandoned registration with no files behind it — it is not us, and installing it fails.

Quick start

from needlesearchai import NeedleSearch

ns = NeedleSearch(api_key="nsk_...", base_url="https://your-instance.com/v1")

# Is it us or is it you? Needs no key.
print(ns.status().status)

# Semantic search, optionally with an AI overview over the matches
results = ns.search("termination clauses")
answer = ns.search("How long is the notice period?", include_overview=True)

# Browse. Prefer iterate() over offset paging: it follows the cursor, which
# is stable while other people are uploading.
for item in ns.items.iterate():
    print(item.name, item.status)

# Deep research. The async form survives proxies that cut long connections.
job = ns.jobs.create("What are the risks in this contract?", item_ids=[doc_id])
print(ns.jobs.wait(job.id).answer)

Document status

item.status is the published lifecycle — queued, processing, ready, failed, cancelled — and is the one to branch on. item.status_detail carries the pipeline’s internal stage; it is useful in a log or a support ticket and carries no compatibility promise.

item = ns.items.get(doc_id)
if item.is_ready:
    text = ns.items.content(doc_id)["markdown"]
elif item.is_terminal:
    print("gave up:", item.error)

Errors

Every failure raises a subclass of NeedleSearchError carrying the API’s machine-readable code:

from needlesearchai import NeedleSearchError, RateLimitError

try:
    ns.search("...")
except RateLimitError as e:
    time.sleep(e.retry_after)
except NeedleSearchError as e:
    print(e.error_code, e.status_code)  # e.g. "insufficient_scope", 403

Webhooks

Verify every delivery before trusting it. verify_signature does the two things a hand-rolled check usually gets wrong — a constant-time comparison, and rejecting a replayed capture by its timestamp:

from needlesearchai import verify_signature

# `raw_body` must be the bytes as received; re-serializing the parsed JSON
# changes them and the signature will not match.
if not verify_signature(secret, raw_body, request.headers["X-NeedleSearch-Signature"]):
    return 400

Retries

Send an Idempotency-Key on anything you might retry — research runs and webhook registrations both cost something you do not want to pay twice:

job = ns.jobs.create("...", idempotency_key="batch-2026-02-01-run-7")
hook = ns.webhooks.create("https://example.com/hook", idempotency_key="hook-v1")

License

MIT