AUN SDK Python - 最佳实践

---

1. 幂等的连接初始化

每次启动时安全地确保已认证并连接，无论是首次还是重复运行：

async def ensure_connected(client: AUNClient, aid: str) -> str:
    try:
        await client.auth.create_aid({"aid": aid})
    except Exception as e:
        print(f"创建 AID 失败: {e}")
        raise
    auth = await client.auth.authenticate({"aid": aid})
    await client.connect(auth, {"auto_reconnect": True})
    return aid

2. 安全关闭多个客户端

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

3. E2EE 幂等运行

每次运行前清除旧 prekey 缓存并跳过历史消息，避免计数器冲突：

# 清除旧 prekey 缓存
sender.e2ee.invalidate_prekey_cache(peer_aid=receiver_aid)
receiver.e2ee.invalidate_prekey_cache(peer_aid=sender_aid)

# 跳过旧消息（获取当前最大 seq 作为 cursor 起点）
r = await receiver.call("message.pull", {"after_seq": 0, "limit": 1})
recv_cursor = r.get("latest_seq", 0)

4. 多 AID 管理

一个 aun_path 可管理多个 AID，每个 AID 数据隔离在 {aun_path}/AIDs/{aid}/ 下：

# 推荐：用应用名作为 aun_path，多个 AID 共存于同一目录
client = AUNClient({"aun_path": "~/.aun/myapp"})

# 创建不同的 AID，各自数据自动隔离
await client.auth.create_aid({"aid": "alice.agentid.pub"})
await client.auth.create_aid({"aid": "bob.agentid.pub"})
# 数据分别在 ~/.aun/myapp/AIDs/alice.agentid.pub/ 和 bob.agentid.pub/ 下

注意：不要用 AID 名称作为 aun_path（如 ~/.aun/{aid}），否则会产生冗余嵌套。

5. 环境变量驱动配置

import os
from pathlib import Path

DATA_ROOT = os.environ.get("AUN_DATA_ROOT", str(Path.home() / ".aun"))

6. 资源清理

async def main():
    client = AUNClient()
    try:
        await ensure_connected(client, aid)
        # ... 业务逻辑
    finally:
        await client.close()

参考

• 协议文档：../src/aun_core/docs/protocol/
• SDK 架构文档：../src/aun_core/docs/skill/sdk-core/
• RPC 方法手册：../src/aun_core/docs/skill/rpc-manual/
• 可运行示例：../src/aun_core/docs/skill/examples/
• GitHub：https://github.com/ModelUnion/aun-sdk-core