silon-sdk 0.2.0

Python SDK for the Silon messaging platform API

Silon Python SDK

Python client for the Silon messaging platform API — send messages on any channel (WhatsApp, SMS, email, push, web push, voice), manage CRM contacts and groups, run bulk campaigns, consume events, and verify webhooks. Sync and async, fully typed.

Installation

pip install silon-sdk

Requires Python 3.10+. Built on httpx and pydantic v2.

Quickstart

from silon import Silon

client = Silon(
    api_key="sk_live_...",   # Settings → API keys; or set SILON_API_KEY
    workspace="acme",        # => https://acme.silon.tech; or set SILON_WORKSPACE / SILON_BASE_URL
)

sent = client.messages.send(
    channel="whatsapp",
    to={"client_id": "cust_001"},
    content={"body": "Your order has shipped 📦"},
)
print(sent.id, sent.status)   # e.g. "9f3e..." "queued"

Async is a mirror image:

from silon import AsyncSilon

async with AsyncSilon(api_key="sk_live_...", workspace="acme") as client:
    sent = await client.messages.send(
        channel="sms",
        to={"phone_number": "[phone]"},
        content={"body": "Your code is 424242"},
    )

Sending

One grammar, every channel. messages.send targets a single recipient with to; messages.send_batch sends many independent personalised messages in one call — up to 500 inline rows via messages, or an uploaded CSV of any size via file; broadcasts.create fans one piece of content out to an audience (client_group, client_ids, or an inline recipients list).

# Approved WhatsApp template to a raw number
client.messages.send(
    channel="whatsapp",
    to={"phone_number": "[phone]"},
    whatsapp_template={"name": "order_confirmed", "language": "en",
                       "variables": {"body_1": "Sara", "body_2": "ORD-42"}},
    provider="meta_cloud",
)

# Email broadcast to a client group
result = client.broadcasts.create(
    channel="email",
    audience={"type": "client_group", "slug": "vip"},
    content={"subject": "We saved you a seat", "body": "<h1>Hello</h1>"},
)
print(result.target_count, result.skipped_count)
if result.skipped:            # per-reason breakdown; skipped_count is the sum
    print(result.skipped.suppressed, result.skipped.wrong_channel,
          result.skipped.duplicate)

# SMS broadcast to an inline ad-hoc list (max 1,000 rows; duplicates
# are deduped into skipped.duplicate)
client.broadcasts.create(
    channel="sms",
    audience={"type": "recipients", "recipients": [
        {"phone_number": "[phone]"},
        {"phone_number": "[phone]"},
        {"client_id": "cust_001"},
    ]},
    content={"body": "Flash sale ends tonight"},
)

# Track it
broadcast = client.broadcasts.retrieve(result.id)
for delivery in client.broadcasts.deliveries(result.id).auto_paging_iter():
    print(delivery.client_id, delivery.status)

# Personalised batch — every row its own recipient and content (max 500 rows;
# rows are the same shape as messages.send minus audience, and a row's
# channel= overrides the top-level default). Validation is all-or-nothing:
# one bad row 422s the batch with its index and nothing is queued.
batch = client.messages.send_batch(
    channel="sms",
    messages=[
        {"to": {"phone_number": "[phone]"},
         "content": {"body": "Sara, your table for 2 is confirmed for 7pm."}},
        {"channel": "email", "to": {"email": "[e-mail]"},
         "content": {"subject": "Confirmed", "body": "<p>Table for 4 at 9pm.</p>"}},
    ],
)
for row in batch.messages:               # per-row envelopes, in request order
    print(row.id, row.channel, row.status)
status = client.messages.retrieve(batch.messages[0].id)  # each row pollable

# File batch — upload a CSV once, then run it through the same endpoint.
# Request-level fields act as row defaults; CSV columns override per row
# ({{variable}} columns render into the default content). Rows expand
# asynchronously, so the response is the aggregate batch object
# (batch.messages is None) and batch.id is the bulk batch id — read
# per-row status back via client.bulk.retrieve(batch.id) and the reports.
uploaded = client.bulk.files.upload("contacts.csv")
batch = client.messages.send_batch(
    file=uploaded.name,
    channel="sms",
    content={"body": "Hello {{name}}, sale ends tonight."},
)
print(batch.id, batch.status, batch.row_count)   # e.g. "17" "queued" 1200

Every messages.send / messages.send_batch / broadcasts.create / otp.send call carries an Idempotency-Key header (auto-generated UUID unless you pass idempotency_key=), so automatic retries can never double-send.

messages.retrieve(id) returns the status envelope — read the modern keys id / object / channel / status and timeline, the ordered list of attested {status, at, provider?} transitions (its vocabulary adds a per-recipient delivered, which appears only in the timeline, never as the top-level status):

