Python SDK

broadcast-python is the official Python client for the Broadcast API. It covers every API endpoint and works against any Broadcast instance.

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

Installation

pip install broadcast-python

Warning

The distribution is broadcast-python; the module you import is broadcast_python. A package named broadcast already exists on PyPI, so this client deliberately does not claim that name: import broadcast will either fail or hand you something unrelated.

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 and select only the permissions your integration needs
  5. Copy the token: it is shown once

See API Authentication for the full permission list.

Quick Start

from broadcast_python import Broadcast

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

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>",
)

Warning

host is required. There is no default. Broadcast is self-hosted first, so every instance lives at its own domain. A built-in guess would be wrong for almost everyone, so the client asks you to be explicit.

BROADCAST_HOST and BROADCAST_API_TOKEN are the same names the Agents CLI uses in ~/.config/broadcast/config.

Configuration

Broadcast(
    api_token="...",
    host="https://mail.example.com",
    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-supplied Retry-After
    warnings_mode="log",      # "log" | "raise" | "ignore"
    logger=logging.getLogger(__name__),
    debug=False,
    broadcast_channel_id=42,  # admin/system tokens only
)

A typo in a setting name raises TypeError rather than being silently ignored, so a misspelled timout fails immediately instead of leaving you wondering why the timeout had no effect.

Timeouts, rate-limited requests, and server errors are retried with backoff. A validation error (422) is never retried: it is deterministic, so a retry only adds latency.

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                  # parsed warnings, if any
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"].

Python and PHP can both do this; the Node client cannot, and documents its workaround.

Warnings

A 2xx response can carry warnings: the API accepted your request but ignored part of it, such as an unrecognised parameter or a filter it could not parse. A mistyped created_after silently widens a result set rather than failing.

for warning in result.warnings:
    logger.warning(str(warning))

Set warnings_mode="raise" to turn them into exceptions instead. Note the write has already happened by then: nothing is rolled back. See API Response 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

rate_limit is None when the server sends no limit headers, and also when the limit header cannot be parsed: a value the client cannot read makes the whole block untrustworthy, so it reports nothing rather than a partial answer.

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

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.unsubscribe("ada@example.com")
client.subscribers.redact("ada@example.com")   # irreversible

See Subscribers API for every filter and field.

Transactional email

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

Passing an idempotency_key makes a retry safe: the server stores the response for 24 hours and replays it rather than sending a second email. Check result.idempotent_replay to tell a replay from a fresh send.

Full details in Transactional Email API.

Broadcasts

broadcast = client.broadcasts.create(subject="Weekly update", body="<p>Hello</p>")

client.broadcasts.schedule(
    broadcast["id"],
    scheduled_send_at="2026-08-01T09:00:00Z",
    scheduled_timezone="UTC",
)

Warning

client.broadcasts.send(id) sends immediately to the whole audience. There is no unsend. Call client.status() first: if readiness["broadcasts"] is false, the channel has no usable email server or sender identity.

Sequences, segments, templates, opt-in forms

client.sequences.get(id, include_steps=True)
client.sequences.add_subscriber(id, email="ada@example.com")
client.segments.create(name="VIPs")
client.templates.create(label="Welcome", subject="Hi")
client.opt_in_forms.analytics(id, start_date=date(2026, 1, 1))

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

Topics

Subscriber Topics use the subscriber permissions.

topic = client.topics.create(name="Webinars", custom_data_key="sub_webinars")
client.broadcasts.create(subject="Webinar", body="...", topic_id=topic["id"])
# Change only the keys you send; None deletes a key
client.subscribers.update("jane@example.com", custom_data={"sub_webinars": False}, custom_data_mode="merge")

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

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.

Suppressions

client.suppressions.check("ada@example.com")   # will this address receive mail?
client.suppressions.list(page=1, email="example.com")
client.suppressions.add("bounced@example.com")
client.suppressions.remove("bounced@example.com")
client.suppressions.bulk_add(["a@example.com", "b@example.com"])   # up to 10,000
client.suppressions.bulk_remove(["a@example.com"])

check reads across both the channel’s list and the installation-wide global list, so it answers the question integrations actually ask: will this address receive mail? Adding an already-suppressed address is a success (200 rather than 201), so there is no need to check first.

The global list is a separate resource and needs an admin (system) token:

