A Coding Implementation to Design an Enterprise AI Governance System Using OpenClaw Gateway Policy Engines, Approval Workflows and Auditable Agent Execution

!apt-get update -y
!apt-get install -y curl
!curl -fsSL https://deb.nodesource.com/setup_22.x | bash –
!apt-get install -y nodejs
!node -v
!npm -v
!npm install -g openclaw@latest
!pip -q install requests pandas pydantic

import os
import json
import time
import uuid
import secrets
import subprocess
import getpass
from pathlib import Path
from typing import Dict, Any
from dataclasses import dataclass, asdict
from datetime import datetime, timezone

import requests
import pandas as pd
from pydantic import BaseModel, 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)
WORKSPACE = OPENCLAW_HOME / “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()
while time.time() – start < timeout:
try:
r = requests.get(f”{GATEWAY_URL}/”, timeout=5)
if r.status_code in (200, 401, 403, 404):
return True
except Exception:
pass
time.sleep(2)
return False

assert wait_for_gateway(), Path(gateway_log).read_text()[-6000:]

headers = {
“Authorization”: f”Bearer {GATEWAY_TOKEN}”,
“Content-Type”: “application/json”
}

def openclaw_chat(messages, user=”demo-user”, agent_id=”main”, temperature=0.2):
payload = {
“model”: f”openclaw:{agent_id}”,
“messages”: messages,
“user”: user,
“temperature”: temperature,
“stream”: False
}
r = requests.post(
f”{GATEWAY_URL}/v1/chat/completions”,
headers=headers,
json=payload,
timeout=180
)
r.raise_for_status()
return r.json()

class ActionProposal(BaseModel):
user_request: str
category: str
risk: str
confidence: float = Field(ge=0.0, le=1.0)
requires_approval: bool
allow: bool
reason: str

def classify_request(user_request: str) -> ActionProposal:
text = user_request.lower()

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 text for t in 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 any(t in text for t in amber_terms):
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”
)

return 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”:
approved = False
note = “Rejected automatically in demo for red-risk request”
elif proposal.risk == “amber”:
approved = True
note = “Approved automatically in demo for amber-risk request”
else:
approved = True
note = “No approval required”
return {
“approved”: approved,
“reviewer”: “simulated_manager”,
“note”: note
}

@dataclass
class TraceEvent:
trace_id: str
ts: str
stage: str
payload: Dict[str, Any]

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

def append(self, event: TraceEvent):
with open(self.path, “a”) as f:
f.write(json.dumps(asdict(event)) + “\n”)

def read_all(self):
rows = []
with open(self.path, “r”) as f:
for line in f:
line = line.strip()
if line:
rows.append(json.loads(line))
return rows

trace_store = TraceStore()

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

SYSTEM_PROMPT = “””
You are an enterprise OpenClaw assistant operating under governance controls.

Rules:
– Never claim an action has been executed unless the governance layer explicitly allows it.
– For low-risk requests, answer normally and helpfully.
– For moderate-risk requests, propose a safe plan and mention any approvals or checks that would be needed.
– For high-risk requests, refuse to execute and instead provide a safer non-operational alternative such as a draft, checklist, summary, or review plan.
– Be concise but useful.
“””

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()))

approval = None
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 result

if proposal.risk == “amber” and not approval[“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 result

messages = [
{“role”: “system”, “content”: SYSTEM_PROMPT},
{“role”: “user”, “content”: f”Governance classification: {proposal.model_dump_json()}\n\nUser 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”: result[“status”],
“response_preview”: assistant_text[:500]
}))
return 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 in results:
print(“=” * 120)
print(“TRACE:”, r[“trace_id”])
print(“STATUS:”, r[“status”])
print(“RISK:”, r[“proposal”][“risk”])
print(“APPROVAL:”, r[“approval”])
print(“RESPONSE:\n”, r[“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(
f”{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])