Module 03 ยท Lesson 3 โ The Safety Gate
All Tool Calls Go Through One Function. No Exceptions. You do not call tool.execute() directly from the agent loop. You call execute_with_approval(). Always. This function is the single chokepoint where the three access levels are enforced and every action is logged. The agent loop does not need to know whether a tool is READ, WRITE, or ADMIN โ the gate handles that. --- The Audit Log First, the logging function. Every action โ approved or rejected โ gets a record. import json import datetime AUDIT_LOG_PATH = "audit.jsonl" def _log_action(tool_name: str, params: dict, approved: bool, operator: str): entry = { "timestamp": datetime.datetime.utcnow().isoformat() + "Z", "tool": tool_name, "params": params, "approved": approved, "operator": operator, } with open(AUDIT_LOG_PATH, "a") as f: f.write(json.dumps(entry) + "\n") JSONL format โ one JSON object per line. Easy to ship to a SIEM. Easy to search with grep or jq. Each line is self-contained. Think of this as your debug ip ospf events equivalent โ a timestamped trail of everything the agent did or tried to do. --- The Full Safety Gate def execute_with_approval(tool: BaseTool, params: dict, operator: str = "unknown") -> ToolResult: """The safety gate. Every tool call goes through here.""" if tool.category == READ: result = tool.execute(**params) _log_action(tool.name, params, approved=True, operator=operator) return result if tool.category == WRITE: print(f"\n{'='*55}") print(" WRITE OPERATION REQUESTED") print(f"{'='*55}") print(f" Tool : {tool.name}") print(f" Params: {json.dumps(params, indent=4)}") if "diff" in params: print(f"\n Config diff:\n{params['diff']}") print(f"{'='*55}") answer = input(" Approve? (y/n): ").strip().lower() approved = (answer == "y") _log_action(tool.name, params, approved=approved, operator=operator) if not approved: return ToolResult(success=False, data={}, error="Operator rejected.")