AgentRaaS wrap

agentraas（Python SDK）

为 AI Agent 的操作提供可靠性保障。Agent 执行的每项操作（收费、发送电子邮件、更新 CRM、调用工具）即使遇到重试，也只会执行一次；执行前可根据规则检查，并且会留下记录。此软件包为你的代码提供这些保障：

• @exactly_once 和 protect_tool（LangChain/LangGraph、CrewAI）：无论是你的重试、框架的重试，还是不稳定网络导致的重试，都不会造成重复收费或重复发送消息。无需服务器。
• agentraas wrap：为任意 MCP 服务的写入工具提供相同保障。
• agentraas scan：列出 MCP 服务中哪些工具可以转移资金、删除数据、运行代码或发送消息，不会调用任何工具。
• agentraas chaos：在投入生产之前，找出 Agent 中哪些调用会在重试时重复执行。
• Client：将调用指向 AgentRaaS 服务，以使用操作策略、人工审批、支出上限、熔断和审计跟踪。

适用于任何 AgentRaaS 部署：可自托管（./install.sh，所有层级均免费且不限量）或使用 AgentRaaS Cloud。

安装

pip install agentraas

库模式：无需服务器

为任何具有副作用的函数（例如扣款、发送电子邮件、创建 CRM 记录）添加封装，使重试和并发重复调用只执行一次。无需部署任何内容：状态默认存储在本地 SQLite 文件中，也可以存储在你自己的 Redis 中。

from agentraas.local import exactly_once, RedisStore

@exactly_once()                                   # SQLite file, one machine
def charge(customer, amount):
    return stripe.Charge.create(customer=customer, amount=amount).id

charge("cus_1", 4200)   # runs for real
charge("cus_1", 4200)   # retry: returns the cached id, no second charge

# Across machines, with your own idempotency key and window:
@exactly_once(store=RedisStore("redis://localhost:6379/0"),
              key=lambda order: order["id"], ttl=3600)
async def ship(order):
    ...

• 如果函数引发异常，则不会缓存任何内容，重试时会再次执行。即使服务提供商已完成操作但响应丢失，也会如此。因此，请为函数提供 idempotency_key 参数并将其发送给服务提供商（例如 Stripe 的 Idempotency-Key 标头）。装饰器会在每次重试时填入相同的值：

  @exactly_once()
  def charge(customer, amount, idempotency_key=None):
      return stripe.Charge.create(customer=customer, amount=amount,
                                  idempotency_key=idempotency_key).id
  

• 如果重复请求在首次调用仍在运行时到达，它会等待首次调用的结果（最多等待 wait=30 秒，之后会引发 InFlightError）。

• 返回结果必须可序列化为 JSON（返回 ID 或字典，而非 SDK 对象）。

• RedisStore 需要运行 pip install redis。

如需添加仪表板、审计跟踪、人工审批或熔断功能，可使用下方的 Client 将相同调用指向 AgentRaaS 服务。

LangChain / LangGraph：工具调用只执行一次

运行 pip install "agentraas[langchain]"，然后为具有副作用的工具添加封装。适用于 LangChain Agent、create_react_agent 和 ToolNode。

from agentraas.langchain import protect_tool

tools = [protect_tool(charge_card), protect_tool(send_email), search]
agent = create_react_agent(model, tools)

当模型使用相同参数再次调用受保护的工具，或者图在崩溃或重试后重新运行工具节点时，会返回已存储的结果，而不会再次扣款。支持与上文 @exactly_once 相同的 store=、key= 和 ttl= 选项（例如，多个工作进程可使用 store=RedisStore(...)，以及 key=lambda order_id, **_: order_id）。工具结果必须可序列化为 JSON。

CrewAI：工具调用只执行一次

运行 pip install "agentraas[crewai]"（CrewAI 需要 Python 3.10+），然后为具有副作用的工具添加封装，可以是 BaseTool 子类，也可以是 @tool 函数：

from agentraas.crewai import protect_tool

agent = Agent(role="Billing", goal="...", backstory="...",
              tools=[protect_tool(ChargeCardTool()), protect_tool(send_email)])

