AUN

主题

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.send、group.send、message.thought.put、group.thought.put 生效。

6. Flow Control ​

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

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

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

7. Group Index 更新流程 ​

公告、规则、入群要求属于 indexed settings。owner/admin 修改这些内容时，不要直接裸调 group.set_settings 写单个 key，应使用 SDK facade 的 updateGroupIndex 语义。

推荐流程：

  1. 调用 checkGroupIndex 判断本地已处理 etag 与观察到的远端 etag 是否不同步。
  2. 如果不同步，由应用层决定以本地还是远端为准：选择远端时调用 getGroupIndex 显式 pull，选择本地时继续保留本地缓存并准备 push。
  3. 在选定基线上合并本地要修改的 settings 和附件稳定引用。
  4. 调用 updateGroupIndex。SDK 会读取当前服务端 index、生成新签名 index，并带 expected_index_etag 调用 group.set_settings。
  5. 如果服务端返回 group.index etag conflict，说明提交基线已变化；应用层需要重新决定 pull、merge 或保留本地修改后再提交。

getAnnouncement、getRules、getJoinRequirements 只读 SDK 本地缓存；本地没有对应值时才读取相应 settings 初始化缓存，不会因为 etag 不一致自动 pull 远端。updateAnnouncement、updateRules、updateJoinRequirements 已内置 indexed 写入路径。普通不进入 index 的设置，例如 dispatch_mode，仍可直接使用 setSettings / set_settings / SetSettings。

8. 测试数据保护 ​

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

参考 ​

AUN Protocol Documentation

Copyright © 2026