Python SDK

workwae is a typed Python client for every /v1 operation, built from the same contract as the API. It has a sync and an asyncio client, and ships helpers for cursors, errors and webhook signatures. The TypeScript client is in TypeScript SDK.

InstallCopy Anchor Link

The SDK needs Python 3.10 or later. Its runtime dependencies are httpx and pydantic (v2).

pip install workwae

Make a callCopy Anchor Link

import os

from workwae import Workwae

workwae = Workwae(api_key=os.environ["WORKWAE_API_KEY"])

account = workwae.accounts.get(account_id=account_id)
print(account.email, account.status)

Every answer is a pydantic model validated against the contract, with snake_case fields and date-time fields as datetime. Every contract type (Account, Message, Event, ...) is in workwae.types, under the name the API reference uses. EVENT_TYPES lists every event type at runtime.

Calling conventionsCopy Anchor Link

Methods are named after the operation ID in the API reference, with each dot segment in snake_case: accounts.get is workwae.accounts.get, calendarEvents.list is workwae.calendar_events.list, suppressions.entries.add is workwae.suppressions.entries.add.

Path, query and body fields are keyword arguments, also in snake_case. A field named from is spelled from_.

workwae.messages.list(account_id=account_id, mailbox_id=mailbox_id, unread=True, limit=50)

Nested objects take a model or a plain dict with the same snake_case keys. Each model X has an XParam TypedDict in workwae.types for it. The keys of a free-form map, such as custom headers, are sent as written.

queued = workwae.outbox.create(
    account_id=account_id,
    to=[{"name": "Ada Lovelace", "address": "ada@example.com"}],
    subject="Hello",
    text="Sent with Workwae.",
)

Four DELETEs that take query options are keyword arguments like the rest, with no params and query split: accounts.delete, messages.delete, calendar_events.delete and suppressions.delete.

workwae.accounts.delete(account_id=account_id, purge=True)

Omitted and NoneCopy Anchor Link

An argument left out is not sent. None is sent as null, which clears a nullable field:

workwae.accounts.update(account_id=account_id, external_ref=None)

The default of every optional argument is NOT_GIVEN, exported from workwae.

Binary answersCopy Anchor Link

An operation without a JSON answer returns None. A raw message, an attachment, a contact photo or an export returns a BinaryResponse with content, content_type and filename.

Client optionsCopy Anchor Link

Workwae takes these options:

Async clientCopy Anchor Link

AsyncWorkwae takes the same options, and its methods are coroutines. Its api_key may be an async function and its http_client an httpx.AsyncClient. Both clients close their HTTP client with close() or as a context manager.

from workwae import AsyncWorkwae

async with AsyncWorkwae() as workwae:
    account = await workwae.accounts.get(account_id=account_id)

Per-call optionsCopy Anchor Link

request_options is a dict with these optional keys:

AuthenticationCopy Anchor Link

API keys are created in the console under API keys, with the scopes the integration needs (read, write, send, admin). They reach every account of the tenant, so keep them on your server. See API keys.

To let a browser or mobile app read one account directly, mint a short-lived account token on your server and hand it over. It is limited to that one account:

token = workwae.accounts.mint_token(account_id=account_id, ttl_seconds=900)
print(token.token, token.expires_at)

Pagination helpersCopy Anchor Link

A listing answers items and next_cursor. Pass next_cursor back as cursor until it is None, or let a helper do it. paginate, pages and changes take the list method itself, so its arguments and item type stay checked. For an AsyncWorkwae method they return an async iterator.

Pass the method's arguments by keyword.

from workwae import changes, pages, paginate

for message in paginate(workwae.messages.list, account_id=account_id, unread=True):
    print(message.subject)

for page in pages(workwae.accounts.list, limit=200):
    print(len(page.items), page.next_cursor)

feed = changes(workwae.changes.list, cursor=load_cursor())
for event in feed:
    handle(event)
save_cursor(feed.cursor)

With AsyncWorkwae, iterate with async for:

async for message in paginate(workwae.messages.list, account_id=account_id):
    print(message.subject)

feed.cursor moves past a page only after its last event was consumed. Saving it after processing gives at-least-once delivery: stopping mid-page replays that page next time. See Pagination and Change feed.

ErrorsCopy Anchor Link

A refusal raises an APIError:

from workwae import APIError

try:
    workwae.outbox.create(account_id=account_id, to=[], text="hi")
except APIError as error:
    if error.code == "invalid_input":
        print(error.data["issues"])
    else:
        print(error.code, error.status, error.request_id)
        raise

workwae.ErrorReason is the union of every reason. A network failure is raised by httpx as it is (httpx.HTTPError). An answer that does not match its model raises pydantic.ValidationError.

Every reason and its status is in Errors.

RetriesCopy Anchor Link

The client retries an answer of 429 or 503, which the API sends before acting on a request, so a retry cannot apply it twice. Every attempt of a POST carries the same Idempotency-Key.

• The wait is the answer's Retry-After when it has one. Otherwise it is an exponential backoff with jitter.
• If Retry-After is longer than max_retry_delay, the client does not wait: it raises the error.
• Any other status, or the last attempt's answer, is returned or raised as is.

workwae = Workwae(max_retries=4, max_retry_delay=30)

Rate limits lists the limits behind 429.

Idempotency keysCopy Anchor Link

Every POST is sent with an Idempotency-Key, generated per call and reused across the client's own retries, so a retried send never goes out twice. To make a call safe to repeat across process restarts, such as a job queue retrying a send, pass a key of your own derived from your data:

workwae.outbox.create(
    account_id=account_id,
    to=[{"address": "ada@example.com"}],
    subject="Invoice 1042",
    text="...",
    request_options={"idempotency_key": "invoice-1042"},
)

How the API treats keys is in Idempotency.

WebhooksCopy Anchor Link

parse_webhook(secret, headers, body) checks the Standard Webhooks signature and returns the event, typed by type, with its date-times as datetime. Pass the raw request body exactly as received. headers is any mapping in any key case: a dict, or the request headers of Django, Flask, Starlette or httpx.

• A type this SDK knows narrows on event.type.
• A type added after this SDK version is an UnknownEvent, and its data is passed on unchecked.
• A failed check raises WebhookVerificationError. Its failure is missing_headers, timestamp, signature or payload (signed, but not a valid event).

import os

from workwae import UnknownEvent, WebhookVerificationError, parse_webhook


@app.post("/webhooks/workwae")
async def receive(request: Request) -> Response:
    body = await request.body()
    try:
        event = parse_webhook(os.environ["WORKWAE_WEBHOOK_SECRET"], request.headers, body)
    except WebhookVerificationError:
        return Response(status_code=400)
    if isinstance(event, UnknownEvent):
        return Response(status_code=204)
    if event.type == "send.failed":
        print(event.data.outbox_id, event.data.error.code)
    return Response(status_code=204)

To check only the signature, verify_webhook returns a boolean and check_webhook_signature returns the failure, or None when the signature is valid. All three take tolerance_seconds (default WEBHOOK_TOLERANCE_SECONDS, 300). See Webhook signatures.

TypesCopy Anchor Link

Every model of the API is in workwae.types, under the name the OpenAPI document gives it:

from workwae.types import Account, CalendarEvent, Event, EventType, Message, OutboxEntry

The SDK is versioned with semver. While it is 0.x, a minor version may carry breaking changes, which the changelog lists.