OpenClaw Gateway policy engines, approval workflows and auditable agent execution are used to implement a coding implementation for an enterprise AI governance system.

Download the latest version of!apt update.
!apt-get install -y curl
!curl -fsSL https://deb.nodesource.com/setup_22.x | bash -
!apt-get install -y nodejs
!node -v
!npm -v
Openclaw version :?npm install openclaw@latest
Install pandas with pydantic using!pip


Import os
Download json
import time
import uuid
Import Secrets
Import subprocess
import getpass
From pathlib import Path
You can import any text by typing in Dict.
Import dataclass import asdict
Import datetime and time zone from datetime


Import requests
import pandas as pd
BaseModel and Field


try:
 From Google.colab, import Userdata
   OPENAI_API_KEY = userdata.get("OPENAI_API_KEY")
Except Exception
 OPENAI_API_KEY= None


If not OPENAI_API_KEY
   OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY")


If not OPENAI_API_KEY
   OPENAI_API_KEY = getpass.getpass("Enter your OpenAI API key (hidden input): ").strip()


assert OPENAI_API_KEY != "", "API key cannot be empty."


OPENCLAW_HOME = Path("/root/.openclaw")
OPENCLAW_HOME.mkdir(parents=True, exist_ok=True)
OPENCLAW_HOME = WORKSPACE "workspace"
WORKSPACE.mkdir(parents=True, exist_ok=True)


GATEWAY_TOKEN = secrets.token_urlsafe(48)
GATEWAY_PORT = 18789
GATEWAY_URL=f"http://127.0.0.1:{GATEWAY_PORT}"

config = {
   "env": {
       "OPENAI_API_KEY": OPENAI_API_KEY
   },
   "agents": {
       "defaults": {
           "workspace": str(WORKSPACE),
           "model": {
               "primary": "openai/gpt-4.1-mini"
           }
       }
   },
   "gateway": {
       "mode": "local",
       "port": GATEWAY_PORT,
       "bind": "loopback",
       "auth": {
           "mode": "token",
           "token": GATEWAY_TOKEN
       },
       "http": {
           "endpoints": {
               "chatCompletions": {
                   "enabled"True
               }
           }
       }
   }
}


config_path = OPENCLAW_HOME / "openclaw.json"
config_path.write_text(json.dumps(config, indent=2))


doctor = subprocess.run(
   ["bash", "-lc", "openclaw doctor --fix --yes"],
   capture_output=True,
   text=True
)
print(doctor.stdout[-2000:])
print(doctor.stderr[-2000:])


gateway_log = "/tmp/openclaw_gateway.log"
Gateway_cmd=f"OPENAI_API_KEY='{OPENAI_API_KEY}' OPENCLAW_GATEWAY_TOKEN='{GATEWAY_TOKEN}' openclaw gateway --port {GATEWAY_PORT} --bind loopback --token '{GATEWAY_TOKEN}' --verbose > {gateway_log} 2>&1 & echo $!"
gateway_pid = subprocess.check_output(["bash", "-lc", gateway_cmd]).decode().strip()
print("Gateway PID:", gateway_pid)

def wait_for_gateway(timeout=120):
 "Start" = "time.time()
 Time.() - Start 

def classify_request(user_request: str) -> ActionProposal:
 Lowercase text: user_request()


   red_terms = [
       "delete", "remove permanently", "wire money", "transfer funds",
       "payroll", "bank", "hr record", "employee record", "run shell",
       "execute command", "api key", "secret", "credential", "token",
       "ssh", "sudo", "wipe", "exfiltrate", "upload private", "database dump"
   ]
   amber_terms = [
       "email", "send", "notify", "customer", "vendor", "contract",
       "invoice", "budget", "approve", "security policy", "confidential",
       "write file", "modify", "change"
   ]


 If any (t in the text for t under red_terms),
 Return ActionProposal (
           user_request=user_request,
           category="high_impact",
           risk="red",
           confidence=0.92,
           requires_approval=True,
           allow=False,
           reason="High-impact or sensitive action detected"
       )


 If applicable(t for amber_terms in text)
 Return ActionProposal (
           user_request=user_request,
           category="moderate_impact",
           risk="amber",
           confidence=0.76,
           requires_approval=True,
           allow=True,
           reason="Moderate-risk action requires human approval before execution"
       )


 ActionProposal
       user_request=user_request,
       category="low_impact",
       risk="green",
       confidence=0.88,
       requires_approval=False,
       allow=True,
       reason="Low-risk request"
   )


def simulated_human_approval(proposal: ActionProposal) -> Dict[str, Any]:
   if proposal.risk == "red":
 The word "approved" is a false statement.
 Note: "Rejected automatically in demo for red-risk request"
   elif proposal.risk == "amber":
 Considered True
 Note: "Approved automatically in demo for amber-risk request"
   else:
 No, it is not true
 Note:Return "No approval required"
   return {
       "approved": approved,
       "reviewer": "simulated_manager",
       "note""Note"
   }


