Python SDK

Installation

Install

pip install notilens

Setup

Initialize

Pass credentials on first use — they are saved to ~/.notilens_config.json.

import notilens

# Pass credentials directly
nl = notilens.init(
    name="my-app",
    token="YOUR_TOKEN",   # required on first use
    secret="YOUR_SECRET"  # required on first use
)

# Or use environment variables: NOTILENS_TOKEN / NOTILENS_SECRET
nl = notilens.init(name="my-app")

# Multiple sources
scraper = notilens.init(name="scraper", token="TOKEN_A", secret="SECRET_A")
mailer  = notilens.init(name="mailer",  token="TOKEN_B", secret="SECRET_B")

Notify

Notify

Send a one-off notification — no task or run context needed. Use this for standalone events that don't need a lifecycle: a payment received, a report ready, a system alert. No start() or complete() required.

Real-world uses: Shopify order lands → fire immediately. Nightly report finishes → attach the download link. Disk space hits 90% → send a warning before it's critical.

# E-commerce — order just placed
nl.notify("order.placed", "Order #4821 — $149.00 from Jane Smith")

# Infrastructure — proactive warning
nl.notify("disk.space.low", "Only 4 GB remaining on prod-db-01",
    level="warning", tags="infra,production")

# Scheduled report — with download link
nl.notify("report.ready", "Weekly revenue report is ready",
    download_url="https://example.com/reports/week-22.pdf",
    tags="report,weekly",
)

Task Lifecycle

Task Events

Wrap any long-running or multi-step process — AI agent runs, batch jobs, data pipelines, cron jobs. The run tracks duration, state transitions, and metrics end-to-end. Always close a run with a terminal method so NotiLens can compute total duration and clean up state.

Real-world uses: An AI research agent (queue → start → progress per step → complete/fail). A nightly invoice import (start → loop per batch → metric records_processed → complete). A user-triggered export (start → wait for large file → output_generated → complete).

# AI research agent
run = nl.task("research-agent")
run.start()
run.progress("Searching web sources...")
run.progress("Summarising 12 articles...")
run.metric("sources_found", 12)
run.output_generated("Research summary ready")
run.complete("Done — 12 sources, 3 key findings")

# Nightly invoice import
run = nl.task("invoice-import")
run.start()
for batch_num, batch in enumerate(batches, 1):
    run.loop(f"Processing batch {batch_num} of {len(batches)}")
    process(batch)
    run.metric("records_processed", len(batch))
run.complete(f"Imported {total} invoices")

# Terminal states — use the right one
run.fail("Stripe API returned 500")   # unrecoverable error
run.timeout("Exceeded 5 min limit")   # took too long
run.cancel("User cancelled export")   # user action
run.terminate("OOM — killed by OS")   # process killed

Input & Output

Input & Output Events

Signal human-in-the-loop checkpoints and AI output milestones. Use input_required when your agent needs a decision before continuing — content approval, confirmation dialogs, review gates. Use output events to track when something meaningful is produced so you can catch silent generation failures.

Real-world uses: AI drafts a blog post → output_generated → editor gets notified → approves or rejects → agent continues or revises. AI generates an invoice → input_required → finance confirms → input_approved → invoice sent.

Human-in-the-loop

# AI drafted a blog post — needs editor approval before publishing
run.output_generated("Draft ready: '10 Tips for Better APIs'")
run.input_required("Editor review needed before publish")

# Editor responds
run.input_approved("Approved — publishing now")
# or
run.input_rejected("Rejected — tone needs work")

Output events

# AI successfully generated a PDF report
run.output_generated("Q3 financial summary — 14 pages")

# Generation failed silently (model returned empty)
run.output_failed("Model returned empty response after 3 retries")

Metrics

Metrics

Attach numeric values to a run for cost tracking and anomaly detection — tokens, cost, records processed, API calls. Numeric values accumulate across calls; strings replace. NotiLens uses these to baseline your workload and alert when cost spikes, throughput drops, or error rate climbs.

Real-world uses: Track LLM token spend across a multi-step agent run. Monitor records processed in a nightly ETL job — NotiLens alerts if the count drops 40% vs last week. Watch API error rate — if errors per run suddenly double, something upstream broke.

