Dialpad client

Overview

The module exposes a high-level Client, resource objects such as User and CallCenter, and a lower-level client.endpoints interface. It uses aiohttp_ws.AiohttpResponse for HTTP and toolbox.CachedProperty for async caching.

Imports and setup: install requests and aiohttp, and make the distributed pmblue_update, toolbox, and aiohttp_ws modules importable. Importing dialpad calls pmblue_update.self_update(primary_mod=True), which can check the network, write an updated module and updater cache, or prompt to publish a locally newer/missing module. Dependencies also have import-time updater behavior. See the updater guide before importing in an unattended service.

Quick start

1. Create one client for the Dialpad environment.
2. Optionally declare webhook subscriptions before initialization.
3. Call initialize() once at application startup.
4. Use high-level client or resource methods for normal work.

import asyncio
import os
from dialpad import Client, Subscription

async def run():
    client = Client(os.environ["DIALPAD_API_KEY"], sandbox=True)

    webhook = await client.create_webhook(
        "https://example.com/webhooks/dialpad",
        os.environ["DIALPAD_WEBHOOK_SECRET"],
    )
    webhook.add_subscription(Subscription.agent_status)
    webhook.add_subscription(Subscription.call)

    await client.initialize(verbose=True)

    user = await client.find_user("[email protected]")
    if user:
        await user.initiate_call(
            "+1 (202) 555-0123",
            custom_data="CRM lead 123",
        )

asyncio.run(run())

This example uses the sandbox. Set sandbox=False only for a client intended to use production. Calls, SMS, callbacks, and webhook reconciliation make remote changes when awaited.

Core concepts

Users and groups

The client loads a maximum of 50 users plus all returned departments and call centers. Deleted groups are omitted.

users = await client.users
departments = await client.departments
contact_centers = await client.contact_centers  # alias of call_centers
groups = await client.groups                    # departments + call centers

by_id = await client.get_user_from_id(123456)
by_email = await client.find_user("[email protected]")
by_name = await client.find_user("Ada Lovelace")

find_user() accepts a User, integer ID, numeric string, email address, or full name. Email and name comparisons are case-insensitive and ignore spaces. It returns None when nothing matches.

Group membership state

Department and CallCenter inherit mapping-like behavior from Group. Membership tests accept a user object or ID; indexed values are booleans representing the locally known active state.

support = await client.get_group_from_id(987654)
agent = await client.find_user("Ada Lovelace")

if agent is not None and support is not None and agent in support and support[agent]:
    print("Agent is active in Support")

if agent is not None:
    for group in await agent.groups:
        print(group.name, group.group_type)

Calls and callbacks

Phone numbers are normalized with toolbox.PhoneNumber(...).e164 before they are sent. The same operations are available from the client, a user, or a group. The following snippets assume the client has been initialized and directory lookups returned a user/group; check for None when using real input.

Initiate an outbound call

user = await client.find_user("[email protected]")
group = await client.get_group_from_id(987654)

await client.initiate_call(
    user,
    "(202) 555-0123",
    group=group,
    custom_data="customer=42",
)

# Equivalent convenience call
await user.initiate_call("202-555-0123", group=group)

user may be a User, integer ID, numeric string, or exact email address. group may be a Group, a group ID, {"group_id": ..., "group_type": ...}, or [group_id, group_type]. If supplied, custom_data must be a string.

Ring and consult calls

await user.initiate_call_via_ring(
    "+12025550123",
    group=group,
    is_consult=True,
    custom_data="warm transfer",
)

The ring variant posts to /call. With is_consult=True, the request includes Dialpad's consult-call flag.

Queue a callback

from dialpad import CallCenter

center = await client.get_group_from_id(987654)
if isinstance(center, CallCenter):
    await center.enqueue_callback("+12025550123")

SMS and media

SMS is currently exposed on the endpoint layer.

await client.endpoints.send_sms(
    "Your appointment is confirmed.",
    to_numbers=["+12025550123", "+12025550124"],
    user_id=123456,
    infer_country_code=False,
)

await client.endpoints.send_sms(
    "See attachment",
    to_number="+12025550123",
    media="receipt.png",
)

Webhooks and subscriptions

