ai-newsletter 1.0.3

Official Python SDK for the Signal AI / AI Newsletter public REST API.

ai-newsletter

Official Python SDK for the Signal AI public REST API (/v1). Python 3.9+, built on httpx.

• PyPI: https://pypi.org/project/ai-newsletter/
• Source: https://github.com/ai-newsletter/sdks/tree/main/python
• Interactive API reference: https://qdqyoolnfejkojdcufkk.supabase.co/functions/v1/v1-docs?format=html
• OpenAPI JSON: https://qdqyoolnfejkojdcufkk.supabase.co/functions/v1/v1-docs

---

Install

pip install ai-newsletter
# or
uv pip install ai-newsletter
poetry add ai-newsletter

Requires Python 3.9+.

---

Get an API key

1. Sign in at https://ai-newsletter.app.
2. Open Settings → Public API (/settings/api).
3. Click Create key, pick the scopes you need, copy the key — it is shown once.

Key prefixes:

Prefix	Environment	Notes
sk_live_…	Production	Hits real subscribers, real SES sends, real webhooks.
sk_test_…	Sandbox	Reads/writes parallel test tables. Sends never invoke SES.

Available scopes

Scope	Allows
read:me	Read account context (account.me)
read:subscribers	List subscribers
write:subscribers	Add subscribers, batch import, unsubscribe
send:newsletters	Trigger transactional / broadcast sends
read:sends	Read send job status
manage:webhooks	CRUD webhook endpoints, list & replay deliveries
read:analytics	Read newsletter and campaign analytics

---

Initialize the client

from ai_newsletter import AiNewsletter

with AiNewsletter(api_key="sk_live_…") as client:  # also accepts sk_test_…
    print(client.is_test)  # True when using an sk_test_ key
    me = client.account.me()

The client wraps an httpx.Client. Use it as a context manager (with …) or call client.close() manually when you are done.

Options

client = AiNewsletter(
    api_key="…",                                    # required
    base_url="https://…/functions/v1",              # override the default base URL
    timeout=30.0,                                   # per-request timeout (seconds)
    max_retries=3,                                  # retries for 408/425/429/5xx + network
)

---

Endpoint reference

All methods return the unwrapped data field of the response envelope as plain dicts/lists, and raise AiNewsletterError on non-2xx responses.

Account — client.account

account.me() -> dict

me = client.account.me()
# {"user_id": "...", "plan": "pro", "display_name": "...", "newsletters": [...]}

---

Subscribers — client.subscribers

list(*, newsletter_id, status=None, cursor=None, limit=None)

page = client.subscribers.list(
    newsletter_id="d5…",
    status="subscribed",  # "subscribed" | "unsubscribed" | "pending"
    limit=100,            # default 50, max 100
)
# {"items": [...], "next_cursor": "..." | None}

iterate(*, newsletter_id, status=None, limit=None) — generator

for sub in client.subscribers.iterate(newsletter_id="d5…"):
    print(sub["email"])

create(*, newsletter_id, email, name=None)

sub = client.subscribers.create(
    newsletter_id="d5…",
    email="[e-mail]",
    name="Jane",
)

batch(*, newsletter_id, subscribers)

Bulk-import up to 1,000 subscribers per call. Idempotent for 24h via the auto-generated Idempotency-Key.

result = client.subscribers.batch(
    newsletter_id="d5…",
    subscribers=[
        {"email": "[e-mail]"},
        {"email": "[e-mail]", "name": "B"},
    ],
)
# {"created": …, "skipped": …, "failed": …, "total": …, "results": [...]}

Per-row status is one of created / duplicate / suppressed / invalid.

unsubscribe(id)

Soft unsubscribe. The subscriber is never hard-deleted.

client.subscribers.unsubscribe("sub_…")

---

Sends — client.sends

create(**body)

Accepts newsletter_id, type ("transactional" or "broadcast"), and either to + subject + html/text, or a draft_id.

# Transactional
job = client.sends.create(
    newsletter_id="d5…",
    type="transactional",
    to="[e-mail]",
    subject="Welcome!",
    html="<p>Hi Jane</p>",
)

# Broadcast from an existing draft
client.sends.create(
    newsletter_id="d5…",
    type="broadcast",
    draft_id="draft_…",
)

retrieve(id)

job = client.sends.retrieve("snd_…")
# {"id", "status", "type", "newsletter_id", "recipient_count", "error", "created_at", "completed_at"}