# LLM pipeline — track cost across multiple calls in one run
run.metric("tokens", result1.usage.total_tokens)   # call 1: 350
run.metric("tokens", result2.usage.total_tokens)   # call 2: 280 → total: 630
run.metric("cost",   result1.usage.cost)           # $0.0014
run.metric("cost",   result2.usage.cost)           # $0.0011 → total: $0.0025
run.metric("model",  "gpt-4o-mini")                # strings replace, not add

# ETL job — track throughput and errors
run.metric("records_processed", len(batch))
run.metric("records_failed",    failed_count)
run.metric("api_latency_ms",    response_time_ms)

Custom Events

Custom / Generic Events

Send any business or system event that doesn't fit task lifecycle — orders, signups, deploys, state changes. Source-level when outside a run; run-level to attach the event to a specific task's timing and metrics context.

Real-world uses: Stripe webhook arrives → nl.track("payment.received", ...) to monitor payment volume. User signs up mid-agent run → run.track("user.signup", ...) to correlate signups with the agent that triggered them. Deploy finishes → nl.track("deploy.complete", ...) to watch for anomalies after each release.

# Stripe webhook handler — monitor payment volume
@app.route("/webhook/stripe", methods=["POST"])
def stripe_webhook():
    event = stripe.Webhook.construct_event(...)
    if event["type"] == "payment_intent.succeeded":
        nl.track("payment.received",
                 f"${event['data']['object']['amount'] / 100:.2f} from {event['data']['object']['customer']}",
                 meta={"amount": event['data']['object']['amount'] / 100,
                       "customer_id": event['data']['object']['customer']})

# Inside an agent run — attach business event to this run's context
run = nl.task("onboarding-agent")
run.start()
create_account(user)
run.track("user.signup", f"New signup: {user.email}",
          meta={"plan": user.plan, "source": user.utm_source})
send_welcome_email(user)
run.complete("Onboarding complete")

Auto-Patch

Auto-Patch AI Frameworks

Zero-instrumentation option: set patch=True and every AI call is tracked automatically — no changes to your existing code. Best for quick observability or legacy codebases. Combine with manual run.metric() calls if you also want custom business metrics alongside the auto-captured token usage.

nl = notilens.init(
    name="my-app",
    token="YOUR_TOKEN",
    secret="YOUR_SECRET",
    patch=True  # auto-instruments all AI calls
)

# Every AI call now fires ai.call.start + ai.call.complete
from openai import OpenAI
client = OpenAI()
client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Hello"}]
)

Only one source can own patch=True at a time.

Full Example

Complete Working Example

An AI contract analyser — reads a PDF, extracts key clauses using an LLM, flags risks, and waits for legal review before marking complete.

import notilens
from openai import OpenAI

nl  = notilens.init(name="contract-analyser", token="TOKEN", secret="SECRET")
openai = OpenAI()

def analyse_contract(pdf_url: str, contract_id: str):
    run = nl.task("contract-analysis")
    run.start()

    try:
        # Step 1 — fetch and parse
        run.progress(f"Fetching contract {contract_id}")
        text = fetch_pdf(pdf_url)

        # Step 2 — LLM extraction (multi-call, accumulate tokens)
        run.progress("Extracting key clauses...")
        for section in split_sections(text):
            result = openai.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": f"Extract risks from: {section}"}]
            )
            run.metric("tokens", result.usage.total_tokens)
            run.metric("cost",   result.usage.total_tokens * 0.00015 / 1000)

        # Step 3 — output ready, flag for legal review
        run.output_generated(f"Analysis ready — {len(risks)} risks flagged")
        run.input_required(f"Legal review needed for contract {contract_id}")

        # Step 4 — wait for reviewer (external signal updates state)
        decision = wait_for_legal_decision(contract_id)
        if decision == "approved":
            run.input_approved("Approved by legal team")
            run.complete(f"Contract {contract_id} cleared")
        else:
            run.input_rejected(f"Flagged: {decision['reason']}")
            run.fail("Contract rejected — manual review required")

    except TimeoutError:
        run.timeout("LLM took longer than 60s")
    except Exception as e:
        run.fail(str(e))

force_send

force_send Parameter

Controls whether a notification bypasses NotiLens's ML-based filtering layer or routes through it.

Defaults by method

Override explicitly

# Force deliver a progress event (bypass ML)
run.progress("Fetching records...", force_send=True)

# Let ML decide whether to surface a failure
run.fail("Unrecoverable", force_send=False)

# Works on notify() too
nl.notify("order.placed", "Order #1234", force_send=False)