Capture browser network events

export BROWSER_SESSION_ID="your-active-browser-id"
# BROWSER_USE_API_KEY must already be set. Do not print the resulting CDP URL.
set -o pipefail
BROWSER_CDP_URL="$(curl --fail --silent --show-error \
  "https://api.browser-use.com/api/v4/browsers/$BROWSER_SESSION_ID" \
  -H "X-Browser-Use-API-Key: $BROWSER_USE_API_KEY" \
  | jq --exit-status --raw-output '.cdpUrl // empty')" || exit 1
export BROWSER_CDP_URL

import asyncio
import json
import os
from urllib.parse import urlsplit

from playwright.async_api import async_playwright


def http_origin(value):
    try:
        url = urlsplit(value)
        if url.scheme not in {"http", "https"} or not url.hostname:
            return None
        host = f"[{url.hostname}]" if ":" in url.hostname else url.hostname
        port = url.port
    except ValueError:
        return None  # A malformed URL must not interrupt the capture.
    default_port = {"http": 80, "https": 443}[url.scheme]
    suffix = f":{port}" if port not in (None, default_port) else ""
    return f"{url.scheme}://{host}{suffix}"


async def main():
    async with async_playwright() as p:
        cdp_url = os.environ["BROWSER_CDP_URL"]
        try:
            browser = await p.chromium.connect_over_cdp(cdp_url)
        except Exception:
            # Transport errors can include the credential-bearing CDP URL.
            raise RuntimeError("Check the live CDP URL and browser state") from None
        context = browser.contexts[0]
        sessions = []
        fd = os.open("network.jsonl", os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)

        with os.fdopen(fd, "w") as output:
            def write(row):
                output.write(json.dumps(row) + "\n")
                output.flush()

            async def observe(page, page_number):
                cdp = await context.new_cdp_session(page)
                sessions.append(cdp)

                def record(event, params):
                    row = {"event": event, "page": page_number,
                           "requestId": params["requestId"]}
                    if event == "request":
                        request = params["request"]
                        row.update(method=request["method"],
                                   origin=http_origin(request["url"]),
                                   resourceType=params.get("type"))
                    elif event == "response":
                        response = params["response"]
                        row.update(status=response["status"],
                                   mimeType=response["mimeType"])
                    elif event == "finished":
                        row["bytes"] = params["encodedDataLength"]
                    else:
                        row["error"] = params.get("errorText")
                    write(row)

                events = {
                    "requestWillBeSent": "request", "responseReceived": "response",
                    "loadingFinished": "finished", "loadingFailed": "failed",
                }
                for name, event in events.items():
                    cdp.on(f"Network.{name}",
                           lambda params, event=event: record(event, params))
                await cdp.send("Network.enable")

            try:
                for number, page in enumerate(context.pages, start=1):
                    await observe(page, number)
                print(f"Observing {len(sessions)} existing page(s) for 30 seconds")
                await asyncio.sleep(30)
            finally:
                for cdp in sessions:
                    try:
                        await cdp.detach()
                    except Exception:
                        pass  # The page or browser may already have closed.


asyncio.run(main())