integrate-arcjet-guard-google-adk-py

Integrate Arcjet Guard into Python Google ADK

arcjet.guard.google_adk wraps the agent's existing Arcjet client. It never talks to the Arcjet API itself. Shared Guard fundamentals (client, rules, labels, decisions, capture, registration) live in ../arcjet/references/guards_python.md. Load that reference for anything that is not Google ADK-specific.

Official Python google-adk>=2.0.0,<3 only — not JS @google/adk (@arcjet/guard/google-adk/v2, docs https://docs.arcjet.com/guards/google-adk/), not @google/genai, not Go / Java ADK. Importing arcjet.guard.google_adk does not load LangChain, CrewAI, or the JS adapter. The extra is safe (no chromadb).

Exports: guard_tool, guard_plugin, google_adk_context. There is no inbound helper and no approval helper.

Three surfaces, one decision rule:

• An authored `LlmAgent` → guard_tool. Assign the returned

callback to LlmAgent(before_tool_callback=...). This is not a wrap around FunctionTool.

• Runner-wide / tools you did not attach a callback to →

guard_plugin. A Runner BasePlugin whose before_tool_callback is the run-wide gate. Put it first.

• Correlation → google_adk_context reads a caller-owned id.

It never mints. It never reads invocation_id. It never reads trace_id. It never reads session-service auto-ids.

DENY is a skip dict with arcjetDenied: true (ArcjetDenialResult, camelCase keys). ADK treats any mapping — including {} — as skip and uses the dict as the tool result, except {} is falsy in the callback chain and the tool would run. None is the only allow. Never return {} to allow. Do not raise — a throw is a plugin / callback error, not skip.

There is no /guards/google-adk-py/ docs page. Do not invent one. The JS adapter page is https://docs.arcjet.com/guards/google-adk/ — this skill is the Python teaching. Example: `examples/fastapi-google-adk-guard` on arcjet-py main. Do not invent a second example name.

The gate is before_tool_callback skip dict

Both helpers evaluate Guard and, on DENY or unevaluated Guard under the default on_guard_error="deny", return a deny dict so the original tool function never runs. Fail closed: always return that deny dict on error — do not return None (that executes the tool) and do not raise. On ALLOW they return None. Same fail-closed default as #196: only "allow" fails open; a DENY always blocks. Core guard() still fails open (has_failed_open()).

guard_tool is the positional callback (tool, args, tool_context). guard_plugin is the keyword-only plugin callback (*, tool, tool_args, tool_context). Same skip dict. guard_tool requires action. guard_plugin defaults action to "{tool_name}.invoked".

PluginManager is first-win on a returned dict. If another plugin returns a dict first, Guard never runs. On ALLOW the plugin returns None and a later agent before_tool_callback still runs — do not stack guard_plugin and guard_tool on the same tools or Guard is called twice.

require_confirmation is not a policy gate

require_confirmation / request_confirmation / adk_request_confirmation / ADK SecurityPlugin / confirmation resume is human-in-the-loop. After a human yes, Guard still runs on the tool call. Same trap as CrewAI human_input, JS requireConfirmation, LangGraph interrupt(), OpenAI Agents needs_approval, and Genkit interrupt(). There is no inbound helper and no approval helper.

Screen inbound before runner.run_async

There is no inbound helper. Agent / model callbacks (before_model_callback, before_run_callback, on_user_message_callback) are not this policy gate. Call aj.guard(...) in the application and act on the decision. Core guard() fails open: ALLOW is not proof the rules ran. Gate on decision.has_failed_open() if this call site must fail closed; guard_tool / guard_plugin already default to that.

actor / inputs — omit and remote rules never fire

A remote policy that declares actor or typed inputs only evaluates those values when this call sends them. Pass actor= and inputs= on guard_tool / guard_plugin (and on inbound aj.guard(...)). Build each input with server_input / local_input from arcjet.guard — that is the Python policyInput equivalent. Do not import JS policyInput and do not invent a second helper.

• Omit `actor` / `inputs` and a remote policy that requires

them never fires. The call still contacts Guard; the remote rules just have nothing to read.

• A resolver that throws is degraded and fail-closes under the

default on_guard_error="deny". The tool does not run.

• `actor` is the authenticated caller, never a model-produced

tool argument. An order id or email to= the model chose is not who is acting. Take actor from a signed session, verified token, or other server-side identity.

Resolvers see the tool-call envelope: model args plus tool_name (applied last so a tool argument of that name cannot hide the callback's name).

Questions to ask the human first

Ask only what you cannot infer from the code; suggest defaults.

1. Which tools are risky (external side effects, irreversible,

spends money, sends messages)? An agent you authored gets guard_tool. Runner-wide / unwrapped tools get guard_plugin.

1. What limits? (e.g. "10 lookups/min per user" →

TokenBucket.)

1. Who is the user for metadata — an opaque user/tenant ID

(never PII)? Default: none. That same authenticated id is actor. Put the conversation / session id you already have on guard_tool / guard_plugin (session_id=...) and on runner.run_async(..., session_id=...). That id is the correlation id, not the user. Do not use invocation_id or a session-service auto-id.

1. Is an Arcjet outage unacceptable? Every helper defaults to

on_guard_error="deny". Ask explicitly about inbound screening before runner.run_async: failing closed there means the run does not start, so "allow" is a routine and legitimate choice at that one call site.

The things readers get wrong

1. This is not JS `@arcjet/guard/google-adk/v2`. Import

arcjet.guard.google_adk. The JS adapter is plugin-only (no guardTool).

1. There is no inbound helper. Screen with core guard()

before runner.run_async.

1. require_confirmation / request_confirmation is HITL, not

policy.** Do not use SecurityPlugin as the Arcjet gate.

1. **Fail closed = always return a deny dict with arcjetDenied

on error.** Do not raise, do not return None, and never return {}.

1. `guard_tool` is a callback, not a FunctionTool wrap. Assign

it to LlmAgent(before_tool_callback=...).

1. **Do not stack guard_plugin and guard_tool on the same

tools.** Plugin ALLOW still reaches the agent callback.

1. Put Arcjet first on Runner(..., plugins=).
2. Correlation is read, never minted. Never invocation_id,

never trace_id, never session auto-ids.

1. Key rate limits on the authenticated caller, not a

model-supplied order id. Same for actor.

1. Omit `actor` / `inputs` and remote rules never fire. A

resolver throw fail-closes. Use server_input / local_input, not JS policyInput.

1. Do not hand-wrap every ADK tool with raw `guard()`.

Step 1: Install and find the guard client

Install the published extra. Use the package manager the project already uses — do not pin a git SHA:

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

The extra pulls google-adk>=2.0.0,<3. If the agent has no guard client yet, launch one once at module scope:

import os
from arcjet.guard import launch_arcjet

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

Step 2: Gate an authored agent — guard_tool

Build the callback (and the agent) from the current request's authenticated caller. Do not freeze user_id / conversation_id at import time.

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",
    ),
)

