integrate-arcjet-guard-google-adk-py

Python Google ADK に Arcjet Guard を統合する

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

公式の Python google-adk>=2.0.0,<3 のみが対象です。JS の @google/adk （@arcjet/guard/google-adk/v2、ドキュメント https://docs.arcjet.com/guards/google-adk/）、`@google/genai`、 Go／Java ADK は対象外です。arcjet.guard.google_adk をインポートしても、 LangChain、CrewAI、JS アダプターは読み込まれません。この追加パッケージは安全です（ chromadb は含まれません）。

エクスポート：guard_tool、guard_plugin、google_adk_context。インバウンドヘルパーも承認ヘルパーもありません。

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

• 作成した LlmAgent → guard_tool。返されたコールバックを LlmAgent(before_tool_callback=...) に設定します。これは FunctionTool を ラップするものではありません。
• Runner 全体／コールバックを設定していないツール → guard_plugin。Runner の BasePlugin が、実行全体に適用される before_tool_callback ゲートになります。最初に配置してください。
• 相関付け → google_adk_context は呼び出し元が管理する ID を読み取ります。 ID を生成することはありません。invocation_id、trace_id、 セッションサービスが自動生成する ID は読み取りません。

DENY は arcjetDenied: true を含むスキップ用ディクショナリです（ArcjetDenialResult、 キーはキャメルケース）。ADK は {} を含むあらゆるマッピングをスキップとして扱い、 そのディクショナリをツールの結果として使用します。ただし、コールバックチェーンでは {} は偽値になるため、ツールが実行されてしまいます。許可を示す値は None だけです。 許可を示すために {} を返してはいけません。例外を送出してはいけません。送出すると スキップではなく、プラグイン／コールバックのエラーになります。

/guards/google-adk-py/ というドキュメントページはありません。存在しないページを 作らないでください。JS アダプターページは https://docs.arcjet.com/guards/google-adk/ です。このスキルが Python の解説を担います。例： arcjet-py の main にある examples/fastapi-google-adk-guard。 別の例の名前を作らないでください。

ゲートは before_tool_callback のスキップ用ディクショナリです

どちらのヘルパーも Guard を評価し、DENY の場合、またはデフォルトの on_guard_error="deny" のもとで Guard の評価が完了しなかった場合には、拒否用 ディクショナリを返すため、元のツール関数は実行されません。フェイルクローズにしてください。 エラー時には必ず拒否用ディクショナリを返してください。None を返すと（ツールが 実行されるため）、また例外を送出してもいけません。ALLOW の場合は None を返します。 これは #196 と同じフェイルクローズの デフォルトです。フェイルオープンになるのは "allow" だけで、DENY は常にブロックします。 コアの guard() は引き続きフェイルオープンです（has_failed_open()）。

guard_tool は位置引数を取るコールバック (tool, args, tool_context) です。guard_plugin はキーワード専用のプラグインコールバック (*, tool, tool_args, tool_context) です。スキップ用ディクショナリは同じです。 guard_tool には action が必要です。guard_plugin の action のデフォルトは "{tool_name}.invoked" です。

PluginManager は、最初に返されたディクショナリを採用します。別のプラグインが先に ディクショナリを返すと、Guard は実行されません。ALLOW の場合、プラグインは None を 返すため、後続のエージェントの before_tool_callback も実行されます。同じツールに guard_plugin と guard_tool を重ねて設定しないでください。Guard が 2 回呼び出されます。

require_confirmation はポリシーゲートではありません

require_confirmation／request_confirmation／ adk_request_confirmation／ADK の SecurityPlugin／確認後の再開は、ヒューマン・イン・ザ・ループの仕組みです。 人間が「はい」と答えた後も、ツールの呼び出し時に Guard は実行されます。これは CrewAI の human_input、JS の requireConfirmation、LangGraph の interrupt()、 OpenAI Agents の needs_approval、Genkit の interrupt() と同じ落とし穴です。 インバウンドヘルパーも承認ヘルパーもありません。

runner.run_async の前にインバウンドをスクリーニングする

インバウンド用のヘルパーはありません。エージェント／モデルのコールバック （before_model_callback、before_run_callback、 on_user_message_callback）はこのポリシーゲートではありません。アプリケーションで aj.guard(...) を呼び出し、その判断に基づいて対応してください。 コアの guard() はフェイルオープンします。ALLOW はルールが実行された証拠では ありません。この呼び出し箇所でフェイルクローズが必要な場合は、 decision.has_failed_open() で判定してください。guard_tool ／ guard_plugin は デフォルトでそのように動作します。

actor ／ inputs — 省略するとリモートルールは実行されません

