Metadata-Version: 2.4
Name: notiformer
Version: 1.2.4
Summary: Official Python SDK for Notiformer — human-in-the-loop approval gates and push notifications for AI agents.
Author-email: Notiformer <hello@notiformer.com>
License: MIT
Project-URL: Homepage, https://notiformer.com
Project-URL: Documentation, https://notiformer.com/docs
Project-URL: Repository, https://github.com/notiformer/notiformer-python
Project-URL: Issues, https://github.com/notiformer/notiformer-python/issues
Keywords: notiformer,human-in-the-loop,ai-agents,approval,notifications,llm
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.8
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: requests>=2.25
Provides-Extra: dev
Requires-Dist: pytest>=7.0; extra == "dev"
Dynamic: license-file

# notiformer (Python)

Official Python SDK for [Notiformer](https://notiformer.com) — approval
gates, multi-option decisions, real-time push notifications, and feature
gates for AI agents. Mirrors the [Node.js SDK](https://www.npmjs.com/package/notiformer)
1:1: same methods, same config, same REST API underneath.

## Install

```bash
pip install notiformer
```

Requires Python 3.8+.

## Get started

**1. Create a free account at [app.notiformer.com](https://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

<details>
<summary><strong>Click to expand — key scopes, domain behavior, rate limits, and kill switch</strong></summary>

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."

</details>

---

**3. Quick start**

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

```python
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.

```python
n.event(
    "payments",                        # required: groups related notifications — auto-created on first use
    "payment_success",                 # required: machine-readable event name
    description="$49.00 — john@example.com",  # 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=["cto@company.com"],    # 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.**

```python
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()`.

```python
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.

```python
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:

```python
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**

```python
# 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()`**

```python
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**

```python
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`

```python
# ✅ 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")
```

```python
# ✅ 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:
>
> ```python
> 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()`.

```python
# ✅ 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.

```python
n.event(
    "payments",       # required — auto-created on first use
    "payment_success",  # required — machine-readable event name
    description="$49.00 — john@co.com",
    icon="💳",
    tags={"plan": "pro", "userId": "usr_42"},
    value="$49.00",   # highlighted in the feed
    notify=True,      # default True — False = store silently
    recipients=["cto@company.com"],  # 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.

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

```python
# 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

```python
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

```python
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)

```python
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](https://notiformer.com/docs).

## Links

- [Documentation](https://notiformer.com/docs)
- [Dashboard](https://app.notiformer.com)
- [Node.js SDK](https://www.npmjs.com/package/notiformer)
- [GitHub](https://github.com/notiformer)
- Support: hello@notiformer.com

## License

MIT
