integrate-arcjet-guard-claude-managed-agents-py

Python Claude Managed Agents に Arcjet Guard を統合する

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

公式の anthropic>=0.92.0,<2 のみが対象です。claude-agent-sdk、 arcjet.guard.claude_agent_sdk、JS の @anthropic-ai/claude-agent-sdk アダプター、JS Claude Managed Agents（@arcjet/guard/claude-managed-agents/v0、ドキュメント https://docs.arcjet.com/guards/claude-managed-agents/）は対象外です。 arcjet.guard.claude_managed_agents をインポートしても、Claude Agent SDK、 LangChain、CrewAI は読み込まれません。追加パッケージは安全に使用できます（chromadb は含まれません）。

これはホスト型 Claude Managed Agents ハーネス（client.beta.sessions）です。Anthropic がエージェントのループと組み込みツールセット（bash、files、web_*）を実行します。エージェントのツールセットはデフォルトで always_allow なので、bash/files に対する顧客側の実行前処理はありません。agent.tool_use / agent.tool_result が発生するのは、組み込みツールが実行された後です。PreToolUse はありません。このギャップを always_ask でごまかさないでください。これは Claude Agent SDK のローカルな query() / guard_hooks ではありません。

エクスポート：guard_custom_tool、guard_events、 claude_managed_agents_context。guard_inbound、guard_tool、 guard_tool_confirmation はありません。

3 つの対象に、1 つの判定ルールを適用します。

• 受信テキスト（user.message / initial_events）→ guard_events が sessions.events.send をラップします。DENY の場合は ArcjetDeniedError / ArcjetUnavailableError を送出し、send は呼び出しません。
• 自分で実行するカスタムツール（agent.custom_tool_use）→ guard_custom_tool(run=…)。DENY の場合、元の run は実行されません。 ヘルパーはスキーマフィールド is_error を指定して user.custom_tool_result を送信します。
• 相関付け → claude_managed_agents_context は呼び出し元が所有する correlation_id / session_id を読み取ります。ID を生成することはありません。 Anthropic の session.id / sevt_… を読み取ることもありません。

ドキュメント：https://docs.arcjet.com/guards/claude-managed-agents/（JS と Python 共通のページ）。/guards/claude-agent-sdk/ は使用しないでください。個別の /guards/claude-managed-agents-py/ ページはありません。JS アダプターは @arcjet/guard 1.12.0 で提供されます。

実際に判定を適用するのは受信 user.message とカスタムツールです

guard_events(send=client.beta.sessions.events.send, …) は send 呼び出し可能オブジェクトをラップするため、元の send が実行される前に user.message / initial_events が評価されます。これは、ホスト型ハーネスがプロンプトを読み取る前にターンを拒否できる唯一の場所です。受信 rules には message_arguments() から {"prompt", "content", "type"} が渡されます。JS の { text, events } ではありません。guard_custom_tool(run=…) は await handler(event, send=…, anthropic_session_id=…) を返します。ハンドラーは anthropic_session_id（Anthropic の ses_…）を user.custom_tool_result の送信にのみ使用します。ラップ時に指定する session_id= は呼び出し元が所有する相関 ID です。組み込みツールがこのハンドラーに渡されることはありません。省略可能な tool= を指定すると、セルフホスト型の @beta_tool の run も同じ方法でラップします。CLI ワーカーはカスタムツールを登録できません。

カスタムツールの拒否は is_error 付きの user.custom_tool_result です

DENY（またはデフォルトの on_guard_error="deny" における Guard の未評価）では、元の run は呼び出されません。ヘルパーは実際の user.custom_tool_result（custom_tool_use_id、content 上の ArcjetDenialResult の JSON、is_error を送信します。このフィールドは events スキーマにあります。別のフィールドを追加しないでください）。ホスト型ハンドラーから例外を送出しないでください。例外を送出すると、セッションがアイドル状態のままになります。is_error を省略すると成功したように見えます。これは Claude Agent SDK の structuredContent ではありません。 #196 と同じフェイルクローズのデフォルトです。コアの guard() は引き続きフェイルオープンです。

always_ask + user.tool_confirmation はポリシーとしての HITL ではありません

権限ポリシーが適用されるのはエージェントのツールセットと MCP であり、カスタムツールには適用されません。CrewAI の human_input、JS の canUseTool、LangGraph の interrupt() と同じ落とし穴です。MCP Guard が適用されるのは、自分でホストするサーバーのみです。Anthropic は MCP クライアントです。web_search / web_fetch は常に Anthropic で実行されます。

まず人に確認する質問

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

1. アプリが agent.custom_tool_use で実行するカスタムツールはどれですか？それらには guard_custom_tool を適用します。always_allow の組み込みツールには適用できません。
2. 上限は何ですか？（例：「ユーザーごとに毎分10回のルックアップ」→ TokenBucket。）
3. メタデータ上のユーザーは誰ですか。PII ではない、不透明なユーザー／テナント ID ですか？デフォルトはなしです。Anthropic の session.id / sevt_… は相関 ID ではありません。
4. Arcjet の停止は許容できませんか？すべてのヘルパーのデフォルトは on_guard_error="deny" です。インバウンドの guard_events については明示的に確認してください。そこでフェイルクローズにするとターンが送信されないため、その呼び出し箇所に限っては "allow" も一般的で妥当な選択肢です。

読者が誤解しやすい点

