Python SDK

本页介绍 Python 服务端 SDK 的安装、创建客户端和全部选项、第一次调用（签发登录凭证、发送消息），以及 SDK 怎样获取、缓存和刷新 App Token。调用接口的通用规则见调用接口，接收回调见接收回调，全部方法见服务与方法。

安装

bash

pip install deeprespond-im-server-sdk

运行环境：Python 3.11 以上（CPython），同步和异步程序都可以使用。
依赖：直接依赖只有 httpx（>=0.27,<1），不依赖 Pydantic、requests、aiohttp，不会与你项目中这些库的版本冲突。
导入名为 deeprespond_im，包内带类型标注（py.typed），在 mypy --strict 和 Pyright 下可以直接检查你的调用。

接收回调的框架适配（Flask、Django、FastAPI、Starlette）和 Redis 存储都在单独的子模块中，用到时才导入对应的库，SDK 本身不依赖它们。安装时可以顺带安装这些库，只是为了方便：

bash

pip install "deeprespond-im-server-sdk[redis]"            # 顺带安装 redis（>=5）
pip install "deeprespond-im-server-sdk[flask,redis]"      # 可选：flask、django、fastapi、starlette、redis

模块	内容
`deeprespond_im`	客户端 `DRClient`、`AsyncDRClient`，错误类型，分页与批量结果，`RawJSON`、`UNSET`、`RateLimit`、`Hooks`
`deeprespond_im.types`	响应的数据类（`User`、`Message`、`Group` 等）和批量接口每一项的输入类型
`deeprespond_im.callback`	接收回调：校验签名、防重放、去重、分派，见接收回调
`deeprespond_im.callback.flask`、`.django`、`.fastapi`、`.starlette`	回调的框架适配
`deeprespond_im.redis`	App Token、回调随机数和事件去重的 Redis 存储
`deeprespond_im.testing`	测试辅助：假的 OpenAPI、带签名的回调请求

创建客户端

在控制台取得应用的 Client ID 和 Client Secret，连同接入地址、org_name、app_name 创建客户端：

python

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

异步程序（FastAPI、Starlette、aiohttp 等）使用 AsyncDRClient。两者的参数、服务、方法和行为完全相同，异步客户端的方法是协程，要加 await：

python

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())

一个客户端对应一个应用：客户端只调用创建时给出的应用的接口。管理多个应用时，每个应用各建一个。
在进程中复用：客户端持有连接池和 App Token，请在程序启动时创建一个，全程共用，不要每次请求新建。同步客户端可以被多个线程同时使用。
创建时不发请求：只检查参数，第一次调用接口时才换取 App Token。参数不合规时抛出 ValueError，如 base_url 不是 http:// 或 https:// 的绝对地址，或带了查询参数。为了不以明文发出凭据，http:// 只允许本机和内网地址（localhost、127.0.0.0/8、10.0.0.0/8、172.16.0.0/12、192.168.0.0/16 和不带点的主机名），其他地址必须用 https://。
多进程：gunicorn、uWSGI 的工作进程、Celery 的 worker 都是派生（fork）出来的进程，连接池不能跨进程使用。请在每个工作进程中创建客户端（如 gunicorn 的 post_fork、Celery 的 worker_process_init，或第一次使用时再创建）；在另一个进程中使用父进程创建的客户端时，SDK 抛出 RuntimeError。客户端也不能 pickle。
异步客户端绑定事件循环：异步客户端可以在事件循环启动之前创建（如模块级的变量），但只能在第一次发出请求时所在的事件循环中使用。每个事件循环（如每次 asyncio.run()、pytest-asyncio 的每个用例）各建一个，否则抛出 RuntimeError，见在 asyncio 中使用。

不要在浏览器、App 或小程序中使用 SDK：Client Secret 和 App Token 等同应用管理员的密码。

全部选项

DRClient 和 AsyncDRClient 的参数相同，都必须写参数名：

