integrate-arcjet-guard-claude-managed-agents-py

将 Arcjet Guard 集成到 Python Claude Managed Agents

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

仅支持官方 anthropic>=0.92.0,<2——不支持 claude-agent-sdk、arcjet.guard.claude_agent_sdk、JS @anthropic-ai/claude-agent-sdk 适配器，也不支持 JS Claude Managed Agents（@arcjet/guard/claude-managed-agents/v0，文档地址： https://docs.arcjet.com/guards/claude-managed-agents/）。 导入 arcjet.guard.claude_managed_agents 不会加载 Claude Agent SDK、 LangChain 或 CrewAI。该 extra 可安全使用（不包含 chromadb）。

这是托管的 Claude Managed Agents 执行框架 （client.beta.sessions）。Anthropic 运行 Agent 循环和内置 工具集（bash、文件、web_*）。Agent 工具集默认设为 always_allow，因此 bash/文件操作没有客户执行前检查—— agent.tool_use / agent.tool_result 会在内置工具已经 运行后触发。不存在 PreToolUse。不要用 always_ask 掩盖这个 缺口。它不是 Claude Agent SDK 本地的 query() / guard_hooks。

导出项：guard_custom_tool、guard_events、 claude_managed_agents_context。没有 guard_inbound、没有 guard_tool，也没有 guard_tool_confirmation。

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

• 入站文本（user.message / initial_events）→ guard_events 包装 sessions.events.send。DENY 会引发 ArcjetDeniedError / ArcjetUnavailableError，且不会调用 send。
• 由你执行的自定义工具（agent.custom_tool_use）→ guard_custom_tool(run=…)。DENY 时不会运行原始 run；该 helper 会发送带有模式字段 is_error 的 user.custom_tool_result。
• 关联 → claude_managed_agents_context 读取调用方拥有的 correlation_id / session_id。它不会生成 ID，也不会读取 Anthropic 的 session.id / sevt_…。

文档：https://docs.arcjet.com/guards/claude-managed-agents/（JS+Python 共用页面）。不要使用 /guards/claude-agent-sdk/。没有单独的 /guards/claude-managed-agents-py/ 页面。JS 适配器随 @arcjet/guard 1.12.0 一同发布。

真正的拦截点是入站 user.message 和自定义工具

guard_events(send=client.beta.sessions.events.send, …) 包装 send 可调用对象，使 user.message / initial_events 在原始 send 运行之前进行评估——这是托管执行框架读取提示词之前唯一可以 拒绝一个轮次的位置。入站 rules 从 message_arguments() 接收 {"prompt", "content", "type"}，而不是 JS 的 { text, events }。guard_custom_tool(run=…) 返回 await handler(event, send=…, anthropic_session_id=…)。handler 仅使用 anthropic_session_id（Anthropic 的 ses_…）来发送 user.custom_tool_result；包装时的 session_id= 是调用方拥有的 关联 ID。内置工具不会进入该 handler。可选的 tool= 会以相同方式 包装自托管的 @beta_tool run；CLI worker 无法注册自定义工具。

自定义工具被拒绝时会发送带 is_error 的 user.custom_tool_result

遇到 DENY（或默认 on_guard_error="deny" 下 Guard 未评估）时， 不会调用原始 run。该 helper 会发送真正的 user.custom_tool_result （custom_tool_use_id，content 中为 ArcjetDenialResult 的 JSON，is_error—— 该字段属于 events 模式；不要另造一个字段）。不要从托管 handler 中抛出异常：抛出异常会让会话进入空闲状态。省略 is_error 看起来就像成功了。这不是 Claude Agent SDK 的 structuredContent。 与 #196 相同的默认失败关闭行为。 核心 guard() 仍然失败开放。

always_ask + user.tool_confirmation 并非基于策略的 HITL

权限策略适用于 Agent 工具集和 MCP，而不适用于自定义 工具。这与 CrewAI human_input、JS canUseTool 和 LangGraph interrupt() 遇到的是同一个陷阱。MCP Guard 仅适用于你托管的服务——Anthropic 是 MCP 客户端。web_search / web_fetch 始终在 Anthropic 上运行。

首先要向用户询问的问题

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

1. 应用会在 agent.custom_tool_use 上执行哪些自定义工具？这些工具需要使用 guard_custom_tool。always_allow 下的内置工具无法进行此类防护。
2. 有哪些限制？（例如，“每位用户每分钟查询 10 次” → TokenBucket。）
3. 元数据中的用户是谁——不透明的用户／租户 ID（绝不能是个人身份信息）？默认值：无。Anthropic 的 session.id / sevt_… 不是关联 ID。
4. Arcjet 故障是否不可接受？每个辅助函数的默认值都是 on_guard_error="deny"。请明确询问入站 guard_events 的处理方式：在此处采用故障关闭意味着不会发送该轮请求，因此在这个调用位置选择 "allow" 也是常见且合理的做法。

读者常见的误解

