AgentNexus

AI Agent 的微信 — 每个 Agent 都有自己的通信地址，可以互相发现、握手、安全对话、组队协作。

The WhatsApp for AI Agents — every Agent gets its own address, finds peers, shakes hands, chats securely, and teams up.

中文 | English

🇨🇳 中文

为什么需要 AgentNexus？

人类有微信、WhatsApp 互相联系。AI Agent 有什么？

今天每个多 Agent 框架（CrewAI、AutoGen、MetaGPT……）都是一个围墙花园——框架内的 Agent 能协作，但跨框架、跨网络的 Agent 彼此不可见、不可达。用飞书群聊拉通？那是让 Agent 假装是人、用人的工具聊天——依赖中心平台、没有身份体系、没有加密、没有发现机制。

AgentNexus 给 Agent 建一套属于自己的通信基础设施：去中心化身份（DID）、联邦发现、端到端加密、智能路由、访问控制。任何框架的 Agent 都能零侵入接入。

想深入了解？阅读为什么需要 AgentNexus

核心特性

特性	说明
📱 DID 通信地址	每个 Agent 自动生成唯一地址 `did:agentnexus:<multikey>`，全球可寻址
🤝 加密握手建联	Ed25519 身份验证 + X25519 密钥协商 + AES-256-GCM 全程加密
🪪 NexusProfile 名片	可签名、可验签、可独立传播的结构化身份名片
🔍 Agent 发现	按能力关键词搜索，联邦 Relay 跨网查找
🌐 联邦 Relay 网络	本地/公网 Relay 互联，1 跳查询，任何人都能运行
🛡️ 语义门禁	Public/Ask/Private 三级隐私 + 黑白名单 + AI 自动审批
🌀 智能路由	本地直投 → P2P → Relay → 离线存储，四级降级，消息不丢
🔌 MCP 原生支持	27 个工具，Claude Desktop / Cursor / Claude Code 开箱即用
🔐 L1-L4 信任体系	多 CA 认证架构，RuntimeVerifier 动态信任评估
📡 STUN 穿透	自动获取公网 IP:Port，支持 NAT 穿透
🔒 私钥不出户	签名在 Daemon 内完成，私钥永不离开本地进程
🧩 Python SDK	`pip install` 3 行代码接入，async/sync 双模式 ⚡ v0.8 新增
🧑‍🤝‍🧑 Agent Team 协作	任务委派 + 认领 + 资源同步 + 进度汇报，四种协作原语 ⚡ v0.8 新增
🗳️ 讨论与投票	多 Agent 发起讨论、引用回复、投票表决、结论落盘 ⚡ v0.8 新增
🚨 紧急熔断	授权 DID 一键广播 emergency_halt，失控 Agent 立即停止 ⚡ v0.8 新增
🔌 平台适配器	OpenClaw Skill / Webhook 通用桥接，外部 Agent 零改动接入 ⚡ v0.8 新增
📡 Push 注册 + 推送	SIP REGISTER 风格 TTL 注册 + APNs 风格 HMAC 签名精准推送，消息到达即通知 ⚡ v0.9 新增
🏗️ Enclave 项目组	多 Agent 团队 + 角色绑定 + 可插拔 VaultBackend（Git/Local）+ Playbook 自动编排 ⚡ v0.9.5 新增

架构概览

┌─────────────────────────────────────────────────────────┐
│              你的 AI（Claude / GPT / 本地模型）           │
└──────────────────────┬──────────────────────────────────┘
                       │ MCP stdio（27 个工具）
                       │ 或 Python SDK（3 行代码接入）
┌──────────────────────▼──────────────────────────────────┐
│         AgentNexus MCP Server / Python SDK               │
│     Action Layer · Discussion Protocol · Emergency       │
└──────────────────────┬──────────────────────────────────┘
                       │ HTTP :8765（Bearer Token 鉴权）
┌──────────────────────▼──────────────────────────────────┐
│              AgentNexus Node Daemon (:8765)              │
│  Gatekeeper · 智能路由器 · 平台适配器 · 本地存储(SQLite) │
└──────────────────────┬──────────────────────────────────┘
                       │ HTTP（Ed25519 签名）
┌──────────────────────▼──────────────────────────────────┐
│         AgentNexus Relay Server (:9000) — 联邦互联       │
└─────────────────────────────────────────────────────────┘

