Python SDK

The official Python client for the DialerBee API, version 1: leads in, results out.

• Package: dialerbee (version 1.0.0rc4), built from API contract 1.0.0-rc4.
• Python 3.10 or later. Depends on httpx, anyio (4.14.2 or later) and pydantic 2.
• Status: release candidate, not published yet. Until DialerBee publishes it on PyPI, install it from a source checkout.

Server-side only. An API key can read your call records and change your campaigns. Keep it in your server's secret store. Never put it in a browser, mobile or desktop app, and never commit it.

Install

pip install dialerbee          # after the first release; for now: pip install ./developer/sdk-python

Authenticate

Create an API client and a key in the DialerBee console. A key looks like dbk_live_<credential id>_<secret> and is shown once. Legacy console keys (eng_live_...) do not work with this API.

export DIALERBEE_API_KEY='dbk_live_...'

from dialerbee import DialerBee

db = DialerBee()                  # or DialerBee(api_key=...)
me = db.me.get()                  # works with any active key, even one with no scopes
print(me.client.name, me.account.name, me.scopes)

The SDK sends the key only in the Authorization: Bearer header. It never puts the key in a URL, a log line, an exception or a repr(), and a client object cannot be pickled. dbk_test_ keys answer 401 invalid_api_key until the sandbox is enabled for your account.

Import leads

Leads go into a draft, ready or paused campaign through the platform's own validation (phone checks, duplicates, field rules, Do-Not-Call screen). At most 200 leads per request.

result = db.leads.import_(
    campaign_id="cmp_0f8fad5b-d9cb-469f-a165-70867728950e",
    leads=[
        {"phone": "+12125550101", "external_id": "CRM-1", "full_name": "Ann Lee"},
        {"phone": "+12125550102", "external_id": "CRM-2", "custom_fields": {"tier": "gold"}},
    ],
    idempotency_key="crm-batch-2026-09-29-001",   # your own id for this batch
)
print(result.summary.accepted, "accepted of", result.summary.received)
for row in result.data:
    if row.status != "accepted":
        print(row.index, row.external_id, row.status, row.reason)

Each row has its own status (accepted, duplicate, on_dnc, invalid, blocked_country, failed). A lead's phone number is never changed by leads.update; import the new number instead.

Lists and pagination

Every list returns one page. Iterate the page to walk all pages; the SDK repeats your filters with the opaque cursor and stops with an error if a cursor ever repeats.

page = db.calls.list(direction="outbound", limit=100)
print(len(page.data), page.has_more)

for call in page:                 # this page, then every following page
    print(call.id, call.status, call.duration_seconds)

calls.list returns every call of the account, from every team, campaign and agent.

Calls, recordings, transcripts

call = db.calls.get("call_0f8fad5b-d9cb-469f-a165-70867728950e")

rec = db.calls.recording(call.id)                      # whole file in memory
rec.write_to("call.mp3")

with db.calls.stream_recording(call.id) as stream:     # large files
    with open("call.mp3", "wb") as fh:
        for chunk in stream.iter_bytes():
            fh.write(chunk)

part = db.calls.recording(call.id, byte_range="bytes=0-65535")   # status_code 206

transcript = db.calls.transcript(call.id)
summary = db.calls.summary(call.id)                    # needs the summaries:read scope
if summary.status == "ready":
    print(summary.summary, summary.next_action)
elif summary.status == "not_generated":
    print("no summary:", summary.not_generated_reason)  # e.g. not_sampled, disabled, expired

Every call of your account answers a summary with a status: ready, processing or not_generated (the reason is in not_generated_reason; treat a reason you do not know as "not generated"). Calls handled by people and calls handled by an AI agent both have one. Answers may gain fields within v1: the SDK keeps a field it does not know (read it as an attribute) and never fails on it. A request with a field the API does not know is refused before anything is sent.

Campaigns

progress = db.campaigns.progress("cmp_0f8fad5b-d9cb-469f-a165-70867728950e")
db.campaigns.pause("cmp_0f8fad5b-d9cb-469f-a165-70867728950e")      # reduces contact: Tier 1

Webhooks

endpoint = db.webhooks.create(url="https://example.com/dialerbee", events=["call.completed"])
secret = endpoint.secret          # shown once: store it in your secret store

The signing secret is in the first answer only: DialerBee never stores the secret. If the same call is replayed (same idempotency_key; endpoint.response_info.idempotency_replayed is True), secret is None. The same holds for db.webhooks.rotate_secret(...). If you lost the secret, rotate it.

Verify every delivery with the exact raw request body, before parsing it:

from dialerbee.webhooks import construct_event, WebhookSignatureError

