A Coding Implementation for Enterprise AI Governance Using the OpenClaw Gateway, Policy Engine, Approval Workflows, and Auditable Agent Execution

by ai-intensify
0 comments
A coding implementation for designing enterprise AI governance systems using the OpenGL Gateway policy engine, approval workflow, and auditable agent execution

Overview

This tutorial builds an enterprise-grade AI governance system using OpenClaw and Python. The approach sets up the OpenClaw runtime and launches the OpenClaw Gateway so a Python environment can interact with a real agent through the OpenClaw API. On top of that, it adds a governance layer that classifies requests by risk, enforces approval policies, and routes only sanctioned actions to the agent for execution. The combination shows how an organization can deploy autonomous AI while preserving visibility, traceability, and operational oversight. For a contrasting, lighter-weight take on agent building, see this guide to creating a first code agent.

Setting up the environment

!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}"

The first step prepares everything an OpenClaw-based governance system needs. It installs Node.js, the OpenClaw CLI, and the required Python libraries so the notebook can communicate with the OpenClaw Gateway and its supporting tools. The OpenAI API key is collected securely through a hidden terminal prompt, and the directories and variables the rest of the workflow depends on are initialized.

Configuring the OpenClaw Gateway

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)

Next, a configuration file defines the agent defaults and gateway settings: the operating scope, model selection, authentication token, and HTTP endpoint. With these in place, the gateway can expose an interface compatible with OpenAI-style requests. Running the OpenClaw diagnostic (“doctor”) utility resolves common compatibility issues before the gateway process that powers the agent is started.

Connecting Python and defining the action schema

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

The workflow waits for the gateway to finish initializing before sending any requests. It then builds the HTTP headers and a helper function that submits chat requests to the gateway through the /v1/chat/completions endpoint. An ActionProposal schema is defined here as well; it represents the governance taxonomy applied to each user request later in the pipeline.

Classifying risk and adding approvals

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)

The governance logic analyzes each incoming request and assigns it a risk level. A classification function labels requests as green, amber, or red according to their potential operational impact. A simulated human-approval step is added for higher-risk actions, along with a trace-event structure that records each governance decision and the action taken.

Running the governed execution workflow

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()}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": 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))

Finally, the pieces are combined into a fully controlled execution workflow around the OpenClaw agent. Every stage of the request lifecycle — classification, approval decision, agent execution, and trace recording — is logged. Running several example requests through the system saves governance traces for auditing and shows how a tool is invoked through the gateway. The result is a practical framework in which an OpenClaw-powered assistant operates under explicit policy enforcement, approval workflows, and audit logging.

Limitations and what to watch

The walkthrough is a demonstration rather than a production deployment, and several gaps would need to be closed before real-world use. The human-approval step is simulated, so a genuine system would require an actual reviewer interface, authentication, and role-based access control. Risk classification based on simple green/amber/red labels is a starting point; production policies typically need richer rules, and a misclassified request could either be blocked unnecessarily or, more seriously, allowed through. Audit traces are only trustworthy if they are tamper-evident and stored securely, and secrets such as API keys must be managed through a proper secrets manager rather than collected in a notebook. As with any agent system, the underlying model can still produce incorrect or unexpected output, so governance reduces risk without eliminating the need for monitoring.

Conclusion

The implementation pairs an OpenClaw agent runtime with a governance layer to show how autonomous AI can run under strict enterprise controls. By combining policy enforcement, approval workflows, and trace logging around an OpenAI-compatible gateway, it illustrates a repeatable pattern for deploying agents where oversight, accountability, and auditability are requirements rather than afterthoughts.

Related Articles