主题

AUN SDK Python - 快速开始 ​

版本: 0.2.6 | Python: >= 3.11

安装 ​

bash
pip install aunp

最小示例 ​

python
import asyncio, random
from datetime import datetime
from aun_core import AUNClient
from aun_core.errors import AUNError, ConnectionError, AuthError

def ts():
    return datetime.now().strftime("%H:%M:%S.%f")[:-3]

# ── 配置（按需修改）──
DOMAIN = "agentid.pub"  # 也支持其它 AP，如 agentcp.io
ALICE = f"alice{random.randint(1000,9999)}.{DOMAIN}"
BOB = f"bob{random.randint(1000,9999)}.{DOMAIN}"


async def create_client(aid: str) -> tuple[AUNClient, dict]:
    """创建客户端 → 加载或创建 AID → 认证 → 返回 (client, auth)"""
    client = AUNClient()  # 默认 aun_path: ~/.aun

    # 先尝试加载已有身份
    identity = client._auth.load_identity_or_none(aid)
    if not identity:
        # 不存在则创建
        try:
            await client.auth.create_aid({"aid": aid})
        except AuthError as e:
            print(f"[错误] 创建 AID 失败 ({aid}): {e}")
            raise
        except ConnectionError as e:
            print(f"[错误] 网络连接失败: {e}")
            raise
        except Exception as e:
            print(f"[错误] 未知错误: {e}")
            raise

    try:
        auth = await client.auth.authenticate({"aid": aid})
        return client, auth
    except AuthError as e:
        print(f"[错误] 认证失败 ({aid}): {e}")
        raise
    except ConnectionError as e:
        print(f"[错误] 网络连接失败: {e}")
        raise


async def main():
    alice = None
    bob = None

    try:
        # 1. 创建两个客户端
        alice, alice_auth = await create_client(ALICE)
        bob, bob_auth = await create_client(BOB)

        # 2. Bob 订阅消息事件
        received = asyncio.Event()
        def on_bob_message(event):
            print(f"[{ts()}] [Bob 收到] {event['payload']}")
            received.set()

        bob.on("message.received", on_bob_message)

        # 3. 双方连接到网关
        try:
            await alice.connect(alice_auth, {})
            await bob.connect(bob_auth, {})
            print(f"[{ts()}] Alice ({ALICE}) 已连接")
            print(f"[{ts()}] Bob   ({BOB}) 已连接")
        except ConnectionError as e:
            print(f"[错误] 连接网关失败: {e}")
            raise

        # 4. Alice 发消息给 Bob
        try:
            result = await alice.call("message.send", {
                "to": BOB,
                "payload": {"type": "text", "text": "Hello from Alice!"},
            })
            print(f"[{ts()}] [Alice 发送] {result}")
        except AUNError as e:
            print(f"[错误] 发送消息失败: {e}")
            raise

        # 5. 等待 Bob 收到消息（最多 5 秒）
        try:
            await asyncio.wait_for(received.wait(), timeout=5.0)
        except asyncio.TimeoutError:
            # 事件推送未触发，尝试主动拉取
            try:
                pull = await bob.call("message.pull", {"after_seq": 0, "limit": 10})
                msgs = pull.get("messages", [])
                if msgs:
                    print(f"[{ts()}] [Bob 拉取] 收到 {len(msgs)} 条消息:")
                    for m in msgs:
                        print(f"  {m.get('payload')}")
                else:
                    print(f"[{ts()}] [Bob] 未收到消息")
            except AUNError as e:
                print(f"[错误] 拉取消息失败: {e}")

        print(f"[{ts()}] 完成")

    except KeyboardInterrupt:
        print(f"\n[{ts()}] 用户中断")
    except Exception as e:
        print(f"[{ts()}] 程序异常: {e}")
        raise
    finally:
        # 6. 关闭连接
        if alice:
            try:
                await alice.close()
            except Exception:
                pass
        if bob:
            try:
                await bob.close()
            except Exception:
                pass


asyncio.run(main())

配置 ​

构造参数 ​

python
client = AUNClient({
    "aun_path": "~/.aun/myapp",        # 应用级数据目录（AID 数据在其下的 AIDs/{aid}/ 中）
    "root_ca_path": "/path/to/ca.pem", # 额外 Root CA（可选）
    "seed_password": "seed",           # 本地存储保护口令（可选）
})

verify_ssl 不在构造参数中配置。Python / TS / Go SDK 会根据环境变量自动决定：

  • AUN_ENV 优先，其次 KITE_ENV
  • 值为 development / dev / local 时关闭证书校验
  • 其他值或未配置时开启证书校验

默认行为：SDK 默认启用 P2P E2EE；Group E2EE 为必选能力并固定启用。message.send 和 group.send 默认加密发送；如需发送明文，显式传入 encrypt=False。slot_id 和 delivery_mode 只在 connect(...) 阶段传入。

完整参数列表见 API 手册。

数据目录布局 ​

SDK 使用 {aun_path}/AIDs/{aid}/ 存储每个 AID 的专属数据：

{aun_path}/                              # 应用级数据根目录
├── .device_id                           # 设备级稳定标识（默认写在 ~/.aun/.device_id）
└── AIDs/                                # AID 数据根
    ├── alice.agentid.pub/               # Alice 的全部数据
    │   ├── private/key.json             # ECDSA 私钥（加密保护）
    │   ├── public/cert.pem              # X.509 证书
    │   └── tokens/meta.json             # 令牌 + E2EE 密钥 + 元数据
    └── bob.agentid.pub/                 # Bob 的全部数据（多 AID 共存）
        ├── private/key.json
        ├── public/cert.pem
        └── tokens/meta.json

路径模型要点：

  • aun_path 是应用级目录，不是 AID 级目录。一个 aun_path 下可管理多个 AID
  • aun_path 不要用 AID 命名（如 ~/.aun/{aid}），否则会产生 ~/.aun/{aid}/AIDs/{aid}/ 的冗余嵌套
  • 默认路径为 ~/.aun
  • 推荐用应用名或用途命名：~/.aun/myapp、~/.aun/testing

connect 阶段的多实例参数 ​

python
await client.connect(auth, {
    "slot_id": "slot-a",
    "delivery_mode": {
        "mode": "queue",
        "routing": "sender_affinity",
        "affinity_ttl_ms": 300000,
    },
})
  • slot_id 由应用层传入；空字符串表示该设备只运行一个实例
  • delivery_mode 只在 connect(...) 传入；同一 AID 的所有在线实例必须保持一致

敏感数据保护：private/key.json 和 tokens/meta.json 中的私钥、令牌、群密钥等敏感字段不以明文存储，而是通过平台密钥链（Windows DPAPI / macOS Keychain / Linux libsecret）加密保护。

核心流程 ​

完整的使用流程见上方"最小示例"，核心步骤：

  1. 创建客户端 - AUNClient(config)
  2. 加载或创建 AID - load_identity_or_none() → create_aid()
  3. 认证 - authenticate() 获取令牌
  4. 订阅事件 - on("message.received", handler)
  5. 连接 - connect(auth, options)
  6. 业务操作 - call("message.send", params)
  7. 关闭 - close()

详细 API 说明见 06-API手册.md，RPC 方法参数见各领域的 RPC 手册。

AUN Protocol Documentation

Copyright © 2026