開発者ガイド
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このコマンドは、保護された関数ツールを 1 つ持つエージェントを構築します。チェックが判断するのは記述された操作だけなので、各ツールが実際に行うことからそのツールの操作を組み立ててください。 guarded_tool_call の呼び出しをマンデートのチェックが必要なすべてのツール内に置き、処理を行う前にその操作を渡します。
すべての結果を処理する
- permit: ラッパーは
execute()を呼び出します。マンデートの条項が提案された行為を許可しているためです。 - deny: ラッパーは拒否文字列を返します。別のツールで行為を試みないでください。
- escalate: ラッパーは拒否文字列を返します。このラッパーの外で運用者に判断を求めてください。
- error: ラッパーは拒否文字列を返します。deny として扱い、理由を報告し、無分別に再試行しないでください。
転送障害、不正な形式の応答、または未知の結果は errorとなり、正確な permit だけが execute() に到達します。