AI Agent Fills Out Web Forms

import asyncio, httpx, json
from playwright.async_api import async_playwright

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

# Portal credentials (from your secrets manager, NOT hardcoded)
PORTAL = {
    "name": "Vendor Portal A",
    "login_url": "https://portal.vendor-a.com/login",
    "email": "finance@yourcompany.com",
    "password": "FROM_SECRETS_MANAGER",
    "dashboard_pattern": "**/dashboard**",
}


async def login_to_portal(page, portal: dict) -> bool:
    """Log into a vendor portal. Returns True on success."""

    await page.goto(portal["login_url"], timeout=30000)
    await page.wait_for_load_state("networkidle")

    # Try common email/password selectors
    email_selectors = [
        'input[type="email"]',
        'input[name="email"]',
        'input[name="username"]',
        'input[id="email"]',
        'input[placeholder*="email" i]',
        'input[placeholder*="username" i]',
    ]

    for sel in email_selectors:
        el = await page.query_selector(sel)
        if el:
            await el.fill(portal["email"])
            break
    else:
        print("Could not find email field. Taking screenshot for debugging.")
        await page.screenshot(path="login-failed-email.png")
        return False

    # Fill password
    pw_field = await page.query_selector('input[type="password"]')
    if pw_field:
        await pw_field.fill(portal["password"])
    else:
        await page.screenshot(path="login-failed-password.png")
        return False

    # Submit
    submit_selectors = [
        'button[type="submit"]',
        'input[type="submit"]',
        'button:has-text("Sign in")',
        'button:has-text("Log in")',
        'button:has-text("Login")',
    ]
    for sel in submit_selectors:
        btn = await page.query_selector(sel)
        if btn:
            await btn.click()
            break

    # Wait for redirect to dashboard
    try:
        await page.wait_for_url(portal["dashboard_pattern"], timeout=15000)
        print(f"Logged into {portal['name']}")
        return True
    except:
        # Check for error messages
        error = await page.query_selector('[class*="error"], [class*="alert-danger"], [role="alert"]')
        if error:
            msg = await error.inner_text()
            print(f"Login failed: {msg}")
        await page.screenshot(path="login-failed-redirect.png")
        return False

import json, os

COOKIE_DIR = "cookies"
os.makedirs(COOKIE_DIR, exist_ok=True)


async def save_cookies(context, portal_name: str):
    """Save cookies after successful login."""
    cookies = await context.cookies()
    path = f"{COOKIE_DIR}/{portal_name}.json"
    with open(path, "w") as f:
        json.dump(cookies, f)
    print(f"Saved {len(cookies)} cookies for {portal_name}")


async def restore_cookies(context, portal_name: str) -> bool:
    """Restore cookies from a previous session. Returns True if cookies existed."""
    path = f"{COOKIE_DIR}/{portal_name}.json"
    if not os.path.exists(path):
        return False
    with open(path) as f:
        cookies = json.load(f)
    await context.add_cookies(cookies)
    print(f"Restored {len(cookies)} cookies for {portal_name}")
    return True


async def smart_login(page, context, portal: dict) -> bool:
    """Try cookies first, fall back to full login."""
    if await restore_cookies(context, portal["name"]):
        # Navigate to the dashboard directly
        await page.goto(portal["login_url"].replace("/login", "/dashboard"))
        await page.wait_for_load_state("networkidle")

        # Check if we're actually on the dashboard (cookies still valid)
        if "dashboard" in page.url or "home" in page.url:
            print(f"Session restored for {portal['name']} (skipped login)")
            return True

    # Cookies expired or didn't exist — do full login
    success = await login_to_portal(page, portal)
    if success:
        await save_cookies(context, portal["name"])
    return success

import pyotp  # pip install pyotp

async def handle_totp_mfa(page, totp_secret: str):
    """Enter a TOTP code if the MFA page appears."""
    # Check if we're on an MFA page
    mfa_field = await page.query_selector(
        'input[name*="code"], input[name*="otp"], input[name*="mfa"], '
        'input[placeholder*="code" i], input[autocomplete="one-time-code"]'
    )
    if not mfa_field:
        return  # No MFA required

    # Generate the current TOTP code
    totp = pyotp.TOTP(totp_secret)
    code = totp.now()
    print(f"Entering TOTP code: {code}")

    await mfa_field.fill(code)

    # Submit
    submit = await page.query_selector(
        'button[type="submit"], button:has-text("Verify"), button:has-text("Continue")'
    )
    if submit:
        await submit.click()
        await page.wait_for_load_state("networkidle")

