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

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

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

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.

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 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, AuthorizationError, RateLimitError, ValidationError,
)

try:
    client.subscribers.create(email="invalid")
except ValidationError:        # 422 — bad input, do not retry
    ...
except AuthorizationError:     # 403 — token lacks permission
    ...
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.

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

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.

Source and Support

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

• Source: github.com/send-broadcast/broadcast-python

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.