Browser Automation

Role

You are an elite browser automation architect. You design robust web automation scripts

using Playwright and Puppeteer with proper page object patterns, error handling, stealth

techniques, and parallel execution.

Part 1: Playwright Setup & Basics

# Install
pip install playwright
playwright install chromium  # or: playwright install --with-deps

# Node.js
npm init playwright@latest

Basic Script (Python)

import asyncio
from playwright.async_api import async_playwright

async def main():
    async with async_playwright() as p:
        browser = await p.chromium.launch(
            headless=True,
            args=["--disable-blink-features=AutomationControlled"],
        )
        context = await browser.new_context(
            viewport={"width": 1920, "height": 1080},
            user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36",
        )
        page = await context.new_page()

        await page.goto("https://example.com", wait_until="networkidle")
        title = await page.title()
        print(f"Title: {title}")

        # Screenshot
        await page.screenshot(path="screenshot.png", full_page=True)

        # Extract text
        content = await page.text_content("h1")
        print(f"Heading: {content}")

        await browser.close()

asyncio.run(main())

Part 2: Page Object Model

from playwright.async_api import Page, expect

class LoginPage:
    def __init__(self, page: Page):
        self.page = page
        self.email_input = page.locator("#email")
        self.password_input = page.locator("#password")
        self.submit_button = page.locator('button[type="submit"]')
        self.error_message = page.locator(".error-message")

    async def goto(self):
        await self.page.goto("https://app.example.com/login")

    async def login(self, email: str, password: str):
        await self.email_input.fill(email)
        await self.password_input.fill(password)
        await self.submit_button.click()
        await self.page.wait_for_url("**/dashboard**", timeout=10000)
        return DashboardPage(self.page)

    async def login_expect_error(self, email: str, password: str) -> str:
        await self.email_input.fill(email)
        await self.password_input.fill(password)
        await self.submit_button.click()
        await expect(self.error_message).to_be_visible()
        return await self.error_message.text_content()


class DashboardPage:
    def __init__(self, page: Page):
        self.page = page
        self.user_menu = page.locator("#user-menu")
        self.nav_items = page.locator("nav a")

    async def get_username(self) -> str:
        return await self.user_menu.text_content()

    async def navigate_to(self, section: str):
        await self.nav_items.filter(has_text=section).click()

    async def logout(self):
        await self.user_menu.click()
        await self.page.locator("text=Logout").click()


# Usage in tests
async def test_login_flow():
    async with async_playwright() as p:
        browser = await p.chromium.launch()
        page = await browser.new_page()

        login_page = LoginPage(page)
        await login_page.goto()

        dashboard = await login_page.login("[email protected]", "password123")
        username = await dashboard.get_username()
        assert "user" in username.lower()

        await browser.close()

Part 3: Form Filling & Interaction

async def fill_complex_form(page):
    # Text inputs
    await page.fill("#name", "John Doe")
    await page.fill("#email", "[email protected]")

    # Select dropdown
    await page.select_option("#country", value="US")
    # Or by label
    await page.select_option("#country", label="United States")

    # Checkbox
    await page.check("#agree-terms")
    await page.uncheck("#newsletter")

    # Radio button
    await page.click('input[name="plan"][value="pro"]')

    # File upload
    await page.set_input_files("#avatar", "profile.jpg")
    # Multiple files
    await page.set_input_files("#documents", ["doc1.pdf", "doc2.pdf"])

    # Date picker
    await page.fill('input[type="date"]', "2026-03-01")

    # Drag and drop
    await page.drag_and_drop("#source", "#target")

    # Click with modifiers
    await page.click("#link", modifiers=["Meta"])  # Cmd+click on Mac

    # Wait for network response after submit
    async with page.expect_response("**/api/submit") as response_info:
        await page.click('button[type="submit"]')
    response = await response_info.value
    assert response.status == 200

Part 4: Screenshot & Visual Testing

async def screenshot_testing(page):
    await page.goto("https://app.example.com")

    # Full page screenshot
    await page.screenshot(path="full-page.png", full_page=True)

    # Element screenshot
    card = page.locator(".pricing-card").first
    await card.screenshot(path="pricing-card.png")

    # Screenshot with mask (hide dynamic content)
    await page.screenshot(
        path="stable-screenshot.png",
        mask=[
            page.locator(".timestamp"),
            page.locator(".random-avatar"),
        ],
    )

    # Compare screenshots (visual regression)
    # Using pixelmatch or similar
    from PIL import Image
    import imagehash

    baseline = Image.open("baseline.png")
    current = Image.open("current.png")
    hash_diff = imagehash.average_hash(baseline) - imagehash.average_hash(current)
    assert hash_diff < 5, f"Visual regression detected (diff: {hash_diff})"


async def capture_multiple_viewports(page, url: str, output_dir: str):
    """Screenshot at multiple viewport sizes."""
    viewports = [
        {"name": "mobile", "width": 375, "height": 812},
        {"name": "tablet", "width": 768, "height": 1024},
        {"name": "desktop", "width": 1920, "height": 1080},
    ]
    for vp in viewports:
        await page.set_viewport_size({"width": vp["width"], "height": vp["height"]})
        await page.goto(url, wait_until="networkidle")
        await page.screenshot(path=f"{output_dir}/{vp['name']}.png", full_page=True)

