呼叫介面

本頁介紹用 Python SDK 呼叫服務端 REST API 的通用規則：REST 介面與 SDK 方法的對應方式、返回值、分頁與自動翻頁、批次介面的結果、錯誤的判斷、SDK 的重試與限流處理、檔案的上傳與下載、應用狀態的影響、日誌與排查，以及在 asyncio 程式和單元測試中的用法。

應用許可權與使用者可見性

SDK 以應用管理員的身份呼叫介面，不知道你的業務系統中當前登入的是誰，也不會替你檢查許可權。瀏覽器的操作到達你的後端後，要檢查使用者的登入狀態、會話是否屬於他、他是否是群成員和管理員。

messages.recall 是管理員撤回，不受普通使用者的角色和撤回時限限制，不能直接開放為“使用者撤回”的介面。
conversations.list_messages、iter_messages 讀到的是管理視角的全部歷史，不會按使用者過濾（各自刪除、清空的訊息，入群前、退群后的訊息）。使用者的聊天介面使用客戶端 SDK，或由後端以使用者的登入令牌（User Token）讀取他能看到的歷史。
Client Secret、App Token 和使用者的登入令牌都儲存在後端，不發給瀏覽器。

服務端 SDK 和以使用者身份呼叫的介面怎樣配合，見 Demo 指南。

各介面的引數和語義見服務端 REST API，每個方法對應哪個介面見服務與方法。下面的示例中，client 是建立客戶端得到的 DRClient；使用 AsyncDRClient 時在呼叫前加 await。

要做什麼	看哪裡
把操作開放給使用者之前要檢查什麼	應用許可權與使用者可見性
找到方法，寫引數、`None` 和 `UNSET`	命名規則
給單次呼叫設定時限和追蹤 ID	每個方法的通用引數
翻頁、遍歷全部資料	分頁
讀取批次介面的逐項結果、批次匯入	批次介面的結果
按錯誤型別處理失敗	錯誤
瞭解哪些請求會自動重試、處理限流	重試與冪等、限流
上傳、下載檔案	上傳檔案
在 FastAPI 等非同步程式中使用	在 asyncio 中使用
日誌、指標和單元測試	日誌與排查、測試自己的程式碼

命名規則

REST 介面按模組分為 13 個服務，每個服務是客戶端的一個屬性：client.users、client.friends、client.blacklist、client.groups、client.messages、client.conversations、client.media、client.push、client.chatrooms、client.moderation、client.calls、client.channels、client.presence。

方法名是 snake_case 的動詞或動詞短語：

REST 介面	SDK 方法
`POST /users` 建立使用者	`client.users.create(...)`
`GET /users/{username}` 查詢使用者	`client.users.get(username)`
`GET /users` 分頁列出使用者	`client.users.list(...)`，自動翻頁為 `client.users.iter(...)`
`GET /groups/{group_id}/members` 列出群成員	`client.groups.list_members(group_id)`，自動翻頁為 `iter_members`
`GET /groups/{group_id}/members/{username}` 查詢一位成員	`client.groups.get_member(group_id, username)`
`GET /users/{username}/push-settings` 查詢推送設定	`client.push.get_settings(username)`
`PUT`、`DELETE /groups/{group_id}/mute-all` 開啟、關閉全員禁言	`client.groups.set_mute_all(group_id, True)`、`set_mute_all(group_id, False)`

返回列表的方法叫 list 或 list_xxx，返回單個物件或一組設定的叫 get 或 get_xxx，分頁的列表另有遍歷方法 iter 或 iter_xxx；統計和彙總類的保留原名，如 conversations.unread_count、media.usage、moderation.stats、chatrooms.online_counts。全部方法見服務與方法。

引數與 REST 介面的欄位同名（服務端的欄位名本來就是 snake_case）：

位置引數：路徑中的標識（使用者名稱、群 ID、訊息 ID 等，按路徑中的順序）；請求體只有一個必填欄位時的那個欄位，如 users.set_password(username, password)、groups.set_role(group_id, username, role)；批次介面的項列表，如 users.batch_create(users)、messages.batch_send(to_users, ...)。例外是 calls.stats(from_, to) 的兩個日期。
關鍵字引數：其餘引數一律必須寫引數名，如 users.create("zhangsan", nickname="张三")。這樣以後介面增加引數時，不會悄悄傳錯位置。
與 Python 關鍵字同名的引數加下劃線：傳送者 from 寫作 from_，請求中仍是 from。type、id 等與內建函式同名的照常使用。
ID 一律是 str：傳 int 的在本地以 invalid_params 拒絕，不會發出請求（服務端不接受數字形式的 ID）。seq、版本號是 int。
時間：引數可以傳帶時區的 datetime（轉為 UTC 的 RFC 3339 字串）或字串；不帶時區的 datetime 在本地以 invalid_params 拒絕，SDK 不猜測時區。響應中的時間是帶 UTC 時區的 datetime。