当 Agent 使用相同参数再次调用受保护的工具，或者 Crew 在崩溃后重新启动时，会返回已存储的结果，而不会再次扣款。支持与上文相同的 store=、key= 和 ttl= 选项。原始工具对象保持不变。

MCP 包装器：为任意 MCP 服务提供恰好一次的工具调用

在 MCP 服务的命令前加上 agentraas wrap --。所有内容都会原样传递，唯一的例外是：在去重窗口内，对 写入 工具发起的相同调用（相同工具、相同参数）会返回首次调用的结果，而不会再次执行。Agent 重试或陷入循环时，也无法重复发送同一封邮件或创建同一个问题。每次工具调用都会追加到审计日志中。

Claude Desktop / Cursor 配置：

{
  "mcpServers": {
    "github": {
      "command": "uvx",
      "args": ["agentraas", "wrap", "--", "npx", "-y", "@modelcontextprotocol/server-github"]
    }
  }
}

• 只读工具永远不会去重，因此查询结果始终是最新的。服务的 readOnlyHint 注解决定工具是否只读；如果没有该注解，则根据 get_*、list_*、search_* 等名称判断。使用 --dedupe TOOL / --no-dedupe TOOL 覆盖设置。
• 失败的调用（JSON-RPC 错误或 isError: true）不会缓存，因此重试时会实际运行。
• 默认窗口：10 分钟（--ttl 600）。状态保存在 ~/.agentraas/mcp-dedup.db，日志保存在 ~/.agentraas/mcp-audit.jsonl（--db、--log）。
• 去重后的结果包含 _meta: {"agentraas/deduplicated": true}。

MCP 安全扫描：哪些工具会影响现实世界

agentraas scan                                    # every server in your Claude Desktop, Cursor, Claude Code configs
agentraas scan -- npx -y @modelcontextprotocol/server-github
agentraas scan --config .mcp.json --fail-on high  # CI: exit 1 on a high-risk tool

按照客户端的方式启动每个 stdio 服务，仅发送 initialize 和 tools/list，并对工具进行分类：根据工具的 readOnlyHint / destructiveHint 注解及其名称中的词语，将其归为转账、删除、运行代码、发送（高风险）、写入（中风险）或读取（低风险）。仅扫描你信任的配置：扫描过程会运行其中的命令。使用 --json 输出机器可读结果。https://agentraas.io/mcp-scan

GitHub Actions：

- uses: agentraas/agentraas/src/scan-action@main
  with:
    config: .mcp.json   # or command: "npx -y @modelcontextprotocol/server-github"; fail-on: high | medium | none

混沌测试器：找出 Agent 会重复执行的调用

agentraas-chaos 位于你的代码与 API 之间。每次收到不同的写入请求（POST/PUT/PATCH/DELETE）时，它会先放行请求并让其执行，然后丢弃响应——这就是会导致重试并造成重复扣费的网络故障。之后，它会报告所有执行次数超过一次的操作。

pip install agentraas

# Safe: a built-in mock answers every call, nothing real is touched
agentraas-chaos --mock -- python my_agent.py

# Against a real API: use TEST-mode keys
agentraas-chaos --upstream https://api.stripe.com -- pytest

你的命令会获得 AGENTRAAS_CHAOS_URL（例如 http://127.0.0.1:8787）；将其用作 API 基础 URL。不指定命令时，它会一直运行，直到按下 Ctrl-C，因此你可以手动将 n8n 工作流或其他工具指向它。

agentraas-chaos: 6 write requests, 3 responses dropped after execution

  DUPLICATES: 1 action(s) executed more than once

    2x  POST /v1/charges  (body sha256 623f14d1b00d)
  ...
  Protected by an idempotency key: 2 retries

如果发现重复操作，它会以代码 1 退出，因此可以让 CI 失败。--fault error-after-commit 会返回 502，而不是断开连接。重试时若重复使用已见过的 Idempotency-Key，该请求会被计为已防护，而不会再次执行。

在 GitHub Actions 中：

- uses: actions/setup-python@v5
  with: { python-version: "3.12" }
- uses: agentraas/agentraas/src/chaos-action@main
  with:
    run: python my_agent.py

