broadcast-python

Official Python client for Broadcast, the self-hosted email marketing platform.

Works with any Broadcast instance — self-hosted or SaaS. Covers 143/143 API operations, verified against the API's generated OpenAPI document.

📖 Python SDK documentation · API reference · All docs

Also available: Ruby · PHP · Node/TypeScript

Installation

pip install broadcast-python

Python 3.9+. No runtime dependencies — the transport is built on the standard library's urllib, so it cannot conflict with a pinned requests or httpx elsewhere in your environment.

The distribution is broadcast-python; the module is broadcast_python. A package named broadcast already exists on PyPI, so the module deliberately does not claim that name.

Getting Your API Token

1. Log in to your Broadcast dashboard
2. Go to Settings > API Keys
3. Click New API Key
4. Name it, select the permissions you need (see Permissions below), and save
5. Copy the token

Quick Start

from broadcast_python import Broadcast

client = Broadcast(
    api_token="...",                  # or BROADCAST_API_TOKEN
    host="https://mail.example.com",  # or BROADCAST_HOST — required
)

client.subscribers.create(email="ada@example.com", first_name="Ada")

client.transactionals.create(
    to="ada@example.com",
    subject="Welcome",
    body="<p>Glad you are here.</p>",
)

host is required

There is no default. Broadcast is self-hosted-first, so every instance lives at its own domain and any built-in guess would be wrong for nearly everyone.

BROADCAST_HOST and BROADCAST_API_TOKEN are the same names the Broadcast CLI uses in ~/.config/broadcast/config, so a machine set up for the CLI needs no extra configuration.

Configuration

Broadcast(
    api_token="...",
    host="https://...",
    timeout=30,              # read timeout, seconds
    open_timeout=10,
    retry_attempts=3,
    retry_delay=1,           # base backoff, multiplied by attempt number
    max_retry_delay=30,      # ceiling on a server-sent Retry-After
    warnings_mode="log",     # "log" | "raise" | "ignore"
    logger=logging.getLogger(__name__),
    debug=False,
    broadcast_channel_id=42, # admin/system tokens
)

A typo in a setting name raises TypeError rather than being silently ignored.

Responses

Response subclasses dict, so a result behaves exactly like the parsed body while carrying transport metadata as attributes:

result = client.subscribers.create(email="ada@example.com")

result["id"]                     # the body
result.status                    # 201
result.warnings                  # [Warning_(code=..., param=..., message=...)]
result.rate_limit.remaining
result.idempotent_replay         # True if the API replayed a stored response

Item access reads the body, attribute access reads metadata — so a body field named status (which broadcasts have) stays reachable as result["status"].

Warnings

A 2xx response can carry warnings: the API accepted your request but ignored part of it. A mistyped filter silently widens a result set unless you look.

# "log" (default) — warn through `logger`, return normally
# "raise"         — raise WarningError. NOTE: the write already happened.
# "ignore"        — say nothing; read them off result.warnings

Rate Limits

Every response carries the current limit state, and 429s are retried automatically, honouring the server's Retry-After (capped at max_retry_delay).

result = client.subscribers.list()

result.rate_limit.limit      # 120
result.rate_limit.remaining  # 118
result.rate_limit.reset      # datetime

if result.rate_limit and result.rate_limit.remaining < 10:
    time.sleep(1)

rate_limit is None when the server sends no limit headers, or when the limit header cannot be parsed — a value we cannot read makes the whole block untrustworthy.

If the retries are exhausted you get a RateLimitError, which carries .retry_after so you can requeue the job sensibly.

Errors

BroadcastError
├── ConfigurationError
├── APIError
│   ├── AuthenticationError   401
│   ├── AuthorizationError    403
│   ├── NotFoundError         404
│   ├── ConflictError         409  idempotency replay still in flight
│   └── RateLimitError        429  carries .retry_after
├── ValidationError           422
├── TimeoutError
├── DeliveryError
└── WarningError                   carries .warnings and .response

ValidationError and TimeoutError are siblings of APIError, not children — matching the Ruby gem. Note broadcast_python.TimeoutError is not the builtin TimeoutError.

Timeouts, 429s, and 5xx are retried with backoff. A 422 is not: it is deterministic, so retrying is pure latency.

Common Tasks

Subscribers