不傳、傳 null 與傳值

修改類方法的可選引數預設值是 UNSET，表示不出現在請求中；傳 None 表示傳送 null；傳其他值照常傳送：

python

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)

引數的型別標註區分了這三種情況（如 str | None | Unset），型別檢查器能發現把 None 傳給不接受 null 的引數；執行時收到的同樣在本地以 invalid_params 拒絕。

原樣傳送 JSON

訊息的 body、ext 等自由內容可以傳 dict，SDK 按緊湊的 UTF-8 JSON 編碼（不把中文轉義為 \uXXXX，拒絕 NaN 和 Infinity）。需要保留數字的寫法（如 1.00）或你對內容做了簽名時，用 RawJSON 原樣傳送，SDK 只檢查它是一個合法的 JSON 物件：

python

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 只能用在自由內容的位置，放進普通引數裡的以 invalid_params 拒絕。服務端儲存訊息時會去掉 JSON 中的空白，對內容簽名的請基於緊湊寫法。

每個方法的通用引數

除了介面自己的引數，每個方法都可以傳以下三個引數：

引數	說明
`timeout`	這次呼叫中每次嘗試的時限（秒），覆蓋客戶端的 `timeout`
`deadline`	這次呼叫的總時限（秒），包括全部重試、限流等待和換取 App Token 的時間。到時丟擲 `DRTimeoutError`，不再重試
`request_id`	追蹤 ID，作為請求頭 `X-Request-ID` 傳送，服務端的日誌和錯誤響應中使用同一個 ID。1 到 64 個字母、數字或 `-`、`_`、`.`，不給時 SDK 生成

python

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

遍歷方法（iter、iter_xxx）只有 timeout 和 request_id，沒有 deadline。

返回值

響應解析為 deeprespond_im.types 中的資料類，欄位名與 REST 文件相同：

python

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

資料類不可修改，== 按業務欄位比較；不可雜湊，不能作為 dict 的鍵或放進 set；
新增的欄位：服務端以後增加的欄位不必等 SDK 升級，從 raw 中讀取。列舉類的欄位是 str，遇到新的取值也不會報錯；
訊息的原文：含訊息的物件（Message、SendResult、回撥事件等）除了解析後的 body、ext（dict），另有 body_raw、ext_raw（RawJSON），是響應中這一段的原始文本，與服務端返回的逐位元組相同，可以直接用於驗籤或轉發；
repr() 不顯示 raw、*_raw 和登入憑證、下載地址等敏感欄位，把物件寫進日誌不會洩露它們；
REST 介面返回 204 No Content 的方法返回 None。

分頁

分頁的列表方法返回一頁結果。按游標分頁的是 Page，next_cursor 為 None 表示沒有下一頁：

python

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)

型別	用於	欄位
`Page[T]`	大多數列表	`items`、`next_cursor`、`raw`
`VersionedPage[T]`	好友列表、黑名單、群成員等帶版本號的列表	`Page` 的欄位，加 `version`、`not_modified`、`count`（好友數、黑名單人數、群成員數）、`max_count`（好友、黑名單的上限）；沒有的為 `None`
`SeqPage[Message]`	會話訊息 `conversations.list_messages`	`items`（按序號從小到大）、`has_more`、`next_after_seq`、`next_before_seq`，按 `around_seq` 取時另有 `has_more_before`、`has_more_after`
`ExportPage[Message]`	訊息匯出 `messages.export`	`items`、`has_more`、`next_after_message_id`

帶版本號的列表可以傳上次得到的 version 作為 known_version，沒有變化時 not_modified 為 True、items 為空，不必重新拉取：

python

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

一頁的條數可能少於 limit，甚至為 0 而仍有下一頁（服務端跳過了已刪除的資料），是否結束只看 next_cursor（會話訊息和匯出看 has_more），不要按條數判斷。

自動翻頁

分頁的列表都有對應的遍歷方法，逐項產出全部結果，只在需要下一頁時才發請求：

python

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

