Python SDK

This page covers installing the Python server SDK, creating a client and every option, the first calls (issuing sign-in tickets, sending messages), and how the SDK gets, caches and refreshes the App Token. For general rules of calling APIs, see Calling the API; for receiving callbacks, see Receiving callbacks; for every method, see Services and methods.

For downloading the full chat example, its minimal dependencies, starting it and configuring callbacks, see the demo guide. The SDK acts as the app; the demo additionally keeps users' sign-ins on its back end, to read the history users can see and to perform users' own recalls.

You want to	See
Install the SDK and optional dependencies	Install
Create a client (sync or async) and use it in multi-process programs	Create a client
Set timeouts, retries, proxies and your own httpx client	All options
Issue sign-in tickets for your app	Issue sign-in tickets
Send messages from a notification account	Send messages
Share the App Token across processes, rotate the Secret	App Token

Install

python -m pip install deeprespond-im-server-sdk==1.0.0a2

It's published on PyPI as version 1.0.0a2. This is the third alpha pre-release (written 1.0.0a2 under Python's versioning rules) and adds game rooms; the install commands in these docs pin this version. The SDK's wheel and source distribution don't include the demo; to try the UI, download the demo ZIP separately.

• Runtime: Python 3.11 or later (CPython), in both sync and async programs.
• Dependencies: the only direct dependency is httpx (>=0.27,<1). It doesn't depend on Pydantic, requests or aiohttp, so it won't conflict with their versions in your project.
• The import name is deeprespond_im, with type annotations included (py.typed), so mypy --strict and Pyright check your calls directly.

Callback framework adapters (Flask, Django, FastAPI, Starlette) and Redis stores live in separate submodules that only import their libraries when used; the SDK itself doesn't depend on them. You can install those libraries alongside, purely for convenience:

python -m pip install "deeprespond-im-server-sdk[redis]==1.0.0a2"
python -m pip install "deeprespond-im-server-sdk[flask,redis]==1.0.0a2"

The extras are flask, django, fastapi, starlette and redis; only install the frameworks you actually use. The demo ZIP has its own requirements.txt with FastAPI, Uvicorn, WebSocket and upload-parsing dependencies needed to run the UI.

Module	Contents
deeprespond_im	The clients DRClient and AsyncDRClient, error types, paging and batch results, RawJSON, UNSET, RateLimit, Hooks
deeprespond_im.types	Response data classes (User, Message, Group and so on) and the input types for each item of batch APIs
deeprespond_im.callback	Receiving callbacks: verifying signatures, preventing replay, de-duplicating, dispatching; see Receiving callbacks
deeprespond_im.callback.flask, .django, .fastapi, .starlette	Callback framework adapters
deeprespond_im.redis	Redis stores for the App Token, callback nonces and event de-duplication
deeprespond_im.testing	Testing helpers: a fake OpenAPI and signed callback requests

Create a client

Get the app's Client ID and Client Secret from the console, and create a client with them plus the API address, org_name and app_name:

import os

from deeprespond_im import DRClient

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"],
)

Async programs (FastAPI, Starlette, aiohttp and so on) use AsyncDRClient. The two have exactly the same parameters, services, methods and behavior; the async client's methods are coroutines and need await:

import asyncio
import os

from deeprespond_im import AsyncDRClient


async def main() -> None:
    async with 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"],
    ) as client:
        user = await client.users.get("zhangsan")
        print(user.nickname)


asyncio.run(main())

• One client per app. A client only calls the APIs of the app it was created for. To manage several apps, create one for each.
• Reuse it across the process. The client holds a connection pool and the App Token, so create one at startup and share it, instead of one per request. The sync client can be used by several threads at once.
• Creating it sends no requests. It only checks parameters; the App Token is fetched on the first API call. Invalid parameters raise ValueError, such as a base_url that isn't an absolute http:// or https:// URL or that has a query string. So credentials are never sent in clear text, http:// is only allowed for the local machine and private networks (localhost, 127.0.0.0/8, 10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16, and hostnames without dots); everything else must use https://.
• Multiple processes. gunicorn and uWSGI workers and Celery workers are forked processes, and a connection pool can't be shared across processes. Create the client in each worker process (such as gunicorn's post_fork, Celery's worker_process_init, or lazily on first use); using a client created in the parent process from another process raises RuntimeError. Clients can't be pickled either.
• The async client is bound to an event loop. You can create it before the event loop starts (such as a module-level variable), but it can only be used in the event loop where it sent its first request. Create one per event loop (such as per asyncio.run(), or per pytest-asyncio test), or you get RuntimeError; see Using it with asyncio.

Don't use the SDK in browsers, apps or mini programs: the Client Secret and App Token are as powerful as the app administrator's password.

All options

DRClient and AsyncDRClient take the same parameters, all keyword-only:

