homeai/homeai-agent/openclaw-http-bridge.py

#!/usr/bin/env python3
"""
OpenClaw HTTP Bridge

A simple HTTP server that translates HTTP POST requests to OpenClaw CLI calls.
This allows Home Assistant (running in Docker on a different machine) to
communicate with OpenClaw via HTTP.

Usage:
    python3 openclaw-http-bridge.py [--port 8081]

Endpoints:
    POST /api/agent/message
    {
        "message": "Your message here",
        "agent": "main"
    }

    Returns:
    {
        "response": "OpenClaw response text"
    }
"""

import argparse
import json
import subprocess
import sys
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse


class OpenClawBridgeHandler(BaseHTTPRequestHandler):
    """HTTP request handler for OpenClaw bridge."""

    def log_message(self, format, *args):
        """Log requests to stderr."""
        print(f"[OpenClaw Bridge] {self.address_string()} - {format % args}")

    def _send_json_response(self, status_code: int, data: dict):
        """Send a JSON response."""
        self.send_response(status_code)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(json.dumps(data).encode())

    def do_POST(self):
        """Handle POST requests."""
        parsed_path = urlparse(self.path)

        # Only handle the agent message endpoint
        if parsed_path.path != "/api/agent/message":
            self._send_json_response(404, {"error": "Not found"})
            return

        # Read request body
        content_length = int(self.headers.get("Content-Length", 0))
        if content_length == 0:
            self._send_json_response(400, {"error": "Empty request body"})
            return

        try:
            body = self.rfile.read(content_length).decode()
            data = json.loads(body)
        except json.JSONDecodeError:
            self._send_json_response(400, {"error": "Invalid JSON"})
            return

        # Extract parameters
        message = data.get("message", "").strip()
        agent = data.get("agent", "main")

        if not message:
            self._send_json_response(400, {"error": "Message is required"})
            return

        # Call OpenClaw CLI (use full path for launchd compatibility)
        try:
            result = subprocess.run(
                ["/opt/homebrew/bin/openclaw", "agent", "--message", message, "--agent", agent],
                capture_output=True,
                text=True,
                timeout=30,
                check=True
            )
            response_text = result.stdout.strip()
            self._send_json_response(200, {"response": response_text})
        except subprocess.TimeoutExpired:
            self._send_json_response(504, {"error": "OpenClaw command timed out"})
        except subprocess.CalledProcessError as e:
            error_msg = e.stderr.strip() if e.stderr else "OpenClaw command failed"
            self._send_json_response(500, {"error": error_msg})
        except FileNotFoundError:
            self._send_json_response(500, {"error": "OpenClaw CLI not found"})
        except Exception as e:
            self._send_json_response(500, {"error": str(e)})

    def do_GET(self):
        """Handle GET requests (health check)."""
        parsed_path = urlparse(self.path)

        if parsed_path.path == "/status" or parsed_path.path == "/":
            self._send_json_response(200, {
                "status": "ok",
                "service": "OpenClaw HTTP Bridge",
                "version": "1.0.0"
            })
        else:
            self._send_json_response(404, {"error": "Not found"})


def main():
    """Run the HTTP bridge server."""
    parser = argparse.ArgumentParser(description="OpenClaw HTTP Bridge")
    parser.add_argument(
        "--port",
        type=int,
        default=8081,
        help="Port to listen on (default: 8081)"
    )
    parser.add_argument(
        "--host",
        default="0.0.0.0",
        help="Host to bind to (default: 0.0.0.0)"
    )
    args = parser.parse_args()

    server = HTTPServer((args.host, args.port), OpenClawBridgeHandler)
    print(f"OpenClaw HTTP Bridge running on http://{args.host}:{args.port}")
    print(f"Endpoint: POST http://{args.host}:{args.port}/api/agent/message")
    print("Press Ctrl+C to stop")

    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nShutting down...")
        server.shutdown()


if __name__ == "__main__":
    main()