integrate-arcjet-guard-strands-agents-py

将 Arcjet Guard 集成到 Python Strands Agents

arcjet.guard.strands_agents 封装 Agent 现有的 Arcjet 客户端。 它本身不会与 Arcjet API 通信。通用 Guard 基础知识 （客户端、规则、标签、决策、捕获、注册）见 ../arcjet/references/guards_python.md。 凡是与 Strands 无关的内容，请查阅该参考文档。

仅支持官方 Python strands / strands-agents>=1.11.0,<2，不支持 JS @arcjet/guard/strands-agents/v1（文档： https://docs.arcjet.com/guards/strands-agents/），也不支持社区分支。 导入 arcjet.guard.strands_agents 不会加载 LangChain、CrewAI 或 JS @arcjet/guard/strands-agents/v1。与 CrewAI（无需额外依赖）不同， 此版本提供 arcjet[strands-agents] extra（pyproject.toml 中的 strands-agents = ["strands-agents>=1.11.0,<2"]）。 该 extra 是安全的（不包含 chromadb）。

导出项：guard_tool、guard_hooks、strands_agent_context。 用于自定义 @tool，以及 Agent(hooks=) / add_hook。

三个入口，同一条决策规则：

• 自定义的 @tool → guard_tool。将其返回的副本交给 Agent； 原始工具仍不受 Guard 保护。
• 未封装的工具 / MCP 工具 → guard_hooks。通过每个工具的 BeforeToolCallEvent.cancel_tool（True 或 str）进行拦截。 已封装的工具会跳过，因此 Guard 不会被调用两次。
• 关联 → strands_agent_context 从 invocation_state 读取调用方持有的 ID。 它不会生成 ID，也不会读取 trace_id。

文档：https://docs.arcjet.com/guards/strands-agents/。示例： examples/fastapi-strands-agents-guard。 不要另行杜撰示例名称。

拦截点是每个工具的 BeforeToolCallEvent.cancel_tool

guard_hooks 会在该事件上注册处理器，因此遇到 DENY （或默认 on_guard_error="deny" 时 Guard 未评估）便不会运行工具。字符串是 取消消息（ArcjetDenialResult 的 JSON）；True 使用 Strands 的默认消息并丢弃 字段。发生错误时必须关闭失败：始终设置 cancel_tool，不要留空，也不要抛出异常。 默认关闭失败的行为与 #196 相同： 只有 "allow" 会在失败时放行；DENY 始终会拦截。核心 guard() 仍会在失败时放行（has_failed_open()）。

event.interrupt() 不是策略拦截点

BeforeToolCallEvent.interrupt() / 恢复流程用于人机协同。这与 CrewAI human_input、JS humanInTheLoopMiddleware、LangGraph interrupt()、OpenAI Agents needsApproval 和 Genkit interrupt() 一样，容易造成误解。 没有入站辅助函数，也没有审批辅助函数。

在 Agent(...) / __call__ / stream_async 之前筛查入站请求

没有入站辅助函数。在应用中调用 aj.guard(...)，并且 根据决策采取措施。核心 guard() 会在失败时放行：ALLOW 并不能证明 规则已运行。如果此调用点必须关闭失败，请检查 decision.has_failed_open()； guard_tool / guard_hooks 已默认采用此设置。

先向用户确认的问题

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

1. 哪些工具属于高风险（有外部副作用、不可逆、会花钱、会发送消息）？这些工具应使用 guard_tool。 你未编写的 MCP / 未封装工具应使用 guard_hooks。
2. 有哪些限制？（例如，“每个用户每分钟查找 10 次” → TokenBucket。）
3. 元数据中的用户是谁——不透明的用户 / 租户 ID（绝不能是 PII）？ 默认：无。将已有的对话 / 会话 ID 同时放入 invocation_state 和 guard_hooks(...)。该 ID 是关联 ID，不是用户 ID。
4. Arcjet 服务中断是否不可接受？所有辅助函数默认均为 on_guard_error="deny"。请明确询问是否要在 Agent 运行前进行入站筛查： 此处关闭失败意味着 Agent 不会运行，因此在这个调用点选择 "allow" 是常见且合理的做法。

