Python SDK

The norq Python package provides a client for sending and linting notification templates from Python applications.

Install

Coming soon. The norq package is not yet published to PyPI. See norq.sh for updates.

The SDK runs entirely in-process — no norq binary or daemon is invoked at runtime.

Quick start

from norq import Norq
 
notif = Norq(project_dir="./my-project")
 
# Send
result = notif.send("transactional/order-shipped",
    to={"email": "gaurav@example.com"},
    data={"user": {"first_name": "Gaurav"}, "order": {"id": "ORD-123"}},
)

Constructor

notif = Norq(
    project_dir="./my-project",        # project root (required)
    strict=True,                       # runtime template validation (default: True)
    logger=my_logger,                  # structured logger
    recipient_resolver=my_resolver,    # user ID -> Recipient
)

Option	Type	Description
`project_dir`	`str`	Path to norq project root. Required — templates are read from this directory and compiled in-process.
`strict`	`bool`	Reject sends that reference missing data or unknown pipes. Default `True`.
`logger`	`Logger`	Structured logger for send/compile events
`recipient_resolver`	`RecipientResolver`	Resolves user ID strings to `Recipient` objects

API

`send(notification, *, to, data=None, sample=None, channels=None, dry_run=False, strict=None, idempotency_key=None)`

result = notif.send("transactional/welcome",
    to={"email": "user@example.com",
        "cc": ["audit@example.com"]},                # email-channel CC/BCC arrays
    data={"user": {"first_name": "Gaurav"}},
    channels=["email"],
    dry_run=False,
    idempotency_key="order-shipped-ORD-123",         # optional — see below
)
 
for r in result.results:
    print(f"{r.channel}: {'OK' if r.success else r.error}")
 
for s in result.skipped:
    print(f"Skipped {s.channel}: {s.reason}")

idempotency_key is forwarded to providers that honour it (Resend, SuprSend) as the Idempotency-Key: header. When omitted, the SDK generates a UUID v4 per send and applies the same value to every channel. Pass a stable user-derived key (order-shipped-ORD-123) when you want dedup across processes too.

to accepts an email-channel sub-shape with cc / bcc arrays — providers that support them (Resend) include them in the request body; others ignore.

attachments accepts a list of dicts: {filename, content (bytes auto-base64'd; str passed through), content_type? (inferred from extension), content_id?, disposition? ("attachment" or "inline")}. Inline images use disposition="inline" + content_id="logo" and are referenced from the HTML as <img src="cid:logo">. Provider caps: Resend 40 MB total, SendGrid 30 MB. See the Attachments reference for the full whitelist of inferred types.

`batch(notification, *, channel)`

Create a batch for sending a single channel template to many recipients. The template is compiled once, then rendered per-recipient. Requests auto-flush at the provider’s max batch size.

batch = notif.batch("transactional/welcome", channel="email")
 
# Add recipients (auto-flushes at provider's max batch size)
for user in users:
    batch.add(to=user.email, data={"user": {"first_name": user.name}})
 
# Flush remaining and get all results
results = batch.flush()

`lint(notification=None)`

results = notif.lint()              # all
results = notif.lint("transactional/welcome")     # one
 
for result in results:
    for diag in result.diagnostics:
        print(f"{diag.severity}: {diag.message} [{diag.rule}]")

Recipient

from norq.types import Recipient
 
recipient = Recipient(
    email="user@example.com",
    phone="+1234567890",
)
 
# Or pass a dict:
result = notif.send("transactional/welcome", to={"email": "user@example.com"}, data={...})

RecipientResolver

To send by user ID instead of passing a full Recipient object, implement a RecipientResolver:

class MyResolver:
    def resolve(self, user_id: str) -> dict:
        user = db.get_user(user_id)
        return {
            "email": user.email,
            "phone": user.phone,
            "slack": {"channel_id": user.slack_id} if user.slack_id else None,
        }
 
notif = Norq(
    project_dir="./my-project",
    recipient_resolver=MyResolver(),
)
 
# Now send by user ID
notif.send("transactional/welcome", to="user-123", data={"user": {"first_name": "Alice"}})

Sending push notifications

Push delivery is per-token. A recipient’s push.tokens list contains every device the user has registered, possibly across ios, android, and web. The SDK fans the send out one HTTP call per token and returns one entry in result.results per token.

Token shape

{
    "token": "...",                       # FCM/APNs/Expo token, or Web Push endpoint URL
    "platform": "ios" | "android" | "web",  # required — Norq routes per platform
    "provider": "apns",                   # optional — overrides routing.push / routing.push.<platform>
    "keys": {"p256dh": "...", "auth": "..."},  # required iff platform == "web"
    "environment": "sandbox" | "production",   # APNs only; overrides provider production flag
}

Mixed-platform recipient

result = notif.send("transactional/order-shipped",
    to={
        "push": {
            "tokens": [
                {"token": "fcm-android-token", "platform": "android"},
                {"token": "apns-ios-token", "platform": "ios", "environment": "production"},
                {
                    "token": "https://updates.push.services.mozilla.com/wpush/v2/...",
                    "platform": "web",
                    "keys": {"p256dh": "BNc...", "auth": "tBHI..."},
                },
                {"token": "ExponentPushToken[xxx]", "platform": "ios", "provider": "expo"},
            ]
        }
    },
    data={"user": {"first_name": "Gaurav"}, "order": {"id": "ORD-123"}},
    channels=["push"],
)

Per-token result fields

Each push result carries two extra fields:

Field	Meaning
`token_index`	Index into the original `recipient["push"]["tokens"]` list — match results back to the token.
`token_status`	Normalised cross-provider status: `"invalid"` (FCM `UNREGISTERED`, APNs `BadDeviceToken`, Web Push 404/410) or `"transient"` (5xx, rate limits). Absent on success.

Reaction guide

`token_status`	What it means	What to do
`"invalid"`	Token is permanently dead.	Drop from DB; don’t retry.
`"transient"`	Provider had a temporary failure.	Retry with backoff; token is still good.
(absent) on success	Delivery accepted.	No action.

for r in result.results:
    if r.channel != "push":
        continue
    token = recipient["push"]["tokens"][r.token_index]
    if r.token_status == "invalid":
        db.delete_push_token(user_id, token["token"])
    elif r.token_status == "transient":
        retry_queue.enqueue(user_id, token, retry_at=time.time() + 60)

See Push provider errors for the full set of per-token failure modes.

Error handling

from norq import NorqError, NorqStrictError
 
try:
    notif.send("transactional/welcome", to="user-123", data={...})
except NorqStrictError as e:
    # Template references missing data or uses an unknown pipe.
    for err in e.errors:
        print(f"{err.kind}: {err.message} (line {err.line})")
except NorqError as e:
    print(f"Norq error: {e}")

Codegen

norq codegen --lang python --out src/generated/norq_types.py

Generates typed dataclasses for each notification’s data shape.

Agent Toolkit

The Python SDK ships an Agent Toolkit module that exposes notifications as typed tools to LLM agents (OpenAI Chat Completions tool-calls, plus any framework that consumes JSON Schema). Locked recipient, allowlist scoping, deterministic idempotency keys, raw-HTTP-body redaction, and minimal-by-default result shape — see the Agent Toolkit page for the full API.