integrate-arcjet-guard-crewai

将 Arcjet Guard 集成到 CrewAI

arcjet.guard.crewai 封装 Agent 现有的 Arcjet 客户端。它本身不会与 Arcjet API 通信。Guard 的通用基础知识（客户端、规则、标签、决策、捕获、注册）位于 ../arcjet/references/guards_python.md。 对于非 CrewAI 特有的内容，请加载该参考文档。

仅适用于官方 crewai，不适用于社区分支、LangChain Crew 封装或 npm CrewAI 移植版。没有 arcjet[crewai] extra：CrewAI 硬依赖 chromadb，而它包含未修补的 RCE CVE-2026-45829，因此 Arcjet extra 不得将其引入。请自行安装 CrewAI。导入 arcjet.guard.crewai 不会加载 LangChain。

导出项：register_arcjet_hooks、unregister_arcjet_hooks、 ArcjetCrewAIHooks、guard_tool、ToolPolicy、sanitize_tool_name、 free_text_arguments。没有 guard_crew。

两种接入方式，一条决策规则：

• 由 crew、LiteAgent、MCP 适配器或注入 crew 的列表执行的工具 → register_arcjet_hooks + ToolPolicy。仅通过进程级 PRE_TOOL_CALL 进行门控。DENY 会引发 HookAborted(reason=..., source="arcjet")。
• 由你自行调用的 CrewAI BaseTool → guard_tool。这是 唯一会引发 ArcjetDeniedError / ArcjetUnavailableError 的 CrewAI 路径。BaseTool.run 永远不会分派 PRE_TOOL_CALL。

文档：https://docs.arcjet.com/guards/crewai/。

门控是进程级 PRE_TOOL_CALL，且只执行一次

register_arcjet_hooks 会在 CrewAI 的分派器上注册钩子。遇到 DENY（或在默认设置 on_guard_error="deny" 下遇到未经评估的 Guard）时，会引发 HookAborted(reason=..., source="arcjet")，因此工具不会运行。CrewAI 会吞掉其他所有异常——如果钩子引发 ArcjetDeniedError / ArcjetUnavailableError，工具反而会运行。钩子路径仅支持同步——请传入 launch_arcjet_sync / ArcjetGuardSync。注册时会拒绝异步客户端（引发 ArcjetMisconfiguration）。同一进程中第二次调用 register_arcjet_hooks 也会引发 ArcjetMisconfiguration（CrewAI 的注册表会追加钩子，导致重复评估）；请先对句柄调用 unregister()。

永远不会注册 POST_TOOL_CALL

只安装 PRE。决策在 PRE 中捕获。POST 不是策略接入点。Agent 始终会看到 Tool execution blocked by hook. Tool: {name}。HookAborted.reason 仅用于遥测。继续执行仍会记录为 success——不要将 CrewAI 钩子捕获结果解读为 guard_action / LangChain 辅助函数使用的五值 metadata.outcome 表。

human_input 不是策略门控

Agent/Task 的 human_input 和 request_human_input 属于人在回路流程。 这与 JS humanInTheLoopMiddleware、LangGraph interrupt()、 OpenAI Agents needsApproval 和 Genkit interrupt() 存在相同陷阱。没有入站辅助函数，也没有审批辅助函数。

首先要向用户询问的问题

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

1. 哪些工具有风险（会产生外部副作用、不可逆、花费资金或发送消息）？由 crew 执行的工具使用 register_arcjet_hooks。由你自行调用的独立 BaseTool 使用 guard_tool。
2. 有什么限制？（例如，“每位用户每分钟查询 10 次” → TokenBucket。）
3. 元数据中的用户是谁——不透明的用户/租户 ID（绝不使用 PII）？ 默认：无。Crew、task 和 agent 名称属于元数据，绝不会被创建为关联 ID。请使用调用方拥有的 correlation_id / arcjet_sequence。
4. Arcjet 故障是否不可接受？每个辅助函数的默认值都是 on_guard_error="deny"。请明确询问是否要在 crew.kickoff 前进行入站筛查： 在此处失败时关闭意味着 crew 不会运行，因此在这个调用点选择 "allow" 是常规且合理的做法。

读者常见误解

1. 没有 arcjet[crewai] 这个 extra。 请自行安装 crewai>=1.15.3,<2。 arcjet.guard.crewai 随 PyPI 中的 arcjet 1.0.0 一起发布。
2. 没有 guard_crew。 请使用 register_arcjet_hooks。
3. Hook 路径仅支持同步调用。 请传入 launch_arcjet_sync。
4. 在 Hook 中引发 HookAborted，不要引发 Arcjet 错误。 CrewAI 会吞掉其他所有错误，工具仍会运行。
5. human_input 属于 HITL，不是策略。
6. 在调用 crew.kickoff 之前，使用核心 guard / guard_sync 检查入站内容。 核心 guard() 会在失败时放行。
7. 请根据经过身份验证的调用方设置速率限制，而不是根据模型提供的订单 ID。sanitize_tool_name 可以匹配 Send Email 和 send_email。
8. 不要给每个 CrewAI 工具都手动套上原始 guard()。
9. 支持 inputs=。 Hook 解析器的参数为 (arguments, ctx)。 缺少决策不代表拒绝——请在 Console/CLI 中验证。

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

