为 OpenAI Agents SDK 智能体添加操作前检查

检查的作用

检查返回 permit、 deny或 escalate ，并附上决定结果的条款和签名回执。检查仅提供建议，真正拒绝执行的是函数工具包装器。托管检查在内存中评估请求，不保留请求内容。

代码

这是经过审计的示例，与以下位置发布的内容完全一致： public/examples/frameworks/.

"""OpenAI Agents SDK tool wrapper that checks HANRIA before execution."""

from __future__ import annotations

import json
from collections.abc import Callable
from typing import Any
from urllib.error import URLError
from urllib.request import Request, urlopen

from agents import Agent, function_tool

CHECK_URL = "https://check.hanria.ai/v1/check"
MANDATE = {
    "schema_version": "0.2-draft",
    "mandate_id": "read-text-files",
    "purpose": "Permit reads of text files under the example directory.",
    "default": "deny",
    "clauses": [
        {
            "id": "permit-example-read",
            "effect": "permit",
            "match": {
                "kind": ["file"],
                "verb": ["read"],
                "target_prefix": ["/tmp/example/"],
            },
        }
    ],
}


def check_action(action: dict[str, Any], check_url: str = CHECK_URL) -> dict[str, Any]:
    """Return a HANRIA decision, mapping every transport or response failure to error."""
    body = json.dumps({"mandate": MANDATE, "action": action}).encode()
    request = Request(check_url, data=body, headers={"content-type": "application/json"})
    try:
        with urlopen(request, timeout=10) as response:
            decision = json.load(response)
    except (OSError, URLError, ValueError, json.JSONDecodeError) as error:
        return {"outcome": "error", "reason": f"HANRIA check failed: {error}"}
    if not isinstance(decision, dict) or decision.get("outcome") not in {
        "permit", "deny", "escalate", "error"
    }:
        return {"outcome": "error", "reason": "HANRIA returned an invalid decision"}
    return decision


def guarded_tool_call(
    tool_name: str,
    operation: dict[str, Any],
    execute: Callable[[], Any],
    check_url: str = CHECK_URL,
) -> Any:
    """Call execute only after HANRIA returns permit for the proposed tool call."""
    action = {
        "schema_version": "0.1-draft",
        "requested_by": {"agent": "openai-agents-example"},
        "operation": operation,
        "justification": f"Run {tool_name} through the OpenAI Agents SDK.",
    }
    decision = check_action(action, check_url)
    if decision.get("outcome") != "permit":
        return (
            f"Tool refused. HANRIA outcome={decision.get('outcome', 'error')}; "
            f"reason={decision.get('reason', 'no reason returned')}. Do not retry blindly."
        )
    return execute()


def build_agent(check_url: str = CHECK_URL) -> Agent:
    """Construct an agent whose file tool is wrapped by the HANRIA check."""

    @function_tool
    def read_text_file(path: str) -> str:
        """Read one UTF-8 text file after the mandate permits it."""
        return guarded_tool_call(
            "read_text_file",
            {"kind": "file", "verb": "read", "target": path},
            lambda: open(path, encoding="utf-8").read(),
            check_url,
        )

    return Agent(
        name="HANRIA guarded reader",
        instructions="Use read_text_file only when it is needed.",
        tools=[read_text_file],
    )


if __name__ == "__main__":
    agent = build_agent()
    print(f"Constructed {agent.name} with {len(agent.tools)} guarded tool.")

运行

保存 openai_agents.py，然后在保存该文件的文件夹中运行以下命令。

uv run --with openai-agents==0.23.1 python openai_agents.py

该命令构造一个带有受控函数工具的智能体。检查只判断你描述的操作，因此应根据每个工具实际要做的事情来构建其操作。将对 guarded_tool_call 的调用保留在每个需要授权书检查的工具内，并在执行工作之前传入该操作。

处理每一种结果

• permit: 包装器调用 execute() ，因为授权书条款许可了拟议操作。
• deny: 包装器返回拒绝字符串。不要通过其他工具尝试该操作。
• escalate: 包装器返回拒绝字符串。请在该包装器之外请求运营者作出决定。
• error: 包装器返回拒绝字符串。将其视为 deny，报告原因，并且不要盲目重试。

传输故障、格式错误的响应或未知结果都会变为 error，只有准确的 permit 才会到达 execute().

参考资料

试用检查 · 验证签名回执 · OpenAPI 文档 · 面向智能体的摘要