client.subscribers.list(page=1, is_active=True, tags=["vip"])
client.subscribers.find("ada@example.com")
client.subscribers.create(email="ada@example.com", tags=["vip"])
client.subscribers.update("ada@example.com", first_name="Ada")
client.subscribers.add_tags("ada@example.com", ["beta"])
client.subscribers.remove_tags("ada@example.com", ["beta"])
client.subscribers.activate("ada@example.com")
client.subscribers.deactivate("ada@example.com")
client.subscribers.unsubscribe("ada@example.com")
client.subscribers.resubscribe("ada@example.com")
client.subscribers.redact("ada@example.com")   # irreversible

created_after / created_before that fail to parse are ignored by the server rather than rejected — they return a parameter_ignored warning, so a bad timestamp silently widens your result set. Check result.warnings.

Broadcasts

client.broadcasts.create(subject="Weekly", body="<p>Hi</p>")
client.broadcasts.send(id)                       # no undo
client.broadcasts.schedule(id, scheduled_send_at="2026-08-01T09:00:00Z", scheduled_timezone="UTC")
client.broadcasts.cancel_schedule(id)
client.broadcasts.statistics(id)
client.broadcasts.statistics_timeline(id)
client.broadcasts.statistics_links(id)

Sequences

client.sequences.get(id, include_steps=True)
client.sequences.add_subscriber(id, email="ada@example.com")
client.sequences.remove_subscriber(id, "ada@example.com")
client.sequences.create_step(id, subject="Day 1")
client.sequences.move_step(id, step_id, under_id)

Segments, templates, opt-in forms

client.segments.create(name="VIPs")
client.templates.create(label="Welcome", subject="Hi", body="...")
client.opt_in_forms.analytics(id, start_date=date(2026, 1, 1))
# How often the same visitor sees a popup: one of TRIGGER_FREQUENCIES, or ValidationError (422).
client.opt_in_forms.update(id, trigger_settings={"frequency": "once_per_day"})
client.opt_in_forms.create_variant(id, name="B", weight=50)
client.opt_in_forms.duplicate(id, label="Copy")

Reading a segment recounts its members server-side, so segments.get is not free.

Topics

A topic is one kind of email subscribers opt in to or out of (webinars, offers). Its value lives in a top-level custom_data key (True, False, or no value) or in a tag. unset_receives decides, at send time, whether a subscriber with no value receives it; Broadcast never writes a default. Topics use the subscriber permissions.

topic = client.topics.create(name="Webinars", custom_data_key="sub_webinars")
client.topics.create(name="News", storage="tag", tag_name="news")
client.topics.list()
client.topics.update(topic["id"], unset_receives=False)
client.topics.delete(topic["id"])  # 422 while a broadcast or sequence uses it

# Send to a topic: (segments) AND topic
client.broadcasts.create(subject="Webinar", body="...", segment_ids=[2], topic_id=topic["id"])

# Send to everyone except a segment (also on sequences and autopilots)
client.broadcasts.create(subject="Offer", body="...", excluded_segment_ids=[competitors["id"]])

# Migrate someone who already unsubscribed, in one request
client.subscribers.create(email="opted-out@example.com", unsubscribed_at="2025-03-14T09:30:00Z")

# Change only the keys you send (None deletes a key); the default replaces custom_data
client.subscribers.update("jane@example.com", custom_data={"sub_webinars": False}, custom_data_mode="merge")

The subscriber JSON has topics: each topic's stored value (True, False, or None). Topic changes fire subscriber.preferences_updated as well as subscriber.updated.

Channel design (brand kit)

kit = client.channel_design.get()
kit["colors"]["accent"]         # "#2563eb"
kit["typography"]["font_stack"] # email-safe CSS font stack
kit["brand"]["logo_url"]        # public URL, or None

For the token's own channel, fully resolved (defaults filled in). Reading needs templates_read (or, from Broadcast 2.46.0, channel_settings_read). Block emails built with the drag-and-drop editor inherit these values.

From Broadcast 2.46.0 you can change it. This needs channel_settings_write:

kit = client.channel_design.update({
    "colors": {"accent": "#ff5500"},
    "typography": {"font": "georgia"},
    "brand": {"logo_url": "https://your-broadcast.example/files/eyJ...-logo.png"},
})
kit.warnings   # e.g. a key the server did not recognize

Send only what changes, in the shape get() returns. An empty string or None puts a value back to its default, and social_links replaces the list. font_stack and width are read-only. The logo must be the logo_url of an image in the channel's file library; the server never fetches a URL. A value the Brand Kit page would refuse raises ValidationError (422), and nothing is saved.

Channel settings

settings = client.channel_settings.get()
settings["unsubscribe_page"]["legal_notice_url"]   # "https://example.com/impressum"

