Connect to OpenClaw

npm install -g openclaw@latest
openclaw onboard --install-daemon

async def exec_local_command(
    self,
    command: str | dict,
    target_id: str | None = None,
    timeout: float = 10.0,
)

# Basic
response = await self.capability_worker.exec_local_command(user_inquiry)

# Long-running command — longer timeout
response = await self.capability_worker.exec_local_command(
    "compile large project",
    timeout=30.0,
)

# Specific target device
response = await self.capability_worker.exec_local_command(
    "check battery status",
    target_id="laptop",
)

# Trigger: "start coding session"
# Opens IDE, starts local servers, opens documentation
async def first_function(self):
    commands = [
        "open Visual Studio Code",
        "start local dev server on port 3000",
        "open browser to localhost:3000",
    ]
    for cmd in commands:
        await self.capability_worker.exec_local_command(cmd)
    await self.capability_worker.speak("Development environment is ready.")
    self.capability_worker.resume_normal_flow()

# Trigger: "check system health"
async def first_function(self):
    metrics = [
        ("CPU usage", "get cpu usage"),
        ("Memory usage", "get memory usage"),
        ("Disk space", "get disk usage"),
        ("Battery level", "get battery level"),
    ]
    report = []
    for name, cmd in metrics:
        response = await self.capability_worker.exec_local_command(cmd)
        report.append(f"{name}: {response}")
    await self.capability_worker.speak(", ".join(report))
    self.capability_worker.resume_normal_flow()

# Trigger: "take screenshot of active window"
async def first_function(self):
    user_inquiry = await self.capability_worker.wait_for_complete_transcription()

    if "full screen" in user_inquiry.lower():
        cmd = "screenshot fullscreen save to ~/Desktop"
    elif "active window" in user_inquiry.lower():
        cmd = "screenshot active window save to ~/Desktop"
    else:
        cmd = "screenshot selection save to ~/Desktop"

    response = await self.capability_worker.exec_local_command(cmd, timeout=15.0)
    await self.capability_worker.speak(f"Screenshot saved: {response}")
    self.capability_worker.resume_normal_flow()

# Trigger: "close all browsers"
async def first_function(self):
    response = await self.capability_worker.exec_local_command("list open browsers")

    if "none" in response.lower():
        await self.capability_worker.speak("No browsers are open.")
        self.capability_worker.resume_normal_flow()
        return

    await self.capability_worker.speak(f"Found: {response}. Close all?")
    confirmation = await self.capability_worker.user_response()

    if "yes" in confirmation.lower():
        await self.capability_worker.exec_local_command("close all browsers")
        await self.capability_worker.speak("All browsers closed.")
    else:
        await self.capability_worker.speak("Cancelled.")

    self.capability_worker.resume_normal_flow()

DANGEROUS_COMMANDS = ["rm -rf", "format", "delete system", "shutdown -h now"]

if any(danger in user_inquiry.lower() for danger in DANGEROUS_COMMANDS):
    await self.capability_worker.speak("I can't execute that for safety reasons.")
    return

if "restart" in user_inquiry.lower() or "shutdown" in user_inquiry.lower():
    confirmed = await self.capability_worker.run_confirmation_loop(
        "This will restart your computer. Are you sure?"
    )
    if not confirmed:
        await self.capability_worker.speak("Cancelled.")
        return

# Default is fine for quick commands
response = await self.capability_worker.exec_local_command("open Chrome")

# Long-running — bump the timeout
response = await self.capability_worker.exec_local_command(
    "compile entire project",
    timeout=60.0,
)

response = await self.capability_worker.exec_local_command("get battery level")
# Raw: "Battery: 73% (charging, 2:15 remaining)"
# Spoken: "Battery is at 73 percent."
battery = extract_percentage(response)
await self.capability_worker.speak(f"Battery is at {battery} percent.")

try:
    response = await self.capability_worker.exec_local_command(user_inquiry, timeout=15.0)
    if "error" in response.lower() or "failed" in response.lower():
        await self.capability_worker.speak("That command didn't work. Try something else.")
    else:
        # process success
        ...
except asyncio.TimeoutError:
    await self.capability_worker.speak("That took too long. It might still be running.")
except Exception as e:
    self.worker.editor_logging_handler.error(f"Command failed: {e}")
    await self.capability_worker.speak("Something went wrong. Check the logs.")

workflow = [
    ("Opening calendar", "open Calendar app"),
    ("Starting video", "open Zoom"),
    ("Opening notes", "open Notes app"),
]
for description, command in workflow:
    await self.capability_worker.speak(description)
    await self.capability_worker.exec_local_command(command)
    await asyncio.sleep(1)

await self.capability_worker.speak("Ready for your meeting.")

Voice Input → OpenHome Ability → exec_local_command()
                                        ↓
                                 OpenClaw Client
                                   (via WebSocket)
                                        ↓
                                 OpenClaw Daemon
                                  (with LLM API)
                                        ↓
                                Local System Execution
                                  (apps, files, etc.)
                                        ↓
                        Response ← AI Formatting ← Template