integrate-arcjet-guard-openai-agents-py

将 Arcjet Guard 集成到 Python OpenAI Agents

arcjet.guard.openai_agents 包装 Agent 现有的 Arcjet 客户端。它 不会直接与 Arcjet API 通信。Guard 的通用基础知识（客户端、 规则、标签、判定、捕获、注册）位于 ../arcjet/references/guards_python.md。 对于非 OpenAI Agents 特有的内容，请参阅该文档。

仅支持官方 openai-agents>=0.19.0,<1，不支持 JS @openai/agents 适配器（@arcjet/guard/openai-agents/v0，文档地址为 https://docs.arcjet.com/guards/openai-agents/），也不支持社区分支。 导入 arcjet.guard.openai_agents 不会加载 LangChain。

导出项：guard_tool、openai_agents_context。仅支持自定义的 FunctionTool / @function_tool。不支持托管工具、MCP、Computer / Shell / ApplyPatch、交接或 Agent.as_tool()。

两个接口，共用一条判定规则：

• 自定义的 FunctionTool → guard_tool。门控使用 FunctionTool.tool_input_guardrails + reject_content（ArcjetDenialResult 的 JSON）。 不要抛出异常。
• 关联 → openai_agents_context 读取调用方持有的会话 / 对话 ID。它不会生成 ID，也不会读取 trace_id。

文档：https://docs.arcjet.com/guards/openai-agents/。

门控使用 tool_input_guardrails + reject_content

guard_tool 返回一个副本，其输入保护措施以 Arcjet 开始，因此在 DENY（或默认 on_guard_error="deny" 下未评估的 Guard）时，不会运行 on_invoke_tool。 拒绝结果是 ToolGuardrailFunctionOutput.reject_content，其中包含 ArcjetDenialResult 的 JSON（{ arcjetDenied: true, … }）。不要抛出异常—— raise_exception() 会触发中止，而从 on_invoke_tool 抛出的异常会被 default_tool_error_function 吞掉。这与 #196 中的默认失败关闭行为相同： 只有 "allow" 会失败开放；DENY 始终会拦截。核心 guard() 仍会失败开放（has_failed_open()）。

needs_approval 不是策略门控

needs_approval 用于人工介入（state.approve / state.reject）。 这与 JS OpenAI Agents 的 needsApproval、LangGraph 的 interrupt()、 以及 Genkit 的 interrupt() 一样，容易造成误解。这里没有入站辅助函数，也没有审批辅助函数。 RunConfig.tool_execution.pre_approval_tool_input_guardrails=True 仅由应用选择启用——此辅助函数不会设置它。

在 Runner.run 之前筛查入站内容

这里没有入站辅助函数。SDK 的 input_guardrails / output_guardrails / 工具输出保护措施是 SDK 自身的拦截机制，不是 Arcjet。请在应用中调用 aj.guard(...)，并根据判定采取行动。核心 guard() 会失败开放：ALLOW 并不能证明规则已运行。如果此调用点必须失败关闭， 请通过 decision.has_failed_open() 进行门控； guard_tool 已默认采用此行为。

首先要向用户确认的问题

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

1. 哪些工具风险较高（会产生外部副作用、不可逆、花费资金或发送消息）？这些工具应使用 guard_tool。托管工具 / MCP / 交接 / as_tool 不在支持范围内。
2. 有什么限制？（例如“每位用户每分钟查询 10 次” → TokenBucket。）
3. 元数据中的用户是谁——不透明的用户 / 租户 ID（绝不包含个人身份信息）？ 默认值：无。将已有的对话 / 会话 ID 放到传递给 Runner.run 的应用上下文中。该 ID 是关联 ID，不是用户 ID。
4. Arcjet 服务中断是否不可接受？每个辅助函数默认使用 on_guard_error="deny"。请明确询问是否要在 Runner.run 之前进行入站筛查：在那里失败关闭意味着 Agent 不会运行，因此在这一个调用点选择 "allow" 是常见且合理的做法。

读者容易误解的地方

1. 这不是 JS 适配器。 不要导入 @arcjet/guard/openai-agents/v0。
2. 没有入站辅助函数。 SDK 防护栏不是 Arcjet。在 Runner.run 之前，先用核心 guard() 进行筛查。
3. needs_approval 是 HITL，不是策略。
4. 只有 reject_content 表示拒绝。 抛出异常会触发跳闸式中止，或被 default_tool_error_function 吞掉。
5. 只读取关联 ID，不生成。 绝不使用 trace_id。绝不构造 OpenAIConversationsSession()。
6. 按经过身份验证的调用者设置速率限制，而不是按模型提供的订单 ID。将 guard_tool 返回的副本交给 Agent——原始工具仍未受防护。
7. 不要用原始 guard() 手动包装每个工具。
8. 接受 inputs=。 缺少决策不代表拒绝——请在 Console/CLI 中验证。

第 1 步：安装并找到防护客户端

此模块随 PyPI arcjet 1.0.0 一同发布：

pip install "arcjet[openai-agents]"

如果 Agent 还没有防护客户端，请在模块作用域中启动一个，且只启动一次：

import os
from arcjet.guard import launch_arcjet

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

第 2 步：为编写的工具设置门控——guard_tool

from agents import Agent, Runner, function_tool
from arcjet.guard import DetectPromptInjection, TokenBucket, launch_arcjet
from arcjet.guard.openai_agents import guard_tool, openai_agents_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,
)
user_id = authenticated_user_id

@function_tool  # needs_approval=... is HITL — not this policy gate
def lookup_order(order_number: str) -> dict:
    """Look up an order by number."""
    return {"order_number": order_number, "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",
)

agent = Agent(
    name="support-agent",
    instructions="Help the user.",
    tools=[lookup_order],
)

对 guard_tool 使用 action + rules。

第 3 步：在 Runner.run 之前筛查入站内容

inbound = DetectPromptInjection()
app_context = {"session_id": conversation_id}
derived = openai_agents_context(app_context)
decision = await aj.guard(
    label="message.received",
    rules=[inbound(user_text)],
    correlation_id=derived.correlation_id,
)
if decision.conclusion == "DENY":
    raise Exception("message blocked")
if decision.has_failed_open():
    raise Exception("inbound guard unavailable")

await Runner.run(agent, user_text, context=app_context)

没有入站辅助函数。

第 4 步：关联

openai_agents_context 会读取由调用者提供的 ID。优先级：runContext.context / 普通应用对象上的字段（correlation_id，然后是 session_id，然后是 conversation_id，然后是 group_id，支持蛇形或 驼峰命名），接着是信封上的同名字段，然后是 correlation_id= / session_id= 关键字参数，最后是外层的 arcjet_sequence。它会返回 OpenAIAgentsContext——将 .correlation_id 传给 guard()。它不会生成 ID，不会读取 trace_id，也不会构造 OpenAIConversationsSession()。不要 为每一轮凭空生成关联 ID。

验证集成

1. python -m py_compile（或项目的类型检查）通过。
2. 测试入站 PI（在 Runner.run 之前）、工具拒绝 （使用 reject_content，而不是抛出异常）、速率限制和故障关闭（防护服务 无法访问）。确认绝不会将 needs_approval 当作门控。
3. 在 Arcjet Console / CLI 中确认各项决策共用调用者提供的 session / conversation ID，而不是 trace_id。
4. 使用真实 ARCJET_KEY 进行手动端到端测试，仍待运行验证。

示例 fastapi-openai-agents-guard 仍归 Runtime 所有——不要 另起示例名称。不要在此 skills 仓库中添加示例。