integrate-arcjet-guard-strands-agents-py

Python Strands Agents に Arcjet Guard を統合する

arcjet.guard.strands_agents は、エージェントが既存の Arcjet クライアントをラップします。 Arcjet API と直接通信することはありません。Guard の共通の基礎 （クライアント、ルール、ラベル、判定、キャプチャ、登録）は ../arcjet/references/guards_python.md にあります。 Strands 固有でない内容については、そのリファレンスを読み込んでください。

公式 Python strands / strands-agents>=1.11.0,<2 のみが対象です。JS @arcjet/guard/strands-agents/v1（ドキュメント https://docs.arcjet.com/guards/strands-agents/）や、コミュニティ製フォークは対象外です。 arcjet.guard.strands_agents をインポートしても、LangChain、CrewAI、 JS @arcjet/guard/strands-agents/v1 は読み込まれません。追加の依存関係がない CrewAI とは異なり、 指定バージョンには arcjet[strands-agents] エクストラがあります （pyproject.toml の strands-agents = ["strands-agents>=1.11.0,<2"]）。 このエクストラは安全です（chromadb は含まれません）。

エクスポート：guard_tool、guard_hooks、 strands_agent_context。 作成した @tool と Agent(hooks=) / add_hook を使用します。

3 つの適用箇所に、1 つの判定ルールがあります。

• 作成した @tool → guard_tool。これが返すコピーをエージェントに渡してください。元のツールには Guard が適用されません。
• ラップされていないツール / MCP ツール → guard_hooks。ゲートはツールごとの BeforeToolCallEvent.cancel_tool（True または str）です。すでにラップされたツールはスキップされるため、Guard が二重に呼び出されることはありません。
• 相関 → strands_agent_context は invocation_state から呼び出し元が設定した ID を読み取ります。ID を生成することはありません。trace_id は読み取りません。

ドキュメント：https://docs.arcjet.com/guards/strands-agents/。例： examples/fastapi-strands-agents-guard。 別の例の名前を作らないでください。

ゲートはツールごとの BeforeToolCallEvent.cancel_tool

guard_hooks はこのイベントに登録されるため、DENY （またはデフォルトの on_guard_error="deny" で未評価の Guard）の場合、ツールは実行されません。 文字列はキャンセルメッセージ（ArcjetDenialResult の JSON）です。True を指定すると、 Strands のデフォルトメッセージが使われ、フィールドは破棄されます。フェイルクローズにします。エラー時には必ず cancel_tool を設定してください。未設定のままにしたり、例外を送出したりしないでください。#196 と同じフェイルクローズのデフォルトです。 "allow" の場合に限りフェイルオープンし、DENY は常にブロックします。コアの guard() は引き続きフェイルオープンです（has_failed_open()）。

event.interrupt() はポリシーゲートではありません

BeforeToolCallEvent.interrupt() / resume は、Human-in-the-loop 用です。CrewAI の human_input、JS の humanInTheLoopMiddleware、LangGraph の interrupt()、OpenAI Agents の needsApproval、Genkit の interrupt() と同じ落とし穴があります。 受信時のヘルパーも、承認用ヘルパーもありません。

Agent(...) / __call__ / stream_async の前に受信内容を検査する

受信時のヘルパーはありません。アプリケーション内で aj.guard(...) を呼び出し、 判定に応じた処理をしてください。コアの guard() はフェイルオープンです。ALLOW はルールが実行された証明にはなりません。この呼び出し箇所をフェイルクローズにする必要がある場合は、 decision.has_failed_open() でゲートしてください。guard_tool / guard_hooks はすでにこの動作をデフォルトにしています。

まず人に確認する質問

コードから推測できないことだけを尋ね、デフォルトを提案してください。

1. どのツールがリスクを伴いますか（外部への副作用、取り消し不能、金銭の支出、メッセージの送信など）？それらには guard_tool を適用します。自分で作成していない MCP ツール / ラップされていないツールには guard_hooks を適用します。
2. 制限は何ですか？（例：「ユーザーごとに 1 分あたり 10 回の検索」→ TokenBucket。）
3. メタデータ上のユーザーは誰ですか。不透明なユーザー ID / テナント ID（PII は不可）ですか？デフォルトはなしです。すでにある会話 ID / セッション ID を invocation_state と guard_hooks(...) の両方に設定してください。この ID は相関 ID であり、ユーザー ID ではありません。
4. Arcjet の停止は許容できませんか？すべてのヘルパーはデフォルトで on_guard_error="deny" です。エージェントの実行前に行う受信内容の検査については、明示的に確認してください。そこでフェイルクローズにするとエージェントは実行されないため、その呼び出し箇所では "allow" を選ぶことも一般的かつ妥当です。

読者が誤解しやすい点