读者容易误解的地方

1. 这不是 JS @arcjet/guard/strands-agents/v1。 请导入 arcjet.guard.strands_agents。
2. 没有入站辅助函数。 在运行 Agent(...) / __call__ / stream_async 之前， 使用核心 guard() 进行筛查。
3. event.interrupt() 用于 HITL，不是策略。 拒绝时使用 cancel_tool（True 或 str）。
4. 故障时默认拒绝 = 出错时始终设置 cancel_tool。 不要抛出异常， 也不要让它保持未设置状态。
5. 关联 ID 只读取，不生成。 绝不使用 trace_id、agent.id 或 SessionManager 自动生成的 ID。
6. guard_hooks 会跳过已包装的工具，以免重复调用 Guard。请将 guard_tool 返回的副本交给 Agent。
7. 速率限制应以通过身份验证的调用方为键，而不是模型提供的 订单 ID。
8. 不要对每个 Strands 工具都用原始 guard() 手动包装。
9. 接受 inputs=。 缺少决策不代表拒绝——请在 Console/CLI 中 验证。

步骤 1：安装并获取 Guard 客户端

需要 PyPI arcjet 1.1.0（1.0.0 中没有此 extra）：

pip install "arcjet[strands-agents]"

如果 Agent 尚无 Guard 客户端，请在模块作用域中启动一次：

import os
from arcjet.guard import launch_arcjet

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

步骤 2：筛查自定义工具 — guard_tool

from strands import Agent, tool
from arcjet.guard import DetectPromptInjection, TokenBucket, launch_arcjet
from arcjet.guard.strands_agents import guard_hooks, guard_tool, strands_agent_context

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

@tool  # event.interrupt() is HITL — not this policy gate
def lookup_order(order_id: str) -> dict:
    """Look up an order by ID."""
    return {"order_id": order_id, "status": "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",
)

步骤 3：筛查未包装的 / MCP 工具 — guard_hooks

mcp_tools = []  # from an MCP client you did not wrap with guard_tool
agent = Agent(
    tools=[lookup_order, *mcp_tools],
    # The agent-wide gate for tools guard_tool did not wrap.
    # Already-wrapped tools are skipped, so Guard is not called twice.
    # BeforeToolCallEvent.cancel_tool is True or a str (JSON of the payload).
    hooks=[
        guard_hooks(
            guard=aj,
            action="mcp.invoked",
            rules=[mcp_limit(key=user_id, requested=1)],
            session_id=conversation_id,
            on_guard_error="deny",
        ),
    ],
)

步骤 4：在 Agent 运行前筛查入站内容

invocation_state = {"sessionId": conversation_id}
derived = strands_agent_context(invocation_state)
decision = await aj.guard(
    label="message.received",
    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")

agent(user_text, invocation_state=invocation_state)

没有入站辅助函数。

步骤 5：关联 ID

strands_agent_context 从 invocation_state 中读取由调用方管理的 ID，依次查找： correlationId、sessionId、requestId。 它不会生成 ID，也不会读取 trace_id。它不会读取 agent.id 或 SessionManager 自动生成的 ID。不要为每轮对话虚构关联 ID。 请在调用中以及 guard_hooks(...) 中使用同一个 ID。

验证集成

1. python -m py_compile（或项目的类型检查）通过。
2. 测试入站 PI（在 Agent 运行前）、guard_tool 拒绝、未包装工具上的 guard_hooks 拒绝（cancel_tool 为 True 或字符串）、速率限制，以及故障时默认拒绝（Guard 无法访问）。确认不会将 event.interrupt() 用作门控。
3. 在 Arcjet Console / CLI 中确认，各项决策共享由调用方管理的会话 / 请求 ID， 而不是 trace_id 或 agent.id。
4. 使用真实 ARCJET_KEY 进行的手动端到端测试仍待验证，直到你实际运行测试。

示例： examples/fastapi-strands-agents-guard。 不要虚构第二个示例名称，也不要在此 skills 仓库中添加示例。