# Field mapping for Vendor Portal A's monthly report form
FIELD_MAP = {
    "report_period":   'select[name="period"]',
    "department":      'select[name="dept"]',
    "total_amount":    'input[name="amount"]',
    "invoice_number":  'input[name="inv_no"]',
    "notes":           'textarea[name="notes"]',
    "attachment":      'input[type="file"]',
}

async def fill_form_by_selector(page, data: dict, field_map: dict):
    """Fill a form using CSS selector mappings."""
    for field_name, selector in field_map.items():
        value = data.get(field_name)
        if value is None:
            continue

        el = await page.query_selector(selector)
        if not el:
            print(f"  WARNING: Field not found: {field_name} ({selector})")
            continue

        tag = await el.evaluate("el => el.tagName.toLowerCase()")

        if tag == "select":
            await page.select_option(selector, value)
            print(f"  Selected: {field_name} = {value}")

        elif tag == "input":
            input_type = await el.get_attribute("type") or "text"

            if input_type == "file":
                await el.set_input_files(value)
                print(f"  Uploaded: {field_name} = {value}")

            elif input_type == "checkbox":
                checked = await el.is_checked()
                if bool(value) != checked:
                    await el.click()
                print(f"  Checkbox: {field_name} = {value}")

            elif input_type == "date":
                await el.fill(value)  # Format: YYYY-MM-DD
                print(f"  Date: {field_name} = {value}")

            else:
                await el.fill("")  # Clear first
                await el.fill(str(value))
                print(f"  Filled: {field_name} = {value}")

        elif tag == "textarea":
            await el.fill(str(value))
            print(f"  Filled: {field_name} = {value[:50]}...")

    print("Form filled.")

import base64, anthropic

client = anthropic.Anthropic()

async def fill_form_with_vision(page, data: dict):
    """Use a vision model to understand the form and fill it."""

    # Take a screenshot of the form
    screenshot = await page.screenshot(full_page=False)
    b64 = base64.b64encode(screenshot).decode()

    # Ask the model to identify form fields and map data to them
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": [
                {"type": "image", "source": {
                    "type": "base64", "media_type": "image/png", "data": b64
                }},
                {"type": "text", "text": f"""Look at this form. I need to fill it with this data:

{json.dumps(data, indent=2)}

For each form field visible in the screenshot, tell me:
1. The CSS selector to target it (be specific)
2. What value to enter from my data
3. The interaction type: fill, select, click, or upload

Return JSON array: [{"selector": "...", "value": "...", "action": "fill"}]
Only include fields you can see in the screenshot."""},
            ],
        }],
    )

    # Parse the model's response and execute actions
    text = response.content[0].text
    # Extract JSON from the response (handle markdown code blocks)
    if "```" in text:
        text = text.split("```json")[-1].split("```")[0]
    actions = json.loads(text)

    for action in actions:
        sel = action["selector"]
        val = action["value"]
        act = action.get("action", "fill")

        try:
            if act == "select":
                await page.select_option(sel, val)
            elif act == "click":
                await page.click(sel)
            elif act == "upload":
                await page.set_input_files(sel, val)
            else:
                await page.fill(sel, val)
            print(f"  {act}: {sel} = {val}")
        except Exception as e:
            print(f"  FAILED: {sel} - {e}")

