Playwright Meets Cloud Browser

# Launch a Chrome browser in the cloud
curl -s -X POST "https://sandbox.dev.ab0t.com/api/browsers" \
  -H "Authorization: Bearer $SANDBOX_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "browser_type": "chrome",
    "homepage_url": "https://example.com",
    "idle_timeout_minutes": 30,
    "enable_debug_port": true
  }' | jq .

{
  "container_id": "ctr-7f3a2b1c",
  "browser_type": "chrome",
  "status": "running",
  "state": "provisioning",
  "cdp_url": "ws://54.186.23.117:9222/devtools/browser/a1b2c3d4-e5f6-...",
  "access_url": "https://ctr-7f3a2b1c.browser.ab0t.com?token=tok_9x8y7z...",
  "debug_port": 9222,
  "session_token": "tok_9x8y7z...",
  "created_at": "2026-03-27T14:30:00Z",
  "from_pool": false,
  "mode": "interactive+agent"
}

# pip install playwright httpx
# playwright install chromium  (only needed for type stubs, not the actual browser)

import asyncio
import httpx
from playwright.async_api import async_playwright

SANDBOX_URL = "https://sandbox.dev.ab0t.com"
API_KEY = "ab0t_sk_live_YOUR_KEY_HERE"


async def launch_cloud_browser(
    browser_type: str = "chrome",
    homepage: str = "",
) -> dict:
    """Launch a cloud browser and return the connection details."""
    async with httpx.AsyncClient(timeout=60) as http:
        resp = await http.post(
            f"{SANDBOX_URL}/api/browsers",
            headers={"Authorization": f"Bearer {API_KEY}"},
            json={
                "browser_type": browser_type,
                "homepage_url": homepage,
                "idle_timeout_minutes": 30,
            },
        )
        resp.raise_for_status()
        return resp.json()


async def wait_for_ready(container_id: str, max_wait: int = 60) -> None:
    """Poll until the container is ready (CDP available)."""
    async with httpx.AsyncClient(timeout=10) as http:
        for _ in range(max_wait // 2):
            resp = await http.get(
                f"{SANDBOX_URL}/api/containers/{container_id}/status",
                headers={"Authorization": f"Bearer {API_KEY}"},
            )
            data = resp.json()
            state = data.get("state", "")
            if state in ("ready", "assigned"):
                return
            await asyncio.sleep(2)
    raise TimeoutError(f"Container {container_id} didn't become ready in {max_wait}s")


async def terminate_browser(container_id: str) -> None:
    """Terminate a cloud browser. Billing stops immediately."""
    async with httpx.AsyncClient(timeout=10) as http:
        await http.delete(
            f"{SANDBOX_URL}/api/containers/{container_id}",
            headers={"Authorization": f"Bearer {API_KEY}"},
        )


async def main():
    # 1. Launch
    browser_info = await launch_cloud_browser("chrome", "https://news.ycombinator.com")
    container_id = browser_info["container_id"]
    cdp_url = browser_info["cdp_url"]
    print(f"Launched: {container_id}")
    print(f"Watch live: {browser_info['access_url']}")

    # 2. Wait for ready
    await wait_for_ready(container_id)
    print("Browser ready.")

    # 3. Connect Playwright
    async with async_playwright() as pw:
        browser = await pw.chromium.connect_over_cdp(cdp_url)
        context = browser.contexts[0]
        page = context.pages[0] if context.pages else await context.new_page()

        # 4. The page is already on Hacker News (set by homepage_url)
        await page.wait_for_load_state("networkidle")
        title = await page.title()
        print(f"Page title: {title}")

        # 5. Extract the top 30 stories
        stories = await page.evaluate("""() => {
            return Array.from(document.querySelectorAll('.athing')).map(row => {
                const titleEl = row.querySelector('.titleline > a');
                const subtext = row.nextElementSibling;
                const scoreEl = subtext?.querySelector('.score');
                const ageEl = subtext?.querySelector('.age');
                return {
                    rank: row.querySelector('.rank')?.textContent?.trim(),
                    title: titleEl?.textContent?.trim(),
                    url: titleEl?.href,
                    points: scoreEl?.textContent?.trim() || '0 points',
                    age: ageEl?.textContent?.trim() || '',
                };
            });
        }""")

        print(f"\nTop {len(stories)} stories:")
        for s in stories[:10]:
            print(f"  {s['rank']:>3s} {s['points']:>12s}  {s['title']}")

        # 6. Take a screenshot
        await page.screenshot(path="hn-frontpage.png", full_page=True)
        print(f"\nScreenshot saved: hn-frontpage.png")

        # 7. Click into the top story and extract content
        if stories:
            first_url = stories[0]["url"]
            if first_url and first_url.startswith("http"):
                await page.goto(first_url)
                await page.wait_for_load_state("domcontentloaded")
                body_text = await page.evaluate("() => document.body.innerText.slice(0, 500)")
                print(f"\nFirst story preview:\n{body_text}...")

        await browser.close()

    # 8. Terminate
    await terminate_browser(container_id)
    print(f"\nBrowser {container_id} terminated. Billing stopped.")


asyncio.run(main())

// npm install puppeteer-core
const puppeteer = require('puppeteer-core');

const CDP_URL = 'ws://54.186.23.117:9222/devtools/browser/a1b2c3d4...';

(async () => {
    const browser = await puppeteer.connect({ browserWSEndpoint: CDP_URL });
    const pages = await browser.pages();
    const page = pages[0] || await browser.newPage();

    await page.goto('https://news.ycombinator.com', { waitUntil: 'networkidle2' });

    // Extract stories
    const stories = await page.evaluate(() => {
        return Array.from(document.querySelectorAll('.athing')).slice(0, 10).map(row => ({
            title: row.querySelector('.titleline > a')?.textContent,
            url: row.querySelector('.titleline > a')?.href,
        }));
    });

    stories.forEach((s, i) => console.log(`${i+1}. ${s.title}`));

    await page.screenshot({ path: 'hn.png', fullPage: true });
    await browser.close();
})();

from playwright.sync_api import sync_playwright

cdp_url = "ws://54.186.23.117:9222/devtools/browser/a1b2c3d4..."

with sync_playwright() as pw:
    browser = pw.chromium.connect_over_cdp(cdp_url)
    page = browser.contexts[0].pages[0]

    page.goto("https://example.com")
    print(page.title())
    page.screenshot(path="example.png")

    browser.close()

# Extract product listings from an e-commerce page
await page.goto("https://store.example.com/products")
await page.wait_for_selector(".product-card")

products = await page.evaluate("""() => {
    return Array.from(document.querySelectorAll('.product-card')).map(card => ({
        name: card.querySelector('.product-name')?.textContent?.trim(),
        price: card.querySelector('.price')?.textContent?.trim(),
        rating: card.querySelector('.stars')?.getAttribute('data-rating'),
        url: card.querySelector('a.product-link')?.href,
        in_stock: !card.querySelector('.out-of-stock'),
        image_url: card.querySelector('img')?.src,
    }));
}""")

print(f"Found {len(products)} products")
for p in products[:5]:
    print(f"  {p['name']:40s} {p['price']:>10s} {'In stock' if p['in_stock'] else 'OUT'}")

# Log into a vendor portal
await page.goto("https://vendor-portal.example.com/login")
await page.fill('input[name="email"]', "finance@yourcompany.com")
await page.fill('input[name="password"]', "secure-password-from-vault")
await page.click('button[type="submit"]')

# Wait for redirect to dashboard
await page.wait_for_url("**/dashboard**", timeout=15000)
print("Logged in.")

# Navigate to the invoice section
await page.click('a[href="/invoices"]')
await page.wait_for_selector(".invoice-table")

# Download the latest invoice PDF
async with page.expect_download() as download_info:
    await page.click('.invoice-row:first-child .download-btn')
download = await download_info.value
await download.save_as(f"invoices/{download.suggested_filename}")
print(f"Downloaded: {download.suggested_filename}")

import base64
import anthropic  # pip install anthropic

# Take a screenshot of the current page
screenshot_bytes = await page.screenshot(full_page=False)
b64_image = base64.b64encode(screenshot_bytes).decode()

# Send to Claude's vision model
client = anthropic.Anthropic()
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{
        "role": "user",
        "content": [
            {
                "type": "image",
                "source": {"type": "base64", "media_type": "image/png", "data": b64_image},
            },
            {
                "type": "text",
                "text": "What products are shown on this page? List them with prices.",
            },
        ],
    }],
)
print(response.content[0].text)