1. これは JS の @arcjet/guard/strands-agents/v1 ではありません。 arcjet.guard.strands_agents をインポートしてください。
2. インバウンド用ヘルパーはありません。 Agent(...)／__call__／stream_async の前に、コアの guard() でスクリーニングしてください。
3. event.interrupt() は HITL 用であり、ポリシーではありません。 拒否には cancel_tool（True または str）を使います。
4. フェイルクローズでは、エラー時に必ず cancel_tool を設定します。 例外を発生させたり、未設定のままにしたりしないでください。
5. 相関 ID は読み取るものであり、生成するものではありません。 trace_id も agent.id も SessionManager の自動 ID も使用しないでください。
6. すでにラップされたツールは guard_hooks によってスキップされるため、Guard は二度呼び出されません。guard_tool が返すコピーをエージェントに渡してください。
7. レート制限には、モデルが指定した注文 ID ではなく、認証済みの呼び出し元をキーとして使ってください。
8. Strands のすべてのツールを、生の guard() で個別にラップしないでください。
9. inputs= は受け付けられます。 判定がないことは拒否を意味しません。Console／CLI で確認してください。

ステップ 1：Guard クライアントをインストールして見つける

PyPI の arcjet 1.1.0 が必要です（追加機能は 1.0.0 には含まれていません）。

pip install "arcjet[strands-agents]"

エージェントに Guard クライアントがまだない場合は、モジュールのスコープで一度だけ起動してください。

import os
from arcjet.guard import launch_arcjet

aj = launch_arcjet(key=os.environ["ARCJET_KEY"])

ステップ 2：作成したツールをゲートする — guard_tool

from strands import Agent, tool
from arcjet.guard import DetectPromptInjection, TokenBucket, launch_arcjet
from arcjet.guard.strands_agents import guard_hooks, guard_tool, strands_agent_context

aj = launch_arcjet(key=os.environ["ARCJET_KEY"])
lookup_limit = TokenBucket(
    label="order.looked-up",
    bucket="lookups",
    refill_rate=10,
    interval_seconds=60,
    max_tokens=10,
)
mcp_limit = TokenBucket(
    label="mcp.invoked",
    bucket="mcp-access",
    refill_rate=20,
    interval_seconds=60,
    max_tokens=20,
)
inbound = DetectPromptInjection()
user_id = authenticated_user_id

@tool  # event.interrupt() is HITL — not this policy gate
def lookup_order(order_id: str) -> dict:
    """Look up an order by ID."""
    return {"order_id": order_id, "status": "shipped"}

lookup_order = guard_tool(
    guard=aj,
    tool=lookup_order,
    action="order.looked-up",
    rules=[lookup_limit(key=user_id, requested=1)],
    on_guard_error="deny",
)

ステップ 3：ラップされていないツール／MCP ツールをゲートする — guard_hooks

mcp_tools = []  # from an MCP client you did not wrap with guard_tool
agent = Agent(
    tools=[lookup_order, *mcp_tools],
    # The agent-wide gate for tools guard_tool did not wrap.
    # Already-wrapped tools are skipped, so Guard is not called twice.
    # BeforeToolCallEvent.cancel_tool is True or a str (JSON of the payload).
    hooks=[
        guard_hooks(
            guard=aj,
            action="mcp.invoked",
            rules=[mcp_limit(key=user_id, requested=1)],
            session_id=conversation_id,
            on_guard_error="deny",
        ),
    ],
)

ステップ 4：エージェントの実行前にインバウンドをスクリーニングする

invocation_state = {"sessionId": conversation_id}
derived = strands_agent_context(invocation_state)
decision = await aj.guard(
    label="message.received",
    rules=[inbound(user_text)],
    correlation_id=derived.correlation_id,
)
if decision.conclusion == "DENY":
    raise RuntimeError("message blocked")
if decision.has_failed_open():
    raise RuntimeError("inbound guard unavailable")

agent(user_text, invocation_state=invocation_state)

インバウンド用ヘルパーはありません。

ステップ 5：相関 ID

strands_agent_context は、呼び出し元が所有する ID を invocation_state から読み取ります。順に correlationId、sessionId、requestId を参照します。ID を生成することはありません。trace_id を読み取ることもありません。agent.id や SessionManager の自動 ID も読み取りません。ターンごとに相関 ID を作らないでください。同じ ID を呼び出しと guard_hooks(...) の両方に設定してください。

統合を検証する

1. python -m py_compile（またはプロジェクトの型チェック）が成功することを確認します。
2. インバウンド PI（エージェントの実行前）、guard_tool による拒否、ラップされていないツールに対する guard_hooks による拒否（cancel_tool が True または文字列であること）、レート制限、フェイルクローズ（Guard に到達できない場合）を試します。ゲートとして event.interrupt() が一度も呼び出されないことを確認します。
3. Arcjet Console／CLI で、判定に共通の、呼び出し元が所有するセッション ID／リクエスト ID が使われており、trace_id や agent.id が使われていないことを確認します。
4. 実際の ARCJET_KEY を使った手動の E2E は、実行するまでは未検証です。

動作例： examples/fastapi-strands-agents-guard。 別の例の名前を作らないでください。この skills リポジトリに例を追加しないでください。