Calling the API

This page covers the general rules for calling the server REST API with the Python SDK: how REST APIs map to SDK methods, return values, paging and automatic paging, batch results, checking errors, how the SDK retries and handles rate limits, uploading and downloading files, how app status affects calls, logs and troubleshooting, and use in asyncio programs and unit tests.

App permissions and what users can see

The SDK calls APIs as the app administrator. It doesn't know who is signed in to your system and doesn't check permissions for you. Once a browser action reaches your back end, check the user's sign-in, that the conversation belongs to them, and whether they're a group member or admin.

• messages.recall is an admin recall, not bound by regular users' roles or the recall time limit, so don't expose it directly as a "user recall" endpoint.
• conversations.list_messages and iter_messages read the full admin history and don't filter per user (messages they deleted or cleared, messages from before they joined or after they left a group). A user's chat screen should use a client SDK, or have the back end read what they can see with the user's own sign-in token (User Token).
• The Client Secret, App Token and users' sign-in tokens stay on the back end and are never sent to the browser.

For how the server SDK and APIs called as the user work together, see the demo guide.

For each API's parameters and meaning, see the server REST API; for which API each method maps to, see Services and methods. In the examples below, client is the DRClient from Create a client; with AsyncDRClient, add await before calls.

You want to	See
Know what to check before exposing operations to users	App permissions and what users can see
Find a method, and write parameters, None and UNSET	Naming rules
Set a time limit and trace ID for one call	Parameters every method takes
Page through results, iterate over everything	Paging
Read per-item results of batch APIs, import in bulk	Batch results
Handle failures by error type	Errors
Know which requests are retried, handle rate limits	Retries and idempotency, Rate limits
Upload and download files	Uploading files
Use it in async programs such as FastAPI	Using it with asyncio
Logs, metrics and unit tests	Logs and troubleshooting, Testing your own code

Naming rules

REST APIs are split by area into 13 services, each an attribute of the client: client.users, client.friends, client.blacklist, client.groups, client.messages, client.conversations, client.media, client.push, client.chatrooms, client.moderation, client.calls, client.channels and client.presence.

Method names are snake_case verbs or verb phrases:

REST API	SDK method
POST /users create a user	client.users.create(...)
GET /users/{username} get a user	client.users.get(username)
GET /users list users by page	client.users.list(...), with automatic paging as client.users.iter(...)
GET /groups/{group_id}/members list group members	client.groups.list_members(group_id), with automatic paging as iter_members
GET /groups/{group_id}/members/{username} get one member	client.groups.get_member(group_id, username)
GET /users/{username}/push-settings get push settings	client.push.get_settings(username)
PUT and DELETE /groups/{group_id}/mute-all turn mute-all on and off	client.groups.set_mute_all(group_id, True), set_mute_all(group_id, False)

Methods returning lists are called list or list_xxx, and those returning one object or a set of settings get or get_xxx; paged lists also have an iterator method iter or iter_xxx. Stats and summaries keep their own names, such as conversations.unread_count, media.usage, moderation.stats and chatrooms.online_counts. For every method, see Services and methods.

Parameters have the same names as the REST fields (server field names are already snake_case):