# Capture all API responses the page makes
api_responses = []

async def capture_api(response):
    if "/api/" in response.url and response.status == 200:
        try:
            body = await response.json()
            api_responses.append({
                "url": response.url,
                "method": response.request.method,
                "data": body,
            })
        except:
            pass

page.on("response", capture_api)

# Navigate — the page's JavaScript will make API calls
await page.goto("https://app.example.com/dashboard")
await page.wait_for_timeout(5000)  # Let it load fully

print(f"Captured {len(api_responses)} API responses")
for r in api_responses:
    print(f"  {r['method']:>4s} {r['url']}")

# The api_responses list now contains the actual JSON data
# that the page's frontend fetched — often more structured than
# what's visible in the DOM. This is the scraping shortcut for
# single-page apps.

# Paginate through search results and collect all items
all_results = []
page_num = 1

await page.goto("https://search.example.com?q=ai+agents")

while True:
    await page.wait_for_selector(".result-item")

    # Extract results from this page
    results = await page.evaluate("""() =>
        Array.from(document.querySelectorAll('.result-item')).map(el => ({
            title: el.querySelector('.title')?.textContent?.trim(),
            url: el.querySelector('a')?.href,
            snippet: el.querySelector('.snippet')?.textContent?.trim(),
        }))
    """)
    all_results.extend(results)
    print(f"Page {page_num}: {len(results)} results (total: {len(all_results)})")

    # Check for next page
    next_btn = await page.query_selector('a.next-page:not(.disabled)')
    if not next_btn or page_num >= 10:
        break

    await next_btn.click()
    await page.wait_for_load_state("networkidle")
    page_num += 1

print(f"\nCollected {len(all_results)} total results across {page_num} pages")

import json

# After logging in successfully, save cookies
cookies = await context.cookies()
with open("vendor-cookies.json", "w") as f:
    json.dump(cookies, f)
print(f"Saved {len(cookies)} cookies")

# In a later session, restore cookies before navigating
with open("vendor-cookies.json") as f:
    saved_cookies = json.load(f)
await context.add_cookies(saved_cookies)

# Now navigate directly to the protected page — no login needed
await page.goto("https://vendor-portal.example.com/invoices")
# If cookies are valid, you're already logged in

try:
    await page.goto(url, timeout=30000)
    await page.wait_for_load_state("networkidle")
except Exception as e:
    await page.screenshot(path=f"error-{container_id}.png")
    print(f"Navigation failed: {e}. Screenshot saved for debugging.")
    # Continue to next URL instead of crashing the whole run