遍歷方法的引數與列表方法相同，只是沒有 cursor（和 known_version）；limit 是每頁的條數；
返回的迭代器也是上下文管理器，with 結束時立即釋放；提前 break 不會留下進行中的請求；
某一頁最終失敗時丟擲異常並結束遍歷，已產出的項仍然有效。每一頁照常自動重試；
非同步客戶端的遍歷方法返回非同步迭代器，用 async for，提前結束時用 async with 或 await it.aclose() 立即釋放；
conversations.iter_messages 給出 after_seq 時向新的方向遍歷，否則從最新的訊息向舊的方向，整體從新到舊產出；
chatrooms.iter_members 按使用者名稱去重：遍歷期間離開又進入的使用者只產出一次。

另有兩個方法替你迴圈呼叫：friends.delete_all(username) 迴圈刪除直到沒有剩餘的好友，返回刪除的總數；moderation.process_pending(...) 逐條處理待稽核的記錄，見服務與方法。

批次介面的結果

批次介面（users.batch_create、friends.batch_add、messages.batch_send、messages.batch_recall 等）逐項處理，整個請求成功時返回結果物件，失敗的項在結果中，不丟擲異常：

python

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)

每個批次介面有自己的結果型別（如 BatchCreateUsersResult、BatchSendResult），欄位與 REST 文件相同；
所有結果型別都有 ok（全部成功）、failed_items（失敗的項，ItemError 的列表）和 raise_for_failures()（有失敗的項時丟擲 BatchError，否則什麼也不做）；
ItemError 有 key（這一項的標識，如使用者名稱、訊息 ID）、code、message、details 和 reason；
BatchError 不是 DRError 的子類：請求本身成功了，只是有項失敗。它的 failed 是失敗的項，result 是整個結果；
每一項的輸入是普通的 dict，型別檢查器按 deeprespond_im.types 中的 TypedDict（如 CreateUserInput）檢查鍵名和型別。

超過上限的批次

帶 _all 字尾的方法（users.batch_create_all、friends.batch_add_all、messages.batch_send_all）接受任意可迭代物件（包括生成器），按每次請求的上限拆分、依次提交、合併結果，不必先把全部資料讀進記憶體：

python

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

某一批整個請求失敗時停止，丟擲那一批原來的錯誤（如 RateLimitError），它的 partial 屬性是已完成部分的合併結果；
應用設定了更小的每批上限、服務端以 too_many_items 拒絕時，SDK 按服務端給出的上限重新拆分；
messages.batch_send_all 的各批使用同一個 client_msg_id，整體重試也不會重複傳送。

錯誤

呼叫失敗時丟擲 DRError 或它的子類。服務端返回的錯誤（見錯誤碼）原樣轉為 DRError：

屬性	說明
`code`	錯誤碼，如 `not_found`、`rate_limited`，或 SDK 自己的錯誤碼（見下文）。按它判斷，不要解析 `message`
`reason`	`details.reason`，沒有時為 `None`
`details`	錯誤響應中的 `details`，原樣
`message`	說明文字，只用於日誌
`status_code`	HTTP 狀態碼；SDK 自己的錯誤為 `None`
`request_id`	請求 ID，聯絡技術支援時提供
`retry_after`	`Retry-After` 響應頭的秒數
`op`	介面名，如 `users.create`（見服務與方法的“介面名”一列）
`attempts`	這次呼叫一共嘗試了幾次
`retried`	這個錯誤來自自動重試後的嘗試，之前的嘗試可能已經執行，見重試與冪等
`partial`	只在 `_all` 方法中途失敗時有：已完成部分的結果
`is_temporary`	網路錯誤、超時、`internal`、`5xx` 和 `rate_limited`：稍後可以再試
`is_service_unavailable`	租戶或應用暫停服務、只讀（`tenant_unavailable`、`app_unavailable`，以及冷卻期中由它們引起的 `client_unavailable`）
`is_permanently_unavailable`	應用已刪除或租戶已登出

錯誤按 code 分為以下子類，可以按型別捕獲。同一個錯誤碼永遠屬於同一個子類，不認識的錯誤碼（服務端以後新增的）是 DRError 本身：

子類	錯誤碼
`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`，以及 SDK 的 `invalid_params`
`ContentRejectedError`	`message_rejected`、`content_rejected`
`ServerError`	`internal`
`NetworkError`	SDK 的 `network_error`，同時是內建的 `ConnectionError`
`DRTimeoutError`	SDK 的 `timeout`，同時是內建的 `TimeoutError`
`InvalidResponseError`	SDK 的 `invalid_response`
`ClientUnavailableError`	SDK 的 `client_unavailable`，另有 `cause_code`