status = client.messages.retrieve(sent.id)
print(status.status)                                  # queued | sent | failed | ...
for entry in status.timeline:
    print(entry.at, entry.status, entry.provider)

The legacy event_id / is_sent / messages keys still decode but are deprecated (accessing them warns) — prefer id / timeline.

Scheduling and cancellation

Pass send_at= to messages.send, broadcasts.create, or the file form of messages.send_batch to schedule instead of dispatching immediately — an aware datetime or an ISO-8601 string with a UTC offset (naive date-times are rejected), strictly in the future and at most 90 days ahead (422 send-at-invalid otherwise). The envelope comes back with status="scheduled" and its id stays stable through dispatch, so the same id works with retrieve before and after the send actually runs:

from datetime import datetime, timezone

scheduled = client.messages.send(
    channel="sms",
    to={"phone_number": "[phone]"},
    content={"body": "Doors open in an hour!"},
    send_at=datetime(2026, 8, 1, 9, 0, tzinfo=timezone.utc),   # or an ISO-8601 string
)
print(scheduled.status)                    # "scheduled"

client.messages.retrieve(scheduled.id)     # resolves while still scheduled

# Change of plans — allowed while status is still "scheduled":
canceled = client.messages.cancel(scheduled.id)
print(canceled.status)                     # "canceled" — it will never dispatch

broadcasts.cancel(id) works the same way (on a scheduled broadcast, target_count / skipped_count may be None until the audience resolves at dispatch time). Cancel is idempotent by nature: repeating it on an already-canceled send returns 200 with the canceled envelope again, so no Idempotency-Key is sent. Canceling a send that already dispatched raises ConflictError (409 not-cancellable); scheduled creates themselves stay always-keyed like immediate ones.

send_at on the inline batch form is rejected with 422 batch-invalid — there is no batch cancel resource by design; schedule those rows individually via messages.send, or use the file form (rows expand and send at dispatch time).

Scheduled sends add scheduled and canceled to the status vocabulary (message: scheduled|queued|sent|failed|canceled, broadcast: scheduled|in_progress|completed|failed|canceled), and cancels emit message.canceled / broadcast.canceled events.

Suppressions

client.suppressions manages the workspace's do-not-contact list. A row matches on (address, channel) or (address, all channels), and is enforced on every send path:

• Single-recipient sends (messages.send with to, otp.send) to a suppressed address raise UnprocessableEntityError (422 recipient-suppressed).
• Fan-outs (broadcasts, batch rows — inline and CSV — and legacy bulk) skip suppressed recipients instead, never an error: the envelope's skipped breakdown itemises them (suppressed / wrong_channel / duplicate, always all three keys) and skipped_count stays the sum. Suppressed inline batch rows are omitted from batch.messages; the file form reports its breakdown via client.bulk.retrieve(batch.id) once async expansion runs.

# Suppress an address on every channel (reason defaults to "manual")
sup = client.suppressions.create(address="[phone]")

# Or scope to one channel, with a reason
client.suppressions.create(
    address="[e-mail]", channel="email", reason="unsubscribe",
)

# Create is idempotent by nature: re-adding the same (address, channel)
# answers 200 with the EXISTING row — never an error.

# List (cursor-paginated) with filters
for sup in client.suppressions.list(reason="stop").auto_paging_iter():
    print(sup.address, sup.channel, sup.reason)

# Lift a suppression
client.suppressions.delete(sup.id)

Addresses are stored normalized (compact E.164 / lowercase email), so any formatting of the same address matches. Rows are mode-scoped: sk_test_ keys list, manage, and enforce test suppressions only, live keys live ones. Listing needs the suppressions:read scope; create/delete need suppressions:write.

For transactional/legal sends that must reach a suppressed recipient (a receipt, a legally required notice), pass override_suppression=True to messages.send — single-recipient sends only. The key must carry the suppressions:override scope (in no preset; grant it explicitly), otherwise the server answers 403 missing-scope. Overridden deliveries are flagged suppression_overridden in the delivery ledger:

client.messages.send(
    channel="email",
    to={"email": "[e-mail]"},        # suppressed, but owed a receipt
    content={"subject": "Your receipt", "body": "..."},
    override_suppression=True,
)

Templates

client.templates manages slug-keyed message templates — the content the send pipeline renders for template={"slug": ...} sends — with an immutable version spine. Every content edit (subject / body / body_md) mints a new version; channel is metadata and never bumps the version. An un-pinned send always renders the latest content; pin an older revision with template={"slug": ..., "version": N}.

