nexus-legal

Official Python SDK for the Nexus Legal API v1 — multi-jurisdictional legal analysis with defensible certainty locks, vector case-law search, async jobs, signed webhooks, organizations and sub-keys.

pip install nexus-legal

Requires Python ≥ 3.9.

Quick start

import os
from nexus_legal import NexusClient

nexus = NexusClient(api_key=os.environ["NEXUS_API_KEY"])

result = nexus.analyze(
    text="Por el presente contrato, el arrendador cede el uso…",
    jurisdiction="ES",
    legal_branch="civil",
)

print(result["analysis"])
print("Rate limit remaining:", result["rate_limit"]["remaining"])

Get an API key at /developers. Full docs at /developers/docs.

If your account runs on your OWN engine, this first call raises

Accounts provisioned as BYO-only never run legal generation on a Nexus engine — that is the point of the setting, not a limitation. Until you register an engine, every generative call (analyze, consulta, chat, drafting, the A/B/C nodes) answers 409 BYO_PROVIDER_REQUIRED. It is a configuration state, not a transient failure: retrying returns the same 409, and no credits are charged.

from nexus_legal import NexusClient, ByoProviderRequiredError

try:
    nexus.analyze(text=text, jurisdiction="SG")
except ByoProviderRequiredError:
    # Register the engine once, then re-run. Nothing was executed or charged.
    nexus.llm_providers.create(
        label="Firm Anthropic",
        kind="anthropic",                 # openai_compatible | anthropic | bedrock
        model_default="claude-opus-4-8",
        api_key=os.environ["FIRM_LLM_KEY"],   # write-only: encrypted, never returned
    )

Read-only surfaces (case-law search, legislation lookup, citation verification, coverage, billing) work normally without an engine — they run no generation.

Features

• Synchronous — built on requests. Async (httpx) coming in a future release.
• Auto idempotency — every POST carries an auto-generated Idempotency-Key (UUID4) reused across retries → no double-charge.
• Smart retries — 429 / 5xx with exponential backoff + jitter, honoring Retry-After.
• Typed errors — RateLimitError, InsufficientCreditsError, IdempotencyConflictError, ValidationError, AuthenticationError, ServerError, TimeoutError, and (v0.20.0+) ByoProviderAccountError, ProviderUnreachableError, ByoProviderRequiredError for failures of the firm's own BYO engine.
• Truth signals (v0.20.0+) — a partial, cut-short or unaudited result has the same shape as a complete one; what tells them apart is a set of dedicated fields (truncated, chunksTotal, corpusPartial, piiResiduals, citationsOrigenC, audited, counts_measured, noteCode, termination…). Typed in DoneTruthSignals; one-line accessor get_truth_signals(result). None is never a zero: chunksTotal: None is "could not count", not "no more".
• Rate-limit telemetry — every response exposes result["rate_limit"].
• Audit-trail parser — extract_audit_trail() returns a structured TypedDict without any YAML dependency.
• Webhook signature verification — verify_webhook_signature() for incoming HMAC-signed deliveries.
• Multi-agent mode=deep (v1.6.0+) — Node A + Node B adversarial auditor. Structured findings in result["audit"]. The engine behind each node is a deployment decision, not a contract: read result["model"] for the run you actually got, and on a BYO account no platform model runs at all (see Analysis modes).
• Citation verification (v0.9.0+) — result["citationChecks"] carries the deterministic anti-hallucination verdict of every citation in the analysis (verified | mismatch | not_found | derogated | not_verifiable | pending), checked against the official corpus (BOE / CENDOJ). Typed accessor: get_citation_checks(result). Zero LLM cost.

Analysis modes

Aliases for backward-compat: agil ⇄ standard, auditoria ⇄ deep. The server normalizes them.

result = nexus.analyze(
    text=contract_text,
    jurisdiction="ES",
    mode="deep",     # 2 credits — Nodo B auditará el output de Nodo A
)

