workwae

The Python client for the Workwae API: one REST API for Gmail, Microsoft 365 and IMAP mailboxes, Google, Microsoft and CalDAV calendars, and contacts. Every method, argument and answer is typed from the API's OpenAPI document.

• Sync and asyncio clients with a method for every /v1 operation
• Answers as pydantic models, with date-times as datetime
• Cursor helpers for every listing and the change feed
• Errors keyed by stable reasons, with the request id
• Automatic retries of 429 and 503 answers, honoring Retry-After
• Idempotency keys on every POST
• Webhook signature verification (Standard Webhooks) and typed event parsing

The full guides and API reference are at https://docs.workwae.com (/reference for every operation, /changelog for what changed).

Install

pip install workwae
# or
uv add workwae

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

Create a client

import os

from workwae import Workwae

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

page = workwae.accounts.list(limit=50)
for account in page.items:
    print(account.email, account.status)

Option	Meaning
api_key	An API key (wk_...), an account token (wat_...), or a function returning one per request. Defaults to WORKWAE_API_KEY.
base_url	Defaults to https://api.workwae.com. Set it only to route through your own proxy or gateway.
headers	Extra headers sent with every request.
timeout	Seconds, or an httpx.Timeout. Default 60.
max_retries	Further attempts after a 429 or 503. Default 2. See Retries.
max_retry_delay	Longest single wait, in seconds. Default 60.
http_client	Replaces the httpx.Client the client makes (proxies, transports, tests).

AsyncWorkwae takes the same options (api_key may be an async function, http_client an httpx.AsyncClient) and its methods are coroutines. Both 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)

Methods are named after the operation, grouped by resource, in snake_case: workwae.accounts.get, workwae.outbox.create, workwae.calendar_events.list, workwae.suppressions.entries.add. Path, query and body fields are keyword arguments, also in snake_case. Nested objects take the model or a plain dict with the same snake_case keys (each model X has an XParam TypedDict for it). from is spelled from_.

message = workwae.messages.get(account_id=account_id, message_id=message_id)

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

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

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

request_options carries per-call options: idempotency_key, timeout and headers.

Answers are pydantic models (workwae.types), whose fields are snake_case. Operations without a JSON answer return None, or a BinaryResponse (content, content_type, filename) for a raw message, an attachment, a contact photo or an export.

Authentication

API keys are created on 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. 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

Listings answer items and next_cursor; pass next_cursor back as cursor until it is None. The helpers take the method itself, so its arguments and item type stay checked, and give an async iterator for an AsyncWorkwae method:

from workwae import changes, pages, paginate

# Every item, to the last page.
for message in paginate(workwae.messages.list, account_id=account_id, unread=True):
    print(message.subject)

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

# The change feed, until caught up. Persist feed.cursor after processing and resume from it:
# a page is only passed once its last event was consumed (at-least-once delivery).
feed = changes(workwae.changes.list, cursor=load_cursor())
for event in feed:
    handle(event)
save_cursor(feed.cursor)

Errors

An API refusal is raised as an APIError whose code is the error's stable reason (account_not_found, invalid_input, plan_limit_sends, ...), status its HTTP status, message a human-readable text, data the reason's details when it has any, defined whether the operation declares the reason, and request_id the answer's X-Request-Id (quote it to support).

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.types.ErrorReason is the union of every reason. An answer whose body is not an API error has code None. Network failures are raised by httpx as they are (httpx.HTTPError); an answer that does not match its model raises pydantic.ValidationError.

Retries

429 (rate_limited, too_many_requests) and 503 (overloaded) answers are refused before the API acts, so the client retries them: twice by default, waiting for Retry-After, or backing off exponentially with jitter when there is none. A Retry-After longer than max_retry_delay is not waited for: the error is raised.

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

Idempotency keys

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 (a job queue retrying a send, say), 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"},
)

The API keeps the first answer for 24 hours per tenant and key and replays it for a repeat with the same body. The same key with a different body is refused with idempotency_key_reused; a repeat while the first call still runs with idempotency_key_in_progress.

Webhooks

Workwae signs deliveries with Standard Webhooks. parse_webhook verifies a delivery and returns its event, typed by type, with date-times as datetime. Pass the raw body exactly as received:

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):
        # An event type added after this SDK version.
        return Response(status_code=204)
    if event.type == "send.failed":
        print(event.data.outbox_id, event.data.error.code)
    return Response(status_code=204)

verify_webhook(secret, headers, body) only checks the signature and answers a boolean; check_webhook_signature answers why it failed (missing_headers, timestamp, signature). Both take tolerance_seconds (default WEBHOOK_TOLERANCE_SECONDS, 300). headers is any mapping in any key case: a dict, or the request headers of Django, Flask, Starlette or httpx.

Types

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

EVENT_TYPES lists every event type at runtime (for webhook endpoint filters).

Versioning

The SDK follows semver from 0.1.0; while it is 0.x, a minor version may carry breaking changes, always listed in the changelog.