notiformer 1.2.4

Official Python SDK for Notiformer — human-in-the-loop approval gates and push notifications for AI agents.

notiformer (Python)

Official Python SDK for Notiformer — approval gates, multi-option decisions, real-time push notifications, and feature gates for AI agents. Mirrors the Node.js SDK 1:1: same methods, same config, same REST API underneath.

Install

pip install notiformer

Requires Python 3.8+.

Get started

1. Create a free account at app.notiformer.com — no credit card required for the Dev plan. Verify your email before creating a project or using the API.

2. Create a project and copy your API key

Dashboard (app.notiformer.com) → Projects → Create a Project → Copy the API key.

Every project's default key is private (ntf_live_...), meant for server-side code — full access to all four methods. Python is typically used server-side, but if part of your stack calls Notiformer directly from a browser or other public client, use a public key there instead — see API keys: public vs private below.

---

API keys: public vs private

Click to expand — key scopes, domain behavior, rate limits, and kill switch

You choose a key's scope once, at creation — it can never be widened afterward (only narrowed, e.g. adding a domain restriction, or disabled entirely). For broader access, create a new key.

	ntf_live_... (private)	ntf_pub_... (public)
Use in	server / backend code	browser, public JS/HTML on a site
Can call	event(), ask(), select(), gate()	event() only
Shown in dashboard	full value once, at creation only — masked (last 4 chars) after that	always shown in full
Scope after creation	fixed — never widened	fixed — never widened

This is enforced server-side on every request, not just hidden in the dashboard UI — a public key calling ask(), select(), or gate() gets 403 Forbidden, regardless of what's calling it (this SDK, the Node.js SDK, a raw GET URL, anything). This Python SDK works with either key type — it just has no reason to use a public one, since Python code is server-side by definition.

Existing ntf_live_... keys are unaffected by any of this — same full access as always, nothing to migrate.

Domain behavior for public keys

• Default ("monitor mode"): any domain can call a public key immediately. Every new domain seen is just logged for visibility in the dashboard, never blocked.
• Opt-in strict whitelist: enable it on a specific key in the dashboard, and from then on only explicitly approved domains go through. Everything else gets a silent 202 { "ok": true } response — no visible error, the event just isn't created.
• Origin/Referer headers can be spoofed by anything that isn't a real browser — this is documented honestly as protection against mass or accidental abuse, not as strong authentication.

Rate limits specific to public keys

In addition to the existing project-level limits:

Level	Limit	Behavior over the limit
Project (existing, all keys)	500 event()/month, 60/min	event saved, notification skipped
Per domain (public keys only)	20/min, ~half the plan's monthly quota	event silently dropped (202, no error)
Per key + IP (public keys only)	10/min	event silently dropped (202, no error)
New domains tracked	20/day/key	beyond that: only counted, never blocking
New-domain digest push	max 1 every 15 min / project	—

Kill switch

Any key, public or private, can be disabled instantly from the dashboard. There's no server-side cache — the effect is immediate on the very next request, not "within a few seconds."

---

3. Quick start

Create the client once, then use any of the four methods below.

from notiformer import Notiformer

n = Notiformer(
    "ntf_live_...",       # required: your project's API key, from the dashboard
    silent=False,         # optional: True = skip every API call locally, return safe defaults — handy in tests/dev (default: False)
    throw_on_error=True,  # optional: False = return a safe default instead of raising on failure (default: True)
    on_error=None,        # optional: fn(NotiformerError) -> None, called on every failed call — e.g. forward to Sentry (default: None)
)

More on throw_on_error, silent, and on_error in Advanced config below.

event() — A simple notification

Fire-and-forget. Your code continues immediately — no waiting, no approval needed.

n.event(
    "payments",                        # required: groups related notifications — auto-created on first use
    "payment_success",                 # required: machine-readable event name
    description="$49.00 — [e-mail]",  # optional: shown in the notification body
    icon="💳",                          # optional: emoji shown next to the notification
    tags={"plan": "pro", "userId": "usr_42"},  # optional: key/value metadata, filterable in the dashboard
    value="$49.00",                    # optional: highlighted value shown in the feed
    notify=True,                       # optional: False = store silently, no push sent (default: True)
    recipients=["[e-mail]"],    # optional: notify specific people only — default: everyone on the project (max recipients: Dev/Pro 1, Business 3)
)

ask() — Stop and approve

Pause your code and wait for a human to Approve or Deny from the Notiformer app, Telegram Bot, or Slack Bot. This is a blocking call.