SDK 自己的錯誤碼：

錯誤碼	說明
`invalid_params`	引數在本地檢查不通過，沒有發出請求，`details.field` 指出引數
`network_error`	連線失敗、連線中斷、TLS 錯誤；底層的 `httpx` 異常在 `__cause__` 中
`timeout`	單次嘗試超時，或 `deadline` 已到
`invalid_response`	響應不是預期的 JSON，如代理返回的 HTML 錯誤頁；`details.body_prefix` 是響應的開頭，`5xx` 時 `is_temporary` 為真
`client_unavailable`	客戶端已關閉（`cause_code` 為 `closed`）、已永久失效，或處於換取 App Token 失敗後的冷卻期（`cause_code` 為最初的錯誤碼）

python

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

異常的字串表示含介面名、HTTP 狀態、錯誤碼、原因和 request_id，不含請求體、令牌和 Client Secret，可以直接寫進日誌；
所有異常都可以 pickle，Celery、multiprocessing 把它們傳回呼叫方時欄位不變；
非同步任務被取消時，asyncio.CancelledError 原樣向上傳遞，不轉為 DRError。

重試與冪等

SDK 只自動重試可以安全重試的請求。每個方法屬於以下一類，服務與方法的“自動重試”一列給出了每個方法的類別：

類別	說明	網路錯誤、超時、`5xx` 時
查詢	只讀的介面	自動重試
冪等	重複執行結果相同，如設定、刪除、加入黑名單	自動重試
冪等鍵	傳送訊息：用 `client_msg_id` 去重，不傳時 SDK 生成	用同一個 `client_msg_id` 自動重試
唯一建立	以唯一標識建立，如建立使用者	自動重試；重試後得到 `already_exists` 時 `e.retried` 為真，說明可能是前一次嘗試已經建立
不重試	每次執行都產生新的結果：`groups.create`、`chatrooms.create`、`media.create_upload`、`moderation.create_word_list`、`moderation.escalate`、`users.set_password`、`users.kick_all`，以及只推線上的傳送（`online_only=True`）、低優先順序的聊天室訊息、帶 `actions` 而不帶 `version` 的 `moderation.decide`	只在請求確定沒有發出（如連線失敗）時重試

重試最多 max_retries 次（預設 3 次），間隔約 0.5、1、2 秒（各乘以 0.5 到 1.5 的隨機數）；5xx 帶 Retry-After 時至少等待這麼久；
超過 deadline 的不再重試；
retried 的含義：自動重試後得到 already_exists、not_found、version_conflict 時，e.retried 為真，前一次嘗試可能已經執行（如建立成功後響應丟失）。這時請查詢確認，不要直接當作失敗；
不重試的方法超時後，請先查詢結果再決定是否重試，例如建群超時後按群主呼叫 groups.list 確認（把業務 ID 寫進 attributes 便於識別）；
400、403、404、409 等錯誤不重試，直接丟擲；
令牌失效（不帶原因的 401 unauthenticated）時，SDK 換取新令牌後重發一次，不計入重試次數。

限流

收到 429 rate_limited 時（見限流），SDK 按 Retry-After 等待（加 0 到 20% 的隨機量，把多個請求錯開）後重發，最多 max_rate_limited_retries 次（預設 3 次），不計入 max_retries。等待的範圍按 details.reason 決定：

被限流的原因	誰等待
應用的 OpenAPI 呼叫額度用完（沒有 `details.reason`）	整個客戶端：所有執行緒或協程的請求一起暫停，直到等待結束，不會各自重試把額度再次用完
應用每分鐘的訊息數（`app_message_rate`）	這個客戶端的全部發送訊息請求（`messages.send`、`messages.batch_send`），等待時間在 `Retry-After` 到它的 2 倍之間
舉報的頻率（`request_rate`）	這個客戶端的全部 `moderation.report` 請求
其他原因，如某個群、某個聊天室的傳送頻率（`group_send_rate`、`room_send_rate`）	只有這個請求

以下情況不再等待，直接丟擲 RateLimitError（e.retry_after 為服務端要求等待的秒數）：服務端要求等待的時間超過 max_retry_after（預設 60 秒）；已經重發了 max_rate_limited_retries 次；等待會超過這次呼叫的 deadline。客戶端處於整體暫停中、而等待會超過 deadline 的呼叫，不發請求，直接丟擲 RateLimitError。