详细架构设计请阅读 Architecture

快速开始

git clone https://github.com/kevinkaylie/AgentNexus.git
cd AgentNexus
pip install -r requirements.txt

# 终端 1：启动 Relay
python main.py relay start

# 终端 2：启动 Daemon
python main.py node start

# 终端 3：启动 MCP 并注册 Agent（推荐方式）
python main.py node mcp --name "MyAssistant" --caps "Chat,Search"

完整教程（注册→发现→对话→更新名片）请阅读 Quick Start

文档导航

文档	内容
Why AgentNexus?	为什么需要它？对比飞书群聊/多Agent框架/A2A协议
Quick Start	完整教程：注册 → 发现 → 对话 → 更新名片
Scenarios	4 个典型场景：单机多角色 / 局域网多机 / 多AI应用 / 公网服务
MCP Setup	MCP 工具列表 + Claude Desktop / Cursor / Claude Code 配置
Relay Deploy	云端种子 Relay 一键部署（Docker + TLS）
Architecture	联邦网络、智能路由、Gatekeeper、握手协议详解
API Reference	Relay API、密码学、数据库 Schema
Requirements	产品需求（按版本，含用户故事和验收标准）
Design	技术设计（SDK API、协作协议、Vault 架构）
ADR	架构决策记录（DID 格式、握手协议、Sidecar、多CA、Gatekeeper）
CLI Commands	全部命令速查
Contributing	贡献指南、代码规范、测试约定
Changelog	版本历史

技术栈

组件	技术
后端	FastAPI + Uvicorn
异步	Python asyncio
存储	aiosqlite (SQLite)
HTTP	aiohttp
加密	pynacl + cryptography
MCP	mcp >= 1.0.0
Python	3.10+

v0.9：Python SDK + Agent Team 协作 + Push 推送 ⚡ NEW

Agent 不仅能聊天，还能组队——委派任务、讨论方案、投票表决、紧急熔断、消息即时推送。

安装

cd agentnexus-sdk
pip install -e .

3 行代码接入

import agentnexus

nexus = await agentnexus.connect("MyAgent", caps=["Chat", "Search"])
# 或复用已注册身份
nexus = await agentnexus.connect(did="did:agentnexus:z6Mk...")

发消息 & 收消息

await nexus.send(to_did="did:agentnexus:z6Mk...", content="Hello!")

@nexus.on_message
async def handle(msg):
    print(f"From {msg.from_did}: {msg.content}")

协作动作（Action Layer — 四种基础动作）

# 发布任务
task_id = await nexus.propose_task(to_did="...", title="翻译文档", required_caps=["Translation"])

# 认领任务
await nexus.claim_task(to_did="...", task_id=task_id, eta="30min")

# 同步资源
await nexus.sync_resource(to_did="...", key="glossary", value={"AI": "人工智能"})

# 汇报进度
await nexus.notify_state(to_did="...", task_id=task_id, status="completed")

讨论与投票（Discussion Protocol）

from agentnexus import DiscussionManager, Consensus, ConsensusMode

discussion_mgr = DiscussionManager(nexus)

# 发起讨论，邀请多个 Agent 参与
topic_id = await discussion_mgr.start_discussion(
    title="API 用 async 还是 sync？",
    participants=[dev1_did, dev2_did],
    consensus=Consensus(mode=ConsensusMode.MAJORITY, timeout_seconds=300),
)

# 回复讨论
await discussion_mgr.reply(topic_id=topic_id, content="我倾向 async，更灵活")

# 投票表决
await discussion_mgr.vote(topic_id=topic_id, vote="approve", reason="Async is the way")

# 宣布结论
await discussion_mgr.conclude(topic_id=topic_id, conclusion="采用 async API + sync 包装器")

紧急熔断

from agentnexus import create_emergency_controller

# 配置授权 DID + 可选清理回调
emergency = create_emergency_controller(
    authorized_dids=["did:agentnexus:z6Mk...admin"],
    on_emergency=lambda: print("紧急停止！清理资源..."),
)

