SafeAgent

# 安全代理 — 人工智能代理执行保护 [](https://github.com/giskard09/argentum-core/blob/main/docs/mycelium-provider-protocol.md)  <!-- mcp-name: io.github.azender1/safeagent --> `POST /claim` · [safeagent-production.up.railway.app](https://safeagent-production.up.railway.app) 仪表盘：[safeagent-dashboard-2.vercel.app](https://safeagent-dashboard-2.vercel.app) ```bash # 申请操作 — 免费，无需支付头部 curl -s -X POST https://safeagent-production.up.railway.app/claim \ -H "Content-Type: application/json" \ -d '{"request_id":"order:TQQQ:buy:6:2026-05-19T13:31:00-04:00","action":"order"}' # → {"status":"PROCEED","request_id":"..."} # 操作完成后结算 curl -s -X POST https://safeagent-production.up.railway.app/settle/order:TQQQ:buy:6:2026-05-19T13:31:00-04:00 \ -H "Content-Type: application/json" \ -d '{"result":{"status":"completed"}}' # → {"status":"committed","request_id":"..."} # 重试相同请求_id — 返回 SKIP curl -s -X POST https://safeagent-production.up.railway.app/claim \ -H "Content-Type: application/json" \ -d '{"request_id":"order:TQQQ:buy:6:2026-05-19T13:31:00-04:00","action":"order"}' # → {"status":"SKIP","request_id":"...","existing":{...}} # 免费测试端点 — 相同逻辑，限制每个 IP 10 次调用 curl -s -X POST https://safeagent-production.up.railway.app/claim/test \ -H "Content-Type: application/json" \ -d '{"agent_id":"bot-1","action_type":"order","scope":"TQQQ:buy:bar:2026-05-19T13:31:00-04:00"}' # → {"status":"PROCEED","request_id":"...","test":true,"calls_remaining":9} ``` 已在 [Bazaar](https://orbisapi.com/proxy/safeagent-execution-guard-bb0b02) 索引。 --- ## Mycelium Trails 此实现是 [Mycelium 提供商](https://github.com/giskard09/argentum-core/blob/main/docs/mycelium-provider-protocol.md)。 每个完成的操作都会向 ARGENTUM 提交一个 trail，带有内容寻址的 `action_ref`（JCS+SHA-256 哈希，包含四个预映像字段）。返回的 `mycelium_trail_id` 包含在每个收据中，以进行独立验证。 --- ## 在 Soma 上验证 安全代理是 [Soma](https://soma-api.rgiskard.xyz/catalog) 上首个经过验证的外部集成商 — Mycelium 代理目录。每次生产执行都通过 Mycelium Trails 锚定在链上，并且无需经过操作员即可独立验证。 - **集成商徽章** — 验证的外部客户端，在生产中运行 Mycelium Trails - **实时 trails** — [https://argentum-api.rgiskard.xyz/dashboard/trails?client=safeagent-prod](https://argentum-api.rgiskard.xyz/dashboard/trails?client=safeagent-prod) - **公开审计** — 任何审计员都可以独立验证安全代理执行 --- ## 功能 安全代理是一种精确执行保护。它可以防止 AI 代理和 SaaS 应用程序触发相同的操作两次 — 在崩溃重试、重复信号、Webhook 重放或跨多个实例的并发执行。 每个操作都会获得一个稳定的 `request_id`，该标识符源于代理正在执行的操作和时间。第一次调用会提交。后续调用返回 `SKIP` 和原始结果。不会重复收费。不会重复发送电子邮件。不会重复下订单。不会重复 Webhook。 **状态机：** `PENDING → COMMITTED | SKIP` **常见故障模式 安全代理可以防止：** | 场景 | 没有安全代理 | 有安全代理 | |---|---|---| | Stripe 充值超时，重试触发 | 客户被收取两次费用 | 第二次充值返回 SKIP | | 注册时欢迎电子邮件重发 | 用户收到两个欢迎电子邮件 | 第二次发送返回 SKIP | | Webhook 传递两次（Stripe/GitHub/Twilio 保证至少一次交付） | 事件处理两次 | 第二次处理返回 SKIP | | 工作空间在重试时配置 | 创建两个工作空间 | 第二次配置返回 SKIP | | AI 代理工具调用在崩溃后重试 | 重复副作用 | 第二次调用返回 SKIP | --- ## 技术栈 安全代理是正式指定的代理执行完整性栈中的精确执行强制层： ``` Polaris（提交门控授权） └── AgentGraph 安全裁决（执行前安全门控） └── 安全代理（精确执行保护）← 这里 └── Nobulex（签名双边收据） └── Mycelium Trails（链上锚点） ``` 每一层都是独立编写的，可以独立验证。层之间没有信任关系。 **欧盟 AI 法案第 12 条** — 执行截止日期为 2026 年 8 月 2 日。安全代理为代理合规提供防篡改执行日志。整个栈满足“防篡改”，独立于平台日志。 --- ## API ### POST /claim 控制操作。第一次调用返回 PROCEED，任何重复调用返回 SKIP。 **请求体：** ```json { "request_id": "order:TQQQ:buy:6:2026-05-19T13:31:00-04:00", "action": "order" } ``` - `request_id` — 源自代理正在执行的操作的稳定标识符。相同的输入 = 相同的键。 - `action` — 被控制的操作类型（例如 `payment.send`、`email`、`trade`、`order`） **响应：** ```json { "status": "PROCEED", "request_id": "order:TQQQ:buy:6:2026-05-19T13:31:00-04:00" } ``` 重试相同负载： ```json { "status": "SKIP", "request_id": "order:TQQQ:buy:6:2026-05-19T13:31:00-04:00", "existing": {...} } ``` ### POST /settle/{request_id} 在操作触发后结算请求。将状态从 PENDING 推进到 COMMITTED。 ```bash curl -s -X POST https://safeagent-production.up.railway.app/settle/order:TQQQ:buy:6:2026-05-19T13:31:00-04:00 \ -H "Content-Type: application/json" \ -d '{"result":{"status":"completed","txid":"abc123"}}' ``` ```json { "status": "committed", "request_id": "..." } ``` ### POST /claim/test 免费测试端点 — 相同逻辑，无需支付。每个 IP 限制 10 次调用。 ```bash curl -s -X POST https://safeagent-production.up.railway.app/claim/test \ -H "Content-Type: application/json" \ -d '{"agent_id":"bot-1","action_type":"order","scope":"TQQQ:buy:6:bar:2026-05-19T13:31:00-04:00"}' ``` ```json { "status": "PROCEED", "request_id": "...", "test": true, "calls_remaining": 9 } ``` ### GET /audit 完整的请求历史记录。按 `agent_id`、`action`、`status` 或时间戳范围过滤。 ```bash curl "https://safeagent-production.up.railway.app/audit?agent_id=bot-1&status=COMMITTED" ``` ```json {"items": [...], "total": 5, "limit": 100, "offset": 0} ``` 参数：`agent_id`、`action`、`status`、`from_ts`、`to_ts`、`limit`（最大 1000）、`offset`。 --- ## 集成 ### Python ```python import requests def claim(request_id: str, action: str) -> dict: r = requests.post( "https://safeagent-production.up.railway.app/claim", headers={"Content-Type": "application/json"}, json={"request_id": request_id, "action": action} ) return r.json() # {"status": "PROCEED"|"SKIP", "request_id": "..."} def settle(request_id: str, result: dict) -> dict: r = requests.post( f"https://safeagent-production.up.railway.app/settle/{request_id}", headers={"Content-Type": "application/json"}, json={"result": result} ) return r.json() # 用例 response = claim("payment:customer-123:invoice-456", "payment.send") if response["status"] == "PROCEED": result = send_payment(...) settle("payment:customer-123:invoice-456", {"status": "completed"}) elif response["status"] == "SKIP": print("已经执行，跳过") ``` ### SaaS（Stripe / 电子邮件 / Webhook） ```python import requests def safe_stripe_charge(customer_id, amount, idempotency_key): request_id = f"stripe:charge:{customer_id}:{idempotency_key}" response = requests.post( "https://safeagent-production.up.railway.app/claim", headers={"Content-Type": "application/json"}, json={"request_id": request_id, "action": "stripe_charge"} ).json() if response["status"] == "SKIP": return response.get("existing") charge = stripe.PaymentIntent.create(amount=amount, customer=customer_id) requests.post( f"https://safeagent-production.up.railway.app/settle/{request_id}", headers={"Content-Type": "application/json"}, json={"result": {"charge_id": charge.id}} ) return charge ``` **Webhook 去重** — Stripe、GitHub 和 Twilio 保证至少一次交付。安全代理将至少一次转换为精确一次： ```python def handle_stripe_webhook(event): response = requests.post( "https://safeagent-production.up.railway.app/claim", headers={"Content-Type": "application/json"}, json={"request_id": event["id"], "action": "stripe_event"} ).json() if response["status"] == "SKIP": return {"ok": True} provision_subscription(event) requests.post( f"https://safeagent-production.up.railway.app/settle/{event['id']}", headers={"Content-Type": "application/json"}, json={"result": {"processed": True}} ) ``` ### CrewAI ```python from crewai import Agent import sqlite3 guard_con = sqlite3.connect("safeagent_orders.db", check_same_thread=False) # ... 相同的保护模式，按 tool_name + input hash + session_id 键控 ``` PR [crewAIInc/crewAI#5822](https://github.com/crewAIInc/crewAI/pull/5822) 添加了可插拔的幂等性后端。安全代理的 SQLite 模式兼容。 ### WisePick [WisePick](https://github.com/w2jmoe/WisePick) 是一个用于代理运行时确定性路由的层。集成将 **能力选择** 与 **持久、幂等执行** 分开 — WisePick 回答 *什么* 和 *哪个提供商*，安全代理回答 *此逻辑工作是否已经运行*。 **栈：** ``` WisePick /v1/decide → DashClaw → 安全代理 → Mycelium Trails → Base/Arbitrum ``` - 适配器：[`adapters/safeagent_adapter.py`](https://github.com/w2jmoe/WisePick/blob/main/adapters/safeagent_adapter.py) - 集成文档：[`docs/integrations/safeagent.md`](https://github.com/w2jmoe/WisePick/blob/main/docs/integrations/safeagent.md) - 集成线程： [安全代理 #9](https://github.com/azender1/SafeAgent/issues/9) --- ## 本地保护（嵌入式 SQLite） 相同的保证，无需网络调用。将此代码放入任何 Python 代理： ```python import sqlite3 _SA_DB = "safeagent_orders.db" _sa_con = sqlite3.connect(_SA_DB, check_same_thread=False) _sa_con.execute("""CREATE TABLE IF NOT EXISTS orders ( request_id TEXT PRIMARY KEY, result TEXT, status TEXT DEFAULT 'PENDING', created_at TEXT DEFAULT (datetime('now')) )""") _sa_con.commit() def place_order_with_guard(symbol, qty, side, bar_ts): request_id = f"order:{symbol}:{side}:{qty}:{bar_ts}" _sa_con.execute( "INSERT OR IGNORE INTO orders (request_id, status) VALUES (?, 'PENDING')", (request_id,) ) _sa_con.commit() row = _sa_con.execute( "SELECT status, result FROM orders WHERE request_id = ?", (request_id,) ).fetchone() if row and row[0] == 'COMMITTED': print(f"安全代理跳过：{request_id}") return row[1] result = place_order(symbol, qty, side) _sa_con.execute( "UPDATE orders SET status='COMMITTED', result=? WHERE request_id=?", (json.dumps(str(result)), request_id) ) _sa_con.commit() return result ``` --- ## 案例研究：崩溃重试重复预防 交易机器人发出市场订单，购买 6 股 TQQQ。经纪商接受订单。机器人在更新状态之前崩溃。重启后 — 相同的信号，相同的柱 — 机器人再次触发。经纪商再次成交。代理现在持有 12 股，而其预期持有 6 股。 这并非理论问题。在操作提交和状态持久性之间的任何未处理异常都会发生这种情况。 安全代理在与经纪商交互之前派生一个稳定的键： ``` request_id = "order:TQQQ:buy:6:2026-05-19T13:31:00-04:00" ``` 首次调用：PROCEED。副作用触发。结算写入 COMMITTED。 崩溃和重试：SKIP。经纪商从未再次触及。 --- ## 案例研究：实时重复阻止 — 2026 年 5 月 21 日 来自完整栈的六个确认 SKIP 事件：DashClaw、安全代理、Mycelium Trails、Base/Arbitrum、经纪商 Alpaca。 - 0942 ET：重复购买 TQQQ 数量 = 6 阻止，$452 - 0947 ET：重复添加 TQQQ 数量 = 6 阻止，$452 - 0949 ET：重复出售 TQQQ 数量 = 12 次翻转，$902 - 1000 ET：重复进入 TQQQ 数量 = 6 阻止，$454 - 1014 ET：重复出售 TQQQ 数量 = 18 次 V20 翻转，$1,350 - 1106 ET：重复 SQQQ 添加期间扩展，$43 完整会话数据：https://gist.github.com/azender1/b9112b6519c935df4a75cb05cd250e26 --- ## 一致性 安全代理参与 A2A 跨实现一致性语料库。四个独立实现之间的字节验证固定装置： - kenneives（agentgraph）— verifier-attestation-v0：30/30 ✓ - evidai（LemonCake）— gated-preflight-v1：33/33 ✓ - haroldmalikfrimpong-ops（agentid）— 独立验证器 ✓ - 安全代理 — 精确一次-v1.1 ✓ A2A #1920 已关闭。所有四个实现都满足新鲜度和重放要求。 --- ## 规范的 `action_ref` 派生 JCS（RFC 8785），与 argentum-core、Nobulex 和 APS 对齐： ```python import hashlib import rfc8785 def compute_action_ref( agent_id: str, action_type: str, scope: str, timestamp: str, # RFC 3339 UTC，3 位毫秒： "2026-05-15T10:00:00.123Z" ) -> str: payload = { "agent_id": agent_id, "action_type": action_type, "scope": scope, "timestamp": timestamp, } canonical = rfc8785.dumps(payload) return hashlib.sha256(canonical).hexdigest() ``` `timestamp` 是带有精确 3 位毫秒数字的 RFC 3339 UTC 字符串。末尾的 `Z` 是强制的。 一致性装置： [giskard09/argentum-core tag action-ref-v1.0](https://github.com/giskard09/argentum-core) --- ## 部署 Railway · 无服务器关闭 · 始终在线 PyPI：`pip install safeagent-exec-guard` npm：`npm install n8n-nodes-safeagent` MCP 注册表：`io.github.azender1/safeagent` 仪表盘：[safeagent-dashboard-2.vercel.app](https://safeagent-dashboard-2.vercel.app) --- ## 许可证 Apache-2.0 <!-- mcp-name: io.github.azender1/safeagent -->