A WebhookManager describes the desired remote webhook state. During initialization it reuses an exact URL-and-secret match or creates one, removes undesired subscriptions, and adds missing subscriptions.

manager = await client.create_webhook(WEBHOOK_URL, WEBHOOK_SECRET)
manager.add_subscription(Subscription.agent_status)
manager.add_subscription(Subscription.call)
manager.add_subscription(Subscription.changelog)

await client.initialize()

manager.initialize() is idempotent for that manager instance after it becomes active. manager.delete() removes its remote webhook and marks the manager inactive.

Inbound event processing

Register callbacks

@client.event
async def on_agent_status(event):
    print(event.user_id, event.on_duty_status)

@client.event
async def on_call(event):
    print(event.call_id, event.state)

@client.event
async def on_change(event):
    print(event.action)

Handlers must use exactly one of those three names and must be async. on_call runs after call aggregation, so it can look up the updated call. on_agent_status and on_change run before their respective local-state updates. Exceptions in callbacks propagate to the caller of process().

Process the request

from aiohttp import web

async def dialpad_webhook_handler(request):
    try:
        event = await manager.process(request)
    except (ValueError, TypeError, IndexError, KeyError):
        return web.Response(status=400, text="Invalid webhook payload")

    if event is None:
        return web.Response(status=400, text="Unsupported event")

    return web.Response(text="ok")

Classification and state changes

Recordings

recording and admin_recording events inspect recording_url, admin_call_recording_share_links, and compatible recording_details. The first HTTPS URL becomes a File; absent or unsupported URLs leave recording=None.

from dialpad import CallLog

# Use the event returned by the verified handler; process each event once.
if isinstance(event, CallLog) and event.recording is not None:
    await event.recording.download("recording.mp3")
    # Alternatively: recording_bytes = await event.recording.download()

File.download() calls raise_for_status() before writing. Its async context manager returns downloaded bytes and propagates exceptions from the block. Requests to recording URLs include the client's bearer token, so only use recording URLs from trusted, verified events.

Client API

Construction and lifecycle

Lookups and collections

Actions

Resource objects

JWT helpers

Endpoint API

Use client.endpoints when the high-level interface does not expose the operation you need.

The internal Client._request(path, method="GET", **kwargs) adds authorization and accept headers, accepts absolute URLs for downloads, and supports GET, POST, and DELETE.

Caching and consistency

With asynchronous_clients=False, inner _cache entries can outlive the decorator expiration: a refreshed property may return the same list without another API request. The module mutates these local objects as events arrive. With it set to True, directory properties query more often, but some changelog paths also write directly into local cache attributes. Treat the option as a consistency hint, not a distributed cache protocol.

Errors, limits, and caveats

Common exceptions

Object handbook

The client owns API configuration and caches; directory resources point back to that client. Webhook managers connect incoming signed requests to event objects and update the same cached users, groups, and calls. Constructors that accept d expect a parsed response mapping, not an ID. Obtain resources from the client unless you deliberately need to parse fixture data.

Client and endpoint adapter

Client(api_key, sandbox=True, asynchronous_clients=False) creates local state without fetching directory data. Its public settings are api_key, sandbox, and asynchronous_clients; url derives the API base from the sandbox setting. endpoints is a Client._Endpoints instance bound to this client. live_calls is initialized as a separate list but is not populated by the call aggregation code; read await client.calls for tracked calls.

await initialize(verbose=False) loads departments, call centers, then users and initializes registered webhook managers. It returns None. await force_update() drops and reloads the three directory caches; use it only after initialization with the cache assumptions described above. Neither method starts an HTTP listener or background polling loop. There is no client close() method; request transport is delegated to the HTTP wrapper.

The endpoint adapter's list_* methods return resource lists. create_webhook(hook_url, secret=None) and create_subscription(sub_type, endpoint_id) return newly parsed resource objects. Deletion methods return None; they make a remote request when awaited. Calls, callbacks, screen pops, and SMS return raw responses. Await the operation first, then inspect its status_code, call raise_for_status(), or read json(). The endpoint adapter delegates missing attributes to its parent client.

User

User(d, c) requires directory fields including id, company_id, is_admin, is_online, is_available, first_name, last_name, license, and a nonempty emails list; c is the owning client. The object stores the first email as email, converts id to an integer, preserves data, and exposes computed name. Duty status and reason start as None and are populated by agent-status events.