@dataclass
Class TraceEvent
 Trace_id:
 Ts:
 Stage: Str
 Payload : Dict[str, Any]

TraceStore class:
   def __init__(self, path="openclaw_traces.jsonl"):
 "self.path" = path
       Path(self.path).write_text("")


 Def append(self: event: traceEvent)
 Open(self.path) "a"As f:
           f.write(json.dumps(asdict(event)) + "n")


   def read_all(self):
 The rows are equal to []
 Open(self.path) "r"( f as a:
 For line in f
 Line = strip()
 If line:
                   rows.append(json.loads(line))
 Return rows


TraceStore: trace_store()


Now, def():
   return datetime.now(timezone.utc).isoformat()


SYSTEM_PROMPT = """
OpenClaw Assistants are enterprise-wide assistants who operate under strict governance.


Rules:
Never declare that a particular action has been completed unless your governance layer expressly permits it.
If the request is low risk, respond normally and with helpfulness.
If the request is moderately risky, you should propose a plan of action and include any necessary approvals or verification.
If the request is high risk, you should refuse to carry out and provide an alternative that's less dangerous, such as a checklist, draft, review plan, summary or a list of possible alternatives.
Useful but concise.
"""


def governed_openclaw_run(user_request: str, session_user: str = "employee-001") -> Dict[str, Any]:
   trace_id = str(uuid.uuid4())


   proposal = classify_request(user_request)
   trace_store.append(TraceEvent(trace_id, now(), "classification", proposal.model_dump()))


 No approvalThe result is =
   if proposal.requires_approval:
       approval = simulated_human_approval(proposal)
       trace_store.append(TraceEvent(trace_id, now(), "approval", approval))


   if proposal.risk == "red":
       result = {
           "trace_id": trace_id,
           "status": "blocked",
           "proposal": proposal.model_dump(),
           "approval": approval,
           "response": "This request is blocked by governance policy. I can help by drafting a safe plan, a checklist, or an approval packet instead."
       }
       trace_store.append(TraceEvent(trace_id, now(), "blocked", result))
 Return to result


   if proposal.risk == "amber" Not approvalThe result is =["approved"]:
       result = {
           "trace_id": trace_id,
           "status": "awaiting_or_rejected",
           "proposal": proposal.model_dump(),
           "approval": approval,
           "response": "This request requires approval and was not cleared."
       }
       trace_store.append(TraceEvent(trace_id, now(), "halted", result))
 Return to result


 The message =The result is = [
       {"role": "system", "content": SYSTEM_PROMPT},
       {"role": "user", "content": f"Governance classification: {proposal.model_dump_json()}nnUser request: {user_request}"}
   ]


   raw = openclaw_chat(messages=messages, user=session_user, agent_id="main", temperature=0.2)
   assistant_text = raw["choices"][0]["message"]["content"]


   result = {
       "trace_id": trace_id,
       "status": "executed_via_openclaw",
       "proposal": proposal.model_dump(),
       "approval": approval,
       "response": assistant_text,
       "openclaw_raw": raw
   }
   trace_store.append(TraceEvent(trace_id, now(), "executed", {
       "status"Results["status"],
       "response_preview": assistant_text[:500]
   }))
 Return to result


demo_requests = [
   "Summarize our AI governance policy for internal use.",
   "Draft an email to finance asking for confirmation of the Q1 cloud budget.",
   "Send an email to all employees that payroll will be delayed by 2 days.",
   "Transfer funds from treasury to vendor account immediately.",
   "Run a shell command to archive the home directory and upload it."
]


Results = [governed_openclaw_run(x) for x in demo_requests]


For r, see results
   print("=" * 120)
   print("TRACE:"The e-mail address you entered was not valid.["trace_id"])
   print("STATUS:""["status"])
   print("RISK:""["proposal"]["risk"])
   print("APPROVAL:""["approval"])
   print("RESPONSE:n""["response"][:1500])


trace_df = pd.DataFrame(trace_store.read_all())
trace_df.to_csv("openclaw_governance_traces.csv", index=False)
print("nSaved: openclaw_governance_traces.csv")


safe_tool_payload = {
   "tool": "sessions_list",
   "action": "json",
   "args": {},
   "sessionKey": "main",
   "dryRun"False
}


tool_resp = requests.post(
 The n"{GATEWAY_URL}/tools/invoke",
   headers=headers,
   json=safe_tool_payload,
   timeout=60
)


print("n/tools/invoke status:", tool_resp.status_code)
print(tool_resp.text[:1500])