将规则保存在 Git 中：agentraas apply

使用 YAML 描述服务的去重规则、验证规则、操作策略和支出上限，像审查代码一样审查这些规则，并从 CI 中应用：

pip install "agentraas[apply]"

# agentraas.yaml
apiVersion: agentraas.io/v1alpha1
kind: DedupRule
metadata: {name: charge-once, org: org_acme}
spec: {service: stripe, action: charges.create, fields: [customer, amount], ttlSeconds: 300}
---
apiVersion: agentraas.io/v1alpha1
kind: SpendCap
metadata: {org: org_acme}
spec: {service: stripe, action: refunds.create, window: day, maxCalls: 100, onExceed: block}

export AGENTRAAS_URL=https://agentraas.example.com [email protected] AGENTRAAS_PASSWORD=...
agentraas apply -f agentraas.yaml --dry-run   # print the plan
agentraas apply -f agentraas.yaml             # create what's missing, replace what changed
agentraas apply -f agentraas.yaml --prune     # also delete rules the file doesn't list

类型：DedupRule、ValidationRule、ActionPolicy、SpendCap。Spec 键对应 API 的 camelCase 字段。重复运行不会产生任何变化。

本地隧道：在笔记本电脑上测试 Webhook

agentraas tunnel 会打印一个公开 URL，并将发送到该 URL 的每个请求转发到本地端口，让你可以将 Stripe、GitHub 或 n8n Webhook 指向尚未部署的代码。需要免费的 AgentRaaS 账户（Agent API 密钥）。

pip install "agentraas[tunnel]"
agentraas tunnel --port 3000 --org org_acme --agent agent_1 --key ar_live_...

每个组织只能建立一个隧道；会话时长为 2 小时，本地响应超时时间为 30 秒。

从 Shell 发送单个操作：agentraas call

所有命令也可以使用短名称 agtrs。call 会通过 AgentRaaS 服务发送一个操作，并以 JSON 格式打印结果：去重、策略、审批和审计均与 Client 相同。适用于脚本，也方便手动检查规则。

export AGENTRAAS_KEY=ar_live_... AGENTRAAS_ORG=org_acme AGENTRAAS_AGENT=agent_1
export AGENTRAAS_URL=https://agentraas.io      # default http://localhost:13000
agtrs call stripe charge.create '{"amount": 4999, "currency": "usd"}'
echo '{"text": "hi"}' | agtrs call slack message.post -

--key、--org、--agent、--url 和 --idempotency-key 可覆盖环境变量。调用被拒绝时，会打印错误及其提示，并以代码 1 退出。

从命令行批准暂挂的调用

这与仪表板的等待批准列表相同，适用于你有写入权限的任何组织。 登录方式与 apply 相同（使用 AGENTRAAS_EMAIL / AGENTRAAS_PASSWORD 登录 AGENTRAAS_URL，默认为 https://agentraas.io）。

agtrs approvals                  # req id, service, action, agent, payload
agtrs approve req_abc123         # runs the call; asks first (--yes to skip)
agtrs deny req_abc123

如果调用已运行但失败，approve 会以状态码 1 退出（此时该调用会出现在“失败的请求”中）。 如果没有终端，它会拒绝执行，除非传入 --yes，因此脚本不会意外批准调用。

从命令行读取审计日志

你所属组织中的每个调用，按时间从新到旧排列，登录方式与 approvals 相同：

agtrs audit                      # last 50: time, status, service action, agent, org, duration, req id, error
agtrs audit -n 200 --json        # adds run/step ids, dedup hash, redacted payload preview (max 1,000)
agtrs audit --csv > audit.csv    # the dashboard's CSV export (newest 10,000)

从命令行查看失败的调用

已失败且保留的调用（仪表板中的失败的请求），登录方式与 approvals 相同：

agtrs dlq                        # id, time, service action, agent, org, req id, error
agtrs dlq replay 42              # sends the call again with the same idempotency key; asks first (--yes to skip)
agtrs dlq dismiss 42             # drop it without sending

如果服务再次失败，replay 会以状态码 1 退出；该条目仍会留在列表中。

将仪表板规则移入 Git

