integrate-arcjet-guard-crewai

CrewAI に Arcjet Guard を統合する

arcjet.guard.crewai は、エージェントが既に使用している Arcjet クライアントをラップします。Arcjet API と直接通信することはありません。クライアント、ルール、ラベル、判定、キャプチャ、登録など、Guard に共通する基本事項については、../arcjet/references/guards_python.md に記載されています。CrewAI 固有ではない内容については、このリファレンスを参照してください。

公式の crewai のみが対象です。コミュニティフォーク、LangChain Crew ラッパー、npm 版 CrewAI ポートは対象外です。arcjet[crewai] の extras はありません。CrewAI は chromadb に強く依存しており、これには未修正の RCE CVE-2026-45829 が含まれているため、Arcjet の extra でこれを取り込んではなりません。CrewAI はご自身でインストールしてください。arcjet.guard.crewai をインポートしても、LangChain は読み込まれません。

エクスポート：register_arcjet_hooks、unregister_arcjet_hooks、ArcjetCrewAIHooks、guard_tool、ToolPolicy、sanitize_tool_name、free_text_arguments。guard_crew はありません。

2 つの利用方法と 1 つの判定ルール：

• crew、LiteAgent、MCP アダプター、または crew に渡されたリストが実行するツール → register_arcjet_hooks + ToolPolicy。ゲートはプロセス全体で共有される PRE_TOOL_CALL のみです。DENY の場合は HookAborted(reason=..., source="arcjet") が発生します。
• 自分で呼び出す CrewAI の BaseTool → guard_tool。ArcjetDeniedError / ArcjetUnavailableError が発生する CrewAI の経路は、これだけです。BaseTool.run は PRE_TOOL_CALL をディスパッチしません。

ドキュメント：https://docs.arcjet.com/guards/crewai/。

ゲートはプロセス全体で共有される PRE_TOOL_CALL のみ。登録は 1 回

register_arcjet_hooks は CrewAI のディスパッチャーに登録します。DENY（または、デフォルトの on_guard_error="deny" のもとで判定されなかった Guard）が発生すると、HookAborted(reason=..., source="arcjet") が発生し、ツールは実行されません。CrewAI はそれ以外の例外をすべて握りつぶすため、フックから ArcjetDeniedError / ArcjetUnavailableError を発生させると、ツールが実行されてしまいます。フック経由の処理は同期のみです。launch_arcjet_sync / ArcjetGuardSync を渡してください。非同期クライアントは登録時に拒否されます（ArcjetMisconfiguration）。同じプロセス内で register_arcjet_hooks を 2 回呼び出すことも ArcjetMisconfiguration になります（CrewAI のレジストリは登録を追加するため、判定が二重に実行されます）。先にハンドルの unregister() を呼び出してください。

POST_TOOL_CALL は登録されません

PRE のみが登録されます。判定は PRE でキャプチャされます。POST はポリシーを適用する場所ではありません。エージェントには常に Tool execution blocked by hook. Tool: {name} が表示されます。HookAborted.reason はテレメトリー専用です。処理が続行された場合は success が記録されます。CrewAI のフックキャプチャを、guard_action / LangChain ヘルパーで使われる 5 値の metadata.outcome テーブルとして読み取らないでください。

human_input はポリシーを適用するゲートではありません

Agent/Task の human_input と request_human_input は、人間が介在する処理用です。JS の humanInTheLoopMiddleware、LangGraph の interrupt()、OpenAI Agents の needsApproval、Genkit の interrupt() と同じ落とし穴があります。インバウンド用ヘルパーも承認用ヘルパーもありません。

まずユーザーに確認する質問

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

1. どのツールがリスクを伴いますか（外部への副作用、取り消し不可、金銭の消費、メッセージの送信など）？crew が実行するツールには register_arcjet_hooks を使用します。自分で呼び出すスタンドアロンの BaseTool には guard_tool を使用します。
2. 制限はありますか？（例：「ユーザーごとに 1 分あたり 10 回の検索」→ TokenBucket。）
3. メタデータ上のユーザーは誰ですか（PII ではない、不透明なユーザー／テナント ID）？デフォルトはなしです。crew、タスク、エージェントの名前はメタデータであり、相関 ID として生成されることはありません。呼び出し元が所有する correlation_id / arcjet_sequence を使用してください。
4. Arcjet の障害は許容できませんか？すべてのヘルパーはデフォルトで on_guard_error="deny" を使用します。crew.kickoff の前にインバウンドのスクリーニングを行うかどうかは、必ず明示的に確認してください。そこでフェイルクローズにすると crew は実行されないため、その 1 か所では "allow" を選ぶことも一般的で妥当です。

読者が誤解しがちな点

