Python SDK

talki is the Python client for /v1. It is generated from the same OpenAPI document as the API reference, and adds errors, retries, paging, webhook verification and OAuth on top.

pip install --pre talki   # or: uv add --prerelease=allow talki

Python 3.11 or later. It depends on httpx, attrs and anyio. The package is in beta: --pre is needed until 1.0, and its own interface may still change between beta releases. Source, changelog and the full README are on GitHub.

Create a client

from talki import Talki

with Talki(api_key="talki_test_…") as talki:
    account = talki.account.get()
    print(account.business_name, account.livemode)

• Pass api_key= for an API key, or access_token= for an OAuth access token. With neither, Talki() reads the key from the TALKI_API_KEY environment variable.
• with closes the client’s connections at the end of the block. A server can instead keep one Talki for the life of the process, which reuses its connections, and call talki.close() when done.
• Every request sends Talki-Version: talki.API_VERSION, the dated version the package was generated from. Talki(version="…") pins another; version=None sends none, so the version pinned on the key or app applies.

Call an operation

Talki has one attribute per tag of the API reference (talki.account, talki.appointments, talki.calls, …) and one method per operation, named after the operation ID without the tag’s noun: listCalls is talki.calls.list, getCallTranscript is talki.calls.get_transcript.

appointment = talki.appointments.create(
    body={"client_name": "Ada", "client_surname": "Lovelace", "start_at": "2026-11-02T09:30:00+01:00"}
)
talki.appointments.cancel(appointment.id)

• Path parameters are positional; query parameters and body= are keywords.
• A method returns the operation’s model, or None for 204 No Content, and raises for an error.
• body= takes the request model (talki.models.AppointmentCreate(...)) or a dict with the API’s keys, typed so a type checker catches a wrong key or value. The API validates it and answers a wrong one with 400.
• In an update body, talki.types.UNSET (or a key left out of a dict) leaves a field as it is, and None sends null, which clears a field that allows it.

Async

AsyncTalki takes the same arguments and has the same methods, to await:

from talki import AsyncTalki

async with AsyncTalki(api_key="talki_test_…") as talki:
    account = await talki.account.get()
    async for call in talki.calls.list(limit=100):
        print(call.id)

Status and headers

For the status or a header (ETag, Idempotency-Replayed, the rate-limit headers, X-Request-Id), call the same method on with_raw_response. It returns a RawResponse with status_code, headers, content, request_id and parse():

raw = talki.account.with_raw_response.get()
account, etag = raw.parse(), raw.headers["ETag"]

# Refused with 412 if the account changed since that read.
talki.account.update(body={"greeting": "Buongiorno"}, if_match=etag)

talki.with_options(timeout=5, max_retries=0) returns the same client with other settings for the calls made on it. It shares the original’s connections and token renewal.

Errors

A method raises a subclass of talki.ApiError, chosen by the error envelope’s type (see Errors and rate limits): InvalidRequestError, AuthenticationError, PermissionDeniedError, ConflictError, RateLimitError (with retry_after) and ServerError. Each carries status, type, code, message, request_id, headers and body. Branch on code, and quote request_id to support.

from talki import ConflictError

try:
    talki.appointments.create(body=booking)
except ConflictError as error:
    if error.code == "slot_unavailable":
        ...

An answer with no envelope, such as a proxy’s, is classed by its HTTP status, with code set to None.

Retries and idempotency

Talki retries up to max_retries times (default 2):

• a request that never left (connection refused, connect or pool timeout);
• a request answered 429 or 409 request_in_progress;
• a read answered 500, 502, 503 or 504, or whose connection broke after sending;
• a write with an Idempotency-Key answered 502, 503 or 504 by a gateway.

A write the API itself answered with a 5xx, or whose connection broke after sending, is never retried, since it may have been saved: read the resource before writing again. Waits follow Retry-After up to 60 seconds; a longer one raises instead (RateLimitError for a 429). timeout applies to each attempt.