agtrs rules 会将服务器上已有的规则打印为 apply 可接受的文件，因此立即应用它不会产生任何变化：

agtrs rules --org org_acme > agentraas.yaml
agtrs apply -f agentraas.yaml --dry-run   # no changes

快速入门

1. 从 AgentRaaS 仪表板连接 Agent（+ Connect Agent）—— 这会提供一个 agentraas_key，以及你的 org_id 和 agent_id。
2. 为要调用的服务添加凭据（Credentials 面板） ——AgentRaaS 会使用这些凭据转发请求到实际 API，你无需通过此 SDK 传递 原始上游 API 密钥。

import agentraas

client = agentraas.Client(
    agentraas_key="ar_live_...",
    org_id="acme-corp",
    agent_id="billing-bot",
    base_url="http://localhost:13000",  # or your Cloud/self-hosted URL
)

result = client.call("stripe", "charge.create", {"amount": 5000, "currency": "usd"})

也可以使用点号表示法简写：

stripe = client.service("stripe")
result = stripe.charge.create({"amount": 5000, "currency": "usd"})

调用你注册的自定义操作：

result = client.custom("my-internal-webhook", {"foo": "bar"})

为什么 org_id / agent_id 很重要

省略它们后，服务端会让所有未标记的 SDK 调用方共享一个不受约束的身份——你的每个 Agent 的速率限制和审计记录将无法区分你的流量与其他人的流量。连接仪表板中的 Agent 后设置它们即可；只需额外传入两个关键字参数。

重试是安全的

AgentRaaS 会先在服务端原子性地声明一个去重槽位，然后才转发任何内容。如果 client.call(...) 因超时或连接中断而引发异常，使用相同的负载再次调用是安全的——它要么正常完成，要么返回实际成功调用的缓存结果。此 SDK 不会自动重试；你可以按需积极使用自己的重试逻辑（或 Agent 框架的重试逻辑）。

有一种情况有所不同：如果提供方未能及时响应 AgentRaaS，你会收到 504 和 "outcome": "unknown"，在问题解决前，相同的调用会收到 409。这是有意为之，重试不会再次运行调用，因为它可能已经执行。请向提供方确认，然后在仪表板的失败的请求中重放或驳回该调用 (步骤)。如果确定调用没有运行，请使用你自己的密钥将其作为新操作重新发送：

client.call("stripe", "charge.create", payload, idempotency_key="order-1042-retry-1")

录制与重放

让 Agent 针对真实提供方运行一次，然后重放它（在测试中、CI 中或更改提示词后），而无需再次调用这些提供方：

recorder = Client(agentraas_key, org_id, agent_id, record="refund-flow")
# ... run the agent with recorder ...

replayer = Client(agentraas_key, org_id, agent_id, replay="refund-flow", run_id="ci-42")
# each call is answered from the recording, in order; nothing reaches the provider

每个重放的回答都会附带 replay: {recording, position, payload_matches}； 如果录制中没有某个调用，则会引发 404。新的 run_id 会从头开始。详情请参阅：agentraas.io/docs#record-replay。

错误处理

from agentraas import AgentRaaSError

try:
    client.call("stripe", "charge.create", {"amount": 5000, "currency": "usd"})
except AgentRaaSError as err:
    print(err.status_code, err.req_id, str(err))
    if err.status_code == 504:
        pass  # outcome unknown: don't retry blindly, resolve it first

如果负载未通过验证规则，会引发 422；所有未通过验证的字段都会列在 err.validation_errors 中（包含 field、rule、message、hint）。 请求不会发送，而且 str(err) 会列出每条提示，因此 Agent 看到错误后可以在一次重试中修复所有问题：

[422] amount must be number, got string (reqId=...)
- amount: Send "amount" as a JSON number: 49.99, without quotes.
- currency: Send "usd" (exact spelling) in "currency".
Nothing was sent. Fix every field listed in validation_errors, following each hint, and send the call again.

许可证

MIT — 请参阅 LICENSE。（AgentRaaS 服务端核心同样采用 MIT 许可证，其 Team/Enterprise 功能采用源码可用许可；请参阅 主仓库。）