await user.groups resolves the IDs in the user's original data["group_details"] to Department/CallCenter resources, with a 120-second decorator cache. await user.screen_pop(uri) asks Dialpad to display a URI for that user. await user.initiate_call(phone_number, group=None, custom_data=None) and await user.initiate_call_via_ring(phone_number, group=None, is_consult=False, custom_data=None) forward the user ID and return raw responses. These methods make remote changes. User equality compares IDs against another user, an integer, or a string.

Group, Department, and CallCenter

Group(d, c) parses integer id, name, and raw data; it starts with an empty operator-state mapping. Department(d, c) sets group_type="department". CallCenter(d, c) sets group_type="call_center", which outgoing call methods normalize to the API's callcenter.

user in group and group[user] accept either a User or ID and inspect locally tracked membership/availability. A missing indexed ID raises KeyError. Assignment and deletion should use IDs: group[user.id] = True and del group[user.id] modify only local state, without changing remote membership. User construction and webhook processing normally maintain this map for you.

await group.initiate_call(user, phone_number, custom_data=None) and the corresponding ring method supply this group automatically. await center.enqueue_callback(phone_number) is available on CallCenter and passes its ID to the client's callback method. Use concrete Department/CallCenter objects for calls: the base Group constructor does not assign group_type. Group removal helpers and destructors affect local cache membership; they are not API methods for deleting remote departments or call centers.

WebhookManager, Webhook, and Subscription

WebhookManager(url, secret, c) can be constructed directly, but await client.create_webhook(url, secret) also registers the manager for client initialization. Its client, url, and secret describe the desired connection; webhook begins as None and later refers to a remote Webhook. add_subscription(sub_type) accepts the three Subscription constants, avoids local duplicates, and returns None.

await manager.initialize() first matches a webhook by URL and secret, creates one if needed, and reconciles subscriptions. The first successful call returns its Webhook; subsequent calls return None when already active. Reconciliation may delete subscriptions on that matched remote webhook if they are absent from the manager's desired list. await manager.subscriptions requires initialization and delegates to its webhook. await manager.delete() marks the manager inactive and delegates remote deletion; the existing deletion caveat still applies.

Webhook(d, c) expects webhook JSON with hook_url, id, and signature; its adapter argument is normally client.endpoints. It exposes hook_url/url, id, secret, and data. await webhook.subscriptions filters the client's subscriptions by webhook ID. await webhook.create_subscription(sub_type) returns a Subscription and makes a remote change.

Subscription(d, sub_type, c) likewise normally receives the endpoint adapter. Its id, sub_type/type, and optional webhook_id identify the remote subscription. await subscription.delete() deletes it remotely and removes a matching entry from its typed local list, returning None. The class constants are plain strings: agent_status, call, and changelog.

JWT, CallLog, AgentStatus, and Changelog

JWT(raw, secret) takes token text and a shared secret, exposes raw, parts, partial_msg, parsed data, and boolean verify. Mapping access jwt[key], key in jwt, and jwt.get(key, default=None) reads the decoded payload. String field values that themselves contain JSON are decoded again. Construction alone does not enforce validity; manager.process(request) does.

CallLog(d, client=None) requires state and call_id. It exposes the master/entry-point/operator IDs, direction, external number, custom data, and optional recording. It falls back to call_id when the master ID is absent. AgentStatus(d, client=None) requires target ID, duty fields, and a group-details mapping; it exposes user_id, on_duty_status, on_duty_status_reason, and groups. on_duty is true for available, occupied, wrapup, and wrapup-end. await agent_event.user resolves a User or None.

Changelog(d, client=None) requires action, keeps data, and normalizes an absent/empty changed_by to None. await change.changed_by_obj resolves a user only when the actor type is exactly User; other types return None. Supported processing includes selected user/group refreshes, admin-flag changes, and local group deletion. Some actions, including operator_added, currently have no state-update implementation.

These constructors parse data without dispatching callbacks or updating client state. The normal input path is await manager.process(request), which verifies, classifies, dispatches, and processes the event. It returns the event object, or None for an unrecognized signed payload.

Call and File

