Python

talqing is the Python client for the whole /v1 API, generated from the same OpenAPI document as the TypeScript SDK. Python 3.10+, httpx as the only runtime dependency, typed with a py.typed marker.

pip install talqing

Creating a client

import os

from talqing import Talqing

with Talqing(token=os.environ["TALQING_API_KEY"], base_url="https://api.talqing.com") as talqing:
    agents = talqing.agents.list()

Argument	Default	Meaning
`token`	—	Required. Personal access token; missing or blank raises `ValueError`
`base_url`	—	Required, keyword-only. Trailing slashes trimmed
`timeout`	`30.0`	Seconds, or an `httpx.Timeout`
`headers`	`None`	Sent on every request, over the SDK’s `Authorization` and `User-Agent`
`transport`	`None`	An `httpx.BaseTransport` (`httpx.AsyncBaseTransport` for the async client)

Talqing.from_env(**overrides) reads TALQING_API_KEY and TALQING_BASE_URL; anything passed wins. AsyncTalqing has the same shape with every operation a coroutine:

import asyncio

from talqing import AsyncTalqing

async def main() -> None:
    async with AsyncTalqing.from_env() as talqing:
        print(await talqing.agents.list())

asyncio.run(main())

Use either as a context manager, or call talqing.close() (await talqing.close() on the async client) when done.

The default 30-second timeout is shorter than an agent task’s budget. talqing.task_runs.create waits for the run, whose timeout_seconds defaults to 300 and can be 600. A longer run raises httpx.ReadTimeout on your side while it continues, and is billed, on ours. Raise the timeout past the task’s budget, or pass background=True and collect the result later:

import httpx

with Talqing.from_env(timeout=httpx.Timeout(660.0, connect=10.0)) as talqing:
    run = talqing.task_runs.create(task_id=task_id, vars={"domain": "acme.com"})
    print(run["status"], run["output"])

The call shape

Path parameters are positional; body fields and query parameters are keyword arguments. Responses are plain decoded dicts described by TypedDicts, so read fields with response["field"].

agent = talqing.agents.create(
    config={
        "name": "Support line",
        "channel": "text",
        "prompt": "You are the support desk for Acme. Be brief and specific.",
        "llm": {"provider": "openai", "model": "gpt-5.6-luna"},
    }
)
talqing.agents.publish(agent["id"])
talqing.agents.versions.rollback(agent["id"], 3)
talqing.telephony.phone_numbers.assign(number_id, agent_id=agent["id"])

An argument you do not pass is not sent

Optional arguments default to the sentinel OMIT and are left out of the request. That lets a PATCH distinguish “clear this field” from “leave it alone”:

talqing.telephony.phone_numbers.update(number_id, label=None)          # sends "label": null
talqing.telephony.phone_numbers.update(number_id, can_outbound=False)  # label untouched

OMIT is exported from talqing for building arguments dynamically. The one renamed method is telephony.phone_numbers.import_().

Errors

Every non-2xx raises TalqingAPIError with str(error) (the message), .status_code, .errors (always a list), .detail and .response. A non-JSON error body, such as a proxy’s 502 page, still raises it with the raw text as the message; connection errors and timeouts raise httpx’s own exceptions. See errors.

from talqing import TalqingAPIError

try:
    talqing.agents.publish(agent_id)
except TalqingAPIError as error:
    print(error.status_code, error, error.errors)

Pagination

paginate(method, *, limit=200, **filters) takes the bound list method itself and yields items across pages. paginate_async is the async for form.

from talqing import paginate

for call in paginate(talqing.calls.list, agent_id=agent_id, outcome="failure"):
    print(call["id"], call["close_reason"])

Streams

conversations.events and copilot.agents.stream / tools.stream / tasks.stream return a Stream (an AsyncStream on the async client): an iterator of decoded frames that is also a context manager. Opening one is not a coroutine on either client.

with talqing.conversations.events(conversation_id) as events:
    for frame in events:
        if frame["event"] == "assistant.delta":
            print(frame["text"], end="", flush=True)
        elif frame["event"] == "turn" and frame["status"] != "running":
            break

Use the context manager whenever you may exit early; it closes the connection. A stream that fails to open raises TalqingAPIError immediately. A finished stream cannot be reopened (RuntimeError). Frame shapes are in streaming.

Escape hatches

Member	Use
`talqing.request(method, path, *, query=None, body=None)`	Call an endpoint newer than your SDK version, with the same auth, `OMIT` filtering and errors
`talqing.http`	The underlying `httpx.Client` (or `AsyncClient`), already carrying base URL and token
`talqing.oauth_start_url(provider, integration_id=None)`	The URL to send a browser to for an integration’s OAuth
`talqing.base_url`	The normalized base URL

Every request and response shape is importable by its API name (from talqing import AgentConfig, AgentResponse, Page).

SDK overview

Every resource namespace.

Streaming

Every frame the event streams emit.

SDKs

Creating a client

The call shape

An argument you do not pass is not sent

Errors

Streams

Escape hatches

Next

SDK overview

Streaming

​Creating a client

​The call shape

​An argument you do not pass is not sent

​Errors

​Pagination

​Streams

​Escape hatches

​Next

SDK overview

Streaming

Creating a client

The call shape

An argument you do not pass is not sent

Errors

Pagination

Streams

Escape hatches

Next