integrate-arcjet-guard-claude-agent-sdk-py

Python Claude Agent SDK に Arcjet Guard を統合する

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

公式の claude-agent-sdk>=0.2.127,<1 のみが対象です。JS の @anthropic-ai/claude-agent-sdk アダプター （@arcjet/guard/claude-agent-sdk/v0、ドキュメント https://docs.arcjet.com/guards/claude-agent-sdk/）、Vercel AI、コミュニティ製フォーク、 Claude Managed Agents のホスト型 client.beta.sessions （arcjet.guard.claude_managed_agents）は対象外です。 arcjet.guard.claude_agent_sdk をインポートしても、LangChain、CrewAI、JS アダプターは読み込まれません。 追加パッケージは安全です（chromadb は含まれません）。

エクスポート：guard_tool、guard_hooks、claude_agent_context。 自作のツールには、作成した @tool と create_sdk_mcp_server を使用します。 受信する UserPromptSubmit と、ラップされていない組み込みツール／MCP の PreToolUse には guard_hooks を使用します。 受信テキスト用のヘルパーも guard_can_use_tool もありません。

3 つのサーフェスに、1 つの判定ルール：

• 作成した @tool → guard_tool。拒否時は JSON をコンテンツに含め、 is_error: True を設定します。Python は structuredContent を転送しません （これは JS アダプターの動作です）。例外を発生させないでください。
• 受信テキスト → guard_hooks の UserPromptSubmit（decision: "block"）。モデルがプロンプトを読む前にターンを拒否できるのは、この箇所だけです。
• 組み込みツール／ラップされていない MCP → guard_hooks の PreToolUse （permissionDecision: "deny"）。PostToolUse はキャプチャ専用です。
• 相関付け → claude_agent_context は、呼び出し元が所有する UUID の session_id を読み取ります。UUID を新規発行することはありません。

ドキュメント：https://docs.arcjet.com/guards/claude-agent-sdk/。

ラップされていないツールは guard_hooks の PreToolUse で拒否する

組み込みツール（Bash、Write、…）や guard_tool を通していない MCP ツールには、作成したハンドラーがありません。 それらを拒否できるのは PreToolUse だけです。 PostToolUse では、すでに実行されたツールを取り消せません。 guard_tool でラップしたものはすべて exclude に列挙してください。そうしないと、作成した各ツールに Guard が 2 回適用されます （往復 2 回、クォータ単位 2 つ）。エントリーは報告された名前と一致します。 作成した MCP ツールには {"server": "support", "name": "lookup_order"} を渡してください （mcp__support__lookup_order に解決されます）。組み込みツールには "Bash" のような単独の文字列を渡します。 単独の作成済みツール名は、意図的にすべてのサーバーにある同名ツールとは一致しません。

作成した @tool の拒否は、JSON をコンテンツに含め、is_error: True を設定する

guard_tool は @tool の定義をラップするため、DENY 時（またはデフォルトの on_guard_error="deny" で Guard を評価できなかった時）にハンドラーは実行されません。 モデルには ArcjetDenialResult が content 上の JSON テキストとして is_error: True とともに渡されます。 structuredContent は設定しないでください（JS のみ）。例外を発生させないでください。例外を投げると未加工の例外になり、is_error を省略すると成功したように見えます。 #196 と同じフェイルクローズのデフォルトです。フェイルオープンになるのは "allow" だけで、DENY は常にブロックします。コアの guard() は引き続きフェイルオープンです。

can_use_tool はポリシーゲートではない

can_use_tool は人間が介在する仕組みです。allowed_tools、allow ルール、bypassPermissions／acceptEdits を使用すると、これを経由しません。 CrewAI の human_input、JS の canUseTool、LangGraph の interrupt()、OpenAI Agents の needs_approval と同じ落とし穴です。 受信テキスト用のヘルパーはありません。プロンプトインジェクションは guard_hooks の UserPromptSubmit で検査してください。

最初に人に確認する質問

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

1. どのツールがリスクを伴いますか（外部への副作用、取り消し不能、費用が発生する、メッセージを送信するなど）？それらには guard_tool を使います。組み込みツールとラップされていない MCP ツールには guard_hooks の PreToolUse を使います。
2. 制限はありますか？（例：「ユーザーごとに 1 分あたり 10 回の検索」→ TokenBucket。）
3. メタデータ上のユーザーは誰ですか。不透明なユーザー／テナント ID（PII は使用しません）ですか？デフォルトはなしです。セッション ID は相関 ID であり、ユーザーではありません。Claude SDK では session_id が UUID である必要があり、同じ ID を作成できるのは 1 回だけです。以降のターンでは resume を使います。
4. Arcjet が停止すると許容できませんか？すべてのヘルパーのデフォルトは on_guard_error="deny" です。受信側の UserPromptSubmit については、ここでフェイルクローズにするとエージェントが回答を停止します。そのため、この呼び出し箇所では "allow" を選ぶのが一般的で、妥当な選択肢であることを明確に確認してください。