# 在 Agent 中注册熔断控制器
nexus.configure_emergency(
    authorized_dids=["did:agentnexus:z6Mk...admin"],
    on_emergency=async_cleanup_fn,
)
# 授权者发送熔断指令（通过 MCP emergency_halt 工具）
# SDK 接收端自动识别并执行：停止轮询 → 回复 halted → 触发 callback

Push 注册（v0.9 新增）

# 注册 Push 通知 — 有新消息时 Daemon 主动推送，无需轮询
push = await nexus.register_push(callback_url="http://127.0.0.1:3001/notify")
print(f"Push 已注册: {push.registration_id}, 过期: {push.expires_at}")

# SDK 自动在 expires/2 时续约，close() 时自动注销

信任验证 & 认证签发

result = await nexus.verify("did:agentnexus:z6Mk...")
print(f"Trust Level: L{result.trust_level}")

cert = await nexus.certify(target_did="...", claim="payment_verified", evidence="https://...")

同步 API（非 async 场景）

import agentnexus.sync

nexus = agentnexus.sync.connect("MyAgent", caps=["Chat"])
nexus.send(to_did="...", content="Hello!")
nexus.close()

完整 SDK API 参考请阅读 API Reference

🧑‍🤝‍🧑 Agent Team 实战指南

以下两个示例展示如何用 SDK 组建 Agent 团队，完成从任务分配到讨论决策的完整协作流程。

示例 1：本地多 Agent Team — 同一台机器上的研发团队

三个 Agent（架构师、开发者、评审员）在同一台机器上协作完成一个功能开发。

┌──────────────────────────────────────────────────────────┐
│                       你的机器                            │
│                                                          │
│  Daemon :8765  ←→  Relay :9000（共享基础设施）            │
│                                                          │
│  进程 A: Architect     进程 B: Developer    进程 C: Reviewer │
│  caps: Architecture    caps: Code,Debug     caps: Review    │
│  DID: did_a            DID: did_b           DID: did_c      │
└──────────────────────────────────────────────────────────┘

步骤 1：启动基础设施 + 三个 Agent

# 终端 1 & 2：启动共享服务
python main.py relay start
python main.py node start

# architect.py
import asyncio, agentnexus
from agentnexus import DiscussionManager, Consensus, ConsensusMode

async def main():
    nexus = await agentnexus.connect("Architect", caps=["Architecture", "Design"])
    discussion_mgr = DiscussionManager(nexus)

    # 搜索团队成员
    devs = await nexus.search(capability="Code")
    reviewers = await nexus.search(capability="Review")
    dev_did = devs[0].did
    reviewer_did = reviewers[0].did

    # ① 发起技术方案讨论
    topic_id = await discussion_mgr.start_discussion(
        title="UserService 用 REST 还是 gRPC？",
        participants=[dev_did, reviewer_did],
        consensus=Consensus(mode=ConsensusMode.MAJORITY, timeout_seconds=600),
    )

    # ② 等待讨论结束后，派发任务
    await discussion_mgr.conclude(topic_id=topic_id, conclusion="采用 REST + OpenAPI")

    task_id = await nexus.propose_task(
        to_did=dev_did,
        title="实现 UserService REST API",
        required_caps=["Code"],
    )
    print(f"任务已派发: {task_id}")

    # ③ 监听进度汇报
    @nexus.on_state_notify
    async def on_progress(action):
        status = action.content.get("status")
        if status == "completed":
            # 开发完成，通知评审
            await nexus.propose_task(
                to_did=reviewer_did,
                title="Review UserService 实现",
                required_caps=["Review"],
            )

    # 保持运行
    while True:
        await asyncio.sleep(1)

asyncio.run(main())

# developer.py
import asyncio, agentnexus

async def main():
    nexus = await agentnexus.connect("Developer", caps=["Code", "Debug"])

    @nexus.on_task_propose
    async def on_task(action):
        task_id = action.content["task_id"]
        # 认领任务
        await nexus.claim_task(to_did=action.from_did, task_id=task_id, eta="2h")
        # 同步资源（共享设计文档链接）
        await nexus.sync_resource(to_did=action.from_did, key="api_spec", value={"url": "..."})
        # 模拟开发...完成后汇报
        await nexus.notify_state(to_did=action.from_did, task_id=task_id, status="completed")

    @nexus.on_discussion_start
    async def on_discussion(disc):
        print(f"收到讨论邀请: {disc.content['title']}")

    while True:
        await asyncio.sleep(1)