if result.get("audit"):
    audit = result["audit"]
    if not audit.get("audited", True):
        # No verdict was produced. `findings: []` here is filler, NOT a clean bill.
        raise RuntimeError("the adversarial audit did not run — do not ship this")
    print(f"Overall severity: {audit['overallSeverity']}")
    for finding in audit["findings"]:
        print(f"[{finding['severity']}] {finding['type']}: {finding['description']}")
    print("Recommendations:", audit["recommendations"])

The audit key is present only when mode=deep. Its shape:

{
    "auditor":         "deepseek-v4-pro",
    "audited":         True,                        # v0.20.0+ — see below
    "text":            "...",                       # free-form audit prose
    "findings": [
        {
            "type":        "hallucination",         # | missing_clause | weak_reasoning
                                                    # | citation_error | other
            "severity":    "medium",                # low | medium | high
            "description": "...",
        }
    ],
    "overallSeverity": "medium",
    "recommendations": "...",
    "processingMs":    1234,
}

🔴 Read audited before anything else in that block. audited: False means no usable verdict was produced — the findings: [] and overallSeverity: "low" beside it are filler, byte for byte the same response that means "the analysis is sound". overallSeverity is deliberately not extended with an "unknown" member: it is a published, closed enum and a new value would break every consumer branching on it.

A failure of the audit ENGINE no longer reaches this block at all — it raises ByoProviderAccountError (424) or ProviderUnreachableError (503). See Error handling.

API surface

nexus = NexusClient(
    api_key="nlk_...",
    base_url="https://legal.nexusquantum.legal",   # default
    timeout=120.0,                                  # seconds
    max_retries=3,
    auto_idempotency=True,
)

# Sync
nexus.analyze(text=..., jurisdiction="ES", legal_branch="civil")
nexus.analyze_batch(documents=[...], concurrency=5)

# Sandbox (no credits)
nexus.sandbox.analyze(sample="providencia_aeat")
nexus.sandbox.metadata()

# Case-law search
nexus.jurisprudencia.search(query="...", jurisdiction="ES", top_k=5)

# Vigent legislation — Spain (ES) live: full BOE corpus indexed
# (~356k vigent articles, semantic search). Other jurisdictions may
# still be rails-only; check per-jurisdiction status with coverage().
nexus.normativa.articulo(                      # literal a fecha
    norma="BOE-A-1889-4763",
    articulo="1124",
    fecha="2026-05-21",
    jurisdiccion="ES",
)
nexus.normativa.search(                        # búsqueda semántica
    query="responsabilidad contractual del arrendatario",
    jurisdiccion="ES",
    top_k=10,
)
nexus.normativa.get_versiones("BOE-A-1889-4763")
nexus.normativa.coverage()                     # status + totals, no auth

# DMS — analiza un fichero de Box por id (sin descargarlo tú) y, por
# defecto, escribe el análisis de vuelta a Box como metadata. Requiere
# que el usuario tras la API key tenga una conexión Box activa (OAuth).
box = nexus.dms.box.analyze(
    file_id="1234567890",
    jurisdiction="ES",
    legal_branch="civil",
    # write_metadata=False,   # desactiva el write-back a Box
)
print(box["analysis"], box["source"]["file_name"], box["metadata_written"])

# Catalogs and usage
nexus.jurisdictions.list()
nexus.usage.get(from_="2026-04-01", to="2026-05-01", granularity="day")
nexus.usage.csv(from_="2026-04-01", to="2026-05-01")

# Time tracking (v0.10.0+) — útil para Office add-ins y facturación
cats = nexus.time_categories.list(locale="es")
nexus.time_entries.create(
    case_id="7f3a...",
    category_id=cats["categories"][0]["id"],
    minutes=45,
    description="Revisión de contrato",
)
entries = nexus.time_entries.list(case_id="7f3a...", unbilled_only=True)

