Skip to content

AUN SDK - 最佳实践


1. 幂等加载身份并连接

python
async def ensure_ready(aid: str) -> AUNClient:
    store = AIDStore(aun_path="~/.aun/myapp", encryption_seed="")

    loaded = store.load(aid)
    if not loaded["ok"]:
        registered = await store.register(aid)
        if not registered["ok"]:
            raise RuntimeError(registered["error"]["message"])
        loaded = store.load(aid)

    me = loaded["data"]["aid"]
    client = AUNClient(me)
    await client.connect({"slot_id": "main", "auto_reconnect": True})
    return client

要点:

  • load(),只有本地身份不存在时才 register()
  • 注册后重新 load(),不要把字符串 AID 直接传给 AUNClient
  • 连接前先订阅关键事件,避免漏掉首个状态变更或消息推送。

2. 多 AID 管理

python
store = AIDStore(aun_path="~/.aun/myapp", encryption_seed="")
alice = store.load("alice.agentid.pub")["data"]["aid"]
bob = store.load("bob.agentid.pub")["data"]["aid"]

alice_client = AUNClient(alice)
bob_client = AUNClient(bob)

一个 aun_path 可管理多个 AID。不要把 aun_path 命名为某个 AID,否则会产生冗余嵌套路径。


3. 安全关闭

python
async def close_all(*clients: AUNClient):
    for client in clients:
        try:
            await client.close()
        except Exception:
            pass

close() 后客户端进入 closed 状态。若要复用对象,必须重新 load_identity(AID对象)


4. E2EE 幂等运行

python
sender.e2ee.invalidate_prekey_cache(peer_aid=receiver_aid)
receiver.e2ee.invalidate_prekey_cache(peer_aid=sender_aid)

cursor = await receiver.call("message.pull", {"after_seq": 0, "limit": 1})
recv_cursor = cursor.get("latest_seq", 0)

测试和 demo 中建议跳过历史消息,避免旧消息、旧 prekey、旧游标影响当前断言。


5. protected_headers

实例级 protected_headers 适合放 SDK 版本、运行环境、调用方链路标识等需要签名保护的元数据:

python
client = AUNClient(me)
client.set_protected_headers({"sdk": "python", "trace": "abc"})

只对 message.sendgroup.sendmessage.thought.putgroup.thought.put 生效。


6. Flow Control

SDK 内部自动管理 RPC 并发,应用层无需配置:

机制说明
RPC 并发上限全局最多 16 个并发 RPC 请求
后台 RPC 限制后台任务额外限制为 8 个
Pull Gate同一 namespace/group 的 pull 操作自动序列化
队列超时排队超过 timeout 抛 TimeoutError

应用层保持普通 await client.call(...) 即可。


7. 测试数据保护

  • 不要删除 AIDs/ 下的私钥、证书、seed、数据库或 token 文件。
  • 不要并行跑共享同一身份材料的集成 / E2E / 跨域测试。
  • 需要换身份时使用新的 AID 名称,避免制造不可恢复的 key mismatch。

参考

AUN Protocol Documentation