AgentRaaS wrap

agentraas（Python SDK）

AI エージェントが行う処理の信頼性を高めるレイヤーです。エージェントが行うすべてのアクション（請求、メール送信、CRM の更新、ツール呼び出し）は、リトライされても一度だけ実行され、実行前にルールと照合でき、その記録が残ります。このパッケージを使うと、これらの機能をコードに取り入れられます。

• @exactly_once と protect_tool（LangChain／LangGraph、CrewAI）：自分のコード、フレームワーク、または不安定なネットワークによるリトライが、請求やメッセージの重複につながることはありません。サーバーは不要です。
• agentraas wrap：あらゆる MCP サーバーの書き込みツールに同じ機能を適用します。
• agentraas scan：MCP サーバー内で、金銭の移動、データの削除、コードの実行、メッセージの送信が可能なツールを、呼び出さずに一覧表示します。
• agentraas chaos：本番環境で問題が起きる前に、リトライによって二重実行される可能性のあるエージェント内の呼び出しを検出します。
• 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 にシリアライズ可能である必要があります（SDK オブジェクトではなく、ID や辞書を返してください）。

• RedisStore を使うには pip install redis が必要です。

ダッシュボード、監査証跡、人間による承認、サーキットブレーカーも利用する場合は、以下の Client を使って、同じ呼び出しの接続先を AgentRaaS サーバーに設定してください。

LangChain／LangGraph：ツール呼び出しの一度だけの実行

pip install "agentraas[langchain]" を実行し、副作用のあるツールをラップしてください。LangChain エージェント、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)

モデルが同じ引数で保護されたツールを再度呼び出した場合や、クラッシュまたはリトライ後にグラフがツールノードを再実行した場合、2回目の請求を行う代わりに保存済みの結果が返されます。上記の @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)])

エージェントが同じ引数で保護されたツールを再度呼び出した場合や、クラッシュ後にクルーが再度起動された場合、2回目の請求を行う代わりに保存済みの結果が返されます。上記と同じ store=、key=、ttl= オプションを利用できます。元のツールオブジェクトは変更されません。

MCP wrap：あらゆる MCP サーバーでツール呼び出しを厳密に 1 回だけ実行

MCP サーバーのコマンドの前に agentraas wrap -- を付けます。すべて そのまま通過しますが、重複排除ウィンドウ内に同一の呼び出し（同じ ツール、同じ引数）が書き込みツールに対して行われた場合は、再実行せず、 最初の呼び出しの結果を返します。エージェントが再試行したりループしたりしても、 同じメールを送信したり、同じ課題を 2 回作成したりすることはありません。すべてのツール 呼び出しは監査ログに追記されます。

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

カオステスター：エージェントが 2 回実行する呼び出しを特定

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。仕様のキーには API のフィールドを camelCase で指定します。2 回実行しても変更は発生しません。

ローカルトンネル：ノートパソコンで Webhook をテスト

agentraas tunnel は公開 URL を表示し、そこに送信されたすべてのリクエストを ローカルポートに転送するため、デプロイ前のコードに Stripe、GitHub、n8n の Webhook を 接続できます。無料の AgentRaaS アカウント（エージェント API キー）が必要です。

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

組織ごとに 1 つのトンネル、セッションは 2 時間、ローカルのレスポンスタイムアウトは 30 秒です。

シェルからアクションを 1 つ送信：agentraas call

すべてのコマンドは短縮名 agtrs でも実行できます。call は 1 つの アクションを 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_URL に対して AGENTRAAS_EMAIL / AGENTRAAS_PASSWORD を使用します。デフォルトは 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 ダッシュボードからエージェントを接続します（+ エージェントを接続）。これにより、agentraas_key と org_id、agent_id が付与されます。
2. 呼び出すサービスの認証情報を追加します（認証情報パネル）。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 呼び出し元が、サーバー側で強制されない単一の ID を共有します。その場合、エージェントごとのレート制限と監査証跡で、自分のトラフィックを他のユーザーのものと区別できません。ダッシュボードからエージェントを接続したら設定してください。追加のキーワード引数を 2 つ指定するだけです。

安全な再試行

AgentRaaS は何かを転送する前に、サーバー側でアトミックな重複排除スロットを確保します。タイムアウトや接続切断によって client.call(...) が例外を発生させても、同じペイロードで再度呼び出して安全です。正常に完了するか、実際に処理された呼び出しのキャッシュ済み結果が返されます。この SDK は自動で再試行しません。独自の再試行ロジック（またはエージェントフレームワークの再試行ロジック）は、必要に応じて何度でも再試行できます。

例外は、プロバイダーが時間内に AgentRaaS に応答しない場合です。このとき "outcome": "unknown" を含む 504 が返され、解決するまで同一の呼び出しは 409 になります。すでに処理されている可能性があるため、意図的に再試行しても実行されません。プロバイダーに確認してから、ダッシュボードの 失敗したリクエスト で再実行するか、破棄してください （手順）。処理されていないことが確実な場合は、独自のキーを使って新しいアクションとして再送信してください。

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

記録と再生

実際のプロバイダーに対してエージェントを一度実行した後、再度プロバイダーを呼び出すことなく、その実行を再生できます（テスト、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

ペイロードが検証ルールに違反すると、err.validation_errors に違反したすべてのフィールド（field、rule、message、hint）を含む 422 が発生します。何も送信されていません。また、str(err) は各ヒントを列挙するため、エラーを受け取ったエージェントは、1 回の再試行ですべての問題を修正できます。

[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 機能はソースコードが公開されています。メインリポジトリを参照してください。）