Add a pre-action check to an OpenAI Agents SDK agent

What the check does

The check returns permit, deny, or escalate with the governing clause and a signed receipt. It is advisory, and your function tool wrapper is what refuses execution. The hosted check evaluates the request in memory and keeps no request content.

Code

This is the audited example exactly as published under 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.")

Run it

Save openai_agents.py, then run this command from the folder where you saved the file.

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

The command constructs an agent with one guarded function tool. The check judges only the operation you describe, so build each tool's operation from what that tool will actually do. Keep the call to guarded_tool_call inside every tool that needs the mandate check and pass that operation before doing the work.

Handle every outcome

• permit: the wrapper calls execute() because a mandate clause permits the proposed action.
• deny: the wrapper returns a refusal string. Do not attempt the action through another tool.
• escalate: the wrapper returns a refusal string. Ask the operator for a decision outside this wrapper.
• error: the wrapper returns a refusal string. Treat it as deny, report the reason, and do not retry blindly.

A transport failure, malformed response, or unknown outcome becomes error, and only an exact permit reaches execute().

References

Try the check · Verify signed receipts · OpenAPI document · Agent-facing summary