1. これは Claude Agent SDK ではありません。 guard_tool、guard_hooks、PreToolUse、UserPromptSubmit はありません。arcjet.guard.claude_agent_sdk または @arcjet/guard/claude-agent-sdk/v0 でもラップしないでください。
2. guard_inbound はありません。 ヘルパーは guard_events です。
3. インバウンドの ctx は {"prompt", "content", "type"} です。 JS の { text, events } ではありません。
4. 相関 ID は呼び出し元が管理し、決して生成されません。 Anthropic の session.id / sevt_… / trace_id を相関 ID として渡さないでください。
5. デフォルトの always_allow は制御できません。 Anthropic のクラウド上の bash／ファイル操作をブロックできると主張しないでください。
6. カスタムツールが DENY された場合は、is_error を指定して user.custom_tool_result を送信してください。 例外を発生させないでください。ALLOW の場合、呼び出し元は成功結果を送信しなければならず、そうしなければセッションはアイドル状態になります。
7. always_ask + user.tool_confirmation はオプトインの確認であり、ポリシーではありません。
8. すべてのセッションイベントを生の guard() で個別にラップしないでください。
9. guard_events に inbound= はありません。 action と rules はトップレベルに置き、send= を渡します。返される呼び出し可能オブジェクトは send を置き換え、DENY の場合は例外を発生させます。guard_custom_tool が受け取るのは run= であり、tool=<function> ではありません。判定がないことは拒否を意味しません。

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

PyPI の arcjet 1.1.0 が必要です（この extra は 1.0.0 には含まれていません）。

pip install "arcjet[claude-managed-agents]"

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

import os
from arcjet.guard import launch_arcjet

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

動作する例： examples/fastapi-claude-managed-agents-guard。

ステップ 2：sessions.events.send の前にインバウンドをスクリーニングする

from anthropic import Anthropic
from arcjet.guard import DetectPromptInjection, TokenBucket, launch_arcjet
from arcjet.guard.claude_managed_agents import (
    claude_managed_agents_context,
    guard_custom_tool,
    guard_events,
)

client = Anthropic()
aj = launch_arcjet(key=os.environ["ARCJET_KEY"])
inbound = DetectPromptInjection()
user_id = authenticated_user_id
conversation_id = authenticated_conversation_id
derived = claude_managed_agents_context(session_id=conversation_id)

send = guard_events(
    guard=aj,
    send=client.beta.sessions.events.send,
    action="message.received",
    # Python inbound ctx is {"prompt", "content", "type"} — not JS { text, events }.
    rules=lambda ctx: [inbound(ctx["prompt"])],
    session_id=derived.correlation_id or conversation_id,
    on_guard_error="deny",
)

session = client.beta.sessions.create(agent=agent_id, environment_id=environment_id)
# Anthropic minted session.id — pass it to the sessions API, not as correlation.

# Screen user.message before the hosted harness reads it. DENY raises
# ArcjetDeniedError / ArcjetUnavailableError and does not send.
await send(
    session.id,
    events=[{"type": "user.message", "content": [{"type": "text", "text": user_text}]}],
)

ステップ 3：agent.custom_tool_use でカスタムツールを制御する

lookup_limit = TokenBucket(
    label="order.looked-up",
    bucket="lookups",
    refill_rate=10,
    interval_seconds=60,
    max_tokens=10,
)

async def lookup_order(event) -> dict:
    order_id = event.input["order_id"]
    return {"content": [{"type": "text", "text": f"{order_id}: shipped"}]}

handle_lookup = guard_custom_tool(
    guard=aj,
    action="order.looked-up",
    run=lookup_order,
    rules=[lookup_limit(key=user_id, requested=1)],
    session_id=derived.correlation_id or conversation_id,
    on_guard_error="deny",
)

with client.beta.sessions.events.stream(session_id=session.id) as stream:
    for event in stream:
        if event.type == "agent.custom_tool_use" and event.name == "lookup_order":
            # DENY posts user.custom_tool_result with is_error. Do not raise.
            await handle_lookup(
                event,
                send=client.beta.sessions.events.send,
                anthropic_session_id=session.id,
            )
        # agent.tool_use / always_ask + user.tool_confirmation are not this policy gate

モデルが指定した注文 ID ではなく、認証済みの呼び出し元に対してレート制限を設定します。ALLOW の場合、呼び出し元は成功した user.custom_tool_result を引き続き送信します。

ステップ 4：相関

claude_managed_agents_context は ClaudeManagedAgentsContext（データクラス）を返す読み取り関数です。コンテキストマネージャーでも、contextvar セッターでもありません。呼び出し元が管理する correlation_id / session_id を読み取ります。生成することはありません。Anthropic の id / event_id / session.id（ses_…）/ sevt_… / trace_id を読み取ることもありません。Anthropic の Session オブジェクトを渡しても安全です。生成された ID は無視されます。無効な候補値はスキップされ、有効な値が残らなければ、生成された ID との紐付けは行われず、相関なしになります。ヘルパーは内部でも session_id= / correlation_id= を再読み込みします。

統合を検証する

1. python -m py_compile（またはプロジェクトの型チェック）が成功することを確認します。
2. インバウンド PI（送信が呼び出されない）、カスタムツールの拒否 （is_error を含む user.custom_tool_result。run は呼び出されない）、レート制限、 フェイルクローズ（到達不能なガード）をテストします。 always_ask / user.tool_confirmation がゲートとして扱われず、 組み込みの agent.tool_use が監視のみであることを確認します。
3. Arcjet Console / CLI で、判定が呼び出し元所有の会話 ID を共有していることを確認します。 Anthropic の session.id / sevt_… ではありません。
4. 実際の ARCJET_KEY を使用した手動 E2E は、実行するまで未検証のままです。

実装例： examples/fastapi-claude-managed-agents-guard。 2つ目の実装例の名前を勝手に作らないでください。この skills リポジトリに実装例を追加しないでください。