# SLA status (v0.10.0+) — uptime/latencia del mes en curso + log mensual
sla = nexus.sla.status(limit_history=12)
print(sla["current"]["uptime_pct"], sla["lifetime"]["total_credits"])

# IP allowlist (v0.10.0+) — restringe la API key a tus rangos
# (0 entries = allow all; se activa al añadir la primera)
nexus.security.ip_allowlist.add(cidr="203.0.113.0/24", label="Oficina")
ips = nexus.security.ip_allowlist.list()
nexus.security.ip_allowlist.delete(ips["entries"][0]["id"])

# Async jobs
job = nexus.jobs.create(kind="analyze", payload={"text": ..., "jurisdiction": "ES"})
snap = nexus.jobs.get(job["jobId"])
final = nexus.jobs.run(kind="analyze", payload={...})   # poll helper

# Webhooks
wh = nexus.webhooks.create(url="https://example.com/hook", events=["analysis.completed"])
# Guarda wh["secret"] AHORA en tu vault — se muestra UNA SOLA VEZ.

# v0.4.0+ — inspect deliveries (debug)
deliveries = nexus.webhooks.get_deliveries(wh["id"], limit=100, status="failed")
for d in deliveries["deliveries"]:
    print(d["event"], d["status"], d["last_status_code"], d["attempts"])

# DSR export (GDPR Art.15 + Art.20 — v0.3.0+)
bundle = nexus.account.export_dsr()
# Masters: bundle = nexus.account.export_dsr(for_sub_key="<uuid>")

# Organizations + sub-keys
org = nexus.organizations.create(name="Mi Despacho LATAM")
sk  = nexus.organizations.create_key(org["organization"]["id"], child_label="Despacho Vázquez")
# sk["key"] solo se muestra UNA SOLA VEZ

# White-label config (partner/distributor — v1.7.0+)
nexus.branding.get()
nexus.branding.update(
    branding="partner",
    partner_brand_name="Acme Legal AI",
    partner_legal_name="Acme Legal S.A.",
    partner_logo_url="https://cdn.example.com/logo.png",
    partner_primary_color="#1A5F8B",
    partner_support_email="soporte@example.com",
    partner_website="https://example.com",
)
# Si la key es master de una org, los cambios se propagan a todas las
# sub-keys automáticamente. Sub-keys (child) reciben 403.

Sanctions screening (v0.26.0+)

Party screening runs against local lists by default: downloaded to our own infrastructure, and no party name leaves it. That is what fits Zero Retention and a sovereign deployment.

If you would rather use your own sanctions data, you bring it under your own contract:

provider = nexus.screening_providers.create(
    label="Our OpenSanctions subscription",
    base_url="https://api.example.com",
    api_key="…",                       # WRITE-ONLY: encrypted, never returned
    lists=["EU FSF", "UK OFSI", "OFAC SDN"],
)
provider["key_fingerprint"]            # "sha256:5052032a4bff" — a non-reversible hash

Why it has to be yours and not ours. Screening needs the parties' REAL names, before anonymisation: a sanctions list does not match [PERSON_1]. Sending those names to a third party is a transfer of personal data, and whoever contracts the source has to be whoever answers for it. There is therefore no provider of ours that calls a third party, and its absence is the design.

Read who is actually screening, not just the list

out = nexus.screening_providers.list()
out["providers"], out["effective_provider"], out["local_lists"]

effective_provider exists because an empty providers list was ambiguous:

none is not "configure one" — it is "we do not offer this here yet". Check it before reading an empty list as "no matches". On a truncated response the SDK assumes none, never local.

Two refusals, on purpose

# Rotating a credential in place leaves no trace of when it changed. Delete and recreate.
nexus.screening_providers.update(pid, api_key="new")     # raises before the round trip
nexus.screening_providers.update(pid, enabled=False)     # this is fine

# Withdrawing yours does not switch screening off: it says what you fall back TO.
nexus.screening_providers.delete(pid)["effective_provider"]   # "local" | "none"

