AUN

主题

AUN SDK - 连接与认证 ​

当前公开流程 ​

重构后，注册、身份加载、会话连接分离：

python
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})

关键约束：

  • AIDStore.register(aid) 只负责生成本地密钥、申请证书并落盘。
  • AIDStore.load(aid) 返回 AID 值对象；AID 对象加载后不可变。
  • AUNClient 只接收 AID 对象，不接收字符串 AID 或配置字典里的 aid 字段。
  • connect(options) 内部会按需认证、建立 WebSocket、初始化 E2EE 和后台任务。

AIDStore ​

构造 ​

python
store = AIDStore(
    aun_path="~/.aun/myapp",
    encryption_seed="",
    device_id=None,
    slot_id="default",
    debug=True,
)
参数说明
aun_path应用级数据目录
encryption_seed本地密钥保护种子
device_id可选设备 ID；不传时从 {aun_path}/.device_id 读取或生成
slot_id默认连接槽位
verify_sslNone=自动（AUN_ENV/KITE_ENV 为 dev/local 时关闭），True/False=强制
root_ca_path私有部署时指定自定义根证书路径
debug是否输出 DEBUG 日志和文件日志

常用方法 ​

方法说明
load(aid)从本地加载 AID，返回 Result
register(aid)注册新 AID，返回 Result
list()列出本地身份
exists(aid)HEAD PKI 端点，判断 AID 是否已注册
resolve(aid, opts=None)拉取并缓存对端证书；默认同时下载并验签 agent.md，可传 skip_agent_md=True 跳过
download_agent_md(aid)下载并验签对端 agent.md
check_agent_md(aid, ttl_days=1)检查本地缓存与远端 agent.md 状态
diagnose(aid)本地和远端一致性诊断
renew_cert(aid) / rekey(aid)证书续签和换钥，成功后重新 load()

Result 使用方式：

python
result = await store.exists("alice.agentid.pub")
if result["ok"] and not result["data"]["exists"]:
    await store.register("alice.agentid.pub")

AUNClient ​

构造与身份加载 ​

python
client = AUNClient()       # no_identity
client.load_identity(me)   # standby

client = AUNClient(me)     # standby

load_identity() 只允许在 no_identity 或 closed 状态调用；传入对象必须是私钥有效的 AID。

显式认证 ​

通常不需要单独调用认证，connect() 会自动完成。需要提前获取 token 或检查认证链路时可调用：

python
auth = await client.authenticate()
print(auth["access_token"], auth["gateway"])

认证成功后状态进入 authenticated，随后调用 connect() 建立会话。

连接 ​

python
await client.connect({
    "slot_id": "main",
    "connection_kind": "long",
    "auto_reconnect": True,
    "heartbeat_interval": 30.0,
    "token_refresh_before": 60.0,
    "retry": {
        "initial_delay": 1.0,
        "max_delay": 64.0,
    },
    "timeouts": {
        "connect": 5.0,
        "call": 10.0,
        "http": 30.0,
    },
})
选项说明
slot_id同一设备内的实例槽位；允许 /、:、空格作为共享隔离键分隔符
connection_kind"long" 或 "short"
short_ttl_ms短连接服务端兜底超时
delivery_mode连接级投递语义
auto_reconnect断线后是否自动重连
heartbeat_interval心跳间隔，秒
token_refresh_beforetoken 过期前刷新提前量，秒
retry.initial_delay / retry.max_delay退避重连参数
timeouts.connect/call/http连接、RPC、HTTP 超时

当前稳定实现为 Gateway 模式；Peer / Relay 仍是协议能力定义，SDK 连接层会明确返回未实现。

长短连接共存 ​

同一 (aid, device_id, slotIsolationKey(slot_id)) 隔离槽下允许 1 条长连接 + 最多 10 条短连接。slot_id 的第一个 /、: 或空格之前的部分是共享隔离键，因此 evolclaw daemon、evolclaw cli、evolclaw/netcheck 会共享 evolclaw 隔离槽。

长连接：

python
daemon = AUNClient(me)
await daemon.connect({"connection_kind": "long", "slot_id": "main"})
daemon.on("message.received", lambda e: print(e["payload"]))

短连接：

python
cli = AUNClient(me)
await cli.connect({"connection_kind": "short", "slot_id": "main", "short_ttl_ms": 30000})
await cli.call("message.send", {"to": peer, "payload": {"type": "text", "text": "hello"}})
await cli.close()

同一个 AIDStore / aun_path 下的 token、gateway、证书和 E2EE 材料会被复用。

网关发现 ​

注册、解析和认证会根据 AID 的 issuer 自动发现 Gateway：

  • 生产模式：优先 https://{aid}/.well-known/aun-gateway，失败后回退到 https://gateway.{issuer}/.well-known/aun-gateway
  • 开发模式：优先 gateway.{issuer}，兼容没有泛域名的本地环境

verify_ssl 由 AUN_ENV / KITE_ENV 控制。development、dev、local 会关闭校验，其余情况开启。

发现成功后，SDK 会基于 WebSocket URL 推导 /health 地址，并通过 gateway_health 读取最近健康检查结果。

状态与事件 ​

python
print(client.state)
print(client.can_connect, client.can_send, client.is_online)

常用事件：

事件说明
state_change状态变化，payload 中的 state 是九态公开值
connection.error连接、认证、重连错误
token.refreshedtoken 自动刷新完成
message.received收到消息
group.changed群组事件
message.undecryptable / group.message_undecryptableE2EE 解密失败

建议在 connect() 前注册事件处理器。

Agent Web / agent.md ​

发布自己的 agent.md：

python
uploaded = await store.upload_agent_md("alice.agentid.pub", content)

aid 必须是本地 AID。上传由 AIDStore 读取本地私钥签名，并通过 AuthFlow 获取或复用该 AID 的 access_token。

解析对端并验签 agent.md：

python
resolved = await store.resolve("bob.agentid.pub")
if resolved["ok"]:
    print(resolved["data"]["agent_md"]["verification"]["status"])

只下载对端 agent.md：

python
downloaded = await store.download_agent_md("bob.agentid.pub")

只检查本地缓存与远端状态：

python
state = await store.check_agent_md("bob.agentid.pub", ttl_days=1)

当前公开入口分工固定：

主体公开能力
AIDStoreupload_agent_md(aid, content=None)、download_agent_md(aid)、check_agent_md(aid, ttl_days=1)、resolve(aid, opts)
AIDsign_agent_md(content)、verify_agent_md(content) 低层签验能力
AUNClient连接、事件和 RPC 调用；不再暴露 agent.md 上传入口

本地落盘位置由 SDK 管理。Python / TypeScript / Go 写入 {aun_path}/AIDs/{aid}/agent.md 和同目录 agentmd.json；浏览器 JavaScript 写入 IndexedDB 的等价 logical key，存储不可用时退化为内存缓存。agent.md 不写入 SQLite，也不再使用旧 {aun_path}/AgentMDs 目录。

RPC 调用 ​

认证连接后，所有业务操作通过 client.call()：

python
result = await client.call("message.send", {
    "to": "bob.agentid.pub",
    "payload": {"type": "text", "text": "Hello!"},
})

RPC 方法完整参数和响应格式见：

领域手册
消息09-message-rpc-manual.md
群组09-group-rpc-manual.md
存储09-storage-rpc-manual.md
元信息09-meta-rpc-manual.md

AUN Protocol Documentation

Copyright © 2026