Use action + rules on guard_tool. action may be a function of the tool-call envelope (tool_name plus input). Empty rules still contacts Guard. Key rate limits on the authenticated caller, not a model-supplied order id.

Step 3: Gate the Runner — guard_plugin

Use this for tools you did not attach guard_tool to. Put Arcjet first. Do not also assign guard_tool on those same agents.

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",
        ),
    ],
)

Step 4: Screen inbound before runner.run_async

google_adk_context accepts sessionId or session_id (same for correlationId / conversationId). CamelCase is not required.

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

There is no inbound helper.

Step 5: Correlation

google_adk_context reads a caller-owned id. Preference: correlationId / correlation_id, then sessionId / session_id, then conversationId / conversation_id on the object (or a bare mapping), then the same names on application-owned state, then correlation_id= / session_id= / conversation_id= kwargs, then an enclosing arcjet_sequence. It never mints an id. It never reads invocation_id (ADK always generates it). It never reads trace_id. It never reads toolContext.sessionId / session.id (session auto-ids) and never walks into .session. Do not invent a correlation id per turn. Put the same id on the helper and on runner.run_async(...). If nothing valid remains, the call is uncorrelated rather than joined to a generated id.

Verify the integration

1. python -m py_compile (or the project's type-check) passes.
2. Exercise inbound PI (before run_async, including

has_failed_open()), a guard_tool deny-dict skip (arcjetDenied: true), a guard_plugin deny-dict skip, None execute, first-plugin short-circuit (Arcjet first), no-raise, never-{}, never-mint, omitted actor / inputs (remote rules silent), a resolver throw (fail-closed deny dict), and fail-closed (an unreachable guard → deny dict, never None). Confirm require_confirmation is never treated as the gate and that stacking both helpers does not double-call Guard on ALLOW.

1. Confirm in the Arcjet Console / CLI that decisions share the

caller-owned session / conversation id — not an invocation_id or session auto-id — and that actor is the authenticated caller.

1. Manual E2E with a real ARCJET_KEY is still-to-verify until you

run it.

Worked example: `examples/fastapi-google-adk-guard` on arcjet-py main. Do not invent a second example name. Do not add an example in this skills repo.