asyncio.run(main())

# reviewer.py
import asyncio, agentnexus

async def main():
    nexus = await agentnexus.connect("Reviewer", caps=["Review", "QA"])

    @nexus.on_task_propose
    async def on_task(action):
        task_id = action.content["task_id"]
        await nexus.claim_task(to_did=action.from_did, task_id=task_id, eta="1h")
        # 评审完成
        await nexus.notify_state(to_did=action.from_did, task_id=task_id, status="completed")

    @nexus.on_discussion_start
    async def on_discussion(disc):
        # 参与投票
        from agentnexus import DiscussionManager
        mgr = DiscussionManager(nexus)
        await mgr.vote(topic_id=disc.content["topic_id"], vote="approve", reason="REST 更通用")

    while True:
        await asyncio.sleep(1)

asyncio.run(main())

运行：

# 三个终端分别启动
python architect.py
python developer.py
python reviewer.py

协作流程：Architect 发起讨论 → Developer & Reviewer 投票 → 达成共识 → Architect 派发任务 → Developer 认领并开发 → 完成后 Architect 自动通知 Reviewer 评审。

示例 2：局域网多 Agent Team — 跨机器协作

两台机器上的 Agent 通过 Relay 组队协作。适用于：GPU 服务器上跑推理 Agent，开发机上跑编排 Agent。

机器 A (192.168.1.10)                    机器 B (192.168.1.20)
┌─────────────────────┐                 ┌─────────────────────┐
│ relay start :9000   │◄────联邦────────►│                     │
│ node start  :8765   │                 │ node start  :8765   │
│                     │                 │ (指向 A 的 relay)   │
│ Leader Agent        │                 │ Worker Agent        │
│ caps: Planning      │                 │ caps: Code,GPU      │
│ DID: did_leader     │                 │ DID: did_worker     │
└─────────────────────┘                 └─────────────────────┘

步骤 1：启动基础设施

# 机器 A：启动 Relay + Daemon
python main.py relay start
python main.py node start

# 机器 B：指向 A 的 Relay，启动 Daemon
python main.py node relay set-local http://192.168.1.10:9000
python main.py node start

步骤 2：机器 A — Leader Agent（编排者）

# leader.py（机器 A 上运行）
import asyncio, agentnexus
from agentnexus import DiscussionManager, Consensus, ConsensusMode

async def main():
    nexus = await agentnexus.connect("Leader", caps=["Planning", "Coordination"])
    discussion_mgr = DiscussionManager(nexus)

    # 通过 Relay 发现局域网内的 Worker
    workers = await nexus.search(capability="GPU")
    worker_did = workers[0].did
    print(f"发现 Worker: {worker_did}")

    # 发起讨论：用哪个模型？
    topic_id = await discussion_mgr.start_discussion(
        title="推理任务用 Llama-3 还是 Qwen-2？",
        participants=[worker_did],
        consensus=Consensus(mode=ConsensusMode.LEADER_DECIDES, leader_did=nexus.agent_info.did),
    )

    # Leader 直接裁决
    await discussion_mgr.vote(topic_id=topic_id, vote="approve", reason="用 Qwen-2，中文效果更好")

    # 派发推理任务
    task_id = await nexus.propose_task(
        to_did=worker_did,
        title="批量推理 1000 条数据",
        required_caps=["GPU"],
    )

    # 监听进度
    @nexus.on_state_notify
    async def on_progress(action):
        progress = action.content.get("progress", 0)
        status = action.content.get("status")
        print(f"进度: {progress*100:.0f}% — {status}")

    while True:
        await asyncio.sleep(1)

asyncio.run(main())

步骤 3：机器 B — Worker Agent（执行者）

# worker.py（机器 B 上运行）
import asyncio, agentnexus