# Create at version 1
tmpl = client.templates.create(
    slug="order-shipped",
    channel="sms",                                  # optional routing hint (metadata)
    body_md="Hi {{ client_name }}, order {{ order_id }} has shipped.",
)

# Edit the body → mints version 2 (version 1 stays frozen and pinnable)
tmpl = client.templates.update(
    "order-shipped", body_md="Hi {{ client_name }} — shipped today!",
)
print(tmpl.version, tmpl.versions)              # 2 [1, 2]

# Render the latest on a send…
client.messages.send(
    channel="sms", to={"client_id": "cust_001"},
    template={"slug": "order-shipped", "variables": {"client_name": "Sara"}},
)
# …or pin an immutable revision (works on send / broadcasts.create / batch)
client.messages.send(
    channel="sms", to={"client_id": "cust_001"},
    template={"slug": "order-shipped", "version": 1,
              "variables": {"client_name": "Sara", "order_id": "ORD-42"}},
)

# List (cursor-paginated; filter by channel hint or slug prefix)
for row in client.templates.list(q="order").auto_paging_iter():
    print(row.slug, row.version)

# Archive (soft delete): reads as missing everywhere afterwards, but its
# version history survives and the slug stays reserved.
client.templates.delete("order-shipped")

Listing/retrieving needs the templates:read scope; create/update/delete need templates:write. An unknown or archived slug raises NotFoundError (404 template-not-found); re-creating an archived slug raises ConflictError (409 template-exists).

Resources

Resource	Methods
client.messages	send, send_batch, retrieve, cancel
client.broadcasts	create, retrieve, deliveries (paginated), cancel
client.otp	send, verify
client.clients	list (paginated), create, retrieve, update, replace, delete
client.client_groups	list (paginated), create, retrieve, update, replace, delete
client.bulk	list, retrieve, send (deprecated — use messages.send_batch), files.list, files.upload, recipients.retrieve
client.reports	messages, channels, clients, users, bulks, specific_bulks, subscriptions, aws_usage, balance
client.templates	list (paginated), create, retrieve, update, delete
client.whatsapp_templates	list, retrieve
client.webhook_endpoints	list (paginated), create, retrieve, update, delete, test, list_attempts (paginated)
client.events	list (paginated), retrieve
client.suppressions	list (paginated), create, delete
client.push	subscribe_android, subscribe_ios, upsert_devices, mark_read, list_notifications, subscribe_web
client.profile	retrieve, update, replace
client.auth	signup, login (deprecated)

Pagination

Cursor-paginated lists (events, templates, webhook_endpoints, webhook_endpoints.list_attempts, suppressions, broadcasts.deliveries, clients, client_groups) return a page you can walk manually or drain with auto_paging_iter():

page = client.events.list(type="message.failed", limit=100)
for event in page:                       # this page only
    ...
for event in page.auto_paging_iter():    # every page, lazily
    ...

# async
page = await client.events.list(limit=100)
async for event in page.auto_paging_iter():
    ...

Behavior change (0.2.0). client.clients.list() and client.client_groups.list() now target the canonical plural CRM routes (/crm/clients/, /crm/groups/) and return a cursor page instead of a bare list. Existing for c in client.clients.list(), indexing, and len() call sites keep working unchanged (the page is iterable, indexable, and sized), but a single call now yields one page (default 50) rather than every contact — use .auto_paging_iter() to walk them all.

Errors

Non-2xx responses raise a typed exception with the parsed error payload:

import silon

try:
    client.messages.send(channel="banana", to={"client_id": "x"})
except silon.UnprocessableEntityError as err:   # 422
    print(err.status_code, err.request_id)
    for detail in err.errors:
        print(detail.code, detail.attr, detail.detail)
except silon.RateLimitError as err:             # 429
    print("retry after", err.retry_after, "seconds")

BadRequestError (400), AuthenticationError (401), PermissionDeniedError (403), NotFoundError (404), ConflictError (409, idempotency-key reuse or cancelling a send that already dispatched — not-cancellable), GoneError (410, expired OTP), UnprocessableEntityError (422), RateLimitError (429) and InternalServerError (5xx) all subclass APIStatusError. Network failures raise APIConnectionError / APITimeoutError.

Every v1 error carries a retryable flag, surfaced verbatim as err.retryable (True for 429 / 5xx and an in-flight idempotency twin, False for other 4xx). Branch on it instead of parsing status codes; it is None only on a legacy/non-v1 body that omits the flag — never inferred from the status code.

try:
    client.messages.send(channel="sms", to={"phone_number": "+1"}, content={"body": "hi"})
except silon.APIStatusError as err:
    if err.retryable:
        ...   # transient — safe to retry the same request