updated = client.channel_settings.update({
    "unsubscribe_page": {
        "privacy_policy_url": "https://example.com/privacy",
        "legal_notice_url": "https://example.com/impressum",
    },
})
updated.warnings

Broadcast 2.46.0 or later. The settings are grouped like the Settings pages: general, sender, unsubscribe_page (with the legal links), confirmation_page and opt_in_forms (match_brand_kit: whether new opt-in forms start from the brand kit). Updates are partial: nested text and colours merge, an empty string or None removes a value, and languages replaces the list. Reading needs channel_settings_read, changing needs channel_settings_write; both are off on every token until you tick them.

The match_brand_kit switches (unsubscribe_page, confirmation_page and opt_in_forms) decide where the brand kit is used. They are set here, not with the brand kit update.

File assets

The channel's file library (Broadcast 2.48.0 or later): PNG, JPEG and GIF images under 10 MB, each with a public URL that never expires. Use the URL in emails, or as the brand kit logo.

# Upload from a path (multipart). The content type follows the extension.
file = client.file_assets.upload("logo.png")
file["url"]   # "https://your-broadcast.example/files/eyJ...-logo.png"

# From a file object or bytes, with a filename; link it to this channel only; safe to retry
client.file_assets.upload(png_bytes, filename="banner.gif", channel_restricted=True,
                          idempotency_key="banner-{}".format(campaign_id))

# Already have base64? Send it as JSON.
client.file_assets.upload_base64(filename="logo.png", content_type="image/png",
                                 data=base64.b64encode(png_bytes).decode())

# Use it as the brand kit logo
client.channel_design.update({"brand": {"logo_url": file["url"]}})

client.file_assets.list(filter="logo", limit=20)   # {"data": [...], "total": n}
client.file_assets.get(file["id"])
client.file_assets.delete(file["id"])

A file that is not a PNG, JPEG or GIF, a file of 10 MB or more, or invalid base64 raises ValidationError (422). So does deleting a file that is a channel's brand kit logo: set another logo first. The server never fetches a URL; upload the bytes. Reading needs file_assets_read; uploading and deleting need file_assets_write.

Email servers

Credential redaction guard. The API returns credentials bullet-masked (••••••••). A naive fetch-modify-save would write those bullets back and destroy a working SMTP password. update() strips any credential field whose value matches the redaction pattern and warns:

server = client.email_servers.get(id)
server["smtp_password"]                      # '••••••••'
client.email_servers.update(id, name="Renamed", smtp_password=server["smtp_password"])
# -> sends only {"name": "Renamed"}, warns about the dropped field

Users

Requires an admin API token. Sudo users are read-only through this API — update, deactivate, activate, delete, and permission writes on one raise AuthorizationError. Sudo access itself can never be granted through the API.

client.users.list(status="active", q="ada")
user = client.users.create(email="ada@example.com", first_name="Ada", last_name="L", password="x")
client.users.update(user["id"], first_name="Grace")
client.users.deactivate(user["id"])
client.users.activate(user["id"])
client.users.delete(user["id"])

Channel and system permissions:

client.users.channel_permissions(user["id"])

# PUT replaces the whole channel record -- pass exactly one of
# permissions, role, or preset_id
client.users.set_channel_permissions(user["id"], channel_id, role="Editor")
client.users.set_channel_permissions(user["id"], channel_id, permissions={"subscribers_read": True})
client.users.remove_channel_permissions(user["id"], channel_id)

# Same rule, applied to several channels at once
client.users.bulk_channel_permissions(user["id"], [channel_id, other_channel_id], role="Viewer")

client.users.system_permissions(user["id"])
client.users.update_system_permissions(user["id"], {"user_management": True})

Autopilot

client.autopilots.create(name="Weekly", ai_model="openai/gpt-4o")
client.autopilots.activate(id)
client.autopilots.trigger_run(id)     # 202 — async, poll runs()
client.autopilots.runs(id, limit=10)

activate requires an active source, an API key, and a model. Sources and tone samples have no API endpoints — they live in the web UI, so an autopilot created entirely over the API cannot be activated until a source is added there.

Transactional email

client.transactionals.create(
    to="ada@example.com",
    subject="Receipt",
    body="<p>Thanks</p>",
    idempotency_key="receipt-{}".format(order.id),
)

The server stores the response for 24 hours keyed on (token, key) and replays it rather than sending a second email. The key is part of a fingerprint over method + path + body:

• same key, same payload, still running → ConflictError (409)
• same key, different payload → ValidationError (422)

That 422 means "this key was already used for something else", not that the email was invalid. Do not retry it with the same key.

This is the only endpoint that accepts Idempotency-Key.