list(*, cursor=None, limit=None) and iterate(*, limit=None)

for job in client.sends.iterate():
    print(job["id"], job["status"])

---

Webhooks — client.webhooks

list()

endpoints = client.webhooks.list()["items"]

create(*, url, event_types)

The response includes secret once — store it; it is used to verify incoming deliveries.

endpoint = client.webhooks.create(
    url="https://yourapp.com/webhooks/ai-newsletter",
    event_types=["send.completed", "subscriber.unsubscribed"],
)
print(endpoint["secret"])  # store securely

update(id, **body) — url, event_types, is_active

delete(id)

deliveries(endpoint_id)

deliveries = client.webhooks.deliveries("whk_…")["items"]

replay(endpoint_id, delivery_id)

Clones the original payload into a new pending delivery.

client.webhooks.replay("whk_…", "whd_…")

Supported event types

Event	Triggered when
send.queued	A send job has been accepted
send.completed	A send job finished successfully
send.failed	A send job failed
subscriber.created	A new subscriber was added
subscriber.unsubscribed	A subscriber unsubscribed
newsletter.published	A campaign was published

---

Analytics — client.analytics

Requires the read:analytics scope.

list_newsletters()

items = client.analytics.list_newsletters()["items"]

for_newsletter(id)

stats = client.analytics.for_newsletter("d5…")
print(stats["open_rate"], stats["click_rate"])

for_campaign(id)

stats = client.analytics.for_campaign("cmp_…")

Test keys return zeroed shapes so demos never read live data.

---

Pagination

List endpoints return:

{"items": [...], "next_cursor": "..." | None}

The cursor is opaque (base64 of iso_created_at|id). Pass it back as cursor= to fetch the next page, or use iterate() to let the SDK do it for you.

---

Idempotency

Every non-GET request automatically includes an Idempotency-Key header. Replays within 24h return the original result, so retries during network blips are safe.

subscribers.batch is the most useful target — re-running the same job will not double-import rows.

---

Retries & timeouts

The SDK retries 408, 425, 429, 500, 502, 503, 504, and any network error using exponential backoff with jitter. Retry-After (when present) is honoured.

client = AiNewsletter(api_key="…", max_retries=5, timeout=60.0)

---

Errors

from ai_newsletter import AiNewsletterError

try:
    client.subscribers.create(newsletter_id="d5…", email="bad")
except AiNewsletterError as e:
    print(e.status, e.code, str(e), e.retry_after, e.body)

code	HTTP	Meaning
missing_api_key	401	No Authorization header
invalid_api_key	401	Key not recognized
revoked_api_key	401	Key has been revoked
scope_denied	403	Key lacks the required scope
rate_limited	429	Rate limit or auto-throttle hit. See Retry-After
invalid_request	422	Body or query failed validation
not_found	404	Resource does not exist or is not yours
conflict	409	Idempotency-Key reused with a different payload
internal_error	500	Unexpected server error — safe to retry

---

Verify webhooks

Header format: X-Webhook-Signature: t=<unix_seconds>,v1=<hex_hmac_sha256>.

from flask import Flask, request, abort
from ai_newsletter import verify_webhook_signature

app = Flask(__name__)
SECRET = "whsec_…"

@app.post("/webhooks/ai-newsletter")
def webhook():
    sig = request.headers.get("X-Webhook-Signature", "")
    if not verify_webhook_signature(request.get_data(as_text=True), sig, SECRET):
        abort(401)
    event = request.get_json()
    # handle event["type"] / event["data"]
    return "", 200

verify_webhook_signature(raw_body, header, secret, tolerance_seconds=300) rejects timestamps outside the tolerance window to prevent replay.

---

Rate limits

Plan	Burst / sustained
Free / Starter	60 req/min
Pro	300 req/min
Business	1,000 req/min

Response headers: X-RateLimit-Limit, X-RateLimit-Remaining, Retry-After (on 429). Auto-throttling temporarily blocks keys with abnormal error rates — see your dashboard for status.

---

Test mode

Pass an sk_test_… key — every call hits the parallel sandbox tables, sends short-circuit SES, and webhooks fire to test endpoints only. client.is_test reports the mode.

---

Self-hosting

Override the base URL if you run a fork of the API:

client = AiNewsletter(
    api_key="…",
    base_url="https://your-fork.example.com/functions/v1",
)

---

License

MIT — see LICENSE and CHANGELOG.md.