Requests are retried automatically (default max_retries=2, exponential backoff, honouring Retry-After / RateLimit-Reset) — but only when it is safe: idempotent methods, plus POSTs that carry an Idempotency-Key.

Webhooks

Verify the Silon-Signature header on deliveries with the endpoint's one-time whsec_ secret:

from silon import webhooks

event = webhooks.construct_event(
    payload=request.body,                        # raw bytes
    header=request.headers["Silon-Signature"],
    secret=os.environ["SILON_WEBHOOK_SECRET"],
)
if event.type == "broadcast.completed":
    print(event.data.sent, "delivered,", event.data.failed, "failed")

Send a signed ping to an endpoint to check it is reachable, and inspect its delivery attempts (cursor-paginated). A failing sink is reported in-band (delivered=False with the reason in error), never raised:

result = client.webhook_endpoints.test("we_01J1ABC")
print(result.delivered, result.response_status, result.latency_ms, result.error)

for attempt in client.webhook_endpoints.list_attempts("we_01J1ABC").auto_paging_iter():
    print(attempt.event_type, attempt.attempts, attempt.ok, attempt.next_attempt_at)

test needs the webhooks:write scope, list_attempts needs webhooks:read. The endpoint id must match the key's mode (a live key tests livemode=True endpoints, a test key livemode=False); test pings are never persisted and never appear in list_attempts.

Test mode

An sk_test_ API key exercises the full pipeline — validation, scopes, throttles, idempotency, delivery rows, events, webhooks — but never reaches a provider and never bills. Every affected envelope (message, broadcast, batch, OTP, events, webhook payloads) carries livemode=False; live traffic carries livemode=True.

client = Silon(api_key="sk_test_...", workspace="acme")

sent = client.messages.send(
    channel="sms",
    to={"phone_number": "[phone]"},   # magic recipient: always delivered
    content={"body": "test-mode ping"},
)
assert sent.livemode is False

Magic recipients get deterministic outcomes. Statuses are simulated asynchronously a few seconds after the 202, so polling and webhooks behave realistically; any other recipient in test mode is delivered. In live mode the magic values are rejected with 422 test-recipient-in-live, so test fixtures can never leak into real sends.

Recipient	Outcome
[phone]	delivered
[phone]	failed (simulated provider error)
[e-mail]	delivered
[e-mail]	failed
[phone]	always suppressed (no suppression row needed)
[e-mail]	always suppressed (no suppression row needed)

The always-suppressed fixtures exercise the suppression path end to end: a single send answers 422 recipient-suppressed, a fan-out skips them into skipped.suppressed — exactly like a real suppression row, without creating one.

Test-mode OTPs are never dispatched; the magic code 000000 always verifies (and only it).

Webhook endpoints carry a create-time livemode flag (default True). Test events deliver only to livemode=False endpoints, live events only to livemode=True ones — register one endpoint per mode:

client.webhook_endpoints.create(
    url="https://example.com/hooks/silon-test", livemode=False
)

Configuration

Argument	Env var	Default
api_key	SILON_API_KEY	— (required)
workspace	SILON_WORKSPACE	—
base_url	SILON_BASE_URL	https://<workspace>.silon.tech
timeout	—	30 s
max_retries	—	2

A base URL must be resolvable at construction time, from one of four sources checked in this order — otherwise the constructor raises SilonError immediately:

1. base_url= argument (wins over everything)
2. SILON_BASE_URL env var
3. workspace= argument → https://<workspace>.silon.tech
4. SILON_WORKSPACE env var → same expansion

You can also pass default_headers= or your own http_client= (httpx.Client / httpx.AsyncClient) for full transport control.

On-prem / self-hosted instances

The workspace= shortcut is SaaS-only sugar; everything else in the SDK is host-agnostic. For a self-hosted Silon, point base_url at your instance:

client = Silon(api_key="sk_live_...", base_url="https://silon.customer.internal")

API keys, the error contract, retries, idempotency, and webhook signature verification all behave identically — they ride on the base URL.

Private CA / self-signed TLS. Supply your own transport:

http_client = httpx.Client(verify="/etc/pki/customer-ca.pem", timeout=30)
client = Silon(api_key="...", base_url="https://10.20.0.5", http_client=http_client)

Note that a custom http_client brings its own timeout — set it on the httpx.Client, since the SDK's timeout= only applies to the transport it constructs itself.

Reverse proxies. Cursor pagination never follows the server's opaque next URL directly — the SDK extracts only its query parameters and re-issues the request against your configured base_url, so a proxy that rewrites hostnames can't send pagination to an unreachable internal host.

Development

uv sync
uv run pytest
uv run ruff check src tests