deeprespond-im-server-sdk

深应智能（DeepRespond）IM 的服务端 SDK（Python 版）：在业务服务端调用 OpenAPI（管理用户、好友、群组、消息、文件、聊天室、内容安全等，签发登录凭证），并接收回调（校验签名、防重放、按事件 ID 去重、分派事件回调和同步回调）。

• Python 3.11 以上；直接依赖只有 httpx。
• 同步的 DRClient 和异步的 AsyncDRClient，服务、方法和行为完全相同。
• 不要在浏览器和 App 中使用：Client Secret 和 App Token 等同应用管理员的密码。

用户权限与消息可见性

OpenAPI 以应用管理员身份运行，SDK 不会根据业务系统的当前登录用户自动做授权。 将 SDK 接到浏览器或 App 的业务接口时，业务后端必须验证用户身份，并检查会话参与者、群成员及管理角色。 messages.recall 是不受用户角色和撤回时间窗口限制的管理撤回，不能直接暴露成用户的撤回接口。

conversations.list_messages / iter_messages 是服务端全量视角，不应用用户删除、清空记录、入退群时间的过滤。 需要展示用户自己的聊天记录时，应由客户端 SDK 或已登录设备的 User Token 调用客户端消息接口； 仅在全量结果上检查会话 ID 或 cleared_seq，无法恢复逐用户删除的完整语义。 本仓库的 Web Demo 展示了如何组合服务端管理 SDK 与保存在后端的设备身份。

安装

pip install deeprespond-im-server-sdk==1.0.0a0

当前为 Alpha 预发布版本。1.0.0.alpha 按 Python 版本规则规范为 1.0.0a0。 Demo 源码单独保存在仓库中，不包含在 SDK 安装包内。

调用 OpenAPI

import os
from deeprespond_im import DRClient, NotFoundError

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

# 用户登录业务系统后，为他签发登录凭证，交给 App 用客户端 SDK 登录
ticket = client.users.issue_login_ticket("zhangsan", auto_create=True)

# 以业务的通知账号发单聊；业务单号作为 client_msg_id，重试也不会重复发送
result = client.messages.send(
    from_="notice", to_user="zhangsan", client_msg_id="order-1001-paid",
    type="custom", body={"card": "order", "order_id": "1001"},
)

for user in client.users.iter(status="active"):   # 自动翻页
    ...

client.close()   # 或 with DRClient(...) as client:

异步：方法相同，加 await，遍历用 async for：

from deeprespond_im import AsyncDRClient

async with AsyncDRClient(base_url=..., org_name=..., app_name=..., client_id=..., client_secret=...) as client:
    user = await client.users.get("zhangsan")

• 客户端在进程中复用：它持有连接池和 App Token；App Token 由 SDK 换取、缓存、提前续期，失效后自动重新换取。多进程（gunicorn、Celery）的每个工作进程各建一个。
• 重试：查询、带幂等键的写操作（发消息）、本身幂等的写操作在网络错误、超时、5xx 时自动重试；建群、设置密码等不幂等的写操作发出后不重试。被限流时按 Retry-After 等待后重发：应用额度用完时整个客户端一起等待，某个群、聊天室被限流时只有这个请求等待。
• 三种状态的参数：修改类方法的参数默认为 UNSET（不发送），None 发送 null：client.users.update("zhangsan", attributes={"level": None}) 删除一项属性，attributes=None 清空全部。
• 原样发送 JSON：body=RawJSON('{"price":1.00}')；响应中消息的 body_raw 是原文，逐字节相同。
• 单次调用的选项：每个方法都有 timeout、deadline（总时限）、request_id（追踪 ID）；client.with_options(max_retries=0, timeout=3) 返回共用令牌和连接池的视图。

错误

from deeprespond_im import DRError, RateLimitError, ServiceUnavailableError

try:
    client.users.create("zhangsan")
except DRError as e:
    print(e.code, e.reason, e.request_id)   # 按 code 判断，不解析 message