参数	类型	默认值	说明
`base_url`	`str`	必填	IM 服务的接入地址，如 `https://im.example.com`，末尾的 `/` 会去掉
`org_name`	`str`	必填	租户标识，即 AppKey 中 `#` 之前的部分
`app_name`	`str`	必填	应用名称，即 AppKey 中 `#` 之后的部分
`client_id`	`str`	必填	应用的 Client ID
`client_secret`	`str`	必填	应用的 Client Secret
`timeout`	`float`	`30.0`	单次尝试的时限（秒）。这是一次尝试的总时长，包括连接、发送和读完响应，不是 `httpx` 那样按阶段分别计算
`max_retries`	`int`	`3`	网络错误、超时、`5xx` 后自动重试的次数，不含第一次；`0` 表示不重试，见重试与幂等
`max_rate_limited_retries`	`int`	`3`	收到 `429 rate_limited` 后等待并重发的次数，见限流
`max_retry_after`	`float`	`60.0`	自动等待的 `Retry-After` 上限（秒），服务端要求等待更久时直接抛出 `RateLimitError`
`rate_limit`	`RateLimit \| None`	`None`	本地限速，默认关闭，见本地限速
`token_store`	`TokenStore \| None`	`None`	多个进程、多台服务器共用 App Token 的存储，见多个进程共用 App Token。异步客户端为 `AsyncTokenStore`
`http_client`	`httpx.Client \| None`	`None`	自己的 `httpx` 客户端，见使用自己的 HTTP 客户端。异步客户端为 `httpx.AsyncClient`
`hooks`	`Hooks \| None`	`None`	每次尝试前后调用的钩子，用于追踪和指标，见日志与排查。异步客户端为 `AsyncHooks`
`user_agent_suffix`	`str \| None`	`None`	追加在 `User-Agent` 之后，如你的服务名和版本。默认的 `User-Agent` 为 `deeprespond-im-server-sdk/python/版本号`
`redact_httpx_logs`	`bool`	`True`	给 `httpx` 的日志器加过滤器，把日志中地址的查询参数替换为 `?…`，避免对象存储的签名地址写进日志，见日志与排查

使用自己的 HTTP 客户端

默认情况下 SDK 自己创建 httpx 客户端（不跟随重定向），关闭 SDK 的客户端时一起关闭。需要设置代理、自定义证书或连接池时，传入自己的客户端：

python

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

SDK 照常为每个请求设置时限和请求头，不改动你的客户端的连接池、代理和证书设置；
SDK 不关闭你传入的客户端，client.close() 之后请自己关闭它；
httpx 默认读取 HTTP_PROXY、HTTPS_PROXY 等环境变量。服务器上设置了代理而访问 IM 服务不应经过代理时，传入 trust_env=False 的客户端；
请不要关闭证书校验（verify=False），SDK 也不提供这样的选项。

临时改变选项

client.with_options() 返回一个视图，与原客户端共用 App Token、连接池和限流状态，只改变时限和重试的设置。适合在线请求中需要快速失败的调用：

python

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

可以改变的有 timeout、max_retries、max_rate_limited_retries、max_retry_after。视图的 close() 什么也不做，连接池随原客户端关闭；原客户端关闭后视图也不能再用。只想限制某一次调用的总时长时，用方法的 deadline 参数，见每个方法的通用参数。

签发登录凭证

用户登录你的业务系统后，业务服务端为他签发一个 IM 登录凭证，交给 App 或网页，客户端 SDK 用它登录 IM 服务（见登录凭证和 Web SDK）：

python

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

auto_create=True 时用户不存在则自动创建，不必先同步用户；
凭证 5 分钟内有效、只能使用一次。每次签发得到新的凭证，旧的仍然有效，所以这个调用可以安全地重试，SDK 在网络错误时会自动重试；
返回的 LoginTicket 中，ticket 是凭证，expires_in 是有效期（秒），user_created 表示这次是否创建了用户。凭证不出现在 repr() 中，把对象写进日志不会泄露它。

发送消息

以业务的通知账号（一个普通用户）给用户发一条单聊消息：

python

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)

to_user（单聊）和 to_group（群聊）恰好给一个；群聊不给 from_ 时以系统身份发送；
服务端按“会话、发送者、client_msg_id”去重：重复提交返回第一次写入的消息，result.duplicate 为 True。用业务单号这样的稳定 ID 作为 client_msg_id，业务层的重试也不会重复发送；不传时 SDK 为每次调用生成一个，SDK 自己的自动重试用同一个 ID；
服务端发送的规则（不检查禁言、好友关系等）见发送消息。

App Token

SDK 用 Client ID 和 Client Secret 自动换取 App Token（见鉴权与 App Token），你不需要自己调用换取接口，也不需要在每次调用前判断它是否过期。