Every write without an Idempotency-Key gets a new one, so the client’s own retries are never applied twice. To retry a write yourself, pass your own with idempotency_key= (talki.new_idempotency_key() makes one). auto_idempotency=False turns this off.

Paging and change feeds

A method whose operation takes cursor= returns a talki.pagination.Page: the first page, with data, has_more and next_cursor. Iterating it yields every row of it and of the pages after it, fetched as the loop reaches them. A list without a cursor (talki.locations.list(), talki.tags.list(), …) returns its model whole.

from talki.pagination import position

for call in talki.calls.list(limit=100):
    ...

for page in talki.appointments.list(cursor=stored).iter_pages():
    save(page.data)
    stored = position(page, stored)  # store it with the rows

A change feed returns next_cursor on every non-empty page, the last one included: store every non-null one with the rows and resume from it, as Core concepts describes. position(page, stored) returns the cursor to keep. On AsyncTalki, such a method returns an AsyncPager: await it for the first page, or async for it.

Knowledge base files

A note or a website is a JSON body like any other. A file has methods of its own, which send it as multipart/form-data:

from pathlib import Path

document = talki.documents.upload("price-list.pdf")  # a path, bytes or an open file
talki.documents.replace_file(document.id, "price-list-2027.pdf")
file = talki.documents.download(document.id)
Path(file.file_name or "download").write_bytes(file.payload.read())

The file’s name becomes the document’s title, so bytes need filename=. See the knowledge base guide for formats, limits and processing.

Verify a webhook

talki.webhooks checks X-Talki-Signature on the raw body, before parsing, as Webhooks describes:

from talki.webhooks import SignatureVerificationError, construct_event, signature_of

try:
    event = construct_event(raw_body, signature_of(headers), secret)
except SignatureVerificationError:
    ...  # answer 400 and drop it

• construct_event returns the envelope as a dict; verify_signature only checks. Both take the body as the bytes received, and refuse a signature more than 300 seconds from now.
• signature_of(headers) finds the header whatever its case.
• During a secret rotation either secret verifies; pass a list when you hold more than one.

OAuth apps

talki.oauth covers an app installed on Talki accounts: the consent URL with PKCE, the callback, the code exchange, and a token pair that renews itself.

import secrets
from talki import Talki
from talki.oauth import OAuthApp, Pkce, authorization_url, read_callback

app = OAuthApp(client_id, client_secret)
pkce, state = Pkce.new(), secrets.token_urlsafe(16)
url = authorization_url(client_id, redirect_uri, ["calls:read"], state=state, pkce=pkce)
# Send the customer's browser to url; then, on the callback:
code = read_callback(callback_url, state=state)
tokens = app.exchange_code(code, redirect_uri=redirect_uri, code_verifier=pkce.verifier)
save(tokens)

talki = Talki(tokens=tokens, oauth_app=app, on_token_refresh=save)

• With oauth_app, Talki renews the pair a minute before it expires, and on a 401 with error="invalid_token", then sends the request again.
• The refresh token rotates on every renewal, and presenting a rotated one again revokes the installation. Keep one Talki per installation per process, and store the new pair in on_token_refresh, which runs before the pair is used.
• After invalid_grant (the customer disconnected, or the pair was revoked) or invalid_client, every request raises that OAuthError without asking again: build a new Talki once fixed.
• A code works once: never retry exchange_code with the same code.

Logging

The client logs to the talki logger. Set it to DEBUG to log every answer with its request ID, or INFO for every retry and token renewal; at WARNING and ERROR it reports a token renewal whose outcome is unknown or that was refused. It writes a request by method and path only, never its query, headers, body or credentials. httpx’s own logger writes full URLs at INFO, query included: keep it at WARNING when your root level is INFO or lower.

Talki(…, user_agent="acme-crm/2.1") appends your integration’s name to the User-Agent.

Not covered yet

• The partner routes (/oauth/app/*, /oauth/managed-accounts, /oauth/enrollments) are not in the OpenAPI document, so not in the package. See Partners and Managed billing.
• Webhook payloads have no generated types: construct_event returns a dict. Event payloads lists their shapes.