Call(master_call_id) initializes an aggregate with id, master_call_id, call_id, state/direction unknown, empty call_ids/call_logs, and no external number/latest user. Processing appends logs in arrival order, remembers related IDs, and updates current fields. id stays at its construction value even if master_call_id later changes. latest_user_id updates for a target mapping whose type is exactly lowercase user. Aggregates are in-memory history; there is no sorting, deduplication, persistence, or eviction.

File(url, client) stores a URL and uses that client for an authenticated GET. await file.download(to_file=None) returns bytes with no destination, or writes to a path/open writable binary object and returns None. A supplied file object is closed after writing. async with file as data downloads bytes on entry and does not suppress block exceptions. The module does not stream large recordings to disk; it buffers the response.

Worked examples

Run the asynchronous helpers below from an existing event loop. Examples use sandbox configuration or an already configured client; lookup IDs and phone numbers are examples. Imports retain the updater behavior described earlier.

Return a directory snapshot for an application

async def directory_snapshot(client):
    await client.initialize()
    result = []
    for user in await client.users:
        memberships = []
        for group in await user.groups:
            active = group[user.id] if user.id in group else None
            memberships.append({
                "id": group.id,
                "name": group.name,
                "type": group.group_type,
                "active": active,
            })
        result.append({
            "id": user.id,
            "name": user.name,
            "email": user.email,
            "groups": memberships,
        })
    return result

This produces ordinary dictionaries for rendering or serializing. active=None means the local membership map has no status for that user; it does not prove the user is unavailable. The endpoint still fetches only its first 50 users.

Screen pop and send an SMS for a resolved user

async def notify_agent_and_customer(client, email, customer_number):
    agent = await client.find_user(email)
    if agent is None:
        raise LookupError("Agent was not found in the loaded directory")

    screen = await agent.screen_pop("https://example.com/customer/42")
    screen.raise_for_status()
    message = await client.endpoints.send_sms(
        "Your appointment is confirmed.",
        to_number=customer_number,
        user_id=agent.id,
        infer_country_code=False,
    )
    message.raise_for_status()
    return message.json()

This workflow performs two independent remote actions. A successful screen pop is not rolled back if SMS fails. Pass the SMS number in the expected international format; the SMS wrapper does not call the phone normalizer used by the call methods.

Inspect the updated call inside an event callback

def register_call_observer(client):
    @client.event
    async def on_call(event):
        aggregate = await client.get_call_from_id(event.call_id)
        if aggregate is None:
            return
        print({
            "master_call_id": aggregate.master_call_id,
            "current_call_id": aggregate.call_id,
            "state": aggregate.state,
            "related_ids": list(aggregate.call_ids),
            "events_received": len(aggregate.call_logs),
        })
        if event.recording is not None:
            # Process the returned bytes or choose a local destination.
            recording_bytes = await event.recording.download()
            print("Recording bytes:", len(recording_bytes))
    return on_call

Register this observer before processing incoming requests. The client stores one callback per event name, so registering another on_call replaces the prior callback. In version 1.3 the aggregate has already been updated when on_call runs. Downloading a recording inside the callback delays the webhook response; an application that needs immediate acknowledgment should place verified work in its own queue.

Connect a manager to an aiohttp application

import os
from aiohttp import web
from dialpad import Client, Subscription

async def create_app():
    client = Client(os.environ["DIALPAD_API_KEY"], sandbox=True)
    manager = await client.create_webhook(
        os.environ["DIALPAD_WEBHOOK_URL"],
        os.environ["DIALPAD_WEBHOOK_SECRET"],
    )
    manager.add_subscription(Subscription.call)
    register_call_observer(client)
    await client.initialize()

    async def receive(request):
        try:
            event = await manager.process(request)
        except (ValueError, TypeError, KeyError, IndexError):
            return web.Response(status=400, text="Invalid webhook")
        if event is None:
            return web.Response(status=400, text="Unsupported event")
        return web.Response(status=204)

    app = web.Application()
    app.router.add_post("/webhooks/dialpad", receive)
    return app

# In a dedicated server entry point:
# web.run_app(create_app(), port=8080)

Set the public webhook URL to the external address that reaches this route. Startup reconciles that URL/secret's remote subscriptions to the desired call subscription. The sample handler does not deduplicate deliveries; keep any application event-ID tracking outside this module.