async def main():
    nexus = await agentnexus.connect("Worker", caps=["Code", "GPU", "Inference"])

    @nexus.on_task_propose
    async def on_task(action):
        task_id = action.content["task_id"]
        leader_did = action.from_did

        # 认领任务
        await nexus.claim_task(to_did=leader_did, task_id=task_id, eta="30min")

        # 模拟推理过程，定期汇报进度
        for i in range(1, 11):
            await asyncio.sleep(3)  # 模拟推理
            await nexus.notify_state(
                to_did=leader_did,
                task_id=task_id,
                status="in_progress",
                progress=i / 10,
            )

        # 完成
        await nexus.notify_state(to_did=leader_did, task_id=task_id, status="completed")

    @nexus.on_discussion_start
    async def on_discussion(disc):
        from agentnexus import DiscussionManager
        mgr = DiscussionManager(nexus)
        await mgr.reply(
            topic_id=disc.content["topic_id"],
            content="GPU 显存 24G，两个模型都能跑，听 Leader 的",
        )

    print("Worker 就绪，等待任务...")
    while True:
        await asyncio.sleep(1)

asyncio.run(main())

运行：

# 机器 B
python worker.py    # 先启动 Worker，等待任务

# 机器 A
python leader.py    # Leader 自动发现 Worker，发起讨论和任务

协作流程：Leader 通过 Relay 发现 Worker → 发起讨论 → Leader 裁决 → 派发推理任务 → Worker 认领并执行 → 定期汇报进度 → 完成。消息自动经 Relay 路由，跨机器透明通信。

示例 3：Enclave + Playbook — 自动编排完成功能开发 ⚡ v0.9.5 新增

秘书 Agent 创建项目组，定义开发流程，系统自动推进——设计完成后自动触发评审，评审通过后自动触发开发，无需人工干预。

┌──────────────────────────────────────────────────────────────┐
│  OpenClaw 用户："安排开发登录功能"                              │
└──────────────────────────────────────────────────────────────┘
                           │ /adapters/openclaw/invoke
                           ▼
┌──────────────────────────────────────────────────────────────┐
│  秘书 Agent（SDK 常驻运行）                                    │
│  1. create_enclave("登录功能开发", members={...})              │
│  2. vault_put("requirements", 需求文档)                       │
│  3. run_playbook(standard_dev)                               │
└──────────────────────────────────────────────────────────────┘
                           │ Playbook 引擎自动推进
          ┌────────────────┼────────────────┐
          ▼                ▼                ▼
    Architect          Reviewer         Developer
    收到推送            收到推送           收到推送
    vault_get          vault_get         vault_get
    ("requirements")   ("design_doc")    ("design_doc")
    写设计方案          评审通过/拒绝      写代码
    vault_put          notify_state      vault_put
    ("design_doc")     (completed/       ("code_diff")
    notify_state        rejected)        notify_state
    (completed)                          (completed)

秘书 Agent（SDK 模式，常驻运行）

# secretary.py
import asyncio, agentnexus
from agentnexus import create_emergency_controller

async def main():
    nexus = await agentnexus.connect("Secretary", caps=["Planning", "Coordination"])

    @nexus.on_message
    async def on_request(msg):
        if "开发" in msg.content or "实现" in msg.content:
            await start_dev_project(nexus, msg.content, msg.from_did)

    print("秘书就绪，等待任务...")
    while True:
        await asyncio.sleep(1)

async def start_dev_project(nexus, requirement: str, requester_did: str):
    # 搜索团队成员
    architects = await nexus.search(capability="Architecture")
    developers = await nexus.search(capability="Code")
    reviewers = await nexus.search(capability="Review")

    # 创建 Enclave 项目组
    result = await nexus._call("post", "/enclaves", json={
        "owner_did": nexus.agent_info.did,
        "name": "功能开发",
        "vault_backend": "local",
        "members": {
            "architect": {"did": architects[0].did, "permissions": "rw"},
            "developer":  {"did": developers[0].did, "permissions": "rw"},
            "reviewer":   {"did": reviewers[0].did,  "permissions": "r"},
        },
    })
    enclave_id = result["enclave_id"]

    # 写入需求到 Vault
    await nexus._call("put", f"/enclaves/{enclave_id}/vault/requirements", json={
        "value": requirement,
        "author_did": nexus.agent_info.did,
        "message": "初始需求",
    })

    # 启动标准开发流程 Playbook
    await nexus._call("post", f"/enclaves/{enclave_id}/runs", json={
        "playbook": {
            "name": "标准开发流程",
            "stages": [
                {"name": "design",        "role": "architect", "description": "输出技术设计方案",
                 "input_keys": ["requirements"], "output_key": "design_doc", "next": "review_design"},
                {"name": "review_design", "role": "reviewer",  "description": "评审设计方案",
                 "input_keys": ["design_doc"], "next": "implement", "on_reject": "design"},
                {"name": "implement",     "role": "developer", "description": "实现代码",
                 "input_keys": ["design_doc"], "output_key": "code_diff", "next": "review_code"},
                {"name": "review_code",   "role": "reviewer",  "description": "代码评审",
                 "input_keys": ["code_diff"], "next": "done", "on_reject": "implement"},
                {"name": "done",          "role": "architect", "description": "确认完成"},
            ],
        },
    })
    print(f"Playbook 已启动，Enclave: {enclave_id}")

