Building a Litigation Alert System

Guides · Use Cases

A litigation alert system monitors a defined set of cases and sends notifications when new docket activity appears. The pattern is a scheduled sweep: check each case against a stored last_checked timestamp, collect any new filings, and dispatch alerts through whatever channel your team uses. DocketLayer's last_checked parameter — detailed in the efficient polling guide — makes this efficient: you only receive the delta, not the full docket on every check.

Overview

The system has three components: a watch list that stores the cases to monitor and when they were last checked, a sweep function that queries each case and collects new activity, and a notification layer that dispatches alerts through email, Slack, a webhook, or any other channel.

Everything runs as a scheduled job. The sweep is stateful — it reads and updates last_checked per case on each run so the next sweep only looks at what happened after the last one.

Watch List Structure

Each entry in the watch list needs a court_code, a case_id, a last_checked timestamp, and optionally a set of filing types to alert on. The label is for human-readable notifications.

from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Optional

@dataclass
class WatchedCase:
    court_code: str
    case_id: str
    label: str                        # human-readable name for alerts
    last_checked: str                 # ISO-8601 with timezone offset
    alert_on: list[str] = field(default_factory=list)  # filing types to alert on

# Example watch list
watch_list = [
    WatchedCase(
        court_code="nysd",
        case_id="1:24-cv-09822",
        label="Acme Corp v. Competitor Inc",
        last_checked="2025-01-01T00:00:00-05:00",
        alert_on=["motion", "order", "judgment"],
    ),
    WatchedCase(
        court_code="cacd",
        case_id="2:23-cv-04501",
        label="Smith Patent Litigation",
        last_checked="2025-01-01T08:00:00-08:00",
    ),
]

Alert Sweep

The sweep function iterates over the watch list, queries each case with last_checked, checks the delta for new filings, and builds alert records. After each successful query it advances the stored timestamp. Payment for each query follows the x402 probe-pay-retry pattern.

import httpx, json
from datetime import datetime, timezone

def run_alert_sweep(watch_list: list[WatchedCase]) -> list[dict]:
    """Check all watched cases; return list of alerts."""
    alerts = []
    now = datetime.now(timezone.utc).isoformat()

    for case in watch_list:
        params = {
            "court_code": case.court_code,
            "case_id": case.case_id,
            "last_checked": case.last_checked,
        }
        if case.alert_on:
            params["filing_types"] = ",".join(case.alert_on)

        try:
            # probe
            probe = httpx.get("https://api.docketlayer.ai/v2/case", params=params)
            requirements = json.loads(probe.headers["X-Payment-Requirements"])

            # pay and retry
            encoded = build_payment_tx(
                solana_client, payer,
                requirements["payTo"],
                int(requirements["maxAmountRequired"]) / 1e6,
            )
            resp = httpx.get(
                "https://api.docketlayer.ai/v2/case",
                params=params,
                headers={"x402-payment": encoded},
            )
            resp.raise_for_status()
            result = resp.json()

            # Check for delta activity
            delta = result.get("delta", {})
            new_filings = delta.get("new_filings", [])

            if new_filings:
                alerts.append({
                    "label": case.label,
                    "court_code": case.court_code,
                    "case_id": case.case_id,
                    "filing_count": len(new_filings),
                    "filings": new_filings,
                })

            # Advance last_checked
            case.last_checked = now

        except httpx.HTTPStatusError as e:
            if e.response.status_code == 422:
                # Court coverage pending — skip silently
                continue
            if e.response.status_code == 503:
                # Portal temporarily unavailable — retry next sweep
                continue
            raise

    return alerts


def send_alerts(alerts: list[dict]) -> None:
    """Dispatch alerts through your notification channel."""
    for alert in alerts:
        message = (
            f"New activity in {alert['label']} ({alert['court_code']}):\n"
            f"{alert['filing_count']} new filing(s) detected."
        )
        # Send via email, Slack, webhook, etc.
        print(message)

Handle 422 and 503 silently

422 means the court has pending coverage — skip it and continue. 503 means the court portal was temporarily unreachable — do not advance last_checked so the next sweep covers the same window. Both are billed at $0. See the API error reference for full error semantics.

Scheduling

Run the sweep on a schedule that matches the publishing cadence of the courts you're monitoring. Federal courts typically publish within minutes of receipt; some state portals batch overnight. Hourly is a reasonable default for federal cases.

import schedule, time

def job():
    alerts = run_alert_sweep(watch_list)
    if alerts:
        send_alerts(alerts)

# Run every hour
schedule.every().hour.do(job)

while True:
    schedule.run_pending()
    time.sleep(60)

# Check every hour, log output
0 * * * * /usr/bin/python3 /opt/docketlayer/alert_sweep.py >> /var/log/docketlayer-alerts.log 2>&1

Filtering by Filing Type

Use the filing_types parameter to narrow the delta to filings that matter for your use case. Pass a comma-separated list of filing type identifiers. The delta will include only filings that match those types.

This is particularly useful when you want alerts for significant events — motions, orders, and judgments — but not for routine administrative filings that don't require action. Filtering server-side means you pay $0.99 per check regardless; the value is in cleaner alert output and less noise for your team.