Browser Agent

1. search(task)        → find cached Playwright scripts for this task
2. LLM reviews         → decides: reuse, modify, or generate fresh
3. Execute script      → run Playwright in headless browser
4. upload(task, script) → cache successful scripts for next time

pip install raysurfer anthropic playwright
playwright install chromium

export ANTHROPIC_API_KEY=sk-ant-...
export RAYSURFER_API_KEY=rs_...

import asyncio
import os
from raysurfer import AsyncRaySurfer, FileWritten
from anthropic import AsyncAnthropic

SYSTEM_PROMPT = (
    "You are a browser automation expert. Write Python scripts using "
    "playwright.sync_api with chromium in headless mode. "
    "Print the final result to stdout. "
    "Respond with ONLY a JSON object: {\"script\": \"<python code>\", \"explanation\": \"<one line>\"}"
)

async def run_browser_task(task: str, url: str | None = None):
    rs = AsyncRaySurfer(api_key=os.environ["RAYSURFER_API_KEY"])
    anthropic = AsyncAnthropic()

    # 1. Search for cached scripts
    cached_script = None
    response = await rs.search(task=task, top_k=1, min_verdict_score=0.3)
    if response.matches:
        cached_script = response.matches[0].code_block.source

    # 2. Build prompt — include cached script if found
    url_line = f"\nStarting URL: {url}" if url else ""
    if cached_script:
        user_msg = (
            f"Task: {task}{url_line}\n\n"
            f"A cached script for a similar task:\n```python\n{cached_script}\n```\n\n"
            "Review it: run as-is if it fits, modify for the current task, "
            "or write a new script if not relevant."
        )
    else:
        user_msg = f"Task: {task}{url_line}"

    # 3. LLM generates or approves the script
    llm_response = await anthropic.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        system=SYSTEM_PROMPT,
        messages=[{"role": "user", "content": user_msg}],
    )

    import json
    parsed = json.loads(llm_response.content[0].text)
    script = parsed["script"]

    # 4. Execute the Playwright script
    import subprocess, sys, tempfile
    with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
        f.write(script)
        f.flush()
        result = subprocess.run([sys.executable, f.name], capture_output=True, text=True, timeout=60)

    if result.returncode == 0:
        print(f"Output: {result.stdout.strip()}")
        # 5. Cache the successful script
        await rs.upload(
            task=task,
            file_written=FileWritten(path="browser_script.py", content=script),
            succeeded=True,
            execution_logs=result.stdout,
            tags=["browser-automation", "playwright"],
        )
    else:
        print(f"Error: {result.stderr.strip()}")

    await rs.close()

asyncio.run(run_browser_task("Get the titles and prices of the first 5 books", url="https://books.toscrape.com"))

from playwright.sync_api import sync_playwright
import json

def scrape_books():
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page = browser.new_page()
        page.goto('https://books.toscrape.com', timeout=10000)
        page.wait_for_load_state('networkidle')

        books = []
        for book in page.locator('article.product_pod').all()[:5]:
            title = book.locator('h3 a').get_attribute('title')
            price = book.locator('p.price_color').inner_text()
            books.append({'title': title, 'price': price})

        print(json.dumps(books, indent=2))
        browser.close()

if __name__ == '__main__':
    scrape_books()

from playwright.sync_api import sync_playwright

def scrape_with_guards():
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page = browser.new_page()
        page.goto('https://example-store.com', timeout=15000)
        page.wait_for_load_state('networkidle')

        # Cookie consent banner
        cookie_btn = page.locator('button:has-text("Accept"), #cookie-accept, [data-testid="cookie-accept"]')
        if cookie_btn.count() > 0:
            cookie_btn.first.click()
            page.wait_for_timeout(500)

        # Age verification gate
        age_gate = page.locator('[data-testid="age-gate"], .age-verification')
        if age_gate.count() > 0:
            confirm_btn = age_gate.locator('button:has-text("Yes"), button:has-text("I am over")')
            if confirm_btn.count() > 0:
                confirm_btn.first.click()
                page.wait_for_timeout(500)

        # Dismiss newsletter/promo popup
        popup_close = page.locator('[aria-label="Close"], .modal-close, .popup-close, button:has-text("No thanks")')
        if popup_close.count() > 0:
            popup_close.first.click()
            page.wait_for_timeout(500)

        # "Load more" button — click until all items are visible or button disappears
        while True:
            load_more = page.locator('button:has-text("Load more"), button:has-text("Show more")')
            if load_more.count() == 0 or not load_more.first.is_visible():
                break
            load_more.first.click()
            page.wait_for_timeout(1000)

        # Login wall — skip scraping if login is required
        login_wall = page.locator('form[action*="login"], [data-testid="login-required"]')
        if login_wall.count() > 0:
            print("Login required — cannot proceed")
            browser.close()
            return

        # Now scrape the actual content
        items = page.locator('.product-card').all()
        results = []
        for item in items[:10]:
            name = item.locator('.product-name')
            price = item.locator('.product-price')
            results.append({
                'name': name.inner_text() if name.count() > 0 else 'N/A',
                'price': price.inner_text() if price.count() > 0 else 'N/A',
            })

        import json
        print(json.dumps(results, indent=2))
        browser.close()

if __name__ == '__main__':
    scrape_with_guards()

from raysurfer_browser_agent import RaysurferBrowserAgent

async with RaysurferBrowserAgent() as agent:
    # First run: generates script from scratch, caches it
    result = await agent.run("Get the top 3 stories from Hacker News")

    # Second run: LLM reviews cached script and reuses it
    result = await agent.run("Get the top 3 stories from Hacker News")
    print(result.output)       # scraped data
    print(result.cached)        # True — cached script was found
    print(result.reused_cache)  # True — LLM chose to reuse it