openclaw adapter

Author: rcholic

README.md

@@ -64,7 +64,7 @@ secure_agent.run()
 | LangChain | Supported |
 | Playwright | Supported |
 | PydanticAI | Supported |
-| OpenClaw | Planned |
+| OpenClaw | Supported |
 
 ## Architecture
 

docs/user-manual.md

@@ -482,6 +482,133 @@ secure_agent = SecureAgent(
 
 ---
 
+### OpenClaw
+
+[OpenClaw](https://github.com/openclaw/openclaw) is a local-first AI agent framework that connects to messaging platforms. SecureAgent integrates with OpenClaw CLI via HTTP proxy interception.
+
+#### Architecture
+
+The OpenClaw adapter works by:
+1. Starting an HTTP proxy server that intercepts OpenClaw skill calls
+2. Enforcing authorization policies before forwarding to actual skills
+3. Managing the OpenClaw CLI subprocess lifecycle
+
+```
+Python SecureAgent
+      │
+      ├─── HTTP Proxy (localhost:8788) ───┐
+      │                                    ▼
+      └─── spawns ──────► OpenClaw CLI ──► predicate-snapshot skill
+                              │
+                              └─► Browser actions (authorized)
+```
+
+#### Basic Usage
+
+```python
+from predicate_secure import SecureAgent
+from predicate_secure.openclaw_adapter import OpenClawConfig
+
+# Create OpenClaw configuration
+openclaw_config = OpenClawConfig(
+    cli_path="/usr/local/bin/openclaw",  # Or None to use PATH
+    skill_proxy_port=8788,
+    skill_name="predicate-snapshot",
+)
+
+# Or use a dict:
+# openclaw_config = {
+#     "openclaw_cli_path": "/usr/local/bin/openclaw",
+#     "skill_proxy_port": 8788,
+# }
+
+# Wrap with SecureAgent
+secure_agent = SecureAgent(
+    agent=openclaw_config,
+    policy="policies/openclaw.yaml",
+    mode="strict",
+)
+
+# Run a task
+result = secure_agent.run(task="Navigate to example.com and take a snapshot")
+```
+
+#### Policy Example for OpenClaw
+
+```yaml
+# policies/openclaw.yaml
+rules:
+  # Allow snapshot skill
+  - action: "openclaw.skill.predicate-snapshot"
+    resource: "*"
+    effect: allow
+
+  # Allow clicking elements
+  - action: "openclaw.skill.predicate-act.click"
+    resource: "element:*"
+    effect: allow
+
+  # Allow typing (but not in password fields)
+  - action: "openclaw.skill.predicate-act.type"
+    resource: "element:*"
+    effect: allow
+    conditions:
+      - not_contains: ["password", "ssn"]
+
+  # Block scroll actions
+  - action: "openclaw.skill.predicate-act.scroll"
+    resource: "*"
+    effect: deny
+
+  # Default deny
+  - action: "*"
+    resource: "*"
+    effect: deny
+```
+
+#### Proxy Configuration
+
+The HTTP proxy intercepts requests to OpenClaw skills:
+
+```python
+from predicate_secure.openclaw_adapter import create_openclaw_adapter, OpenClawConfig
+
+config = OpenClawConfig(skill_proxy_port=8788)
+
+# Custom authorizer function
+def my_authorizer(action: str, context: dict) -> bool:
+    # Custom authorization logic
+    if "snapshot" in action:
+        return True
+    print(f"Blocked: {action}")
+    return False
+
+adapter = create_openclaw_adapter(config, authorizer=my_authorizer)
+
+# Start proxy
+adapter.start_proxy()
+print("Proxy running on http://localhost:8788")
+
+# ... run OpenClaw tasks ...
+
+# Cleanup
+adapter.cleanup()
+```
+
+#### Environment Variables
+
+Configure OpenClaw skill to use the proxy:
+
+```bash
+export PREDICATE_PROXY_URL="http://localhost:8788"
+```
+
+The OpenClaw adapter automatically sets this when starting the CLI subprocess.
+
+**Full example:** [examples/openclaw_browser_automation.py](../examples/openclaw_browser_automation.py)
+
+---
+
 ## Modes
 
 SecureAgent supports four execution modes:

examples/openclaw_browser_automation.py

@@ -0,0 +1,191 @@
+"""
+Example: Secure OpenClaw agent with authorization and verification.
+
+This example demonstrates how to wrap an OpenClaw CLI agent with predicate-secure
+to add pre-action authorization and audit logging for browser automation tasks.
+
+Prerequisites:
+    1. OpenClaw CLI installed (npm install -g openclaw)
+    2. predicate-snapshot skill installed in OpenClaw
+    3. Policy file (policies/openclaw_browser.yaml)
+"""
+
+from pathlib import Path
+
+from predicate_secure import SecureAgent
+from predicate_secure.openclaw_adapter import OpenClawConfig
+
+# Example policy file content (create this as policies/openclaw_browser.yaml):
+EXAMPLE_POLICY = """
+rules:
+  # Allow OpenClaw snapshot skill
+  - action: "openclaw.skill.predicate-snapshot"
+    resource: "*"
+    effect: allow
+
+  # Allow clicking on known safe domains
+  - action: "openclaw.skill.predicate-act.click"
+    resource: "element:*"
+    effect: allow
+    conditions:
+      # Only allow on safe domains (you'd check current URL in real policy)
+      - domain_matches: ["*.example.com", "*.trusted-site.com"]
+
+  # Allow typing in form fields (with restrictions)
+  - action: "openclaw.skill.predicate-act.type"
+    resource: "element:*"
+    effect: allow
+    conditions:
+      # Prevent entering sensitive data
+      - not_contains: ["password", "ssn", "credit"]
+
+  # Block scrolling to prevent UI confusion
+  - action: "openclaw.skill.predicate-act.scroll"
+    resource: "*"
+    effect: deny
+
+  # Default deny for safety
+  - action: "*"
+    resource: "*"
+    effect: deny
+"""
+
+
+def main():
+    """Run OpenClaw agent with secure authorization."""
+    # Create OpenClaw configuration
+    openclaw_config = OpenClawConfig(
+        cli_path="/usr/local/bin/openclaw",  # Or None to use PATH
+        skill_proxy_port=8788,  # Port for HTTP proxy
+        skill_name="predicate-snapshot",
+        working_dir=str(Path.home() / ".openclaw"),
+    )
+
+    # You could also use a dict instead of OpenClawConfig:
+    # openclaw_config = {
+    #     "openclaw_cli_path": "/usr/local/bin/openclaw",
+    #     "skill_proxy_port": 8788,
+    #     "skill_name": "predicate-snapshot",
+    # }
+
+    # Create policy file
+    policy_dir = Path("policies")
+    policy_dir.mkdir(exist_ok=True)
+    policy_file = policy_dir / "openclaw_browser.yaml"
+
+    if not policy_file.exists():
+        policy_file.write_text(EXAMPLE_POLICY)
+        print(f"Created example policy at {policy_file}")
+
+    # Wrap OpenClaw with SecureAgent
+    secure_agent = SecureAgent(
+        agent=openclaw_config,
+        policy=str(policy_file),
+        mode="strict",  # Fail-closed mode
+        principal_id="openclaw-agent-01",
+        trace_format="console",
+    )
+
+    print(f"[predicate-secure] Detected framework: {secure_agent.framework.value}")
+    print(f"[predicate-secure] Mode: {secure_agent.config.mode}")
+    print(f"[predicate-secure] Policy: {secure_agent.config.effective_policy_path}")
+
+    # Example task
+    task = "Navigate to example.com and take a snapshot"
+
+    print(f"\n[OpenClaw] Running task: {task}")
+    print("[predicate-secure] Starting HTTP proxy for skill interception...")
+
+    try:
+        # Run the OpenClaw task with authorization
+        result = secure_agent.run(task=task)
+        print(f"\n[OpenClaw] Task completed successfully")
+        print(f"Return code: {result.get('returncode', 'N/A')}")
+        print(f"Output: {result.get('stdout', '')[:200]}...")
+    except Exception as e:
+        print(f"\n[predicate-secure] Task failed: {e}")
+
+
+def example_with_debug_mode():
+    """Run OpenClaw agent in debug mode for troubleshooting."""
+    openclaw_config = OpenClawConfig(skill_proxy_port=8789)
+
+    secure_agent = SecureAgent(
+        agent=openclaw_config,
+        mode="debug",  # Human-readable trace output
+        trace_format="console",
+        trace_colors=True,
+    )
+
+    print("\n[Debug Mode] Running OpenClaw agent with full tracing...")
+
+    task = "Check if example.com loads correctly"
+
+    try:
+        result = secure_agent.run(task=task)
+        print("\n[Debug] Task trace complete")
+    except Exception as e:
+        print(f"\n[Debug] Error occurred: {e}")
+
+
+def example_with_manual_proxy():
+    """
+    Example showing how to manually control the proxy lifecycle.
+
+    Useful when you want to keep the proxy running across multiple tasks.
+    """
+    from predicate_secure.openclaw_adapter import create_openclaw_adapter
+
+    openclaw_config = OpenClawConfig(skill_proxy_port=8790)
+
+    # Create adapter manually
+    def authorizer(action: str, context: dict) -> bool:
+        """Simple authorizer that allows snapshot but blocks act."""
+        if "snapshot" in action:
+            return True
+        print(f"[Authorizer] Blocked action: {action}")
+        return False
+
+    adapter = create_openclaw_adapter(openclaw_config, authorizer=authorizer)
+
+    try:
+        # Start proxy (stays running)
+        adapter.start_proxy()
+        print("[Proxy] Started on http://localhost:8790")
+
+        # Run multiple tasks with same proxy
+        tasks = [
+            "Take snapshot of example.com",
+            "Take snapshot of httpbin.org",
+        ]
+
+        for task in tasks:
+            print(f"\n[Task] {task}")
+            adapter.start_cli(task)
+            # Process would run in background
+            # In real usage, you'd wait for completion
+
+    finally:
+        # Clean up
+        adapter.cleanup()
+        print("\n[Proxy] Stopped")
+
+
+if __name__ == "__main__":
+    print("=" * 60)
+    print("predicate-secure: OpenClaw Agent Example")
+    print("=" * 60)
+
+    # Uncomment the example you want to run:
+
+    # Example 1: Basic usage with policy file
+    # main()
+
+    # Example 2: Debug mode with full tracing
+    # example_with_debug_mode()
+
+    # Example 3: Manual proxy control
+    # example_with_manual_proxy()
+
+    print("\nNote: Uncomment one of the example functions in __main__ to run")
+    print("Make sure OpenClaw CLI is installed and in your PATH")

src/predicate_secure/__init__.py

@@ -410,6 +410,8 @@ def run(self, task: str | None = None) -> Any:
                 result = self._run_langchain(task)
             elif self._wrapped.framework == Framework.PYDANTIC_AI.value:
                 result = self._run_pydantic_ai(task)
+            elif self._wrapped.framework == Framework.OPENCLAW.value:
+                result = self._run_openclaw(task)
             else:
                 raise NotImplementedError(
                     f"run() not implemented for framework: {self._wrapped.framework}"
@@ -481,6 +483,52 @@ def _run_pydantic_ai(self, task: str | None) -> Any:
         """Run PydanticAI agent with authorization."""
         raise NotImplementedError("PydanticAI integration not yet implemented.")
 
+    def _run_openclaw(self, task: str | None) -> Any:
+        """Run OpenClaw CLI agent with authorization."""
+        try:
+            from .openclaw_adapter import OpenClawAdapter, create_openclaw_adapter
+        except ImportError:
+            raise NotImplementedError(
+                "OpenClaw integration requires openclaw_adapter module. "
+                "Ensure all dependencies are installed."
+            )
+
+        # Get or create adapter
+        if not hasattr(self._wrapped, "openclaw_adapter"):
+            # Create adapter from original agent config
+            authorizer = self._create_pre_action_authorizer()
+            adapter = create_openclaw_adapter(self._wrapped.original, authorizer)
+            self._wrapped.metadata["openclaw_adapter"] = adapter
+        else:
+            adapter = self._wrapped.metadata.get("openclaw_adapter")
+
+        if not isinstance(adapter, OpenClawAdapter):
+            raise ValueError("Invalid OpenClaw adapter")
+
+        # Start proxy server
+        adapter.start_proxy()
+
+        try:
+            # Start CLI with task
+            if task is None:
+                raise ValueError("Task is required for OpenClaw agents")
+
+            process = adapter.start_cli(task)
+
+            # Wait for completion
+            stdout, stderr = process.communicate()
+
+            # Check for errors
+            if process.returncode != 0:
+                raise RuntimeError(f"OpenClaw CLI failed: {stderr}")
+
+            return {"stdout": stdout, "stderr": stderr, "returncode": process.returncode}
+
+        finally:
+            # Cleanup
+            adapter.stop_cli()
+            adapter.stop_proxy()
+
     def trace_step(
         self,
         action: str,