• Positional arguments: identifiers in the path (username, group ID, message ID and so on, in path order); the single required body field when there's only one, such as users.set_password(username, password) and groups.set_role(group_id, username, role); and the list of items for batch APIs, such as users.batch_create(users) and messages.batch_send(to_users, ...). The exception is the two dates of calls.stats(from_, to).
• Keyword arguments: everything else must be passed by name, such as users.create("zhangsan", nickname="Zhang San"). That way, when an API gains parameters later, nothing silently shifts position.
• Parameters named like Python keywords get an underscore: the sender from is written from_, and is still from in the request. Names shared with built-ins, such as type and id, are used as is.
• IDs are always str: passing an int is rejected locally with invalid_params without sending a request (the server doesn't accept numeric IDs). seq and version numbers are int.
• Times: parameters accept a timezone-aware datetime (converted to a UTC RFC 3339 string) or a string; naive datetimes are rejected locally with invalid_params, since the SDK doesn't guess timezones. Times in responses are UTC datetimes.

Not sent, null and a value

Optional parameters of update methods default to UNSET, meaning they're left out of the request; None sends null; any other value is sent as is:

from deeprespond_im import DRClient


def update_profile(client: DRClient) -> None:
    # 请求体：{"nickname":"张三","attributes":{"vip":"1","level":null}}
    client.users.update("zhangsan", nickname="张三", attributes={"vip": "1", "level": None})

    # attributes=None 发送 "attributes": null，清空全部自定义属性
    client.users.update("zhangsan", attributes=None)

    # friend_add_mode 是必须出现、可以为 null 的字段：None 表示改用应用的默认值
    client.friends.set_settings("zhangsan", None)

Parameter type annotations distinguish these three cases (such as str | None | Unset), so type checkers catch None passed to parameters that don't accept null; at runtime the same is rejected locally with invalid_params.

Sending JSON as is

Free-form content such as message body and ext can be a dict, which the SDK encodes as compact UTF-8 JSON (without escaping Chinese to \uXXXX, and rejecting NaN and Infinity). To keep how numbers are written (such as 1.00), or when you've signed the content, send it as is with RawJSON; the SDK only checks it's a valid JSON object:

from deeprespond_im import DRClient, RawJSON


def send_price_card(client: DRClient) -> None:
    client.messages.send(
        from_="notice",
        to_user="zhangsan",
        type="custom",
        body=RawJSON('{"card":"price","amount":1.00}'),
    )

RawJSON only works in free-form content positions; in ordinary parameters it's rejected with invalid_params. The server strips whitespace from JSON when storing messages, so base any content signature on the compact form.

Parameters every method takes

Besides the API's own parameters, every method accepts these three:

Parameter	Description
timeout	Time limit per attempt for this call (seconds), overriding the client's timeout
deadline	Total time limit for this call (seconds), including all retries, rate-limit waits and fetching the App Token. When it passes, DRTimeoutError is raised without further retries
request_id	Trace ID, sent as the X-Request-ID header; the server uses the same ID in its logs and error responses. 1 to 64 letters, digits or -, _, .; generated by the SDK if omitted

from deeprespond_im import DRClient


def get_user_for_web(client: DRClient, username: str, trace_id: str) -> str:
    # 在线请求中最多花 2 秒，日志用业务的追踪 ID 关联
    user = client.users.get(username, deadline=2.0, request_id=trace_id)
    return user.nickname

Iterator methods (iter, iter_xxx) only take timeout and request_id, not deadline.

Return values

Responses are parsed into data classes from deeprespond_im.types, with the same field names as the REST docs:

from deeprespond_im import DRClient


def show_user(client: DRClient) -> None:
    user = client.users.get("zhangsan")
    print(user.username, user.nickname, user.status, user.created_at)   # created_at 是带时区的 datetime
    print(user.raw.get("some_new_field"))   # raw：这个对象在响应中的原始内容
    data = user.to_dict()                   # 原始内容的副本，可以直接作为 FastAPI 等的返回值
    print(data["username"])

• Data classes are immutable and compare with == by business fields; they're not hashable, so they can't be dict keys or go in a set.
• New fields: fields the server adds later can be read from raw without waiting for an SDK upgrade. Enum-like fields are str, so new values don't cause errors.
• The original message text: objects containing messages (Message, SendResult, callback events and so on) have, besides the parsed body and ext (dict), body_raw and ext_raw (RawJSON): the raw text of that part of the response, byte for byte as the server returned it, usable directly for signature checks or forwarding.
• repr() doesn't show raw, *_raw, or sensitive fields such as sign-in tickets and download addresses, so logging objects won't leak them.
• Methods whose REST API returns 204 No Content return None.

Paging

Paged list methods return one page. Cursor-paged results are Page, with next_cursor None meaning there's no next page:

from deeprespond_im import DRClient


def all_active_usernames(client: DRClient) -> list[str]:
    names: list[str] = []
    page = client.users.list(status="active", limit=100)
    while True:
        names.extend(user.username for user in page.items)
        if page.next_cursor is None:
            return names
        page = client.users.list(status="active", limit=100, cursor=page.next_cursor)

Type	Used for	Fields
Page[T]	Most lists	items, next_cursor, raw
VersionedPage[T]	Versioned lists such as friends, blocklists and group members	Page's fields, plus version, not_modified, count (friend count, blocklist size, member count) and max_count (limit for friends and blocklists); None where absent
SeqPage[Message]	Conversation messages, conversations.list_messages	items (ascending by seq), has_more, next_after_seq, next_before_seq; with around_seq, also has_more_before and has_more_after
ExportPage[Message]	Message export, messages.export	items, has_more, next_after_message_id

For versioned lists, pass the version you got last time as known_version; when nothing changed, not_modified is True and items is empty, so there's no need to fetch again:

from deeprespond_im import DRClient


def friends_changed(client: DRClient, username: str, known_version: int) -> bool:
    page = client.friends.list(username, known_version=known_version)
    return not page.not_modified

A page may have fewer items than limit, or even none, with a next page still to come (the server skipped deleted data). Decide the end only by next_cursor (by has_more for conversation messages and exports), not by item count.

Automatic paging

Every paged list has an iterator method that yields every result, only requesting the next page when needed:

from deeprespond_im import DRClient


def find_vip(client: DRClient) -> str | None:
    with client.users.iter(status="active", limit=100) as users:
        for user in users:
            if user.attributes.get("vip") == "1":
                return user.username   # 提前结束：不会再取下一页
    return None

• Iterator methods take the same parameters as list methods, without cursor (or known_version); limit is the page size.
• The returned iterator is also a context manager, released as soon as with ends; breaking out early leaves no request in progress.
• If a page finally fails, the exception is raised and iteration ends; items already yielded remain valid. Each page is retried automatically as usual.
• The async client's iterator methods return async iterators used with async for; to release early, use async with or await it.aclose().
• conversations.iter_messages iterates toward newer messages when after_seq is given, otherwise from the latest toward older ones, yielding newest first overall.
• chatrooms.iter_members de-duplicates by username: a user who leaves and re-enters during iteration is yielded once.

Two more methods loop for you: friends.delete_all(username) keeps deleting until no friends are left and returns the total deleted; moderation.process_pending(...) handles pending moderation items one by one; see Services and methods.

Batch results

Batch APIs (users.batch_create, friends.batch_add, messages.batch_send, messages.batch_recall and so on) handle each item separately. When the whole request succeeds you get a result object, with failed items in it rather than raised as exceptions:

from deeprespond_im import DRClient


def import_users(client: DRClient) -> None:
    result = client.users.batch_create([
        {"username": "u1", "nickname": "用户一"},
        {"username": "u2", "password": "S3cret-pass"},
    ])
    print(len(result.created), "个已创建")
    for item in result.failed_items:            # 失败的项：ItemError
        if item.code != "already_exists":
            print(item.key, item.code, item.message)

• Each batch API has its own result type (such as BatchCreateUsersResult and BatchSendResult), with fields matching the REST docs.
• Every result type has ok (all succeeded), failed_items (failed items, a list of ItemError) and raise_for_failures() (raises BatchError when any item failed, otherwise does nothing).
• ItemError has key (the item's identifier, such as a username or message ID), code, message, details and reason.
• BatchError isn't a subclass of DRError: the request itself succeeded, only some items failed. Its failed is the failed items and result the whole result.
• Each item's input is a plain dict, and type checkers check keys and types against the TypedDicts in deeprespond_im.types (such as CreateUserInput).

Batches over the limit

Methods ending in _all (users.batch_create_all, friends.batch_add_all, messages.batch_send_all) accept any iterable (including generators), split by the per-request limit, submit in turn and merge the results, without reading all the data into memory first:

import csv
from collections.abc import Iterator

from deeprespond_im import DRClient, DRError
from deeprespond_im.types import CreateUserInput


def import_from_csv(client: DRClient, path: str) -> None:
    with open(path, newline="", encoding="utf-8") as f:
        rows: Iterator[CreateUserInput] = ({"username": row["id"], "nickname": row["name"]} for row in csv.DictReader(f))
        try:
            result = client.users.batch_create_all(rows)
        except DRError as e:
            done = e.partial   # 已完成部分的合并结果
            print("中途失败：", e.code, "已处理", len(done.created) if done else 0, "个")
            raise
    print("创建", len(result.created), "个，失败", len(result.failed), "个")

• If a whole batch request fails, it stops and raises that batch's original error (such as RateLimitError), whose partial attribute is the merged result of what finished.
• When the app sets a smaller per-batch limit and the server rejects with too_many_items, the SDK splits again by the limit the server gives.
• Every batch of messages.batch_send_all uses the same client_msg_id, so retrying the whole thing never sends twice.

Errors

Failed calls raise DRError or a subclass. Errors returned by the server (see Error codes) are turned into DRError as is:

Attribute	Description
code	Error code, such as not_found or rate_limited, or one of the SDK's own (see below). Branch on this; don't parse message
reason	details.reason, or None
details	The details from the error response, as is
message	Description text, for logs only
status_code	HTTP status; None for the SDK's own errors
request_id	Request ID; give it when contacting support
retry_after	Seconds from the Retry-After response header
op	API name, such as users.create (see the "API name" column of Services and methods)
attempts	How many attempts this call made in total
retried	This error came from an attempt after an automatic retry, and earlier attempts may have run; see Retries and idempotency
partial	Only when an _all method fails partway: the results of what finished
is_temporary	Network errors, timeouts, internal, 5xx and rate_limited: you can try again later
is_service_unavailable	The tenant or app is paused or read-only (tenant_unavailable, app_unavailable, and the resulting client_unavailable during the cooldown)
is_permanently_unavailable	The app was deleted or the tenant closed

Errors are split into the subclasses below by code, so you can catch by type. An error code always belongs to the same subclass, and unknown codes (added by the server later) are plain DRError:

Subclass	Error codes
AuthenticationError	unauthenticated, invalid_client, ip_not_allowed
PermissionDeniedError	permission_denied, user_disabled, not_group_member, group_disabled, chatroom_disabled, file_blocked
NotFoundError	not_found, file_expired
ConflictError	already_exists, version_conflict
LimitExceededError	limit_exceeded
RateLimitError	rate_limited
ServiceUnavailableError	tenant_unavailable, app_unavailable
InvalidRequestError	invalid_argument, payload_too_large, method_not_allowed, and the SDK's invalid_params
ContentRejectedError	message_rejected, content_rejected
ServerError	internal
NetworkError	The SDK's network_error, also the built-in ConnectionError
DRTimeoutError	The SDK's timeout, also the built-in TimeoutError
InvalidResponseError	The SDK's invalid_response
ClientUnavailableError	The SDK's client_unavailable, with an extra cause_code

The SDK's own error codes:

Error code	Description
invalid_params	A parameter failed the local check and no request was sent; details.field names the parameter
network_error	Connection failed or dropped, or a TLS error; the underlying httpx exception is in __cause__
timeout	One attempt timed out, or the deadline passed
invalid_response	The response isn't the expected JSON, such as an HTML error page from a proxy; details.body_prefix is the start of the response, and is_temporary is true for 5xx
client_unavailable	The client is closed (cause_code closed), permanently unavailable, or in the cooldown after a failed App Token fetch (cause_code is the original error code)

from deeprespond_im import ConflictError, DRClient, DRError


def ensure_user(client: DRClient, username: str) -> None:
    try:
        client.users.create(username)
    except ConflictError as e:
        if e.code != "already_exists":   # 已存在的用户视为成功
            raise
    except DRError as e:
        if e.is_temporary:
            print("稍后再试", e.request_id)
        raise

• An exception's string form includes the API name, HTTP status, error code, reason and request_id, without the request body, tokens or Client Secret, so it's safe to log as is.
• All exceptions can be pickled, so their fields are intact when Celery or multiprocessing passes them back to the caller.
• When an async task is cancelled, asyncio.CancelledError propagates as is and isn't turned into DRError.

Retries and idempotency

The SDK only retries requests that are safe to retry automatically. Each method belongs to one category below, given in the "Auto retry" column of Services and methods:

Category	Description	On network errors, timeouts and 5xx
Query	Read-only APIs	Retried automatically
Idempotent	Repeating gives the same result, such as setting, deleting or blocking	Retried automatically
Idempotency key	Sending messages: de-duplicated by client_msg_id, generated by the SDK if not given	Retried automatically with the same client_msg_id
Unique create	Created with a unique key, such as creating users	Retried automatically; if a retry gets already_exists, e.retried is true, meaning an earlier attempt may have created it
No retry	Every call produces a new result: groups.create, chatrooms.create, media.create_upload, moderation.create_word_list, moderation.escalate, users.set_password, users.kick_all, plus online-only sends (online_only=True), low-priority chat room messages, and moderation.decide with actions but no version	Only retried when the request certainly wasn't sent (such as a failed connection)

• Up to max_retries retries (3 by default), about 0.5, 1 and 2 seconds apart (each multiplied by a random 0.5 to 1.5); a 5xx with Retry-After waits at least that long.
• Nothing is retried past the deadline.
• What retried means: when an automatic retry gets already_exists, not_found or version_conflict, e.retried is true, and an earlier attempt may have run (such as creating successfully and then losing the response). Look up and confirm instead of treating it as a failure.
• When a no-retry method times out, look up the result before deciding whether to retry; for example, after creating a group times out, check with groups.list by owner (writing a business ID into attributes makes it easy to recognize).
• Errors such as 400, 403, 404 and 409 aren't retried and are raised directly.
• When the token becomes invalid (401 unauthenticated without a reason), the SDK fetches a new one and resends once, not counted toward retries.

Rate limits

On 429 rate_limited (see Rate limits), the SDK waits for Retry-After (plus a random 0 to 20% to spread requests out) and resends, up to max_rate_limited_retries times (3 by default), not counted toward max_retries. Who waits depends on details.reason:

Why it was limited	Who waits
The app's OpenAPI quota is used up (no details.reason)	The whole client: requests from every thread or coroutine pause together until the wait ends, rather than each retrying and using up the quota again
The app's messages per minute (app_message_rate)	Every message-sending request in this client (messages.send, messages.batch_send), waiting between Retry-After and twice that
Report rate (request_rate)	Every moderation.report request in this client
Other reasons, such as one group's or chat room's send rate (group_send_rate, room_send_rate)	Only this request

In these cases it doesn't wait and raises RateLimitError (with e.retry_after the seconds the server asked for): the server asks to wait longer than max_retry_after (60 seconds by default); it has already resent max_rate_limited_retries times; or waiting would pass this call's deadline. While the whole client is paused, calls whose wait would pass their deadline raise RateLimitError without sending a request.

import time

from deeprespond_im import DRClient, RateLimitError


def send_with_backoff(client: DRClient, username: str, text: str) -> None:
    try:
        client.messages.send(from_="notice", to_user=username, type="text", body={"text": text})
    except RateLimitError as e:
        # SDK 已经等待并重发过；仍被限流时由业务决定稍后再发
        time.sleep(e.retry_after or 1)
        raise

"The whole client waits together" only applies within one process: when several processes are limited at once they each wait, and the random part of the wait spreads them out.

Local rate limiting

Background jobs such as bulk imports and broadcast notifications can throttle themselves locally, leaving the OpenAPI quota for live requests:

import os

from deeprespond_im import DRClient, RateLimit

batch_client = DRClient(
    base_url="https://im.example.com",
    org_name="1100250925",
    app_name="demo",
    client_id=os.environ["IM_CLIENT_ID"],
    client_secret=os.environ["IM_CLIENT_SECRET"],
    rate_limit=RateLimit(per_second=300),   # 每秒最多 300 次，burst 默认等于 per_second（向上取整）
)

• Local rate limiting is a token bucket: per_second is calls per second and burst the allowed burst, max(1, ceil(per_second)) by default.
• Batch APIs that count toward the server quota by item (such as batch creating users and batch sending) also consume the local limit by item (at most burst per call).
• If waiting would pass the deadline, DRTimeoutError is raised.
• Local rate limiting applies within one client; with several processes, divide the quota among them.

Uploading files

media.upload() does the three steps of creating the upload, uploading the content and completing the upload (see Upload files), and returns the file info:

from deeprespond_im import DRClient


def send_image(client: DRClient, path: str) -> None:
    file = client.media.upload(path, purpose="attachment", kind="image")
    client.messages.send(
        from_="notice",
        to_user="zhangsan",
        type="image",
        body={"url": file.url, "width": file.width, "height": file.height, "size": file.size},
    )

Parameter	Description
source	Positional: a file path (str or pathlib.Path), bytes, or a seekable binary file object (opened with rb, read from the current position to the end). Non-seekable streams (such as network streams or stdin) can't resend parts and are rejected with invalid_params; write them to a temporary file first
purpose, kind	Use and kind, such as attachment and image. Their combination is checked locally first
name	File name, taken from the path or file object if omitted. Required for kind file and for group files
content_type, group_id, owner	Same as the fields of Create an upload
concurrency	Concurrent parts, 4 by default
on_progress	Progress callback on_progress(uploaded, total) (bytes). In the sync client it may be called from another thread
request_id	Trace ID

• Up to 32 MB is uploaded in one request; larger files go in concurrent 16 MB parts. Content uploads go straight to object storage without the App Token.
• file.url is the file URL to put in the message body; don't save download addresses, which expire.
• Avatars (user_avatar and so on) can't exceed 5 MB or be empty, checked locally first.
• Expired part upload addresses are requested again automatically; if the server is still merging parts on completion, the SDK waits and retries; if parts are missing, it uploads them and completes again.
• When interrupted with Ctrl+C (KeyboardInterrupt) in the sync client, or when the task is cancelled in the async client, the SDK tries to cancel the upload, then re-raises the exception.

Resuming large files

A multipart upload (over 32 MB) interrupted within 24 hours can continue with media.resume_upload(file_id, source) and the same content: the SDK checks which parts were uploaded, uploads only the missing ones, then completes. The exception from upload() doesn't carry the file ID, so for large files you may need to resume, create the upload yourself, save the file_id, then upload the content with resume_upload():

import os

from deeprespond_im import DRClient
from deeprespond_im.types import File


def upload_big_file(client: DRClient, path: str, saved_file_id: str | None) -> File:
    if saved_file_id is None:
        session = client.media.create_upload(
            purpose="attachment", kind="file", size=os.path.getsize(path), name=os.path.basename(path)
        )
        saved_file_id = session.file_id   # 保存下来，中断后用它续传
    return client.media.resume_upload(saved_file_id, path)

Downloading files

media.download(url, dest) exchanges the file URL for a download address and streams it to a path or writable binary file object, without reading the whole file into memory, returning the bytes written. If the download address has expired (object storage returns 403), it gets a new one once:

from deeprespond_im import DRClient


def save_attachment(client: DRClient, file_url: str) -> int:
    return client.media.download(file_url, "/tmp/attachment.bin")

When you only need the download address (such as for a browser download), use media.get_download_urls([url, ...]).

media.download only writes the file content; it doesn't set the content type, caching or download headers of your own HTTP responses. When proxying attachments to the browser, don't decide the response type from the MIME type the browser sends; content that can run scripts, such as HTML and SVG, should be forced to download or served from a separate domain. The demo only shows content recognized as PNG, JPEG, GIF or WebP inline, sends everything else as a download, and sets nosniff, a sandboxed CSP and no caching.

App and tenant status

When the app is disabled or paused, or the tenant is paused or has unpaid bills (see App and tenant status), calls behave like this:

Status	What the SDK does	Recommendation
Read-only	Queries and actions work; writes that create data or update profiles raise ServiceUnavailableError (app_unavailable, with reason arrears for unpaid bills)	Alert; pause background jobs that write
App disabled or paused, tenant paused	Issued tokens stop working; when the SDK fetches again it gets app_unavailable or tenant_unavailable and raises ServiceUnavailableError; for the next 2 to 5 minutes it doesn't fetch, and calls raise ClientUnavailableError directly (cause_code the original error code)	Alert and check the status in the console; the SDK tries again after the cooldown, so no restart is needed once restored
Wrong or revoked Client Secret, IP not on the allowlist	Raises AuthenticationError (invalid_client, ip_not_allowed); for the next minute calls raise ClientUnavailableError directly	After updating the credentials, call client.update_credentials() to end the cooldown immediately
App deleted, tenant closed	Raises AuthenticationError (reason app_deleted or tenant_closed); the client becomes permanently unavailable and raises ClientUnavailableError from then on	Clear the saved credentials and stop calling

For code that treats this as "service paused", check e.is_service_unavailable, which covers both ServiceUnavailableError and ClientUnavailableError during the cooldown:

from deeprespond_im import DRClient, DRError


def sync_profile(client: DRClient, username: str, nickname: str) -> bool:
    try:
        client.users.update(username, nickname=nickname)
    except DRError as e:
        if e.is_service_unavailable or e.is_permanently_unavailable:
            print("IM 服务不可用，稍后同步：", e)
            return False
        raise
    return True

Logs and troubleshooting

The SDK logs with the standard library's logging, under the logger deeprespond_im, split into deeprespond_im.http (requests, retries, uploads), deeprespond_im.token (App Token fetching) and deeprespond_im.callback (receiving callbacks). The SDK adds no handlers, so whether anything is output depends on your logging config:

• Retries and rate-limit waits are logged at WARNING or DEBUG; final network errors, timeouts, 5xx and authentication failures at ERROR; other 4xx are business outcomes, logged at DEBUG.
• Logs never contain request bodies, the App Token or the Client Secret.
• httpx itself logs every request's full URL at INFO, and signatures of object storage upload and download URLs are in the query string. By default the SDK adds a filter to the httpx logger that replaces query strings with ?… (turn it off with redact_httpx_logs=False). The httpcore logger records response headers at DEBUG, so set it to WARNING or above in production.

Hooks

The hooks option calls your functions before and after each attempt, for injecting trace headers and recording metrics. after_response receives a RequestInfo: the API name op (app.token for fetching the token), method, url (without the query), the attempt number attempt, status_code, error code and reason, request_id, the time taken elapsed, and the rate-limit wait before this attempt rate_limit_wait:

import logging
import os

import httpx

from deeprespond_im import DRClient, Hooks, RequestInfo

log = logging.getLogger("im")


def add_trace(request: httpx.Request) -> None:
    request.headers["traceparent"] = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"


def record(info: RequestInfo) -> None:
    log.info("%s attempt=%d status=%s code=%s %.0fms", info.op, info.attempt, info.status_code, info.code, info.elapsed * 1000)


client = DRClient(
    base_url="https://im.example.com",
    org_name="1100250925",
    app_name="demo",
    client_id=os.environ["IM_CLIENT_ID"],
    client_secret=os.environ["IM_CLIENT_SECRET"],
    hooks=Hooks(before_request=add_trace, after_response=record),
)

• Hooks are called synchronously on the thread making the request, so don't do slow work in them; exceptions they raise propagate as is.
• Don't log the Authorization header or the request body in hooks.
• The async client uses AsyncHooks, where either hook can be a plain function or a coroutine function.

Stats

client.stats() returns OpStats per API name: attempts requests, calls that finally failed failures, retries and resends after rate limits retries, and the total rate-limit wait rate_limit_wait (seconds). Collect it periodically into your monitoring:

from deeprespond_im import DRClient


def report_stats(client: DRClient) -> None:
    for op, s in client.stats().items():
        print(op, s.requests, s.failures, s.retries, round(s.rate_limit_wait, 2))

When contacting support, give the error's request_id, op, code and the time it happened.

Using it with asyncio

Async programs such as FastAPI, Starlette and aiohttp use AsyncDRClient; don't call the sync client from async code (it blocks the event loop). In FastAPI you can create and close the client in the app's lifespan:

import os
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI

from deeprespond_im import AsyncDRClient

im = AsyncDRClient(
    base_url="https://im.example.com",
    org_name="1100250925",
    app_name="demo",
    client_id=os.environ["IM_CLIENT_ID"],
    client_secret=os.environ["IM_CLIENT_SECRET"],
)


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    yield
    await im.aclose()


app = FastAPI(lifespan=lifespan)


@app.post("/im/ticket")
async def im_ticket(user_id: str) -> dict[str, str]:
    ticket = await im.users.issue_login_ticket(user_id, auto_create=True)
    return {"ticket": ticket.ticket}

• The async client can be created at module level, but can only be used in the event loop where it sent its first request. Using it in another event loop (such as a second asyncio.run(), or the new event loop pytest-asyncio creates per test) raises RuntimeError, so create one per event loop.
• Iterate with async for: async for user in im.users.iter(status="active"): ....
• You can limit a whole call with asyncio.timeout(), and cancellation interrupts the request and any waits in progress; or use the method's deadline parameter. A cancelled or timed-out write may already have run, so handle it by its category in Retries and idempotency.
• The async client's App Token store is an AsyncTokenStore (such as deeprespond_im.redis.AsyncRedisTokenStore with redis.asyncio.Redis), and its hooks are AsyncHooks.
• The sync client has no cancellation; limit durations with timeout, deadline and with_options().

Testing your own code

deeprespond_im.testing.FakeOpenAPI is a fake OpenAPI with no network: set responses or errors per API name, and check what requests your code sent.

from deeprespond_im import RateLimitError
from deeprespond_im.testing import FakeOpenAPI


def test_send_notice() -> None:
    fake = FakeOpenAPI()
    fake.respond("messages.send", status=201, json={"message_id": "m1", "client_msg_id": "order-1", "created_at": "2026-10-06T08:00:00Z"})
    client = fake.client()          # 连到假的 OpenAPI 的 DRClient；fake.async_client() 返回 AsyncDRClient

    result = client.messages.send(from_="notice", to_user="zhangsan", client_msg_id="order-1", type="text", body={"text": "hi"})

    assert result.message_id == "m1" and not result.duplicate
    assert fake.calls("messages.send")[0].json["to_user"] == "zhangsan"
    client.close()


def test_rate_limited() -> None:
    fake = FakeOpenAPI()
    fake.fail("users.get", "rate_limited", retry_after=120, times=None)   # 一直返回 429
    client = fake.client()
    try:
        client.users.get("zhangsan")
    except RateLimitError as e:
        assert e.retry_after == 120          # 超过 max_retry_after，不等待
    else:
        raise AssertionError("应当抛出 RateLimitError")
    client.close()

Method	Description
fake.respond(op, json=..., status=200, headers=..., times=None)	Sets an API's response; times is how many times it applies, forever by default; rules set later are used first
fake.fail(op, code, status=..., details=..., retry_after=..., times=1)	Makes an API return an error, with the HTTP status chosen by error code, once by default
fake.handle(op, responder, times=None)	Handles requests with your own function returning an httpx.Response; it can also raise httpx exceptions to simulate network errors
fake.revoke_tokens()	Invalidates App Tokens already fetched, to test fetching again after invalidation
fake.requests, fake.calls(op)	Requests received: API name, method, path, query, headers and body, with .json the parsed body
fake.client(**kwargs), fake.async_client(**kwargs)	Clients connected to it, with other arguments passed to the client as is (such as max_retries=0)

• op is the API name from the "API name" column of Services and methods; for example, the API name of users.list_mutes is users.mutes.
• APIs without a response set return 200 with {} (204 for those with no result). Fields missing from responses parse as zero values (empty string, 0, False, None), so put the fields your test checks into json.
• For APIs where the status code means something, set status as in the REST docs; for example, sending a message returns 201 for a new write, and 200 means a repeated submission (duplicate true).
• With pytest, enable the fixtures with pytest_plugins = ["deeprespond_im.testing.pytest"] in conftest.py to get im_fake, im_client and im_async_client (the last needs pytest-asyncio).

For testing callback code, see Local testing.

Using standalone channels

created = client.channels.create("meeting-demo", name="项目会议", media="video", access_mode="ticket")
issued = client.channels.issue_tickets(created.channel.channel_id, users=[{"username": "alice", "role": "publisher"}])
# 只将 issued.tickets 中目标用户的票据交付该用户，勿记录原文。
page = client.channels.list(status="open", limit=20)
corrections = client.channels.usage_adjustments(limit=20)

• AsyncDRClient has the same services and methods; add await when calling.
• Creating returns a ChannelsCreateResult and querying a RTCChannelResult, and tickets have their own result types; every field is typed and the raw data is kept.
• expires_at uses UNSET, None and a value for not sent, clear and set; stats takes the keyword arguments from_ and to.
• Channel lists have no iter method; page with next_cursor yourself.
• Issuing tickets isn't retried automatically; removing members and revoking tickets use a fixed operation_id.
• These types are available from PyPI 1.0.0a0; for the matching APIs, see Channel methods.

Game rooms

Game rooms give multiplayer games a real-time match channel. Your server uses client.game to create rooms, issue join tickets to players and close rooms when matches end; players join and play through the client SDK. Here is what your server does for one match (client as above):

def start_match(match_id: str, players: list[str]) -> dict[str, str]:
    """匹配成功后建房，给每个玩家签一张加入票据。"""
    try:
        # 房间 ID 用你的对局 ID：重试或重复调用都只有一个房间
        client.game.create_room(match_id, max_players=len(players), start_mode="all_ready", ext={"mode": "ranked"})
    except DRError as e:
        if e.reason == "game_capacity_unavailable":
            raise RuntimeError("游戏服务繁忙，请稍后再试") from e
        if e.reason == "open_rooms":
            raise RuntimeError("同时进行的对局太多") from e
        raise
    # 不会自动重试；超时后直接再签一次，旧票据随之作废
    tickets = client.game.issue_tickets(match_id, users=[{"username": u, "seat": i} for i, u in enumerate(players)])
    # 经你的业务接口分别交给每个玩家，不要写进日志
    return {t.username: t.ticket for t in tickets}


def finish_match(match_id: str, winner: str) -> None:
    """对局结束时带上结算结果结束房间。"""
    res = client.game.close_room(match_id, reason="对局结束", result={"winner": winner})
    if not res.accepted:
        print(match_id, "之前已在结束，这次的结果不生效")


def kick_cheater(match_id: str, username: str) -> bool:
    """把作弊的玩家移出房间；返回 False 表示他已经不在房间里。"""
    return client.game.remove_member(match_id, username, reason="作弊").removed

To do	Method	Notes
Create a room	create_room	Use your match ID as the room ID. Calling again with the same parameters returns the same room (created is false), and network errors are retried automatically
Issue join tickets	issue_tickets	1–10 players per call. Not retried automatically: after a timeout just issue again, which revokes the user's previous unused ticket. The raw ticket is returned only once; don't log it
Start manually	start_room	Only for rooms whose start mode is manual
Close a room	close_room	The result is passed as-is to the players and to the game_room.closed callback. If the room was already closing, accepted is false and this call's reason and result don't apply
Remove a player	remove_member	The SDK generates operation_id and keeps it across retries. If the player isn't in the room, removed is false
Query	get_room, list_rooms / iter_rooms, list_members, list_regions	Lists are newest first; you can pass several statuses separated by commas

• AsyncDRClient has the same methods; await the calls.
• Results are data classes such as GameCreateRoomResult and GameTicket; created, accepted and removed come from the HTTP status code, not the response body. Pass ext and result as a dict.
• Common errors: unavailable (game_capacity_unavailable, ServiceUnavailableError) when the region has no free game node — retry later or give fallback regions when creating the room; limit_exceeded (open_rooms, LimitExceededError) when the number of open rooms hits the limit. See game room errors for the full list.
• ext can be up to 2KB and result up to 4KB; for limits and billing see Console, billing and limits.
• The settlement callback after a match (game_room.closed) is covered in Game room callbacks.
• For the API behind each method see game room methods; for request and response fields see Game rooms: server API. Game room methods are available from PyPI 1.0.0a2.

AI streaming replies

In AI conversations, after a user messages a bot you get an ai.message_received callback, then start a reply with client.ai.start_stream and write it as it's generated. The returned stream object handles numbering and batching deltas (sending a chunk every 20 milliseconds or 4 KB), switching between continuous and batch writes, and resending on failure; after three failed writes in a row it ends with fail("transport"). The sync client returns a Stream (sending on a background thread) and the async client an AsyncStream (sending in a background task), with the same methods; in the async version, flush, finish, fail, close and pipe_* need await. Manage bots with client.ai.create_bot and related methods; see Services and methods.

Start a reply when a message arrives

Callback handlers should return quickly, with the model call on another thread or task. Use the event ID as client_stream_id so a redelivered callback doesn't start a second reply:

import threading

handler = CallbackHandler(secrets={ENDPOINT_ID: [SECRET]}, nonce_store=MemoryNonceStore(), dedup_store=MemoryDedupStore())

@handler.on_ai_message
def on_ai_message(event, data):
    stream = client.ai.start_stream(
        data.bot, data.from_, reply_to_seq=data.message["seq"], client_stream_id=event.id
    )
    question = data.message["body"].get("text", "")
    threading.Thread(target=reply, args=(stream, question), daemon=True).start()

With replace=True, start_stream first ends any reply being generated in the conversation; batch_only=True uses only batch writes (for runtimes without streaming request bodies, such as some cloud functions).

Connect a model

pipe_openai, pipe_openai_responses, pipe_anthropic and pipe_gemini read the model's stream to the end and then finish automatically (with the finish reason, model name and usage); if the model returns an error, they end with fail("model_error", description) and raise that error. The argument can be a stream from an official SDK (openai, anthropic, google-genai) or a raw SSE response (httpx.Response, or an iterator yielding chunks of str or bytes). Models compatible with the OpenAI format (DeepSeek, Qwen, Kimi, Doubao, vLLM, Ollama and so on) all use pipe_openai:

from openai import OpenAI

openai_client = OpenAI()

def reply(stream, question):
    with stream:  # 退出时没有结束的以 fail("canceled") 结束
        completion = openai_client.chat.completions.create(
            model="gpt-4.1",
            messages=[{"role": "user", "content": question}],
            stream=True,
            stream_options={"include_usage": True},
        )
        stream.pipe_openai(completion)

The async version:

from openai import AsyncOpenAI

openai_client = AsyncOpenAI()

async def reply(client: AsyncDRClient, bot: str, user: str, question: str) -> None:
    async with await client.ai.start_stream(bot, user) as stream:
        completion = await openai_client.chat.completions.create(
            model="gpt-4.1", messages=[{"role": "user", "content": question}], stream=True
        )
        await stream.pipe_openai(completion)

The four conversions follow the same rules as the Go SDK.

Write manually

stream.write_reasoning("先查询天气。")
stream.write_tool("call_1", "weather", "查询天气", "running")
# … 调用工具 …
stream.write_tool("call_1", "weather", "查询天气", "done")
stream.write_text("明天上海**多云**，")
stream.write_text("最高 23℃。")
stream.write_image("https://cdn.example.com/a.png", width=800, height=600)
result = stream.finish(model="gpt-4.1", usage={"input_tokens": 120, "output_tokens": 380})

• write_* only buffers in memory without waiting on the network, and needs no await in either version; text of the same type as the previous part is appended to the same part.
• flush() sends immediately and waits for confirmation; finish() confirms all content first, then ends.
• When the model call fails, call stream.fail("model_error", "The model is temporarily unavailable"); content generated so far is kept, and the user sees "Generation interrupted" with your message.

Stop generating

When the user taps stop, the reply times out, or it's replaced by a new reply, stream.cancelled is set (a threading.Event in the sync version, an asyncio.Event in the async one), and stream.cause is a DRError of stream_closed; later writes raise that error, and pipe_* stops reading and closes the model's stream.

The callback handlers (CallbackHandler, AsyncCallbackHandler) stop the matching stream in this process automatically when ai.stream_stopped arrives, with no handler needed; to do more, use @handler.on_ai_stream_stopped. If you handle callbacks yourself (such as taking events from a queue), call client.ai.stream_stopped(stream_id).