result = n.ask(
    "Deploy v2 to production?",          # required: shown as the notification title
    timeout=300,                          # optional: seconds to wait before giving up (default: 300; max: Dev 300s, Pro/Business 900s)
    fallback="deny",                      # optional, but strongly recommended: "deny"|"approve" — used if nobody responds in time. Omit it and a timeout raises NotiformerError instead
    context="Build #442 · 3 services affected",   # optional: shown in the notification body (max 500 chars)
    details="CHANGELOG:\n• Fix: auth token race #912",  # optional: long-form text shown in the app (max 10,000 chars)
)

if result["approved"]:
    deploy()

⚠️ If nobody responds and no fallback was set, this raises NotiformerError(code="timeout") — always, even with throw_on_error=False. See the full ask() section further down for why, and for safe patterns.

select() — Stop and choose an option

Like ask(), but the user picks one of 2–6 custom options instead of Approve/Deny. Same timeout rule as ask().

from notiformer import select_option

result = n.select(
    "How should the agent handle the error?",   # required: shown as the notification title
    [                                             # required: 2 to 6 options
        select_option("retry", "🔄 Retry"),       # value: required, returned when this option is picked · label: required, button text
        select_option("skip", "⏭ Skip"),
        select_option("stop", "🛑 Stop", is_destructive=True),  # is_destructive: optional — renders the button in red (default: False)
    ],
    timeout=300,                                  # optional: same limits as ask() (default: 300)
    fallback="skip",                               # optional, but recommended: must match one of the option values above. Omit it and a timeout raises
    context="Step 4/10 failed — HTTP 503",         # optional: shown in the notification body (max 500 chars)
    details="...",                                 # optional: long-form text shown in the app (max 10,000 chars)
)

gate() — Get a remote variable

A boolean feature flag you toggle from the dashboard — no redeploy needed. Never raises.

Available on all plans, including Dev (free). What changes per plan is how many gates you can have active at once: Dev 2 · Pro 5 · Business 30 · Custom unlimited.

is_enabled = n.gate(
    "new-checkout-flow",  # required: the gate's key, created/toggled from the dashboard
    fallback=False,       # optional: value returned if the gate can't be fetched, e.g. on a network error (default: False)
    cache_ttl=60,         # optional: local in-memory cache duration in seconds — 0 always reads fresh from the server (default: 0)
)

if is_enabled:
    ...  # new behaviour

---

Advanced config

The three optional constructor settings, in more depth:

n = Notiformer(
    "ntf_live_...",

    # throw_on_error (default: True)
    # - True:  event()/gate()'s underlying calls raise NotiformerError on failure
    # - False: they resolve to None / the fallback value instead of raising
    # NOTE: this does NOT apply to ask()/select() timing out with no fallback —
    # that always raises regardless of throw_on_error. See the ask() section.
    throw_on_error=True,

    # silent (default: False)
    # True = every method skips the network call entirely and returns a safe
    # default (event() -> None, ask()/select() -> not approved/selected,
    # gate() -> fallback). Useful so test suites and local dev don't spend
    # quota or need a real key at all.
    silent=os.environ.get("ENV") != "production",

    # on_error (default: None)
    # Called with the NotiformerError on every failure, in addition to (not
    # instead of) raising/returning a default — good for centralized
    # logging regardless of how each call site handles the error locally.
    on_error=lambda err: sentry_sdk.capture_exception(err),
)

Tip: Use Notiformer("ntf_live_test") to try the SDK without a real key. It prints setup instructions and skips all API calls — safe to run as-is, and a quick way to see silent-like behavior without setting it explicitly.

Combining methods — a few realistic patterns

Defense in depth: gate() and ask() for a risky rollout

# Only offer the new flow at all if it's toggled on — then still require
# a human to approve rolling it out to this specific customer.
if n.gate("new-billing-flow"):
    result = n.ask(
        f"Enable new billing flow for {customer.name}?",
        context=f"Customer ID: {customer.id} · MRR: ${customer.mrr}",
        fallback="deny",
    )
    if result["approved"]:
        enable_new_billing_flow(customer)

Audit trail: log the outcome of a select() as an event()

result = n.select(
    f"Build #{build.id} failed at step {step}",
    [
        select_option("retry", "🔄 Retry from this step"),
        select_option("abort", "🛑 Abort pipeline", is_destructive=True),
    ],
    fallback="abort",
)
selected = result["selected"]

# Keep a silent record in the feed regardless of what was chosen
n.event(
    "ci",
    "pipeline_decision",
    description=f"Build #{build.id}: {selected}",
    notify=False,
)

if selected == "retry":
    retry_step()
if selected == "abort":
    abort_pipeline()

Per-environment client: real in prod, silent everywhere else

n = Notiformer(
    os.environ["NOTIFORMER_API_KEY"],
    silent=os.environ.get("ENV") != "production",
    on_error=lambda err: logger.error("notiformer: %s", err),
)

---

ask() — Approval gate (Approve / Deny)