client.global_suppressions.list()
client.global_suppressions.add("spamtrap@example.com")
client.global_suppressions.remove("spamtrap@example.com")
client.global_suppressions.bulk_add([...])
client.global_suppressions.bulk_remove([...])

See Suppressions API for the full semantics.

Channel settings and brand kit

client.channel_settings.get()                      # general, sender, unsubscribe_page, confirmation_page, opt_in_forms
client.channel_settings.update({"unsubscribe_page": {"legal_notice_url": "https://example.com/impressum"}})

client.channel_design.get()                        # the brand kit
client.channel_design.update({"colors": {"accent": "#ff5500"}, "typography": {"font": "georgia"}})

Updates are partial: send only what changes, and the response is the whole object after the change. Read result.warnings for keys the server ignored. Changing either one needs the Channel Settings write permission. The brand kit switches (match_brand_kit in unsubscribe_page, confirmation_page and opt_in_forms) are channel settings, not part of the brand kit update. The brand kit logo is the logo_url of an image in the channel’s file library. Reading and changing channel settings, and changing the brand kit, need Broadcast 2.46.0 or later and the next release of this SDK after 0.5.0. See Channel Settings API and Brand Kit API.

Users

Create and manage the people who sign in to Broadcast, and what they can reach. This needs an admin (system) token with the Users permission, and no channel: a channel token gets AuthorizationError.

client.users.list(q="ada", status="active")
user = client.users.create(email="ada@example.com", first_name="Ada", last_name="Lovelace",
                           send_password_reset=True)   # or password="..."
client.users.get(user["id"])                           # includes system and channel permissions
client.users.update(user["id"], last_name="King")
client.users.deactivate(user["id"])
client.users.activate(user["id"])                      # also clears a lockout
client.users.delete(user["id"])

Channel access. A write replaces the user’s whole record on that channel. Pass exactly one of permissions, role or preset_id:

client.users.channel_permissions(user["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.bulk_channel_permissions(user["id"], [3, 4], preset_id=12)
client.users.remove_channel_permissions(user["id"], channel_id)

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

What to expect:

  • Sudo users are read-only. Changing one, or their permissions, raises AuthorizationError, and sudo access can never be granted through the API.
  • Nothing is silently ignored. A request the API will not apply (such as sudo, active, or permissions inside the user payload) raises ValidationError and changes nothing. Use deactivate and activate for sign-in access.
  • A bulk apply that applies to no channel raises ValidationError; otherwise check its failed list.
  • Passing none, or more than one, of the permission options raises ValueError before any request is sent.

See Users API for every field and rule, and Users & Permissions for what the roles and permissions mean.

Channel Scoping

Admin (system) tokens can address any channel:

with client.with_channel(123):
    client.email_servers.list()

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["BROADCAST_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 32 event names.

See Webhook Endpoints API for creating and managing endpoints.

Errors

from broadcast_python import (
    APIError, AuthenticationError, AuthorizationError, RateLimitError, ValidationError,
)

try:
    client.subscribers.create(email="invalid")
except ValidationError:        # 422: bad input, do not retry
    ...
except AuthenticationError:    # 401: bad token, or it lacks this permission
    ...
except AuthorizationError:     # 403: needs an admin token, or the user is sudo
    ...
except RateLimitError as e:    # 429
    time.sleep(e.retry_after or 5)
except APIError:               # 5xx and other transport failures
    ...

ValidationError and TimeoutError are siblings of APIError, not children, so catching APIError for transport problems will not silently swallow a validation error.

Warning

broadcast_python.TimeoutError mirrors the Ruby gem’s hierarchy and is not Python’s builtin TimeoutError. In a module that imported ours, a bare except TimeoutError will not catch a socket timeout, and vice versa.

Export & Migration

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

client = Broadcast(api_token="...", host="...", broadcast_channel_id=42)

client.migration.manifest()          # sizes the export

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

for row in client.migration.each_record("unsubscribed_emails"):
    ...   # the channel's own suppression list

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

The channel’s own suppression list is a separate collection from suppressions, which is the global list, so export both. It needs Broadcast 2.40.0 or later and version 0.3.0 or later of this SDK.

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.

Source and Support

The client is open source and its README documents every method:

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.

Prefer another language? See the Ruby SDK, the PHP SDK, or the Node SDK.

Last updated

Was this page helpful?

Thanks for your feedback!

Thanks for letting us know. We'll work on improving this page.