Parameter	Type	Default	Description
base_url	str	Required	The IM service address, such as https://im.example.com; a trailing / is removed
org_name	str	Required	The tenant ID, the part of the AppKey before #
app_name	str	Required	The app name, the part of the AppKey after #
client_id	str	Required	The app's Client ID
client_secret	str	Required	The app's Client Secret
timeout	float	30.0	Time limit per attempt (seconds). This is the total for one attempt, including connecting, sending and reading the whole response, not per phase as in httpx
max_retries	int	3	Automatic retries after network errors, timeouts and 5xx, not counting the first attempt; 0 means none. See Retries and idempotency
max_rate_limited_retries	int	3	How many times to wait and resend after 429 rate_limited; see Rate limits
max_retry_after	float	60.0	Longest Retry-After to wait automatically (seconds); when the server asks for longer, RateLimitError is raised
rate_limit	RateLimit | None	None	Local rate limiting, off by default; see Local rate limiting
token_store	TokenStore | None	None	Store for sharing the App Token across processes and servers; see Sharing the App Token across processes. AsyncTokenStore for the async client
http_client	httpx.Client | None	None	Your own httpx client; see Using your own HTTP client. httpx.AsyncClient for the async client
hooks	Hooks | None	None	Hooks called before and after each attempt, for tracing and metrics; see Logs and troubleshooting. AsyncHooks for the async client
user_agent_suffix	str | None	None	Appended to User-Agent, such as your service name and version. The default User-Agent is deeprespond-im-server-sdk/python/version
redact_httpx_logs	bool	True	Adds a filter to the httpx logger that replaces URL query strings with ?…, so object storage signed URLs don't end up in logs; see Logs and troubleshooting

Using your own HTTP client

By default the SDK creates its own httpx client (not following redirects) and closes it when the SDK client closes. To set a proxy, custom certificates or connection pool, pass your own client:

import os

import httpx

from deeprespond_im import DRClient

http = httpx.Client(
    proxy="http://proxy.internal:3128",   # 或 trust_env=False：不读取 HTTP_PROXY 等环境变量
    limits=httpx.Limits(max_connections=200),
)
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"],
    http_client=http,
)

• The SDK still sets the time limit and headers for each request, without changing your client's connection pool, proxy or certificate settings.
• The SDK doesn't close the client you pass in; close it yourself after client.close().
• httpx reads environment variables such as HTTP_PROXY and HTTPS_PROXY by default. If the server has a proxy set but traffic to the IM service shouldn't use it, pass a client with trust_env=False.
• Don't turn off certificate verification (verify=False); the SDK has no option for that either.

Changing options temporarily

client.with_options() returns a view that shares the App Token, connection pool and rate-limit state with the original client, changing only time limits and retry settings. It suits calls in live requests that should fail fast:

from deeprespond_im import DRClient, DRError


def nickname_or_default(client: DRClient, username: str) -> str:
    fast = client.with_options(timeout=3, max_retries=0)
    try:
        return fast.users.get(username).nickname
    except DRError:
        return username

You can change timeout, max_retries, max_rate_limited_retries and max_retry_after. A view's close() does nothing; the connection pool closes with the original client, and the view can't be used after the original is closed. To limit just one call's total time, use the method's deadline parameter; see Parameters every method takes.

Issue sign-in tickets

After a user signs in to your system, your server issues an IM sign-in ticket for them and gives it to the app or web page, where a client SDK signs in to IM with it (see Sign-in tickets and the Web SDK):

from deeprespond_im import DRClient, PermissionDeniedError


def im_ticket_for(client: DRClient, user_id: str) -> str:
    """业务系统“获取 IM 登录凭证”接口的实现：IM 用户名直接用业务系统的用户 ID。"""
    try:
        ticket = client.users.issue_login_ticket(user_id, auto_create=True)
    except PermissionDeniedError as e:
        if e.code == "user_disabled":
            raise PermissionError("账号已被封禁") from e
        raise
    return ticket.ticket

• With auto_create=True, a missing user is created automatically, so you don't need to sync users first.
• A ticket is valid for 5 minutes and can be used once. Each call issues a new ticket and old ones stay valid, so this call is safe to retry, and the SDK retries it automatically on network errors.
• In the returned LoginTicket, ticket is the ticket, expires_in the lifetime (seconds), and user_created whether the user was created this time. The ticket doesn't appear in repr(), so logging the object won't leak it.

Send messages

Send a one-to-one message to a user from your notification account (an ordinary user):

from deeprespond_im import DRClient


def notify_paid(client: DRClient, username: str, order_id: str) -> None:
    result = client.messages.send(
        from_="notice",                          # from 是 Python 的关键字，参数名写作 from_
        to_user=username,
        client_msg_id=f"order-{order_id}-paid",  # 业务上唯一的 ID：重试不会重复发送
        type="custom",
        body={"card": "order", "order_id": order_id},
        push={"title": "订单已支付", "body": f"订单 {order_id} 已支付"},
    )
    if result.duplicate:
        print("这条通知之前已经发送过", result.message_id)