获取与缓存

第一次调用时换取：创建客户端不发请求，第一次调用接口时换取，之后缓存在客户端中，直到快要过期；
只换取一次：多个线程（异步客户端中为多个协程）同时需要令牌时，只有一个去换取，其他的等待它的结果；
提前续期：剩余有效期不到总有效期的 10%（至少提前 5 分钟）时，在后台换取新令牌，换取期间请求照常使用旧令牌，不会因为换取而变慢。已过期或剩余不足 1 分钟时，请求等待换取完成；
失效后重新换取：令牌被吊销（如在控制台吊销了 Secret）时，请求收到不带原因的 401 unauthenticated，SDK 作废这个令牌，换取一次新令牌后把这个请求重发一次；
换取失败的处理：网络错误和 5xx 在 1、2、4 秒后重试；被限流时按 Retry-After 等待。凭据错误（invalid_client）或来源 IP 不在白名单（ip_not_allowed）时，1 分钟内不再换取；租户或应用暂停服务（tenant_unavailable、app_unavailable）时，2 到 5 分钟内不再换取。这段冷却期内的调用直接抛出 ClientUnavailableError，它的 cause_code 是最初的错误码，不会每个请求都去换取，见应用和租户的状态；
永久失效：应用已删除或租户已注销时（401 unauthenticated，details.reason 为 app_deleted、tenant_closed），客户端永久失效，之后的调用都抛出 ClientUnavailableError。

需要排查时，可以用 client.token()（异步客户端为 await client.token()）取得当前有效的令牌（需要时换取），client.app_id 是换取时得到的应用 ID。不要把令牌写进日志。

多个进程共用 App Token

每个进程各自换取和缓存 App Token。进程很多、重启频繁时（换取接口按来源 IP 每分钟限 120 次），可以配置共享的令牌存储，让所有进程和服务器共用一个令牌。deeprespond_im.redis 提供了 Redis 的实现，接受你自己创建的 redis.Redis 客户端（异步客户端用 AsyncRedisTokenStore 和 redis.asyncio.Redis）：

python

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

存储的键由接入地址、org_name、app_name 和 Client ID 计算得出，同一个 Redis 可以存放多个应用的令牌；
令牌失效时，SDK 只在存储中的令牌仍是失效的那个时才删除它，不会删掉别的进程刚换来的新令牌；
存储出错（如 Redis 不可用）时，SDK 记一条警告日志，退回本进程自己换取，调用不受影响。

也可以用其他存储实现，只要提供以下三个方法（异步客户端的三个方法都是协程）：

python

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 时删除

轮换 Client Secret

在控制台轮换 Secret 后，旧 Secret 还有 24 小时的宽限期。把配置更新为新 Secret 后调用 update_credentials()，客户端丢弃当前令牌，下一次调用用新凭据换取，同时结束换取失败后的冷却期：

python

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

关闭客户端

程序退出时关闭客户端，释放连接池。可以用 with（异步客户端为 async with），或调用 close()（异步客户端为 await client.aclose()）：

python

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)

关闭后再调用方法，抛出 ClientUnavailableError（cause_code 为 closed）；
关闭不影响已经签发的 App Token，它在有效期内仍然有效；
传入了自己的 http_client 的，SDK 不关闭它。

下一步

调用接口：命名规则、分页、批量结果、错误、重试与限流、文件上传、在 asyncio 中使用
接收回调：校验签名、去重、处理事件回调和同步回调、与 Web 框架集成
服务与方法：全部方法与 REST 接口的对照

频道版本

独立频道服务及九种强类型回调已写入当前源码，尚未发布本轮新 SDK 制品。安装旧版本时可能没有 Channels / channels 属性，请确认提供的版本包含频道方法，再按调用示例接入。

Python SDK ​

安装 ​

创建客户端 ​

全部选项 ​

使用自己的 HTTP 客户端 ​

临时改变选项 ​

签发登录凭证 ​

发送消息 ​

App Token ​

获取与缓存 ​

多个进程共用 App Token ​

轮换 Client Secret ​

关闭客户端 ​

下一步 ​

频道版本 ​

Python SDK

安装

创建客户端

全部选项

使用自己的 HTTP 客户端

临时改变选项

签发登录凭证

发送消息

App Token

获取与缓存

多个进程共用 App Token

轮换 Client Secret

关闭客户端

下一步

频道版本