1. 这不是 Claude Agent SDK。 没有 guard_tool、没有 guard_hooks、没有 PreToolUse、没有 UserPromptSubmit。也不要再套一层 arcjet.guard.claude_agent_sdk 或 @arcjet/guard/claude-agent-sdk/v0。
2. 没有 guard_inbound。 该辅助函数是 guard_events。
3. 入站 ctx 是 {"prompt", "content", "type"}，不是 JS { text, events }。
4. 关联 ID 由调用方负责，绝不会自动生成。 不要将 Anthropic 的 session.id / sevt_… / trace_id 作为关联 ID 传入。
5. 默认的 always_allow 无法被拦截。 不要声称我们可以阻止 Anthropic 云端的 bash／文件操作。
6. 自定义工具被 DENY 时，要发送带有 is_error 的 user.custom_tool_result。 不要抛出异常。ALOW 时，调用方仍需发送成功结果，否则会话会停滞。
7. always_ask + user.tool_confirmation 是可选确认，不是策略。
8. 不要用原始 guard() 包装每个会话事件。
9. guard_events 没有 inbound=。 action 和 rules 位于顶层；它接收 send=；返回的可调用对象会替代 send，并在 DENY 时抛出异常。guard_custom_tool 接收 run=，而不是 tool=<function>。缺少决策不代表拒绝。

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

需要 PyPI arcjet 1.1.0（此 extra 不在 1.0.0 中）：

pip install "arcjet[claude-managed-agents]"

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

import os
from arcjet.guard import launch_arcjet

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

完整示例： examples/fastapi-claude-managed-agents-guard。

步骤 2：在 sessions.events.send 之前筛查入站内容

from anthropic import Anthropic
from arcjet.guard import DetectPromptInjection, TokenBucket, launch_arcjet
from arcjet.guard.claude_managed_agents import (
    claude_managed_agents_context,
    guard_custom_tool,
    guard_events,
)

client = Anthropic()
aj = launch_arcjet(key=os.environ["ARCJET_KEY"])
inbound = DetectPromptInjection()
user_id = authenticated_user_id
conversation_id = authenticated_conversation_id
derived = claude_managed_agents_context(session_id=conversation_id)

send = guard_events(
    guard=aj,
    send=client.beta.sessions.events.send,
    action="message.received",
    # Python inbound ctx is {"prompt", "content", "type"} — not JS { text, events }.
    rules=lambda ctx: [inbound(ctx["prompt"])],
    session_id=derived.correlation_id or conversation_id,
    on_guard_error="deny",
)

session = client.beta.sessions.create(agent=agent_id, environment_id=environment_id)
# Anthropic minted session.id — pass it to the sessions API, not as correlation.

# Screen user.message before the hosted harness reads it. DENY raises
# ArcjetDeniedError / ArcjetUnavailableError and does not send.
await send(
    session.id,
    events=[{"type": "user.message", "content": [{"type": "text", "text": user_text}]}],
)

步骤 3：在 agent.custom_tool_use 上拦截自定义工具

lookup_limit = TokenBucket(
    label="order.looked-up",
    bucket="lookups",
    refill_rate=10,
    interval_seconds=60,
    max_tokens=10,
)

async def lookup_order(event) -> dict:
    order_id = event.input["order_id"]
    return {"content": [{"type": "text", "text": f"{order_id}: shipped"}]}

handle_lookup = guard_custom_tool(
    guard=aj,
    action="order.looked-up",
    run=lookup_order,
    rules=[lookup_limit(key=user_id, requested=1)],
    session_id=derived.correlation_id or conversation_id,
    on_guard_error="deny",
)

with client.beta.sessions.events.stream(session_id=session.id) as stream:
    for event in stream:
        if event.type == "agent.custom_tool_use" and event.name == "lookup_order":
            # DENY posts user.custom_tool_result with is_error. Do not raise.
            await handle_lookup(
                event,
                send=client.beta.sessions.events.send,
                anthropic_session_id=session.id,
            )
        # agent.tool_use / always_ask + user.tool_confirmation are not this policy gate

速率限制应基于通过身份验证的调用方，而不是模型提供的订单 ID。即使 ALLOW，调用方仍需发送成功的 user.custom_tool_result。

步骤 4：关联

claude_managed_agents_context 是一个读取器，返回 ClaudeManagedAgentsContext（一个数据类）——它不是上下文管理器，也不是 contextvar 设置器。它读取由调用方提供的 correlation_id / session_id。它绝不会自动生成 ID，也绝不会读取 Anthropic 的 id / event_id / session.id（ses_…）/ sevt_… / trace_id。传入 Anthropic Session 对象是安全的——这些自动生成的 ID 会被忽略。无效 候选值会被跳过；如果没有剩余的有效值，该调用就不会关联到任何 ID，而不会与生成的 ID 关联。辅助函数也会在内部重新读取 session_id= / correlation_id=。

验证集成

1. python -m py_compile（或项目的类型检查）通过。
2. 测试入站 PI（不会调用 send）、自定义工具拒绝（使用 user.custom_tool_result 和 is_error；不会调用 run）、速率限制，以及失败时关闭（防护不可达）。确认 always_ask / user.tool_confirmation 从未被当作门控，并且内置的 agent.tool_use 仅用于观察。
3. 在 Arcjet Console / CLI 中确认各项决策使用调用方拥有的对话 ID，而不是 Anthropic session.id / sevt_…。
4. 使用真实 ARCJET_KEY 进行的手动端到端测试，在实际运行之前仍待验证。

示例： examples/fastapi-claude-managed-agents-guard。 不要杜撰第二个示例名称。不要在此 skills 仓库中添加示例。