integrate-arcjet-guard-google-adk-py

将 Arcjet Guard 集成到 Python Google ADK

arcjet.guard.google_adk 封装 Agent 现有的 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。没有入站辅助函数，也没有审批辅助函数。

三种接入方式，一条判定规则：

• 自行编写的 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，也不读取 session-service 自动生成的 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，之后 Agent 的 before_tool_callback 仍会运行——不要在同一批工具上同时使用 guard_plugin 和 guard_tool，否则 Guard 会被调用两次。

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 之前筛查入站请求

没有入站辅助函数。Agent／模型回调 （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= 邮箱地址并不能代表操作者。请从签名会话、已验证令牌或其他服务端身份信息中获取 actor。

解析器会看到工具调用信封：模型参数加上 tool_name（最后应用，因此工具参数中同名的值无法覆盖回调的名称）。

首先要向用户确认的问题

只询问无法从代码推断的信息；同时建议默认值。

1. 哪些工具风险较高（会产生外部副作用、不可逆、花费资金或发送消息）？你编写的 Agent 使用 guard_tool。Runner 范围内／未包装的工具使用 guard_plugin。
2. 有哪些限制？（例如“每位用户每分钟查询 10 次” → TokenBucket。）
3. 元数据中的用户是谁——不透明的用户／租户 ID（绝不使用 PII）？默认：无。同一个经过身份验证的 ID 也是 actor。将已有的会话／会话 ID 传给 guard_tool／guard_plugin（session_id=...），并传给 runner.run_async(..., session_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，而非策略。 不要使用 SecurityPlugin 作为 Arcjet 关卡。
4. 失败时拒绝 = 出错时始终返回带有 arcjetDenied 的拒绝字典。 不要抛出异常，不要返回 None，也绝不要返回 {}。
5. guard_tool 是回调，不是 FunctionTool 包装。 将它赋给 LlmAgent(before_tool_callback=...)。
6. 不要对同一批工具同时使用 guard_plugin 和 guard_tool。 插件的 ALLOW 仍会传递到 Agent 回调。
7. 在 Runner(..., plugins=) 中将 Arcjet 放在第一位。
8. 只读取关联 ID，绝不自行生成。 绝不使用 invocation_id、trace_id 或会话自动生成的 ID。
9. 速率限制应以经过身份验证的调用者为键，而不是模型提供的订单 ID。actor 也一样。
10. 省略 actor／inputs 后，远程规则永远不会触发。 解析器抛出异常会导致失败时拒绝。使用 server_input／local_input，而不是 JS policyInput。
11. 不要给每个 ADK 工具都手动套上原始 guard()。

步骤 1：安装并查找 guard 客户端

安装已发布的扩展。使用项目当前已有的包管理器，不要固定 git SHA：

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

该扩展会引入 google-adk>=2.0.0,<3。如果 Agent 还没有 guard 客户端，请在模块作用域仅启动一次：

import os
from arcjet.guard import launch_arcjet

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

步骤 2：为自定义 Agent 设置防护 — guard_tool

根据当前请求经过身份验证的调用方构建回调（以及 Agent）。不要在导入时固定 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 放在首位。不要再为这些相同的 Agent 指定 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。它不会读取 invocation_id（ADK 始终会生成该 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 拒绝字典的跳过行为 （arcjetDenied: true）、guard_plugin 拒绝字典的跳过行为、None 执行、首个插件短路（Arcjet 位于首位）、不抛出异常、 绝不返回 {}、绝不生成 ID、未提供 actor / inputs（远程 规则静默处理）、解析器抛出异常（失败关闭，返回拒绝字典），以及 失败关闭（Guard 无法访问 → 拒绝字典，绝不返回 None）。 确认 require_confirmation 绝不会被视为防护条件， 并且两个辅助函数叠加使用时，ALLOW 不会导致 Guard 被重复调用。
3. 在 Arcjet Console / CLI 中确认各项决策共用调用方提供的会话 / 对话 ID，而不是 invocation_id 或会话自动 ID；并确认 actor 是经过身份验证的调用方。
4. 使用真实 ARCJET_KEY 进行的手动端到端测试仍待验证，直到你实际运行它。

示例： arcjet-py main 分支上的 examples/fastapi-google-adk-guard。不要另创第二个示例名称。不要在此 skills 仓库中添加示例。