Discovery

client.whoami()   # token identity and permissions
client.status()   # channel readiness — check before a send
client.prime()    # full capability manifest
client.skill()    # plain-text agent skill manifest (a str)

Migration / export

Read-only. Admin tokens only, and every call needs a broadcast_channel_id.

client = Broadcast(..., broadcast_channel_id=42)

client.migration.manifest()

for sub in client.migration.each_record("subscribers"):
    ...   # auto-pages; advances by the limit the server actually applied

# Suppressed addresses live in two lists; export both:
# suppressions = global suppression list, unsubscribed_emails = the channel's own
for row in client.migration.each_record("unsubscribed_emails"):
    ...   # {"id", "email", "broadcast_channel_id", "created_at", "updated_at"}

data = client.migration.download_file_asset(id)   # bytes

On a demo instance this entire API returns 403 for every request, valid token or not — deliberately, so a public demo cannot be used as a token oracle. It surfaces as AuthorizationError.

Channel Scoping

with client.with_channel(123):
    client.email_servers.list()   # scoped to channel 123

The previous scope is restored on exit, including when the block raises. The override lives on the client instance, so concurrent use across threads will interleave — use one client per thread, or pass broadcast_channel_id explicitly.

Webhooks

from broadcast_python import webhook

valid = webhook.verify(
    raw_body,                        # the raw bytes, not a re-serialised dict
    request.headers["X-Broadcast-Signature"],
    request.headers["X-Broadcast-Timestamp"],
    os.environ["WEBHOOK_SECRET"],
)
if not valid:
    return HttpResponse(status=401)

HMAC-SHA256 over timestamp.payload, v1,<base64> header format, 5-minute timestamp tolerance, constant-time comparison. verify returns False for every rejection rather than distinguishing them.

Pass the raw request body. Re-serialising a parsed dict changes the bytes and verification will fail.

broadcast_python.EVENT_TYPES lists all 37 event names. SUBSCRIBER_EVENTS includes subscribers.purged and subscribers.purge_failed: a purge of the whole list sends one of these instead of a subscriber.deleted per subscriber.

API Token Permissions

Each token can be scoped to specific resources. Use the minimum permissions your integration requires.

Troubleshooting

AuthenticationError (401)

• Check the token: it must be an API key from Settings > API Keys, not a password or a session cookie.
• Check the host: pointing at the wrong instance produces a valid-looking 401, because the token is unknown there.

AuthorizationError (403)

The token is valid but lacks the permission for that call. Check the table above, then re-issue the key with the resource enabled — permissions are fixed at creation.

On a demo instance, the entire migration API returns 403 for every request, valid token or not.

ValidationError (422) on a repeated send

If you reused an idempotency_key with a different payload, the API rejects it: the key is fingerprinted over method, path and body. It means "this key was already used for something else", not that the email was invalid. Use a new key.

Emails accepted but never delivered

Call client.status(). If readiness["transactionals"] is false, the channel has no usable email server or sender identity — the API accepts the request and the send stalls. On a demo instance, sends are always accepted and never delivered.

APIError mentioning a redirect

Your host is wrong — usually http instead of https, or a bare apex that redirects to www. The client refuses to follow redirects on writes, and never across hosts, because every request carries your API token. Set host to the final URL.

TimeoutError is not the builtin

broadcast_python.TimeoutError mirrors the Ruby gem's error hierarchy and is a distinct class from Python's builtin. except TimeoutError in a module that imported ours will not catch a socket timeout, and vice versa.

Development

python -m unittest discover -s tests    # mocked HTTP, no network

pip install ruff mypy
ruff check .
mypy

BROADCAST_LIVE_TEST=1 BROADCAST_HOST=http://localhost:3000 \
BROADCAST_API_TOKEN=... python -m unittest tests.test_live

The suite uses only the standard library — no pytest required, and no install step: PYTHONPATH=src:tests is the whole setup.

CI runs the suite on Python 3.9 through 3.14, lints with ruff, type-checks with mypy, and separately builds the wheel and installs it into a clean virtualenv to prove the packaged artifact imports and works.

Documentation

• Python SDK guide — the same material as this README, on the docs site
• API reference — endpoints, parameters, and permissions
• API response warnings — why a 2xx can still tell you something went wrong
• Webhook endpoints — signature format and event types
• Agents CLI — the same credentials, from a terminal

Other SDKs

All four cover the same 104 operations and behave the same way on the wire — the transport contract (warnings, idempotency, rate-limit handling, redirect safety, credential redaction) is identical across languages.

License

MIT