子类：AuthenticationError、PermissionDeniedError、NotFoundError、ConflictError、LimitExceededError、RateLimitError、ServiceUnavailableError、InvalidRequestError、ContentRejectedError、ServerError、NetworkError（同时是 ConnectionError）、DRTimeoutError（同时是 TimeoutError）、InvalidResponseError、ClientUnavailableError。自动重试后得到 already_exists、not_found、version_conflict 时，e.retried 为真：前一次尝试可能已经执行。

文件

file = client.media.upload("/data/a.jpg", purpose="attachment", kind="image")
client.messages.send(from_="notice", to_user="zhangsan", type="image", body={"url": file.url})
client.media.download(file.url, "/tmp/a.jpg")

不超过 32 MB 的一次上传，更大的分片并发上传；中断后可以用 resume_upload(file_id, source) 续传。

media.download 将字节写入目标，不会替 Web 框架设置下载响应头。代理用户上传的附件时， HTML、SVG 等主动内容应通过 Content-Disposition: attachment 下载，或放在隔离的域名； 不要用浏览器传入的 MIME 类型决定同源响应类型。Demo 同时设置 nosniff 和沙箱 CSP，并只内联识别出的光栅图片。

接收回调

from deeprespond_im.callback import CallbackHandler, MemoryNonceStore, MemoryDedupStore, Allow, Reject

handler = CallbackHandler(
    secrets={"1840200000000000007": ["cbs_新密钥…", "cbs_旧密钥…"]},   # 地址 ID → 密钥
    nonce_store=MemoryNonceStore(),        # 多进程、多台服务器部署时用 deeprespond_im.redis 中的实现
    dedup_store=MemoryDedupStore(),
)

@handler.on_event("message.sent")
def archive(event, data):
    save(data.message_id, data.body_raw)   # body_raw 与请求体中的逐字节相同

@handler.on_hook("message.before_send")
def check(request):
    if "广告" in (request.body or {}).get("text", ""):
        return Reject("ad", "消息中有不允许的内容")
    return Allow()

挂到框架上（适配器取原始请求体，签名按原始字节计算）：

# Flask
from deeprespond_im.callback.flask import callback_view
app.add_url_rule("/im/callback", view_func=callback_view(handler), methods=["POST"])

# FastAPI / Starlette（异步处理器 AsyncCallbackHandler）
from deeprespond_im.callback.fastapi import callback_endpoint
app.add_api_route("/im/callback", callback_endpoint(async_handler), methods=["POST"], include_in_schema=False)

# Django
from deeprespond_im.callback.django import callback_view
urlpatterns = [path("im/callback", callback_view(handler))]

• 重复的事件（同一个事件 ID）不再调用处理函数，返回 2xx；一批事件中有处理失败的整批返回 503，服务端稍后重发。
• 同步回调的处理函数在 timeout_ms 减 50 毫秒内没有返回时，返回 503（或 hook_fallback 给出的结果）。

测试自己的代码

from deeprespond_im.testing import FakeOpenAPI, signed_request, event_body

fake = FakeOpenAPI()
fake.fail("messages.send", "rate_limited", retry_after=1)
client = fake.client()                  # 连到假的 OpenAPI 的 DRClient
...
assert fake.calls("messages.send")[0].json["to_user"] == "zhangsan"

文档

完整的说明见仓库中的 Python 文档（行为规则另见 Go 文档）。

独立 RTC 频道管理已提供生成的强类型 Channels/channels API：创建、修改、关闭、票据、成员管理、统计与更正列表。Go 使用 RTCChannel*Params/Result，Node.js/Python 返回具体模型；Python 支持同步/异步。期限保留未设置/null/值三态，票据签发不自动重试。旧 Go 原始 JSON 接口需迁移到新参数/结果；管理 API 不能代替用户登录获取媒体 Token。完整契约及计费规则见仓库 docs/RTC频道/4.实现与部署.md。