async def submit_and_verify(page) -> dict:
    """Submit the form, wait for confirmation, take evidence screenshot."""

    # Screenshot before submit (evidence of what we're submitting)
    await page.screenshot(path="pre-submit.png", full_page=True)

    # Find and click the submit button
    submit_selectors = [
        'button[type="submit"]',
        'button:has-text("Submit")',
        'button:has-text("Save")',
        'input[type="submit"]',
        'button:has-text("Send")',
    ]
    submitted = False
    for sel in submit_selectors:
        btn = await page.query_selector(sel)
        if btn and await btn.is_visible():
            await btn.click()
            submitted = True
            break

    if not submitted:
        return {"success": False, "error": "Submit button not found"}

    # Wait for response
    await page.wait_for_load_state("networkidle")
    await asyncio.sleep(2)  # Let any success/error messages render

    # Check for success indicators
    success_indicators = [
        '[class*="success"]',
        ':has-text("submitted successfully")',
        ':has-text("thank you")',
        ':has-text("confirmation")',
        ':has-text("received")',
    ]
    for sel in success_indicators:
        try:
            el = await page.query_selector(sel)
            if el and await el.is_visible():
                msg = await el.inner_text()
                await page.screenshot(path="confirmation.png", full_page=True)
                return {"success": True, "message": msg.strip()[:200]}
        except:
            continue

    # Check for error indicators
    error_el = await page.query_selector(
        '[class*="error"], [class*="danger"], [role="alert"]'
    )
    if error_el:
        msg = await error_el.inner_text()
        await page.screenshot(path="submit-error.png")
        return {"success": False, "error": msg.strip()[:200]}

    # Ambiguous — take a screenshot and let the human check
    await page.screenshot(path="submit-ambiguous.png", full_page=True)
    return {"success": None, "message": "Submitted but could not confirm. Check screenshot."}

import asyncio, httpx, json
from playwright.async_api import async_playwright

SANDBOX_URL = "https://sandbox.dev.ab0t.com"
API_KEY = "ab0t_sk_live_YOUR_KEY"
HEADERS = {"Authorization": f"Bearer {API_KEY}"}


async def main():
    # 1. Launch a cloud browser
    async with httpx.AsyncClient(timeout=60) as http:
        resp = await http.post(
            f"{SANDBOX_URL}/api/browsers", headers=HEADERS,
            json={"browser_type": "chrome", "idle_timeout_minutes": 15},
        )
        browser_info = resp.json()
        container_id = browser_info["container_id"]
        cdp_url = browser_info["cdp_url"]
        print(f"Browser: {container_id}")
        print(f"Watch live: {browser_info['access_url']}")

    # Wait for ready
    await asyncio.sleep(20)

    # 2. Connect Playwright and do the work
    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()

        # Login
        portal = {
            "name": "expense-portal",
            "login_url": "https://expenses.yourcompany.com/login",
            "email": "finance@yourcompany.com",
            "password": "FROM_VAULT",
            "dashboard_pattern": "**/dashboard**",
        }
        logged_in = await smart_login(page, context, portal)
        if not logged_in:
            print("Login failed. Aborting.")
            return

        # Navigate to the expense form
        await page.click('a[href="/expenses/new"]')
        await page.wait_for_selector('form#expense-form')

        # Fill the form
        expense_data = {
            "report_period": "2026-03",
            "department": "Engineering",
            "total_amount": "4,521.38",
            "invoice_number": "INV-2026-0342",
            "notes": "Cloud infrastructure costs for March 2026",
            "attachment": "/tmp/receipt-march-2026.pdf",
        }
        await fill_form_by_selector(page, expense_data, FIELD_MAP)

        # Submit and verify
        result = await submit_and_verify(page)
        print(f"\nResult: {json.dumps(result, indent=2)}")

        await browser.close()

    # 3. Terminate the browser
    async with httpx.AsyncClient(timeout=10) as http:
        await http.delete(f"{SANDBOX_URL}/api/containers/{container_id}", headers=HEADERS)
    print("Browser terminated.")


asyncio.run(main())

# Run 20 portal submissions in parallel
portals = load_portal_configs()  # 20 portals with credentials
data = load_form_data()          # Data to submit to each portal

async def submit_to_portal(portal, form_data, semaphore):
    async with semaphore:
        # Launch browser, login, fill, submit (same pattern as above)
        ...

semaphore = asyncio.Semaphore(10)  # 10 concurrent portals
tasks = [
    submit_to_portal(portal, data[portal["name"]], semaphore)
    for portal in portals
]
results = await asyncio.gather(*tasks)

# Report
successes = [r for r in results if r["success"]]
failures = [r for r in results if not r["success"]]
print(f"Submitted: {len(successes)}/{len(results)}")
print(f"Failed: {len(failures)} (check screenshots)")

# Total cost: 20 browsers x 5 min each = ~$0.17
# Total time: ~5 minutes (parallel) vs ~6 hours (human, sequential)