A failure of your provider is never substituted. There is nothing of ours to fall back to: sending your names elsewhere because yours was down would be exactly the transfer this design avoids. Screening stops at the first failure and the result says so.

Reading the result

analyze() returns party_screening whenever screening was attempted:

r = nexus.analyze(text=texto, jurisdiction="GB", persona="abogado")
ps = r.get("partyScreening")
ps["screened"]        # parties ACTUALLY checked
ps["failed"]          # attempted and NOT checked
ps["failureReason"]   # verbatim from your provider, e.g. "HTTP 401"
ps["entries"]         # hits only

🔴 Read screened and failed together. entries: [] with failed > 0 is not a clean result: it means nobody looked. Until v1.52.0 of the API these were served identically, and a provider that was down produced the same output as a genuinely clean screening.

Advanced examples

# Cross-reference: statutes cited in a judgment (UUID del corpus)
articulos = nexus.cruce.articulos_de_sentencia(jurisprudencia_id)

# Cross-reference: judgments citing a statute article
sentencias = nexus.cruce.sentencias_de_articulo("400", norma_alias="LEC", top=10)

# Procedural time limits — `jurisdiction` is required and has NO default.
# A period counted under another jurisdiction's rules gives you a date you may act on.
plazo = nexus.plazos.calcular(
    jurisdiction="ES",          # LEC + LOPJ, CGPJ and regional holidays, August non-working
    fecha_base="2026-05-29",
    cantidad=20,
    tipo_computo="habiles",
    ccaa="ES-MD",
)

# England and Wales — CPR r 2.8 (clear days; no August vacation)
appellants_notice = nexus.plazos.calcular(
    jurisdiction="GB-EAW",
    fecha_base="2026-05-01",
    cantidad=21,                # CPR r 52.12(2)(b)
    tipo_computo="naturales",   # returned as `clear_days`
)
# → { jurisdiction, fecha_vencimiento, regla_aplicable, ajustes_aplicados, ... }

# Verify a statute citation against the vigent corpus (anti-hallucination)
cita = nexus.normativa.verify_cita(
    citation="Art. 1902 CC",
    claim_text="El que por acción u omisión causa daño a otro...",
)
# → { status: "verified"|"mismatch"|"not_found"|"derogated"|"pending", similarity, url }

# Verify a case-law citation (ECLI / ROJ)
cita_juris = nexus.jurisprudencia.verify_cita(
    citation="ECLI:ES:TS:2023:1234",
    claim_text="El Tribunal declara la nulidad parcial...",
)

# Corpus coverage statistics
coverage = nexus.corpus.coverage()

# Case email inbox (email-to-matter)
inbox = nexus.cases.emails.list(case_id, limit=20)
alias = NexusClient.inbound_email_alias(case_id)  # sin llamada de red

Error handling

from nexus_legal import (
    InsufficientCreditsError,
    RateLimitError,
    ValidationError,
)

try:
    nexus.analyze(text=text, jurisdiction="ES")
except InsufficientCreditsError as e:
    print(f"Need {e.required}, have {e.balance}")     # your NEXUS credits (402)
except ByoProviderAccountError as e:                     # 424 — v0.20.0+
    # YOUR OWN BYO engine turned it down. Nexus did not fall back to another
    # engine, and retrying does not top up someone's card — so this is NOT retried.
    print(f"{e.provider} rejected it: {e.reason}")    # credit_exhausted, invalid_key…
except ProviderUnreachableError as e:                 # 503 — v0.20.0+
    # That same engine could not be REACHED at all (connect/timeout/http_5xx/
    # ssrf_blocked). Transport, not billing. Subclass of ServerError.
    print(f"engine unreachable: {e.error_class}")
except RateLimitError as e:
    time.sleep(e.retry_after)
except ValidationError as e:
    print(f"Bad request: {e}")
    # e.code is a stable identifier (VALIDATION_ERROR, BATCH_TOO_LARGE, ...)
    print(f"Code: {e.code}, request_id: {e.request_id}")