actor または型付きの inputs を宣言するリモートポリシーは、この呼び出しでそれらの値を 送信した場合にのみ評価されます。guard_tool ／ guard_plugin（およびインバウンドの aj.guard(...)）に actor= と inputs= を渡してください。各入力は arcjet.guard の server_input ／ local_input を使って構築します。これは Python 版の policyInput に相当します。JS の policyInput はインポートせず、別のヘルパーを 独自に作らないでください。

• actor ／ inputs を省略すると、それらを必要とするリモートポリシーは実行されません。 呼び出し自体は Guard に接続しますが、リモートルールが読み取れる値がありません。
• **リゾルバーが例外をスローすると、デグレード状態となり、**デフォルトの on_guard_error="deny" ではフェイルクローズします。ツールは実行されません。
• **actor は認証済みの呼び出し元です。**モデルが生成したツール引数ではありません。 モデルが選んだ注文 ID やメールの to= は、実行者を示すものではありません。署名済み セッション、検証済みトークン、またはその他のサーバー側の ID 情報から actor を取得してください。

リゾルバーにはツール呼び出しのエンベロープが渡されます。内容はモデルの引数と tool_name です（コールバックの名前をツール引数で隠せないように、最後に適用されます）。

まず人に確認すること

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

1. どのツールがリスクを伴いますか（外部への副作用、取り消し不能、金銭の消費、メッセージ送信）？自分で作成したエージェントには guard_tool を使います。Runner 全体のツールやラップされていないツールには guard_plugin を使います。
2. どのような制限が必要ですか？（例：「ユーザーごとに 1 分あたり 10 回の検索」→ TokenBucket。）
3. メタデータでのユーザーとは誰ですか（PII ではない、不透明なユーザー／テナント ID）？デフォルトは指定なしです。同じ認証済み ID が actor になります。すでにある会話／セッション ID を guard_tool ／ guard_plugin に （session_id=...）、また runner.run_async(..., session_id=...) に渡してください。この ID は相関 ID であり、ユーザー ID ではありません。invocation_id やセッションサービスが自動生成した ID は使わないでください。
4. Arcjet の停止は許容できませんか？すべてのヘルパーのデフォルトは on_guard_error="deny" です。runner.run_async の前のインバウンドスクリーニングについては、必ず明示的に確認してください。そこでフェイルクローズすると実行が開始されないため、その呼び出し箇所に限っては "allow" も日常的かつ妥当な選択肢です。

読者が間違えやすいこと

1. **これは JS の @arcjet/guard/google-adk/v2 ではありません。**インポートするのは arcjet.guard.google_adk です。JS アダプターはプラグイン専用で、 guardTool はありません。
2. インバウンド用のヘルパーはありません。runner.run_async の前にコアの guard() でスクリーニングしてください。
3. **require_confirmation ／ request_confirmation は HITL であり、ポリシーではありません。**Arcjet のゲートとして SecurityPlugin を使わないでください。
4. **フェイルクローズとは、エラー時に arcjetDenied を含む deny 辞書を必ず返すことです。**例外を発生させず、None を返さず、{} は絶対に返さないでください。
5. guard_tool はコールバックであり、FunctionTool をラップするものではありません。LlmAgent(before_tool_callback=...) に割り当ててください。
6. **同じツールに guard_plugin と guard_tool を重ねて適用しないでください。**プラグインが ALLOW を返しても、エージェントのコールバックは実行されます。
7. Runner(..., plugins=) では Arcjet を先頭に配置してください。
8. 相関 ID は読み取るもので、生成するものではありません。invocation_id、trace_id、セッションの自動生成 ID は使わないでください。
9. レート制限のキーには、モデルが指定した注文 ID ではなく、認証済みの呼び出し元を使ってください。actor も同様です。
10. **actor ／ inputs を省略すると、リモートルールは実行されません。**リゾルバーが例外をスローするとフェイルクローズします。JS の policyInput ではなく、server_input ／ local_input を使ってください。
11. ADK のすべてのツールを、guard() で個別にラップしないでください。

ステップ 1：guard クライアントをインストールして取得する

公開済みの追加パッケージをインストールします。プロジェクトですでに使用しているパッケージマネージャーを使用し、git SHA は固定しないでください。

pip install "arcjet[google-adk]"
uv add "arcjet[google-adk]"

この追加パッケージは google-adk>=2.0.0,<3 を取り込みます。エージェントにまだ guard クライアントがない場合は、モジュールスコープで 一度だけ 起動します。

import os
from arcjet.guard import launch_arcjet

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

ステップ 2：作成したエージェントを保護する — guard_tool

コールバック（およびエージェント）は、現在のリクエストの 認証済み呼び出し元から作成します。インポート時に user_id / conversation_id を固定しないでください。

from google.adk.agents import LlmAgent
from google.adk.tools import FunctionTool
from arcjet.guard import TokenBucket, launch_arcjet, server_input
from arcjet.guard.google_adk import google_adk_context, guard_plugin, 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,
)
# Per request: authenticated caller + caller-owned conversation id.
user_id = authenticated_user_id
conversation_id = authenticated_conversation_id

