Python SDK

本頁介紹 Python 服務端 SDK 的安裝、建立客戶端和全部選項、第一次呼叫（簽發登入憑證、傳送訊息），以及 SDK 怎樣獲取、快取和重新整理 App Token。呼叫介面的通用規則見呼叫介面，接收回調見接收回調，全部方法見服務與方法。

完整聊天示例的下載、最小依賴、啟動與回撥配置見 Demo 執行指南。SDK 使用應用身份；Demo 另在後端儲存使用者的登入身份，用來讀取使用者能看到的歷史和執行使用者自己的撤回。

要做什麼	看哪裡
安裝 SDK 和可選依賴	安裝
建立客戶端（同步或非同步），用在多程序程式中	建立客戶端
設定超時、重試、代理和自己的 httpx 客戶端	全部選項
為 App 簽發登入憑證	簽發登入憑證
以通知賬號發訊息	傳送訊息
多個程序共用 App Token、輪換 Secret	App Token

安裝

bash

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

當前已釋出到 PyPI，版本為 1.0.0a2。這是第三個 Alpha 預釋出版本（按 Python 版本規則寫作 1.0.0a2），增加了遊戲房間；文件安裝命令固定這一版本。 SDK 的 wheel 和原始碼包不包含 Demo；體驗介面請單獨下載 Demo ZIP。

執行環境：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

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"

可選依賴包括 flask、django、fastapi、starlette、redis，只安裝實際使用的框架。 Demo ZIP 已提供單獨的 requirements.txt，包含執行介面所需的 FastAPI、Uvicorn、WebSocket 和上傳解析依賴。

模組	內容
`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 介面的對照
Demo 執行指南：原始碼 ZIP、最小執行依賴、真實裝置身份、回撥配置和故障排查

頻道版本

1.0.0a0 已包含獨立頻道管理及九種頻道強型別回撥。按頻道方法和呼叫示例接入；使用者加入媒體仍由客戶端完成。

Python SDK ​

安裝 ​

建立客戶端 ​

全部選項 ​

使用自己的 HTTP 客戶端 ​

臨時改變選項 ​

簽發登入憑證 ​

傳送訊息 ​

App Token ​

獲取與快取 ​

多個程序共用 App Token ​

輪換 Client Secret ​

關閉客戶端 ​

下一步 ​

頻道版本 ​

Python SDK

安裝

建立客戶端

全部選項

使用自己的 HTTP 客戶端

臨時改變選項

簽發登入憑證

傳送訊息

App Token

獲取與快取

多個程序共用 App Token

輪換 Client Secret

關閉客戶端

下一步

頻道版本