OpenClaw Gateway

┌─────────────────┐     ┌──────────────────┐     ┌─────────────────┐
│  Agentic AI     │────▶│ OpenClaw Gateway │────▶│   ClawdBot      │
│  (Voice Agent)  │◀────│  (ws://18789)    │◀────│   (Skills)      │
└─────────────────┘     └──────────────────┘     └─────────────────┘
         │                      │                         │
         │                      │                         ▼
    Intent Analysis      JSON-RPC 2.0           ┌─────────────────┐
    "Play Spotify"        Protocol              │  YouTube Skill  │
         │                      │               │  Spotify Skill  │
         └──────────────────────┘               │  Email Skill    │
              sessions_send                     │  Search Skill   │
                                                 │  Message Skill  │
                                                 └─────────────────┘

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "sessions_send",
  "params": {
    "message": {
      "message_type": "action",
      "call_id": "call_abc123",
      "action_type": "execute_command",
      "parameters": {
        "command": "Play Shape of You on Spotify"
      },
      "timestamp": "2024-03-03T12:00:00Z"
    }
  }
}

class GatewayClient:
    def __init__(
        self,
        url: str = "ws://127.0.0.1:18789",
        max_reconnect_attempts: int = 10,
        reconnect_base_delay: float = 1.0,
        reconnect_max_delay: float = 60.0,
    ):
        """Initialize the gateway client."""

    @property
    def is_connected(self) -> bool:
        """Check if connected to gateway."""

    async def connect(self) -> None:
        """Connect to the gateway with automatic reconnection."""

    async def disconnect(self) -> None:
        """Disconnect from the gateway."""

    async def send_message(self, message: GatewayMessage) -> None:
        """Send a message to the gateway.
        
        Queues message if disconnected and sends when reconnected.
        """

import asyncio
from agenticai.gateway.client import GatewayClient
from agenticai.gateway.messages import ActionMessage

async def main():
    client = GatewayClient(url="ws://127.0.0.1:18789")
    
    # Connect
    await client.connect()
    
    # Wait for connection
    await asyncio.sleep(1)
    
    if client.is_connected:
        # Send command
        await client.send_message(
            ActionMessage(
                call_id="test_call",
                action_type="execute_command",
                parameters={
                    "command": "Open YouTube and search for python tutorial"
                }
            )
        )
        
        print("Command sent to ClawdBot!")
    
    # Cleanup
    await asyncio.sleep(5)
    await client.disconnect()

asyncio.run(main())

async def _send_to_clawdbot_async(self, command: str) -> str | None:
    """Send command to ClawdBot agent and wait for response."""
    
    # Use clawdbot agent CLI
    cmd = [
        "clawdbot", "agent",
        "--session-id", "agent:main:main",
        "--message", command,
        "--timeout", "90",
    ]
    
    # Run async and capture output
    process = await asyncio.create_subprocess_exec(
        *cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    
    stdout, stderr = await asyncio.wait_for(
        process.communicate(),
        timeout=95
    )
    
    response = stdout.decode('utf-8').strip()
    return response