def lookup_order(order_id: str) -> dict:
    """Look up an order by ID."""
    return {"order_id": order_id, "status": "shipped"}

# require_confirmation=True is HITL — not this policy gate
lookup = FunctionTool(func=lookup_order)

agent = LlmAgent(
    name="support_agent",
    description="Help the user.",
    instruction="Help the user.",
    tools=[lookup],
    before_tool_callback=guard_tool(
        guard=aj,
        action="order.looked-up",
        # Authenticated caller — never a model-produced order id.
        actor=user_id,
        inputs=lambda call: {
            "order_id": server_input.string(str(call.get("order_id", ""))),
        },
        rules=[lookup_limit(key=user_id, requested=1)],
        session_id=conversation_id,
        on_guard_error="deny",
    ),
)

guard_tool では action と rules を使用します。action はツール呼び出しのエンベロープ（tool_name と input）を受け取る関数にできます。rules が空でも Guard には接続します。レート制限のキーには、モデルから渡された注文 ID ではなく、認証済みの呼び出し元を使用します。

ステップ 3：Runner を保護する — guard_plugin

guard_tool を設定していないツールには、こちらを使用します。Arcjet を 最初に 配置してください。同じエージェントに guard_tool も設定しないでください。

from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService

session_service = InMemorySessionService()
runner = Runner(
    app_name="support",
    agent=agent,
    session_service=session_service,
    # Arcjet first: a deny dict skips the tool before later plugins run.
    plugins=[
        guard_plugin(
            guard=aj,
            action=lambda call: f"{call['tool_name']}.invoked",
            actor=user_id,
            inputs=lambda call: {
                "tool": server_input.string(str(call.get("tool_name", ""))),
            },
            rules=[lookup_limit(key=user_id, requested=1)],
            session_id=conversation_id,
            on_guard_error="deny",
        ),
    ],
)

ステップ 4：runner.run_async の前に受信データをスクリーニングする

google_adk_context は sessionId または session_id を受け付けます（correlationId / conversationId も同様です）。キャメルケースは必須ではありません。

from google.genai import types
from arcjet.guard import DetectPromptInjection

inbound = DetectPromptInjection()
app_context = {"sessionId": conversation_id}
derived = google_adk_context(app_context)
decision = await aj.guard(
    label="message.received",
    actor=user_id,
    inputs={"content": server_input.string(user_text)},
    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")

async for event in runner.run_async(
    user_id=user_id,
    session_id=conversation_id,
    new_message=types.Content(
        role="user",
        parts=[types.Part(text=user_text)],
    ),
):
    _ = event

受信データ用のヘルパーはありません。

ステップ 5：相関

google_adk_context は呼び出し元が所有する ID を読み取ります。優先順位は次のとおりです。オブジェクト（または単独のマッピング）上の correlationId / correlation_id、次に sessionId / session_id、次に conversationId / conversation_id、アプリケーションが所有する state 上の同じ名前、次に correlation_id= / session_id= / conversation_id= キーワード引数、最後に外側の arcjet_sequence です。ID を生成することはありません。ADK が常に生成する invocation_id は読み取りません。trace_id も読み取りません。toolContext.sessionId / session.id（セッションの自動生成 ID）も読み取らず、.session の中も探索しません。ターンごとに相関 ID を新たに作らないでください。同じ ID をヘルパーと runner.run_async(...) の両方に設定します。有効な ID が何も残らない場合、生成された ID と結び付けず、相関なしの呼び出しになります。

インテグレーションを検証する

1. python -m py_compile（またはプロジェクトの型チェック）が成功すること。
2. 受信データの PI（run_async の前。has_failed_open() を含む）、guard_tool による deny dict のスキップ（arcjetDenied: true）、guard_plugin による deny dict のスキップ、None の実行、最初のプラグインによるショートサーキット（Arcjet を最初に配置）、例外を発生させないこと、{} を返さないこと、ID を生成しないこと、actor / inputs が省略された場合（リモートルールは何もせず通過）、リゾルバーが例外を投げる場合（フェイルクローズで deny dict）、およびフェイルクローズ（Guard に到達できない場合は deny dict を返し、None は返さないこと）を確認します。 require_confirmation がゲートとして扱われないこと、また両方のヘルパーを重ねても ALLOW 時に Guard が二重に呼び出されないことを確認します。
3. Arcjet Console / CLI で、判定が invocation_id やセッションの自動生成 ID ではなく、呼び出し元が所有するセッション / 会話 ID を共有し、actor が認証済みの呼び出し元であることを確認します。
4. 実際の ARCJET_KEY を使用した手動の E2E は、実行するまでは未検証です。

実装例： examples/fastapi-google-adk-guard は arcjet-py main にあります。別の例の名前を作らないでください。この skills リポジトリに例を追加しないでください。