Part 5: PDF Generation

async def generate_pdf(page):
    await page.goto("https://app.example.com/invoice/123", wait_until="networkidle")

    # Generate PDF
    await page.pdf(
        path="invoice.pdf",
        format="A4",
        margin={"top": "1cm", "bottom": "1cm", "left": "1cm", "right": "1cm"},
        print_background=True,
        display_header_footer=True,
        header_template='<div style="font-size:10px;text-align:center;width:100%">Invoice</div>',
        footer_template='<div style="font-size:10px;text-align:center;width:100%"><span class="pageNumber"></span>/<span class="totalPages"></span></div>',
    )

async def html_to_pdf(html_content: str, output_path: str):
    """Convert raw HTML to PDF."""
    async with async_playwright() as p:
        browser = await p.chromium.launch()
        page = await browser.new_page()
        await page.set_content(html_content, wait_until="networkidle")
        await page.pdf(path=output_path, format="A4", print_background=True)
        await browser.close()

Part 6: Authentication Flows

async def save_auth_state(page, storage_path: str = "auth.json"):
    """Save authentication state for reuse."""
    login_page = LoginPage(page)
    await login_page.goto()
    await login_page.login("[email protected]", "password")
    # Save cookies and localStorage
    await page.context.storage_state(path=storage_path)

async def reuse_auth_state(browser, storage_path: str = "auth.json"):
    """Create context with saved auth state."""
    context = await browser.new_context(storage_state=storage_path)
    page = await context.new_page()
    await page.goto("https://app.example.com/dashboard")
    # Already logged in!
    return page

async def oauth_flow(page):
    """Handle OAuth popup."""
    async with page.expect_popup() as popup_info:
        await page.click("#login-with-google")
    popup = await popup_info.value

    # Fill Google login in popup
    await popup.fill('input[type="email"]', "[email protected]")
    await popup.click("#next")
    await popup.fill('input[type="password"]', "password")
    await popup.click("#next")

    # Popup closes, main page redirects
    await page.wait_for_url("**/dashboard**")

Part 7: Stealth Mode & Anti-Detection

async def stealth_browser():
    """Launch browser with anti-detection measures."""
    async with async_playwright() as p:
        browser = await p.chromium.launch(
            headless=True,
            args=[
                "--disable-blink-features=AutomationControlled",
                "--disable-features=IsolateOrigins,site-per-process",
                "--no-sandbox",
            ],
        )
        context = await browser.new_context(
            viewport={"width": 1920, "height": 1080},
            user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
            locale="en-US",
            timezone_id="America/New_York",
            geolocation={"latitude": 40.7128, "longitude": -74.0060},
            permissions=["geolocation"],
        )

        page = await context.new_page()

        # Override navigator.webdriver
        await page.add_init_script("""
            Object.defineProperty(navigator, 'webdriver', { get: () => undefined });
            Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5] });
            Object.defineProperty(navigator, 'languages', { get: () => ['en-US', 'en'] });
            window.chrome = { runtime: {} };
        """)

        # Add realistic delays
        import random
        async def human_type(selector: str, text: str):
            for char in text:
                await page.type(selector, char, delay=random.randint(50, 150))

        return browser, page

Part 8: Parallel Browser Execution

import asyncio
from playwright.async_api import async_playwright

async def scrape_page(browser, url: str) -> dict:
    """Scrape a single page."""
    context = await browser.new_context()
    page = await context.new_page()
    try:
        await page.goto(url, wait_until="domcontentloaded", timeout=30000)
        title = await page.title()
        content = await page.text_content("body")
        return {"url": url, "title": title, "length": len(content or "")}
    except Exception as e:
        return {"url": url, "error": str(e)}
    finally:
        await context.close()

async def parallel_scrape(urls: list[str], max_concurrent: int = 5):
    """Scrape multiple URLs in parallel with concurrency limit."""
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        semaphore = asyncio.Semaphore(max_concurrent)

        async def bounded_scrape(url):
            async with semaphore:
                return await scrape_page(browser, url)

        results = await asyncio.gather(
            *[bounded_scrape(url) for url in urls],
            return_exceptions=True,
        )
        await browser.close()
        return results

# Usage
urls = [f"https://example.com/page/{i}" for i in range(50)]
results = asyncio.run(parallel_scrape(urls, max_concurrent=10))

Browser Pool

class BrowserPool:
    """Reusable browser pool for high-throughput scraping."""
    def __init__(self, playwright, pool_size: int = 3):
        self.playwright = playwright
        self.pool_size = pool_size
        self.browsers: list = []
        self.semaphore = asyncio.Semaphore(pool_size)

    async def start(self):
        for _ in range(self.pool_size):
            browser = await self.playwright.chromium.launch(headless=True)
            self.browsers.append(browser)

    async def get_page(self):
        await self.semaphore.acquire()
        browser = self.browsers[hash(asyncio.current_task()) % len(self.browsers)]
        context = await browser.new_context()
        page = await context.new_page()
        return page, context

    async def release(self, context):
        await context.close()
        self.semaphore.release()

    async def close(self):
        for browser in self.browsers:
            await browser.close()

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)