• Give exactly one of to_user (one-to-one) and to_group (group); group messages without from_ are sent as the system.
• The server de-duplicates by conversation, sender and client_msg_id: a repeated submission returns the message from the first write, with result.duplicate True. Use a stable ID such as an order number as client_msg_id so your own retries don't send twice; without one, the SDK generates one per call, and its own automatic retries use the same ID.
• For the server's sending rules (mutes, friendships and so on aren't checked), see Send messages.

App Token

The SDK exchanges the Client ID and Client Secret for an App Token automatically (see Authentication and App Token), so you never call the token API yourself or check whether it's expired before each call.

Getting and caching

• Fetched on the first call. Creating the client sends no request; the token is fetched on the first API call and cached in the client until it's close to expiring.
• Fetched once at a time. When several threads (or coroutines in the async client) need a token at once, only one fetches it and the others wait for its result.
• Renewed early. When less than 10% of the token's lifetime is left (at least 5 minutes early), a new token is fetched in the background while requests keep using the old one, so they don't slow down. When it has expired or has less than 1 minute left, requests wait for the new one.
• Fetched again after it becomes invalid. When the token is revoked (such as revoking the Secret in the console), requests get 401 unauthenticated without a reason; the SDK drops the token, fetches a new one, and resends the request once.
• When fetching fails. Network errors and 5xx are retried after 1, 2 and 4 seconds; rate limits wait for Retry-After. With wrong credentials (invalid_client) or a source IP not on the allowlist (ip_not_allowed), it won't fetch again for 1 minute; when the tenant or app is paused (tenant_unavailable, app_unavailable), it won't fetch again for 2 to 5 minutes. Calls during this cooldown raise ClientUnavailableError directly, with cause_code the original error code, rather than every request trying to fetch; see App and tenant status.
• Permanently unavailable. When the app was deleted or the tenant closed (401 unauthenticated with details.reason app_deleted or tenant_closed), the client becomes permanently unavailable and every later call raises ClientUnavailableError.

For troubleshooting, client.token() (await client.token() in the async client) returns the currently valid token, fetching one if needed, and client.app_id is the app ID obtained when fetching. Never log the token.

Sharing the App Token across processes

Each process fetches and caches its own App Token. With many processes or frequent restarts (the token API allows 120 requests per minute per source IP), configure a shared token store so every process and server uses one token. deeprespond_im.redis provides a Redis implementation that takes your own redis.Redis client (AsyncRedisTokenStore and redis.asyncio.Redis for the async client):

import os

import redis

from deeprespond_im import DRClient
from deeprespond_im.redis import RedisTokenStore

r = redis.Redis.from_url(os.environ["REDIS_URL"])
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"],
    token_store=RedisTokenStore(r, key_prefix="drim:"),
)

• Store keys are computed from the API address, org_name, app_name and Client ID, so one Redis can hold tokens for several apps.
• When a token becomes invalid, the SDK only deletes it from the store if the store still holds that invalid token, never a new one another process just fetched.
• If the store errors (such as Redis being unavailable), the SDK logs a warning and falls back to fetching in this process, without affecting calls.

You can use other storage too, as long as it provides these three methods (all coroutines for the async client):

from datetime import datetime
from typing import Protocol


class TokenStore(Protocol):
    def get(self, key: str) -> tuple[str, datetime] | None: ...          # 令牌和过期时刻，没有时返回 None
    def set(self, key: str, token: str, expires_at: datetime) -> None: ...
    def compare_and_delete(self, key: str, token: str) -> None: ...      # 只在存储中的令牌等于 token 时删除

Rotating the Client Secret

After rotating the Secret in the console, the old Secret has a 24-hour grace period. Once your config has the new Secret, call update_credentials(): the client drops the current token, the next call fetches with the new credentials, and any cooldown after a failed fetch ends:

import os

from deeprespond_im import DRClient


def reload_credentials(client: DRClient) -> None:
    client.update_credentials(os.environ["IM_CLIENT_ID"], os.environ["IM_CLIENT_SECRET"])

Closing the client

Close the client when the program exits to release the connection pool. Use with (async with for the async client), or call close() (await client.aclose() for the async client):

import os

from deeprespond_im import DRClient

with 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"],
) as client:
    for user in client.users.iter(status="active"):
        print(user.username)

• Calling methods after closing raises ClientUnavailableError (cause_code closed).
• Closing doesn't affect App Tokens already issued; they stay valid until they expire.
• If you passed your own http_client, the SDK doesn't close it.

Next steps

• Calling the API: naming rules, paging, batch results, errors, retries and rate limits, file uploads, using it with asyncio
• Receiving callbacks: verifying signatures, de-duplicating, handling event and sync callbacks, integrating with web frameworks
• Services and methods: every method mapped to its REST API
• Demo guide: the source ZIP, minimal runtime dependencies, users' sign-in identities, callback setup and troubleshooting

Channel version

1.0.0a0 already includes standalone channel management and types for the 9 channel callbacks. Integrate them with Channel methods and the usage example; users still join media from the client.