Four failures that look alike and are not — branch on the class, not on the status code: out of Nexus credits (402), your own engine rejected the request (424), your own engine was unreachable (503 PROVIDER_UNREACHABLE), and the request never reached us at all (NetworkError / TimeoutError). Until v0.20.0 the first two arrived as indistinguishable generic errors.

Error format (v1.6.0+) — stable, Stripe-style:

{
  "error": {
    "type":       "invalid_request_error",
    "code":       "VALIDATION_ERROR",
    "message":    "Texto demasiado corto (mín. 50 chars).",
    "request_id": "req_abc-123",
    "details":    { "field": "text", "min": 50, "got": 12 }
  },
  "error_message": "Texto demasiado corto (mín. 50 chars).",
  "code":          "VALIDATION_ERROR"
}

Every response (success or error) includes header X-Request-Id: req_<uuid>. The SDK surfaces it as e.request_id — reference it in support tickets. You can also pre-set your own via X-Request-Id header.

Error types: invalid_request_error, authentication_error, permission_error, rate_limit_error, insufficient_credits_error, not_found_error, conflict_error, api_error.

Webhook signature verification

import os
from nexus_legal import verify_webhook_signature
from flask import Flask, request

app = Flask(__name__)

@app.post("/nexus-webhook")
def webhook():
    raw = request.get_data()                       # bytes, pre-parse
    sig = request.headers.get("X-Nexus-Signature")
    if not verify_webhook_signature(
        raw_body=raw,
        signature=sig,
        secret=os.environ["WEBHOOK_SECRET"],
    ):
        return ("invalid signature", 401)
    payload = request.get_json()
    # ...

Governance (events, ledger, attestations)

Pull the governance event bus (append-only; records every emission even without a subscribed webhook), page the credit ledger, and verify Ed25519 attestations offline.

# Event bus — cursor pagination + filters
page = client.events.list(type=["citation.broken", "spend_cap.reached"])
for ev in client.events.iter(since="2026-07-01T00:00:00Z"):
    print(ev["type"], ev["created_at"], ev["payload"])   # payload is ZR-safe

# Credit ledger — per-sub-key, cursor keyset
for entry in client.credits.iter_ledger(kind="consumption"):
    print(entry["created_at"], entry["amount"], entry["api_key_id"])

Attestations are signed (Ed25519) and verify offline against the public JWKS — a regulator or the end client validates without a Nexus account or a shared secret (unlike the HMAC used for webhooks). Requires the cryptography extra: pip install "nexus-legal[attestation]".

import requests
from nexus_legal import verify_attestation

att = client.outputs.attestation(output_id)["attestation"]
jwks = requests.get(
    "https://legal.nexusquantum.legal/api/v1/attestations/keys"
).json()

res = verify_attestation(att, content=analysis_text, jwks=jwks)
assert res["signature_valid"] and res["content_match"]

content_hash is sha256(NFC(content).strip()), so content_match proves the text you hold is exactly what was attested. For a server-side check that also re-verifies citations against the corpus, use client.attestations.verify(attestation=att, content=analysis_text) → {signature_valid, content_match, citation_recheck}.

Audit-trail parsing

from nexus_legal import extract_audit_trail

trail = extract_audit_trail(result["analysis"])
if trail and trail["levels_emitted"].get("L5-C", 0) > 0:
    # Critical contractual risk — escalate to human reviewer
    ...

The parser handles levels_emitted, review_flags, modules_active, kill_switches_triggered, rag_sources, executive_summary and exposes the raw YAML in trail["raw"] for callers who want to re-parse with PyYAML.

Versioning

This SDK targets API v1.x and follows SemVer:

• MAJOR — drops support for an API breaking change.
• MINOR — new endpoints, new fields, new options.
• PATCH — bug fixes, doc improvements.

See the API changelog.

License

MIT — see LICENSE.

Built by Nexus Legal. Questions? Email support@nexusquantum.legal.