Pause your code and wait for a human to approve or deny from the Notiformer app, Telegram Bot, or Slack Bot. This is a blocking call.

⚠️ Critical: timeout without a fallback raises

If nobody responds in time and you did not set fallback, the SDK raises NotiformerError(code="timeout") — always, even with throw_on_error=False. This is intentional: silently proceeding when no human has actually decided is exactly what causes incidents like "my agent sent 300k emails because nobody had time to respond."

You have two options:

• Set fallback="deny" (recommended for destructive actions) for automatic safe resolution
• Omit fallback and handle the raised error explicitly in a try/except

# ✅ Option A — safe automatic fallback
result = n.ask(
    "Send Black Friday campaign to 3,241 users?",
    context="Campaign ID: bf-2025 · segment A",
    details="Subject: Black Friday Sale\nEstimated revenue: $48,000",
    timeout=300,       # seconds to wait (default: 300)
    fallback="deny",   # auto-deny on timeout — SAFE for destructive actions
)

if result["approved"]:
    send_emails()
else:
    print("Auto-denied (timed out)" if result["timed_out"] else "Denied by human")

# ✅ Option B — explicit error handling, no silent defaults
try:
    result = n.ask(
        "Delete 50,000 rows from production?",
        timeout=120,
        # no fallback — raises if nobody responds
    )
    if result["approved"]:
        db.execute(delete_query)
except NotiformerError as err:
    if err.code == "timeout":
        # Nobody responded — abort and alert.
        # You can respond via: Notiformer App · Telegram Bot · Slack Bot
        alert_team("Approval timed out — action aborted")
    raise

❌ Don't copy this one — missing fallback and no try/except. It raises on timeout and will crash your process unless something upstream catches it. Shown here only to illustrate the mistake, not as something to paste into your code:

result = n.ask("Send emails?")
if result["approved"]:
    send_emails()  # ← never reached if it raises

Parameters

Parameter	Type	Default	Description
message	str	—	Required. Question shown as the notification title.
fallback	"deny"|"approve"	None	What to do automatically when timeout expires. If omitted, timeout raises NotiformerError(code="timeout").
timeout	int	300	Seconds to wait. Max: Dev 300s · Pro/Business 900s.
context	str	None	Optional detail shown in the notification body. Max 500 chars.
details	str	None	Optional long-form text shown in the app. Supports \n. Max 10,000 chars.

Return value

{"approved": bool, "timed_out": bool, "responded_at": str | None}, or raises NotiformerError(code="timeout") if no fallback was set and nobody responded.

---

select() — Multi-option gate

Like ask(), but the user picks from 2–6 custom options instead of Approve/Deny. Same timeout behavior: omitting fallback raises on timeout.

Uses the same monthly quota as ask().

# ✅ With fallback — safe automatic resolution
try:
    result = n.select(
        "How should the agent handle the error?",
        [
            # required — min 2, max 6
            select_option("retry", "🔄 Retry the request"),
            select_option("skip", "⏭ Skip and continue"),
            select_option("stop", "🛑 Stop the pipeline", is_destructive=True),
        ],
        context="Step 4/10 failed — HTTP 503",
        timeout=300,
        fallback="stop",  # ← if nobody responds, stop (safe for pipelines)
        #   omit → raises NotiformerError(code="timeout")
    )

    selected = result["selected"]
    if selected == "retry":
        retry_step()
    if selected == "skip":
        next_step()
    if selected == "stop":
        abort()
except NotiformerError as err:
    if err.code == "timeout":
        # No response and no fallback was set — stop safely
        abort()

Parameters

Parameter	Type	Default	Description
message	str	—	Required. Question shown as the notification title.
options	list[dict]	—	Required. Min 2, max 6. Use select_option(value, label, is_destructive=False) to build each one.
fallback	str	None	Option value to use on timeout. Must exactly match one of the option values. If omitted, timeout raises.
timeout	int	300	Seconds to wait. Same plan limits as ask().
context	str	None	Optional notification body text. Max 500 chars.
details	str	None	Optional long-form text shown in the app. Max 10,000 chars.

Return value

{"selected": str | None, "timed_out": bool, "responded_at": str | None}, or raises NotiformerError(code="timeout") if no fallback was set and nobody responded.

---

event() — Fire-and-forget alert

Send a push notification. Your code continues immediately — no waiting.

n.event(
    "payments",       # required — auto-created on first use
    "payment_success",  # required — machine-readable event name
    description="$49.00 — [e-mail]",
    icon="💳",
    tags={"plan": "pro", "userId": "usr_42"},
    value="$49.00",   # highlighted in the feed
    notify=True,      # default True — False = store silently
    recipients=["[e-mail]"],  # optional — notify specific people only
)

event() never raises by default regardless of throw_on_error. A failed notification will never crash your app.