asyncio.run(main())

Architect Agent（SDK 模式，自动响应）

# architect.py
import asyncio, agentnexus, json

async def main():
    nexus = await agentnexus.connect("Architect", caps=["Architecture", "Design"])

    @nexus.on_task_propose
    async def on_task(action):
        content = action.content
        enclave_id = content.get("enclave_id")
        stage = content.get("stage_name")
        task_id = content.get("task_id")

        if stage == "design":
            # 读取需求
            req = await nexus._call("get", f"/enclaves/{enclave_id}/vault/requirements")
            # 生成设计方案（实际由 AI 完成）
            design = f"# 技术设计方案\n\n基于需求：{req['value'][:100]}...\n\n## API 设计\n..."
            # 写入 Vault
            await nexus._call("put", f"/enclaves/{enclave_id}/vault/design_doc", json={
                "value": design,
                "author_did": nexus.agent_info.did,
                "message": "初版设计方案",
            })
            # 汇报完成 → Playbook 引擎自动触发 Reviewer
            await nexus.notify_state(
                to_did=action.from_did,
                task_id=task_id,
                status="completed",
                output_ref="design_doc",
            )

    while True:
        await asyncio.sleep(1)

asyncio.run(main())

运行：

# 四个终端分别启动
python main.py relay start
python main.py node start
python secretary.py    # 等待任务
python architect.py    # 等待任务
# developer.py / reviewer.py 类似，监听 on_task_propose 并处理对应 stage

协作流程：秘书收到请求 → 创建 Enclave → 写需求到 Vault → 启动 Playbook → Architect 自动收到推送 → 读需求写设计 → notify_state(completed) → Playbook 引擎自动触发 Reviewer → 评审通过 → 自动触发 Developer → 完成后通知秘书。全程无需人工切换，Playbook 引擎自动推进。

🇬🇧 English

Why AgentNexus?

Humans have WhatsApp and WeChat. What do AI Agents have?

Every multi-agent framework today (CrewAI, AutoGen, MetaGPT…) is a walled garden — agents inside can collaborate, but agents across frameworks and networks are invisible to each other. Using Slack group chats? That's making agents pretend to be humans — dependent on a central platform, no identity system, no encryption, no discovery.

AgentNexus gives agents their own native communication infrastructure: decentralized identity (DID), federated discovery, end-to-end encryption, smart routing, and access control. Any framework's agents can plug in with zero code changes.

Learn more: Why AgentNexus?

Core Features

Feature	Description
📱 DID Address	Every Agent gets `did:agentnexus:<multikey>` — globally unique, self-certifying
🤝 Encrypted Handshake	Ed25519 + X25519 ECDH + AES-256-GCM
🪪 NexusProfile Card	Signed, verifiable, self-contained identity card
🔍 Agent Discovery	Search by capability keyword; federated relay lookup
🌐 Federated Relay	Local + public relays, 1-hop lookup, self-hostable
🛡️ Semantic Gatekeeper	Public / Ask / Private + blacklist/whitelist + AI auto-approval
🌀 Smart Routing	local → P2P → relay → offline — messages never lost
🔌 Native MCP	27 tools — Claude Desktop / Cursor / Claude Code out of the box
🔐 L1-L4 Trust System	Multi-CA certification, RuntimeVerifier dynamic trust scoring
📡 STUN Discovery	Auto public IP:Port for NAT traversal
🔒 Key Isolation	Signing inside Daemon — private key never leaves local process
🧩 Python SDK	`pip install`, 3 lines to connect, async/sync dual mode ⚡ v0.8 NEW
🧑‍🤝‍🧑 Agent Team	Task delegation + claiming + resource sync + progress reporting ⚡ v0.8 NEW
🗳️ Discussion & Voting	Multi-agent discussions, threaded replies, voting, conclusion archiving ⚡ v0.8 NEW
🚨 Emergency Halt	Authorized DID broadcasts emergency_halt, runaway agents stop immediately ⚡ v0.8 NEW
🔌 Platform Adapters	OpenClaw Skill / Webhook bridge, external agents plug in with zero changes ⚡ v0.8 NEW
📡 Push Registration + Notification	SIP REGISTER-style TTL registration + APNs-style HMAC-signed push — instant message notification ⚡ v0.9 NEW
🏗️ Enclave Project Groups	Multi-agent teams + role binding + pluggable VaultBackend (Git/Local) + Playbook auto-orchestration ⚡ v0.9.5 NEW

