chrome_cdp.py

"""
Chrome CDP via Extension Relay

Instead of launching Chrome with --remote-debugging-port (which Chrome blocks
on the default profile), we use a Chrome extension + local relay server:

  Browser Use  <--CDP WebSocket-->  Relay (:9222)  <--WebSocket-->  Extension
                                                                       |
                                                                 chrome.debugger
                                                                       |
                                                                  Your real Chrome

The relay server (relay.py) sits on port 9222 and speaks CDP to Browser Use
while forwarding commands to the extension via a separate WebSocket.

Requirements:
  - The Chrome extension (extension/) must be loaded in Chrome
  - Chrome must be running normally (user's real profile)
  - The relay server is started by this module
"""

import asyncio
import os
import subprocess
import sys
import time
from pathlib import Path

import requests

RELAY_PORT = 9222
CDP_URL = f"http://127.0.0.1:{RELAY_PORT}"
CDP_PORT = RELAY_PORT

RELAY_SCRIPT = Path(__file__).parent / "relay.py"
EXTENSION_DIR = Path(__file__).parent / "extension"

_relay_proc = None


def is_cdp_alive() -> bool:
    """Check if the relay is responding on the CDP port."""
    try:
        r = requests.get(f"{CDP_URL}/json/version", timeout=2)
        return r.status_code == 200
    except Exception:
        return False


def is_extension_connected() -> bool:
    """Check if the Chrome extension has connected to the relay."""
    try:
        r = requests.get(f"{CDP_URL}/json/version", timeout=2)
        if r.status_code == 200:
            return r.json().get("extensionConnected", False)
    except Exception:
        pass
    return False


def start_relay() -> subprocess.Popen:
    """Start the relay server as a background process."""
    global _relay_proc

    if is_cdp_alive():
        print("[CDP] Relay already running on port", RELAY_PORT)
        return _relay_proc

    print(f"[CDP] Starting relay server on port {RELAY_PORT}...")
    _relay_proc = subprocess.Popen(
        [sys.executable, str(RELAY_SCRIPT)],
        cwd=str(RELAY_SCRIPT.parent),
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        creationflags=(subprocess.CREATE_NO_WINDOW)
        if sys.platform == "win32" else 0,
    )

    # Wait for relay to start
    for _ in range(20):
        time.sleep(0.5)
        if is_cdp_alive():
            print("[CDP] Relay server ready")
            return _relay_proc
        if _relay_proc.poll() is not None:
            out = _relay_proc.stdout.read().decode("utf-8", errors="replace")
            raise RuntimeError(f"Relay server exited unexpectedly:\n{out}")

    raise TimeoutError("Relay server did not start within 10 seconds")


def stop_relay():
    """Stop the relay server."""
    global _relay_proc
    if _relay_proc and _relay_proc.poll() is None:
        _relay_proc.terminate()
        try:
            _relay_proc.wait(timeout=5)
        except subprocess.TimeoutExpired:
            _relay_proc.kill()
        print("[CDP] Relay server stopped")
    _relay_proc = None


def launch() -> str:
    """
    Start the relay server and wait for the extension to connect.
    Chrome should already be running with the extension installed.
    Returns the CDP WebSocket URL for Browser Use.
    """
    start_relay()

    if is_extension_connected():
        print("[CDP] Extension already connected")
        return f"ws://127.0.0.1:{RELAY_PORT}/devtools/browser"

    print("[CDP] Waiting for Chrome extension to connect...")
    print("[CDP] Make sure Chrome is running with the Agent CDP Relay extension loaded.")
    print(f"[CDP]   Extension dir: {EXTENSION_DIR}")

    for i in range(60):  # 30 seconds
        time.sleep(0.5)
        if is_extension_connected():
            print("[CDP] Extension connected!")
            return f"ws://127.0.0.1:{RELAY_PORT}/devtools/browser"

    raise TimeoutError(
        "Extension did not connect within 30 seconds.\n"
        "Make sure:\n"
        "  1. Chrome is running\n"
        "  2. The Agent CDP Relay extension is loaded (chrome://extensions)\n"
        f"     Extension folder: {EXTENSION_DIR}\n"
    )


def shutdown():
    """Stop the relay server."""
    stop_relay()
    print("[CDP] Shutdown complete.")


if __name__ == "__main__":
    import argparse
    import json

    parser = argparse.ArgumentParser(description="Chrome CDP relay manager")
    parser.add_argument(
        "command",
        choices=["launch", "status", "shutdown"],
        help="launch=start relay & wait for extension, status=check, shutdown=stop relay",
    )
    args = parser.parse_args()

    if args.command == "launch":
        ws_url = launch()
        print(f"CDP WebSocket: {ws_url}")
    elif args.command == "status":
        if is_cdp_alive():
            info = requests.get(f"{CDP_URL}/json/version").json()
            print(json.dumps(info, indent=2))
            if info.get("extensionConnected"):
                print("\nExtension: CONNECTED")
            else:
                print("\nExtension: NOT CONNECTED")
        else:
            print("Relay not running.")
    elif args.command == "shutdown":
        shutdown()