Monthly quotas

Plan	Included / cycle	Overage
Dev	500 (hard stop)	none
Pro	5,000	$0.0005 / event
Business	50,000	$0.0003 / event
Custom	Negotiated	Negotiated

Rate limit: 60 events/minute per project (rateLimited in the response if exceeded).

---

gate() — Feature flags

Toggle features remotely from the dashboard — no redeploy needed. Always reads fresh from the server; the SDK has an optional local in-memory cache only.

Available on all plans, including Dev (free). What changes per plan is how many gates you can have active at once — see the table below.

is_enabled = n.gate("new-checkout-flow")
if is_enabled:
    return new_checkout(req)

# With options:
is_enabled = n.gate(
    "my-gate",
    fallback=False,  # returned if the gate can't be fetched (default: False)
    cache_ttl=60,    # local in-memory cache in seconds (default: 0 — always fresh)
)

# Full details:
result = n.gate_details("my-gate")
# {"key": "my-gate", "enabled": True, "cached": False}

# Clear local cache:
n.clear_gate_cache("my-gate")
n.clear_gate_cache()

Plan	Max active gates / project
Dev	2
Pro	5
Business	30
Custom	Unlimited

---

Error handling

from notiformer import NotiformerError

n = Notiformer("ntf_live_...", throw_on_error=True)

try:
    result = n.ask(
        "Delete records?",
        timeout=120,
        # no fallback → raises if nobody responds
    )
    if result["approved"]:
        delete_records()
except NotiformerError as err:
    if err.code == "timeout":
        # Nobody responded in time, no fallback was configured.
        # Respond via: Notiformer App · Telegram Bot · Slack Bot
        print("No response — action aborted.")
    elif err.code == "cap_reached":
        # Monthly quota exhausted. Resets at err.cycle_resets_at.
        print("Quota reached. Resets:", err.cycle_resets_at)
    elif err.code in ("card_required", "card_locked"):
        # Pro/Business only — Dev plan never receives this.
        print("Payment issue:", err.manage_url)
    elif err.code == "network":
        print("Network error — check your connection.")

Error codes

.code is one of:

Code	HTTP	When
timeout	408	ask()/select() timed out with no fallback set and nobody responded
cap_reached	402	Dev hard quota or Pro/Business overage safety cap reached this cycle
card_required	402	Pro/Business: no payment method on file (Dev plan never receives this)
card_locked	402	Pro/Business: card declined and grace period expired
feature_not_available	403	Feature requires a higher plan
invalid_api_key	401	Missing or invalid API key
rate_limited	429	Event rate limit (60/min per project)
network	—	Cannot reach the API
validation	—	Malformed request (shouldn't happen if you follow the parameter tables above)
internal	500+	Unexpected server error

On cap_reached, .cycle_resets_at, .manage_url, and .upgrade_url may also be set. Request-shape mistakes (missing message, wrong number of options, a fallback that doesn't match any option, etc.) raise a plain ValueError instead of NotiformerError — these are always raised, regardless of throw_on_error.

---

Plans & quotas

Feature	Dev	Pro	Business	Custom
Price	Free	$4.99 / mo	$29.99 / mo	Contact us
Credit card required	✗ No	✓ Yes	✓ Yes	✓ Yes
ask() + select()	15 / cycle	100 incl.	1,500 incl.	Custom
ask() overage	Hard stop	$0.03 / call	$0.02 / call	—
event()	500 / cycle	5,000 incl.	50,000 incl.	Custom
event() overage	Hard stop	$0.0005 / ev	$0.0003 / ev	—
Feature gates	2	5	30	Unlimited
Projects	1	2	3	Unlimited
Max ask() timeout	5 min	15 min	15 min	60 min

Email verification required on all plans. You must verify your email address before creating projects or using the API.

---

Common patterns

Silent analytics

n.event(
    "analytics",
    "page_view",
    tags={"path": request.path, "userId": session.user_id},
    notify=False,  # stored in feed, no push notification
)

Error alert (e.g. in a Flask/Django error handler)

def handle_exception(err, request):
    n.event(
        "errors",
        "unhandled_error",
        description=str(err),
        icon="🔴",
        tags={"path": request.path, "method": request.method},
        notify=True,
    )

More combined examples (gate + ask, select + event, per-environment client) are in Advanced config above.

---

Serverless caveat

n.ask() and n.select() block for up to timeout seconds while polling. This is fine on a standard server or a long-running worker, but most FaaS platforms enforce short execution windows (e.g. Vercel Edge ~25s). On those platforms, call the REST API directly with your own polling loop instead of using this blocking SDK — see the docs.

Links

• Documentation
• Dashboard
• Node.js SDK
• GitHub
• Support: [e-mail]

License

MIT