python

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

“整個客戶端一起等待”只在一個程序內有效：多個程序同時被限流時各自等待，等待時間中的隨機量會把它們錯開。

本地限速

批次匯入、群發通知這類後臺任務，可以用本地限速控制自己的呼叫速率，把 OpenAPI 額度留給線上請求：

python

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（向上取整）
)

本地限速是令牌桶：per_second 是每秒的次數，burst 是允許的突發次數，預設 max(1, ceil(per_second))；
按項數計入服務端額度的批次介面（如批次建立使用者、批次傳送），本地限速也按項數扣減（每次最多扣到 burst）；
等待會超過 deadline 的，丟擲 DRTimeoutError；
本地限速在一個客戶端內有效，多個程序按程序數分攤額度。

上傳檔案

media.upload() 完成建立上傳、上傳內容、完成上傳三步（見檔案上傳），返回檔案資訊：

python

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

引數	說明
`source`	位置引數：檔案路徑（`str` 或 `pathlib.Path`）、`bytes`，或可定位的二進位制檔案物件（以 `rb` 開啟，從當前位置讀到末尾）。不能定位的流（如網路流、標準輸入）不能分片重傳，以 `invalid_params` 拒絕，請先寫入臨時檔案
`purpose`、`kind`	用途和型別，如 `attachment` 和 `image`。用途與型別的組合在本地先檢查
`name`	檔名，省略時取路徑或檔案物件的檔名。型別為 `file` 和群檔案必須有檔名
`content_type`、`group_id`、`owner`	與建立上傳的欄位相同
`concurrency`	分片併發數，預設 4
`on_progress`	進度回撥 `on_progress(uploaded, total)`（位元組數）。同步客戶端中可能從其他執行緒呼叫
`request_id`	追蹤 ID

不超過 32 MB 的一次上傳，更大的分片（每片 16 MB）併發上傳。上傳內容的請求直接發到物件儲存，不帶 App Token；
file.url 是檔案地址，寫進訊息的 body；不要儲存下載地址，它會過期；
頭像（user_avatar 等）不能超過 5 MB、不能上傳空檔案，這些在本地先檢查；
分片的上傳地址過期時自動重新索取；完成上傳時服務端還在合併分片的，SDK 等待後重試；分片不全的，補傳後再完成；
同步客戶端中按 Ctrl+C（KeyboardInterrupt）中斷、非同步客戶端中任務被取消時，SDK 盡力取消這次上傳，再把異常原樣丟擲。

續傳大檔案

分片上傳（超過 32 MB）中斷後，24 小時內可以用 media.resume_upload(file_id, source) 和同一份內容繼續：SDK 查詢已上傳的分片，只補傳缺少的，再完成上傳。upload() 丟擲的異常中不帶檔案 ID，需要續傳的大檔案請自己建立上傳、儲存 file_id，再用 resume_upload() 上傳內容：

python

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)

下載檔案

media.download(url, dest) 用檔案地址換取下載地址，流式寫入路徑或可寫的二進位制檔案物件，不把整個檔案讀進記憶體，返回寫入的位元組數。下載地址過期（物件儲存返回 403）時重新換取一次：

python

from deeprespond_im import DRClient


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

只需要下載地址（如交給瀏覽器下載）時，用 media.get_download_urls([url, ...])。

media.download 只寫出檔案內容，不替你設定業務 HTTP 響應的內容型別、快取和下載頭。把附件轉發給瀏覽器時，不要按瀏覽器傳來的 MIME 型別決定響應型別；HTML、SVG 等可以執行指令碼的內容要強制下載，或放到單獨的域名下。Demo 的做法是：只有按內容識別為 PNG、JPEG、GIF、WebP 的直接顯示，其他都作為下載，並設定 nosniff、沙箱 CSP 和禁止快取。

應用和租戶的狀態

應用被停用、暫停，租戶被暫停或欠費時（見應用和租戶的狀態），呼叫的結果如下：

