62 lines
1.8 KiB
Python
62 lines
1.8 KiB
Python
"""
|
|
Unix socket client — sends commands to the native host relay socket.
|
|
Used by both the CLI and the public Python API.
|
|
"""
|
|
import json
|
|
import socket
|
|
import struct
|
|
import uuid
|
|
from typing import Any
|
|
|
|
SOCKET_PATH = "/tmp/browser-cli.sock"
|
|
|
|
|
|
class BrowserNotConnected(Exception):
|
|
"""Raised when the native host socket is not available."""
|
|
|
|
|
|
def send_command(command: str, args: dict | None = None) -> Any:
|
|
"""Send a command to the browser and return the response data."""
|
|
msg = {
|
|
"id": str(uuid.uuid4()),
|
|
"command": command,
|
|
"args": args or {},
|
|
}
|
|
payload = json.dumps(msg).encode("utf-8")
|
|
framed = struct.pack("<I", len(payload)) + payload
|
|
|
|
try:
|
|
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
|
|
sock.connect(SOCKET_PATH)
|
|
sock.sendall(framed)
|
|
response = _recv_all(sock)
|
|
except (FileNotFoundError, ConnectionRefusedError, OSError):
|
|
raise BrowserNotConnected(
|
|
"Cannot connect to browser.\n"
|
|
"Make sure:\n"
|
|
" 1. The browser-cli extension is installed and enabled\n"
|
|
" 2. The native host is registered: uv run browser-cli install chrome\n"
|
|
" 3. Your browser is running"
|
|
)
|
|
|
|
result = json.loads(response)
|
|
if not result.get("success", True):
|
|
raise RuntimeError(result.get("error", "unknown error from browser"))
|
|
return result.get("data")
|
|
|
|
|
|
def _recv_all(sock: socket.socket) -> bytes:
|
|
raw_len = _recv_exact(sock, 4)
|
|
msg_len = struct.unpack("<I", raw_len)[0]
|
|
return _recv_exact(sock, msg_len)
|
|
|
|
|
|
def _recv_exact(sock: socket.socket, n: int) -> bytes:
|
|
buf = b""
|
|
while len(buf) < n:
|
|
chunk = sock.recv(n - len(buf))
|
|
if not chunk:
|
|
raise ConnectionError("Socket closed before full message received")
|
|
buf += chunk
|
|
return buf
|