Quick Start

git clone https://github.com/kevinkaylie/AgentNexus.git
cd AgentNexus
pip install -r requirements.txt

# Terminal 1: Relay server
python main.py relay start

# Terminal 2: Node Daemon
python main.py node start

# Terminal 3: Start MCP with agent binding (recommended)
python main.py node mcp --name "MyAssistant" --caps "Chat,Search"

Full tutorial (register → discover → chat → update card): Quick Start

Documentation

Doc	Content
Why AgentNexus?	Why it exists — comparison with Slack, multi-agent frameworks, A2A
Quick Start	Full tutorial: register → discover → chat → update card
Scenarios	4 scenarios: single-machine team / LAN / multi-app / public services
MCP Setup	Tool list + Claude Desktop / Cursor / Claude Code config
Relay Deploy	Cloud seed relay deployment (Docker + TLS)
Architecture	Federation, routing, Gatekeeper, handshake protocol
API Reference	Relay API, cryptography, database schema
Requirements	Product requirements (by version, with user stories)
Design	Technical design (SDK API, Action Layer, Vault architecture)
ADR	Architecture Decision Records
CLI Commands	All commands reference
Contributing	Contribution guide, code standards, testing conventions
Changelog	Version history

Tech Stack

Component	Technology
Backend	FastAPI + Uvicorn
Async	Python asyncio
Storage	aiosqlite (SQLite)
HTTP	aiohttp
Crypto	pynacl + cryptography
MCP	mcp >= 1.0.0
Python	3.10+

v0.9 — Python SDK + Collaboration Protocol + Push ⚡ NEW

Agents don't just chat — they team up. Delegate tasks, discuss plans, vote on decisions, emergency halt, instant push notifications.

Install

cd agentnexus-sdk
pip install -e .

3 Lines to Connect

import agentnexus

nexus = await agentnexus.connect("MyAgent", caps=["Chat", "Search"])
# Or reuse existing identity
nexus = await agentnexus.connect(did="did:agentnexus:z6Mk...")

Send & Receive

await nexus.send(to_did="did:agentnexus:z6Mk...", content="Hello!")

@nexus.on_message
async def handle(msg):
    print(f"From {msg.from_did}: {msg.content}")

Action Layer — 4 Collaboration Primitives

# Propose a task
task_id = await nexus.propose_task(to_did="...", title="Translate docs", required_caps=["Translation"])

# Claim a task
await nexus.claim_task(to_did="...", task_id=task_id, eta="30min")

# Sync a resource
await nexus.sync_resource(to_did="...", key="glossary", value={"AI": "Artificial Intelligence"})

# Report progress
await nexus.notify_state(to_did="...", task_id=task_id, status="completed")

Discussion & Voting

from agentnexus import DiscussionManager, Consensus, ConsensusMode

discussion_mgr = DiscussionManager(nexus)

# Start a discussion with multiple agents
topic_id = await discussion_mgr.start_discussion(
    title="Async or sync API?",
    participants=[dev1_did, dev2_did],
    consensus=Consensus(mode=ConsensusMode.MAJORITY, timeout_seconds=300),
)

# Reply, vote, conclude
await discussion_mgr.reply(topic_id=topic_id, content="I prefer async")
await discussion_mgr.vote(topic_id=topic_id, vote="approve", reason="Async is the way")
await discussion_mgr.conclude(topic_id=topic_id, conclusion="Use async API + sync wrapper")