狀態	SDK 的表現	建議
只讀	查詢和處置類操作正常；新增資料、修改資料的寫操作丟擲 `ServiceUnavailableError`（`app_unavailable`，欠費時 `reason` 為 `arrears`）	告警；暫停寫入類的後臺任務
應用停用或暫停、租戶暫停	已簽發的令牌失效，SDK 重新換取時得到 `app_unavailable` 或 `tenant_unavailable`，丟擲 `ServiceUnavailableError`；之後 2 到 5 分鐘內不再換取，呼叫直接丟擲 `ClientUnavailableError`（`cause_code` 為最初的錯誤碼）	告警，在控制台檢視狀態；冷卻期結束後 SDK 自動再試，恢復後不必重啟
Client Secret 錯誤或已吊銷、IP 不在白名單	丟擲 `AuthenticationError`（`invalid_client`、`ip_not_allowed`）；之後 1 分鐘內呼叫直接丟擲 `ClientUnavailableError`	更新憑據後呼叫 `client.update_credentials()`，立即結束冷卻期
應用已刪除、租戶已登出	丟擲 `AuthenticationError`（`reason` 為 `app_deleted`、`tenant_closed`），客戶端永久失效，之後都丟擲 `ClientUnavailableError`	清除儲存的憑據，停止呼叫

按“服務暫停”處理的程式碼，用 e.is_service_unavailable 判斷，它同時涵蓋 ServiceUnavailableError 和冷卻期中的 ClientUnavailableError：

python

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

日誌與排查

SDK 用標準庫 logging 記錄日誌，日誌器為 deeprespond_im，下分 deeprespond_im.http（請求、重試、上傳）、deeprespond_im.token（App Token 的換取）和 deeprespond_im.callback（接收回調）。SDK 不新增處理器，是否輸出由你的日誌配置決定：

重試、限流等待記 WARNING 或 DEBUG；最終失敗的網路錯誤、超時、5xx、鑑權失敗記 ERROR；其他 4xx 是業務結果，記 DEBUG；
日誌不含請求體、App Token 和 Client Secret；
httpx 自己在 INFO 級別為每個請求記一行完整的地址，物件儲存的上傳地址、下載地址中的簽名都在查詢引數裡。SDK 預設給 httpx 日誌器加了過濾器，把地址的查詢引數替換為 ?…（redact_httpx_logs=False 關閉）。httpcore 日誌器在 DEBUG 級別記錄響應頭，生產環境請把它設為 WARNING 以上。

鉤子

hooks 選項在每次嘗試前後呼叫你的函式，可以注入追蹤頭、記錄指標。after_response 收到 RequestInfo：介面名 op（換取令牌為 app.token）、method、url（不含查詢引數）、第幾次嘗試 attempt、status_code、錯誤碼 code 和 reason、request_id、耗時 elapsed 和這次嘗試前的限流等待 rate_limit_wait：

python

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

鉤子在發出請求的執行緒中同步呼叫，不要做耗時的操作；鉤子丟擲的異常原樣向上傳遞；
不要在鉤子中記錄 Authorization 請求頭和請求體；
非同步客戶端用 AsyncHooks，兩個鉤子可以是普通函式，也可以是協程函式。

統計

client.stats() 返回按介面名統計的 OpStats：嘗試次數 requests、最終失敗的呼叫 failures、重試和限流後重發的次數 retries、限流等待的總時長 rate_limit_wait（秒）。可以定期採集到監控系統：

python

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

聯絡技術支援時，請提供錯誤的 request_id、op、code 和發生的時間。

在 asyncio 中使用

FastAPI、Starlette、aiohttp 等非同步程式使用 AsyncDRClient，不要在非同步程式碼中呼叫同步客戶端（它會阻塞事件迴圈）。在 FastAPI 中，可以在應用的生命週期中建立和關閉客戶端：

python

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}

非同步客戶端可以在模組級建立，但只能在第一次發出請求時所在的事件迴圈中使用。在其他事件迴圈中使用（如第二次 asyncio.run()、pytest-asyncio 為每個用例新建的事件迴圈）時丟擲 RuntimeError，請每個事件迴圈各建一個；
遍歷用 async for：async for user in im.users.iter(status="active"): ...；
可以用 asyncio.timeout() 限制整個呼叫，取消會中斷正在進行的請求和等待；也可以用方法的 deadline 引數。被取消或超時的寫操作可能已經執行，按重試與冪等中的類別處理；
非同步客戶端的 App Token 儲存用 AsyncTokenStore（如 deeprespond_im.redis.AsyncRedisTokenStore 和 redis.asyncio.Redis），鉤子用 AsyncHooks；
同步客戶端中沒有取消，只能用 timeout、deadline 和 with_options() 限制時長。

測試自己的程式碼