読者が間違えやすい点

1. これは JS アダプターでも Claude Managed Agents でもありません。 @arcjet/guard/claude-agent-sdk/v0 や arcjet.guard.claude_managed_agents をインポートしないでください。
2. 受信側用のヘルパーはありません。 guard_hooks を使って UserPromptSubmit でスクリーニングします。
3. can_use_tool は HITL であり、ポリシーではありません。
4. Python は structuredContent を転送しません。 作成した拒否応答は JSON を content に含め、is_error: True とします。
5. session_id は UUID で、使用できるのは 1 回だけです。 以降のターンでは resume を使います。claude_agent_context が ID を生成することはありません。
6. すべての guard_tool ラッパーを exclude に列挙してください。 exclude がないと Guard が二重に呼び出されます。
7. レート制限のキーには、モデルが指定した注文 ID ではなく、認証済みの呼び出し元を使ってください。query／create_sdk_mcp_server には guard_tool が返すコピーを渡します。
8. guard() を使ってすべてのツールを直接ラップしないでください。
9. 分離には setting_sources=[] と strict_mcp_config=True の両方が必要です。片方だけでは、Guard を通らない経路が残ります。
10. 異なる ID が 2 つあります。 ClaudeAgentOptions.session_id は実行ごとに固有の UUID です（以降は resume）。Guard の session_id は長期間使うアクター ID です。判定がないことは拒否を意味しません。

ステップ 1：Guard クライアントをインストールして確認する

PyPI の arcjet 1.1.0 が必要です（arcjet[claude-agent-sdk] エクストラは 1.0.0 には含まれていません）。

pip install "arcjet[claude-agent-sdk]"

エージェントに Guard クライアントがまだない場合は、モジュールスコープで 1 回だけ起動します。

import os
from arcjet.guard import launch_arcjet

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

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

from claude_agent_sdk import ClaudeAgentOptions, create_sdk_mcp_server, query, tool
from arcjet.guard import DetectPromptInjection, TokenBucket, launch_arcjet
from arcjet.guard.claude_agent_sdk import guard_hooks, guard_tool

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
# Caller-owned UUID for this conversation. Later turns: resume=session_id.
session_id = conversation_id

@tool("lookup_order", "Look up an order by number", {"order_id": str})
async def lookup_order(args: dict) -> dict:
    return {
        "content": [{"type": "text", "text": f"{args['order_id']}: 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",
)
# can_use_tool=... is HITL — not this policy gate

ステップ 3：受信側をスクリーニングし、ラップされていないツールにゲートを設ける — guard_hooks

async for message in query(
    prompt=user_text,
    options=ClaudeAgentOptions(
        session_id=session_id,  # later turns: resume=session_id instead
        mcp_servers={
            "support": create_sdk_mcp_server(name="support", tools=[lookup_order]),
        },
        hooks=guard_hooks(
            guard=aj,
            session_id=session_id,
            exclude=[{"server": "support", "name": "lookup_order"}],
            inbound={
                "action": "message.received",
                "rules": lambda ctx: [inbound(ctx["prompt"])],
            },
            action="mcp.invoked",
            rules=lambda ctx: [mcp_limit(key=user_id, requested=1)],
            on_guard_error="deny",
        ),
    ),
):
    pass

ステップ 4：相関

claude_agent_context(session_id=session_id) は、呼び出し元が所有する同じ UUID（まずフックの session_id、次に options.session_id）を読み取ります。ID を生成することはありません。すでに持っている ID を guard_hooks と ClaudeAgentOptions に渡し、ターンごとに新しい ID を生成しないでください。Claude SDK でも session_id は UUID である必要があり、同じ ID を作成できるのは 1 回だけです。以降のターンでは resume を使います。

統合を確認する

1. python -m py_compile（またはプロジェクトの型チェック）が成功すること。
2. 受信側の PI（UserPromptSubmit によるブロック）、作成したツールの拒否応答（JSON を content に含め、is_error: True、structuredContent なし）、組み込みツール／ラップされていない MCP ツールの拒否応答（PreToolUse）、レート制限、フェイルクローズ（到達不能な Guard）をテストします。ラップしたツールの呼び出しごとに Guard の判定が1 回だけ行われることを確認します。PreToolUse アクションで 2 回目の判定が行われる場合は、exclude の項目が不足しています。
3. Arcjet Console／CLI で、判定が呼び出し元の UUID session_id を共有していることを確認します。
4. 実際の ARCJET_KEY を使った手動 E2E テストは、実行するまでは未検証です。

例の fastapi-claude-agent-sdk-guard は Runtime にそのまま残してください。新しい例の名前を作らないでください。この skills リポジトリに例を追加しないでください。