pip install arcjet
pip install "crewai>=1.15.3,<2"

对于 Hook 路径，请在模块作用域启动一个同步客户端：

import os
from arcjet.guard import launch_arcjet_sync

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

仅在异步应用代码中，或在调用 arun() 时通过 guard_tool 调用 launch_arcjet。

步骤 2：为 Crew 执行的工具设置门控 — register_arcjet_hooks

ToolPolicy 由 action + rules 组成，以工具名称为键。键和可选的 tools= 过滤器都会经过 sanitize_tool_name（CrewAI 1.15.3+）。没有匹配策略的工具仍会触发 "{sanitized_tool_name}.invoked" 和注册器级别的 rules（即使为空，也仍会联系 Guard），除非你传入 tools=。如果扫描规则只需要自由文本，可以使用 free_text_arguments 去除不透明 ID（tool_call_id、*_id 等）；Hook 本身会将工具自身未经筛选的参数映射传给解析器。已套用 guard_tool 的工具会被跳过，因此 Guard 不会被调用两次。请使用 unregister_arcjet_hooks(hooks) 或 hooks.unregister() 进行清理。

from crewai import Agent, Crew, Task
from arcjet.guard import DetectPromptInjection, TokenBucket, launch_arcjet_sync
from arcjet.guard.crewai import ToolPolicy, register_arcjet_hooks, unregister_arcjet_hooks

aj = launch_arcjet_sync(key=os.environ["ARCJET_KEY"])
lookup_limit = TokenBucket(
    label="order.looked-up",
    bucket="lookups",
    refill_rate=10,
    interval_seconds=60,
    max_tokens=10,
)
inbound = DetectPromptInjection()
user_id = authenticated_user_id

hooks = register_arcjet_hooks(
    guard=aj,
    policies={
        "lookup_order": ToolPolicy(
            action="order.looked-up",
            rules=[lookup_limit(key=user_id, requested=1)],
        ),
    },
    tools=["lookup_order"],
    on_guard_error="deny",
)

agent = Agent(
    role="Support",
    goal="Look up orders",
    backstory="Help the user with order status.",
    # human_input=True is HITL — not this policy gate
)
task = Task(
    description="Look up the user's order",
    expected_output="Order status",
    agent=agent,
)
crew = Crew(agents=[agent], tasks=[task])

decision = aj.guard(
    label="message.received",
    rules=[inbound(user_text)],
)
if decision.conclusion == "DENY":
    raise RuntimeError("message blocked")
if decision.has_failed_open():
    raise RuntimeError("inbound guard unavailable")

crew.kickoff()
unregister_arcjet_hooks(hooks)

步骤 3：为你调用的 BaseTool 设置门控 — guard_tool

将此函数返回的副本交给 Crew（它带有 Hook 会跳过的标记）。原始工具会有意保持未设防状态——如果将其交给 Crew，Hook 仍会保护它。同步调用需要阻塞式客户端；异步调用需要可等待的客户端。

from arcjet.guard.crewai import guard_tool

guarded = guard_tool(
    guard=aj,
    tool=lookup_order,
    action="order.looked-up",
    rules=[lookup_limit(key=user_id, requested=1)],
    on_guard_error="deny",
)
result = guarded.run(order_id=order_id)

步骤 4：关联

使用由调用方提供的 correlation_id / arcjet_sequence。 Crew、任务和 Agent 名称只是元数据，绝不会被生成为 ID。如果没有有效值，调用就不会关联，而不是关联到生成的 ID。

验证集成

1. python -m py_compile（或项目的类型检查）通过。
2. 测试入站 PI（在 kickoff 之前）、PRE_TOOL_CALL 拒绝（HookAborted，Agent 会看到 Tool execution blocked by hook. Tool: {name}）、独立 guard_tool 拒绝（ArcjetDeniedError）、速率限制以及失败时拒绝（Guard 无法访问）。确认未注册 POST_TOOL_CALL，且 human_input 从未被当作门控。
3. 在 Arcjet Console / CLI 中确认各项决策使用相同的调用方提供的关联 ID，而不是 Crew/任务/Agent 名称。
4. 使用真实 ARCJET_KEY 进行手动端到端测试仍待验证，直到你实际运行。

不要在此 skills 仓库中编造 CrewAI 示例名称。