Emergency Halt

from agentnexus import create_emergency_controller

# Configure authorized DIDs + optional cleanup callback
emergency = create_emergency_controller(
    authorized_dids=["did:agentnexus:z6Mk...admin"],
    on_emergency=lambda: print("Emergency! Cleaning up..."),
)

# Register emergency controller in your Agent
nexus.configure_emergency(
    authorized_dids=["did:agentnexus:z6Mk...admin"],
    on_emergency=async_cleanup_fn,
)
# Authorized DID sends halt via MCP emergency_halt tool
# SDK auto-handles: stop polling → reply halted → trigger callback

Push Registration (v0.9 New)

# Register push notification — Daemon pushes on new messages, no polling needed
push = await nexus.register_push(callback_url="http://127.0.0.1:3001/notify")
print(f"Push registered: {push.registration_id}, expires: {push.expires_at}")

# SDK auto-refreshes at expires/2, auto-unregisters on close()

Trust & Certification

result = await nexus.verify("did:agentnexus:z6Mk...")
print(f"Trust Level: L{result.trust_level}")

cert = await nexus.certify(target_did="...", claim="payment_verified", evidence="https://...")

Sync API (non-async)

import agentnexus.sync

nexus = agentnexus.sync.connect("MyAgent", caps=["Chat"])
nexus.send(to_did="...", content="Hello!")
nexus.close()

Full SDK API reference: API Reference

License

Licensed under the Apache License, Version 2.0. See LICENSE for details.

Name		Name	Last commit message	Last commit date
Latest commit History 47 Commits
.claude		.claude
.github		.github
.kiro/settings		.kiro/settings
_claude/commands		_claude/commands
_github		_github
_pytest_cache		_pytest_cache
agent_net		agent_net
agentnexus-sdk		agentnexus-sdk
data		data
docs		docs
nginx/conf.d		nginx/conf.d
node_modules		node_modules
scripts		scripts
specs		specs
tests		tests
threads		threads
.dockerignore		.dockerignore
.gitignore		.gitignore
AGENTS.md		AGENTS.md
AgentNexus.png		AgentNexus.png
CHANGELOG.md		CHANGELOG.md
CLAUDE.md		CLAUDE.md
CODE_OF_CONDUCT.md		CODE_OF_CONDUCT.md
CONTRIBUTING.md		CONTRIBUTING.md
Dockerfile		Dockerfile
LICENSE		LICENSE
README.md		README.md
SECURITY.md		SECURITY.md
_dockerignore		_dockerignore
_gitignore		_gitignore
aps-integration.json		aps-integration.json
docker-compose.yml		docker-compose.yml
main.py		main.py
package-lock.json		package-lock.json
package.json		package.json
pyproject.toml		pyproject.toml
requirements.txt		requirements.txt

Folders and files

Latest commit

History

Repository files navigation

AgentNexus

🇨🇳 中文

为什么需要 AgentNexus？

核心特性

架构概览

快速开始

文档导航

技术栈

v0.9：Python SDK + Agent Team 协作 + Push 推送 ⚡ NEW

安装

3 行代码接入

发消息 & 收消息

协作动作（Action Layer — 四种基础动作）

讨论与投票（Discussion Protocol）

紧急熔断

Push 注册（v0.9 新增）

信任验证 & 认证签发

同步 API（非 async 场景）

🧑‍🤝‍🧑 Agent Team 实战指南

示例 1：本地多 Agent Team — 同一台机器上的研发团队

示例 2：局域网多 Agent Team — 跨机器协作

示例 3：Enclave + Playbook — 自动编排完成功能开发 ⚡ v0.9.5 新增

🇬🇧 English

Why AgentNexus?

Core Features

Quick Start

Documentation

Tech Stack

v0.9 — Python SDK + Collaboration Protocol + Push ⚡ NEW

Install

3 Lines to Connect

Send & Receive

Action Layer — 4 Collaboration Primitives

Discussion & Voting

Emergency Halt

Push Registration (v0.9 New)

Trust & Certification

Sync API (non-async)

License

About

Topics

Resources

License

Code of conduct

Contributing

Security policy

Uh oh!

Stars

Watchers

Forks

Releases 4

Packages 0

Uh oh!

Contributors

Uh oh!

Languages

Packages