deeprespond_im.testing.FakeOpenAPI 是一個假的 OpenAPI，不需要網路：按介面名設定返回值或錯誤，檢查你的程式碼發出了什麼請求。

python

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

方法	說明
`fake.respond(op, json=..., status=200, headers=..., times=None)`	設定介面的返回值；`times` 為生效的次數，預設一直有效；後設置的先用
`fake.fail(op, code, status=..., details=..., retry_after=..., times=1)`	讓介面返回錯誤，HTTP 狀態碼按錯誤碼自動選擇，預設一次
`fake.handle(op, responder, times=None)`	用自己的函式處理請求，返回 `httpx.Response`；也可以丟擲 `httpx` 的異常模擬網路錯誤
`fake.revoke_tokens()`	讓已換取的 App Token 失效，測試令牌失效後的重新換取
`fake.requests`、`fake.calls(op)`	收到的請求：介面名、方法、路徑、查詢引數、請求頭、請求體，`.json` 是解析後的請求體
`fake.client(kwargs)`、`fake.async_client(kwargs)`	連到它的客戶端，其他引數原樣傳給客戶端（如 `max_retries=0`）

op 是介面名，見服務與方法的“介面名”一列，如 users.list_mutes 的介面名是 users.mutes；
沒有設定的介面返回 200 和 {}（沒有結果的返回 204）。響應中缺少的欄位解析為零值（空字串、0、False、None），請在 json 中給出測試要檢查的欄位；
狀態碼有含義的介面要按 REST 文件設定 status，如傳送訊息新寫入時為 201，返回 200 表示重複提交（duplicate 為真）；
使用 pytest 時，可以在 conftest.py 中啟用夾具 pytest_plugins = ["deeprespond_im.testing.pytest"]，得到 im_fake、im_client、im_async_client（後者需要 pytest-asyncio）。

測試接收回調的程式碼見本地測試。

使用獨立頻道

python

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 的服務和方法相同，呼叫時加 await。
建立的結果為 ChannelsCreateResult，查詢的結果為 RTCChannelResult，票據有各自的結果型別；欄位都有型別，並保留原始資料。
expires_at 用 UNSET、None 和值分別表示不傳、清除和設定；stats 用關鍵字引數 from_ 和 to。
頻道列表沒有 iter 方法，用 next_cursor 自己翻頁。
簽發票據不自動重試；移出成員和撤銷票據使用固定的 operation_id。
這些型別從 PyPI 1.0.0a0 起提供，對應的介面見頻道方法。

遊戲房間

遊戲房間給多人線上遊戲提供即時對戰通道。你的服務端用 client.game 建房、給玩家簽發加入票據、對局結束時結束房間；玩家加入房間和對戰由客戶端 SDK 完成。下面是一局對戰在服務端要做的事（client 同前文）：

python

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

要做什麼	方法	注意
建房	`create_room`	房間 ID 用你的對局 ID。同樣的引數重複呼叫返回同一個房間（`created` 為假），網路錯誤時自動重試
簽發加入票據	`issue_tickets`	每次 1–10 人。不自動重試：超時後直接再籤一次，同一使用者之前沒用的票據隨之作廢。票據原文只返回這一次，不要寫進日誌
手動開局	`start_room`	只用於開局方式為 `manual` 的房間
結束房間	`close_room`	結算結果原樣發給玩家和 `game_room.closed` 回撥。房間之前已在結束時 `accepted` 為假，這次的原因和結果不生效
移出玩家	`remove_member`	SDK 自動生成 `operation_id`，重試時不變。玩家不在房間裡時 `removed` 為假
查詢	`get_room`、`list_rooms` / `iter_rooms`、`list_members`、`list_regions`	列表按建立時間從新到舊；狀態可以寫多個，用逗號分隔

AsyncDRClient 的方法相同，呼叫時加 await。
結果是 GameCreateRoomResult、GameTicket 等資料類；created、accepted、removed 由 HTTP 狀態碼得出，不是響應體中的欄位。ext 和 result 傳 dict。
常見錯誤：大區暫時沒有空閒的遊戲節點為 unavailable（game_capacity_unavailable，ServiceUnavailableError），稍後重試，或在建房時給出候選大區；同時進行的房間數達到上限為 limit_exceeded（open_rooms，LimitExceededError）。全部錯誤見遊戲房間的錯誤。
擴充套件欄位 ext 不超過 2KB，結算結果 result 不超過 4KB；上限和計費見控制台、計費與限制。
對局結束後的結算回撥（game_room.closed）見遊戲房間回撥。
每個方法對應的介面見遊戲房間方法，請求和返回的欄位見遊戲房間：服務端介面。遊戲房間的方法從 PyPI 1.0.0a2 起提供。

