#!/usr/bin/env python3 """ openclaw_bridge.py — Missminute OS Async MCP Tool Caller Thin async wrapper around the OpenClaw MCP stdio bridge process. Hermes skills or any custom Python code calls this to route tool invocations through the OpenClaw Gateway without managing the subprocess manually. Protocol: JSON-RPC 2.0 over stdin/stdout (MCP stdio transport). """ import asyncio import json import logging import os import time from typing import Any, Optional log = logging.getLogger("missminute.openclaw_bridge") class BridgeError(Exception): """Raised when the MCP bridge returns an error or is unresponsive.""" class OpenClawBridge: def __init__( self, gateway_url: str = "ws://127.0.0.1:18789", startup_timeout: float = 10.0, call_timeout: float = 30.0, max_restarts: int = 3, ): self.gateway_url = gateway_url self.startup_timeout = startup_timeout self.call_timeout = call_timeout self.max_restarts = max_restarts self._proc: Optional[asyncio.subprocess.Process] = None self._req_id: int = 0 self._restarts: int = 0 self._lock = asyncio.Lock() async def start(self): log.info(f"[bridge] Starting openclaw mcp serve → {self.gateway_url}") env = {**os.environ} try: self._proc = await asyncio.create_subprocess_exec( "openclaw", "mcp", "serve", "--url", self.gateway_url, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, env=env, ) except FileNotFoundError: raise BridgeError("openclaw binary not found. Ensure OpenClaw is installed and in PATH.") deadline = asyncio.get_event_loop().time() + self.startup_timeout while asyncio.get_event_loop().time() < deadline: if self._proc.returncode is not None: stderr = await self._proc.stderr.read() raise BridgeError(f"openclaw mcp serve exited immediately (rc={self._proc.returncode}): {stderr.decode()[:500]}") await asyncio.sleep(0.2) if self._is_alive(): break else: raise BridgeError("Bridge process did not become ready within startup_timeout") log.info(f"[bridge] Ready (PID={self._proc.pid})") async def stop(self): if self._proc and self._is_alive(): log.info(f"[bridge] Stopping (PID={self._proc.pid})") try: self._proc.terminate() await asyncio.wait_for(self._proc.wait(), timeout=5.0) except asyncio.TimeoutError: log.warning("[bridge] SIGTERM timed out — sending SIGKILL") self._proc.kill() await self._proc.wait() except Exception as e: log.warning(f"[bridge] Stop error: {e}") self._proc = None log.info("[bridge] Stopped") async def __aenter__(self): await self.start() return self async def __aexit__(self, *_): await self.stop() async def list_tools(self) -> list[dict]: result = await self._send("tools/list", {}) return result.get("tools", []) async def call_tool(self, tool_name: str, params: dict) -> Any: return await self._send("tools/call", {"name": tool_name, "arguments": params}) async def conversations_list(self) -> list[dict]: return await self._send("conversations/list", {}) async def messages_send(self, conversation_id: str, text: str) -> dict: return await self.call_tool("messages_send", {"conversationId": conversation_id, "text": text}) async def events_poll(self, conversation_id: Optional[str] = None) -> list[dict]: params = {} if conversation_id: params["conversationId"] = conversation_id result = await self._send("events/poll", params) return result.get("events", []) async def permissions_list_open(self) -> list[dict]: result = await self._send("permissions/list_open", {}) return result.get("permissions", []) async def permissions_resolve(self, permission_id: str, approved: bool) -> dict: return await self._send("permissions/resolve", {"id": permission_id, "approved": approved}) def _is_alive(self) -> bool: return self._proc is not None and self._proc.returncode is None async def _ensure_alive(self): if self._is_alive(): return if self._restarts >= self.max_restarts: raise BridgeError(f"Bridge process is dead and restart limit ({self.max_restarts}) has been reached.") log.warning(f"[bridge] Process is dead — restarting (attempt {self._restarts + 1}/{self.max_restarts})") self._restarts += 1 await asyncio.sleep(1.0) await self.start() async def _send(self, method: str, params: dict) -> Any: async with self._lock: await self._ensure_alive() self._req_id += 1 request = json.dumps({ "jsonrpc": "2.0", "id": self._req_id, "method": method, "params": params, }) + "\n" try: self._proc.stdin.write(request.encode()) await asyncio.wait_for(self._proc.stdin.drain(), timeout=self.call_timeout) except (BrokenPipeError, ConnectionResetError) as e: raise BridgeError(f"Bridge stdin broken: {e}") try: raw = await asyncio.wait_for(self._proc.stdout.readline(), timeout=self.call_timeout) except asyncio.TimeoutError: raise BridgeError(f"Bridge did not respond within {self.call_timeout}s for method '{method}'") if not raw: stderr_bytes = b"" if self._proc.stderr: try: stderr_bytes = await asyncio.wait_for(self._proc.stderr.read(1024), timeout=2.0) except asyncio.TimeoutError: pass raise BridgeError(f"Bridge stdout EOF (process may have crashed). Stderr: {stderr_bytes.decode()[:300]}") try: response = json.loads(raw.decode()) except json.JSONDecodeError as e: raise BridgeError(f"Bridge returned invalid JSON: {e}\nRaw: {raw[:300]}") if "error" in response: err = response["error"] raise BridgeError(f"MCP error {err.get('code', '?')}: {err.get('message', err)}") return response.get("result", {}) _default_bridge: Optional[OpenClawBridge] = None async def get_bridge(gateway_url: str = "ws://127.0.0.1:18789") -> OpenClawBridge: global _default_bridge if _default_bridge is None or not _default_bridge._is_alive(): _default_bridge = OpenClawBridge(gateway_url=gateway_url) await _default_bridge.start() return _default_bridge async def shutdown_bridge(): global _default_bridge if _default_bridge is not None: await _default_bridge.stop() _default_bridge = None async def _smoke_test(): logging.basicConfig(level=logging.DEBUG) print("OpenClaw Bridge — smoke test") print(f"Connecting to ws://127.0.0.1:18789 ...") async with OpenClawBridge() as bridge: tools = await bridge.list_tools() print(f"Available tools ({len(tools)}):") for t in tools: print(f" - {t.get('name')}: {t.get('description', '')[:60]}") if __name__ == "__main__": asyncio.run(_smoke_test())