1. arcjet[crewai] という追加パッケージはありません。 crewai>=1.15.3,<2 を自分でインストールしてください。arcjet.guard.crewai は PyPI の arcjet 1.0.0 に含まれています。
2. guard_crew はありません。 register_arcjet_hooks を使用してください。
3. フックの経路は同期のみです。 launch_arcjet_sync を渡してください。
4. フックからは Arcjet のエラーではなく、HookAborted を発生させてください。 CrewAI はそれ以外のすべてを握りつぶすため、ツールが実行されます。
5. human_input は HITL であり、ポリシーではありません。
6. crew.kickoff の前に、コアの guard / guard_sync で受信内容をスクリーニングしてください。 コアの guard() はフェイルオープンです。
7. レート制限のキーには、モデルが指定した注文 ID ではなく、認証済みの呼び出し元を使用してください。 sanitize_tool_name は Send Email と send_email を同じものとして扱います。
8. すべての CrewAI ツールを生の guard() で個別にラップしないでください。
9. inputs= は受け付けられます。 フックのリゾルバーは (arguments, ctx) です。 判定がないことは拒否を意味しません。Console/CLI で確認してください。

Step 1：インストールしてガードクライアントを取得する

pip install arcjet
pip install "crewai>=1.15.3,<2"

フックの経路では、モジュールスコープで同期クライアントを起動します。

import os
from arcjet.guard import launch_arcjet_sync

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

launch_arcjet は、非同期アプリケーションコードから、または arun() を呼び出す guard_tool からのみ使用してください。

Step 2：クルーが実行するツールを制御する — register_arcjet_hooks

ToolPolicy は action と rules の組み合わせで、ツール名をキーにします。キーと 省略可能な tools= フィルターには、sanitize_tool_name が適用されます（CrewAI 1.15.3 以降）。一致するポリシーがないツールにも "{sanitized_tool_name}.invoked" と登録時の rules が適用されます（空の rules でも Guard に問い合わせます）。ただし、tools= を渡した場合を除きます。スキャンルールに自由記述テキストのみを渡す場合は、free_text_arguments によって不透明な ID（tool_call_id、*_id など）が取り除かれます。フック自体は、フィルターされていないツール独自の引数マッピングをリゾルバーに渡します。すでに guard_tool でラップされているツールはスキップされるため、Guard が二重に呼び出されることはありません。unregister_arcjet_hooks(hooks) または hooks.unregister() で登録を解除してください。

from crewai import Agent, Crew, Task
from arcjet.guard import DetectPromptInjection, TokenBucket, launch_arcjet_sync
from arcjet.guard.crewai import ToolPolicy, register_arcjet_hooks, unregister_arcjet_hooks

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

hooks = register_arcjet_hooks(
    guard=aj,
    policies={
        "lookup_order": ToolPolicy(
            action="order.looked-up",
            rules=[lookup_limit(key=user_id, requested=1)],
        ),
    },
    tools=["lookup_order"],
    on_guard_error="deny",
)

agent = Agent(
    role="Support",
    goal="Look up orders",
    backstory="Help the user with order status.",
    # human_input=True is HITL — not this policy gate
)
task = Task(
    description="Look up the user's order",
    expected_output="Order status",
    agent=agent,
)
crew = Crew(agents=[agent], tasks=[task])

decision = aj.guard(
    label="message.received",
    rules=[inbound(user_text)],
)
if decision.conclusion == "DENY":
    raise RuntimeError("message blocked")
if decision.has_failed_open():
    raise RuntimeError("inbound guard unavailable")

crew.kickoff()
unregister_arcjet_hooks(hooks)

Step 3：呼び出す BaseTool を制御する — guard_tool

これが返すコピーをクルーに渡してください（フックがスキップするためのブランドが付いています）。元のツールは意図的にガードされないままです。元のツールをクルーに渡した場合でも、フックによる保護が適用されます。同期呼び出しにはブロッキングクライアントが必要です。非同期呼び出しには await 可能なクライアントが必要です。

from arcjet.guard.crewai import guard_tool

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

Step 4：相関付け

呼び出し元が管理する correlation_id / arcjet_sequence を使用してください。クルー、タスク、エージェントの名前はメタデータであり、ID として生成されることはありません。有効な値が何も残らない場合、生成された ID に結び付けるのではなく、呼び出しは相関付けされません。

統合を検証する

1. python -m py_compile（またはプロジェクトの型チェック）が成功することを確認します。
2. 受信 PI（kickoff の前）、PRE_TOOL_CALL による拒否（HookAborted。エージェントには Tool execution blocked by hook. Tool: {name} と表示）、単独の guard_tool による拒否（ArcjetDeniedError）、レート制限、フェイルクローズ（到達不能なガード）を実行して確認します。 POST_TOOL_CALL が登録されていないこと、および human_input がゲートとして扱われないことを確認します。
3. Arcjet Console / CLI で、判定がクルー／タスク／エージェント名ではなく、呼び出し元が管理する相関 ID を共有していることを確認します。
4. 実際の ARCJET_KEY を使用した手動の E2E は、実行するまでは未検証です。

この skills リポジトリでは、CrewAI の例の名前を創作しないでください。