def handle(raw_body: bytes, headers: dict) -> int:
    try:
        event = construct_event(raw_body, headers, secrets=[secret])
    except WebhookSignatureError as e:
        return 400                                  # e.reason says why
    if already_processed(event.id):                 # deliveries are at-least-once
        return 204
    if event.type == "call.completed":
        print(event.data.call_id, event.data.duration_seconds)
    return 204

• Flask: construct_event(request.get_data(), request.headers, secrets=[...]).
• FastAPI / Starlette: construct_event(await request.body(), request.headers, secrets=[...]).
• Django: construct_event(request.body, request.headers, secrets=[...]).

A delivery is refused when it was signed more than 300 seconds from your clock (tolerance= changes it). After db.webhooks.rotate_secret(...) DialerBee signs with both secrets for 24 hours, so you can switch at your own pace: pass secrets=[new, old] during the switch. Unknown event types come back as WebhookEnvelopeBase; unknown fields are kept. dialerbee.webhooks.sign() builds a valid header to test your receiver.

Errors

Errors are RFC 9457 problems turned into exceptions. Branch on status_code and code, never on detail, and quote request_id to support.

import dialerbee

try:
    db.campaigns.resume(campaign_id)
except dialerbee.TierNotEnabledError:
    ...                                 # Tier-2 action not switched on for your account
except dialerbee.RateLimitError as e:
    print("retry after", e.retry_after)
except dialerbee.APIStatusError as e:
    print(e.status_code, e.code, e.request_id, [(x.pointer, x.code) for x in e.errors])
except dialerbee.APIConnectionError:
    ...                                 # no answer at all (network, DNS, TLS, timeout)

New error codes can appear inside v1; the SDK keeps them in code and picks the class by status.

Retries and idempotency

The SDK retries automatically (by default up to 2 times) only when it is safe:

• only on 429, 502, 503, 504, on 409 conflict (the first request with the key is still running) and on network errors;
• only for GET, for a read-only POST (dnc.check: it writes nothing, so it is retried like a GET and needs no key), or for a write that carries an Idempotency-Key, and always with the same key;
• never on outcome_unknown, idempotency_conflict, daily_cap_reached, replay_limit_reached or tier_not_enabled. replay_limit_reached is a daily cap (5 replays per event per UTC day), so an automatic retry cannot succeed;
• upstream_invalid_response (502) is retried only under the rule above: a GET, a read-only POST, or a write with the same Idempotency-Key, so a retry can never run the write twice;
• it waits Retry-After when sent (and gives up if that is longer than 60 s), otherwise an exponential delay from 0.5 s with full jitter, capped at 8 s.

Every write that needs an Idempotency-Key gets one generated for that call when you pass none. To also survive a crash of your own process, create the key yourself (for example dialerbee.new_idempotency_key()), store it with your record, and pass idempotency_key= on every attempt. Keys are scoped to your API client and kept for 24 hours.

OutcomeUnknownError means DialerBee could not confirm whether the write happened. Do not retry with a new key. Retry later with the same idempotency_key (it is on the exception) to get the settled answer, or contact support with request_id.

max_retries=0 turns retries off; db.with_options(max_retries=0, timeout=5) changes them for a few calls.

Tier 2 actions

Resuming a campaign, creating or moving callbacks and appointments, recording consent and sending SMS can reach or unlock real people. They need their own scope and an account switch that DialerBee keeps off unless it is agreed with you; until then they raise TierNotEnabledError. db.me.get().tier2 shows which are on. SMS sending is blocked for every account in API v1.

API v1 can check and add Do-Not-Call numbers but cannot remove them. Removing a number from the Do-Not-Call list stays a console action for an account administrator.

Configuration

import httpx
from dialerbee import DialerBee

db = DialerBee(
    api_key="dbk_live_...",
    timeout=30,                      # seconds, or an httpx.Timeout (default 60 s, 10 s connect)
    max_retries=2,
    http_client=httpx.Client(proxy="http://proxy.internal:3128"),  # optional; you close it
)

• base_url must be https:// (plain http:// is accepted only for localhost).
• Redirects are never followed, so the key is never sent anywhere else.
• Per call: correlation_id= (sent as X-Correlation-ID, logged with the request, never identity) and timeout=.
• if_match= sends If-Match; the ETag comes from obj.response_info.etag.
• Every returned object has response_info (status_code, request_id, etag, idempotency_replayed).
• Logging uses the dialerbee logger: method, path, status, request id and retries. It never logs the key, headers, query values or bodies.

Async

import asyncio
from dialerbee import AsyncDialerBee

async def main() -> None:
    async with AsyncDialerBee() as db:
        page = await db.calls.list(limit=100)
        async for call in page:
            print(call.id)

asyncio.run(main())

Models

All request and response models are generated from the API contract (dialerbee.models). Response models keep fields that DialerBee adds later inside v1, so an older SDK keeps working. Request models refuse unknown fields and naive datetimes before anything is sent; timestamps go out in UTC.