AI 流式回覆

AI 對話中，使用者給機器人發訊息後你會收到 ai.message_received 回撥，然後用 client.ai.start_stream 開始一條回覆，邊生成邊寫入。返回的流物件替你處理增量的編號與合批（滿 20 毫秒或 4 KB 發一段）、持續寫入與按批寫入的切換、失敗重發；連續三次寫入失敗時以 fail("transport") 結束。同步客戶端返回 Stream（傳送在後臺執行緒中），非同步客戶端返回 AsyncStream（傳送在後臺任務中），方法相同，非同步版的 flush、finish、fail、close、pipe_* 要 await。機器人管理用 client.ai.create_bot 等方法，見服務與方法。

收到訊息後開始回覆

回撥處理函式應儘快返回，模型呼叫放到另一個執行緒或任務中。client_stream_id 用事件 ID，回撥重新投遞時不會開始第二條回覆：

python

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

start_stream 的 replace=True 先結束會話中正在生成的回覆；batch_only=True 只用按批寫入（執行環境不支援流式請求體時，如部分雲函式）。

對接模型

pipe_openai、pipe_openai_responses、pipe_anthropic、pipe_gemini 讀完模型的流後自動 finish（帶上結束原因、模型名和用量）；模型返回錯誤時以 fail("model_error", 说明) 結束並丟擲這個錯誤。引數可以是官方 SDK（openai、anthropic、google-genai）的流，也可以是 SSE 格式的原始響應（httpx.Response，或逐塊產出 str、bytes 的迭代器）。相容 OpenAI 格式的模型（DeepSeek、通義千問、Kimi、豆包、vLLM、Ollama 等）都用 pipe_openai：

python

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)

非同步版：

python

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)

四種轉換的規則與 Go SDK 相同。

手動寫入

python

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_* 只放入記憶體，不等待網路，同步和非同步版都不需要 await；與上一段同類型的文字追加到同一個片段。
flush() 立即發出並等待確認；finish() 先確認全部內容再結束。
模型呼叫失敗時呼叫 stream.fail("model_error", "模型暂时不可用")，已生成的內容保留，使用者看到“生成中斷”和這句說明。

停止生成

使用者點停止生成、回覆超時或被新的回覆替換時，stream.cancelled 被設定（同步版為 threading.Event，非同步版為 asyncio.Event），stream.cause 為 stream_closed 的 DRError；之後的寫入丟擲這個錯誤，pipe_* 停止讀取並關閉模型的流。

回撥處理器（CallbackHandler、AsyncCallbackHandler）收到 ai.stream_stopped 時自動停止本程序中對應的流，不需要註冊處理函式；要額外處理時用 @handler.on_ai_stream_stopped。自己處理回撥（如從佇列中取出事件）時呼叫 client.ai.stream_stopped(stream_id)。

呼叫介面 ​

應用許可權與使用者可見性 ​

命名規則 ​

不傳、傳 null 與傳值 ​

原樣傳送 JSON ​

每個方法的通用引數 ​

返回值 ​

分頁 ​

自動翻頁 ​

批次介面的結果 ​

超過上限的批次 ​

錯誤 ​

重試與冪等 ​

限流 ​

本地限速 ​

上傳檔案 ​

續傳大檔案 ​

下載檔案 ​

應用和租戶的狀態 ​

日誌與排查 ​

鉤子 ​

統計 ​

在 asyncio 中使用 ​

測試自己的程式碼 ​

使用獨立頻道 ​

遊戲房間 ​

AI 流式回覆 ​

收到訊息後開始回覆 ​

對接模型 ​

手動寫入 ​

停止生成 ​

呼叫介面

應用許可權與使用者可見性

命名規則

不傳、傳 null 與傳值

原樣傳送 JSON

每個方法的通用引數

返回值

分頁

自動翻頁

批次介面的結果

超過上限的批次

錯誤

重試與冪等

限流

本地限速

上傳檔案

續傳大檔案

下載檔案

應用和租戶的狀態

日誌與排查

鉤子

統計

在 asyncio 中使用

測試自己的程式碼

使用獨立頻道

遊戲房間

AI 流式回覆

收到訊息後開始回覆

對接模型

手動寫入

停止生成