Vision Model Browser Agent

from dataclasses import dataclass
from typing import Optional, Literal

@dataclass
class BrowserAction:
    """A single action the agent can take in the browser."""
    type: Literal["click", "type", "scroll", "wait", "navigate", "done", "fail"]

    # For click: pixel coordinates
    x: Optional[int] = None
    y: Optional[int] = None

    # For type: text to enter
    text: Optional[str] = None

    # For scroll: direction and amount
    direction: Optional[Literal["up", "down"]] = None
    amount: int = 300  # pixels

    # For navigate: URL
    url: Optional[str] = None

    # For done/fail: result message
    result: Optional[str] = None

    # Reasoning (for debugging)
    reasoning: str = ""

"""
Vision-based browser agent. Takes screenshots, sends to a vision model,
executes actions, loops until done. Works on any website.
"""

import asyncio, base64, json, time
from typing import Optional
import anthropic
import httpx
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}"}

# Vision model client
model_client = anthropic.Anthropic()
MODEL = "claude-sonnet-4-6"  # Best price/performance for vision tasks

# Agent configuration
MAX_STEPS = 30       # Maximum perception-action loops
VIEWPORT_W = 1280    # Browser viewport width
VIEWPORT_H = 900     # Browser viewport height

SYSTEM_PROMPT = """You are a browser automation agent. You can see a screenshot of a web page
and must decide what action to take next to complete the user's task.

AVAILABLE ACTIONS (respond with exactly one JSON object):

{"type": "click", "x": 450, "y": 320, "reasoning": "Clicking the Sign In button"}
{"type": "type", "text": "hello world", "reasoning": "Typing search query into the search box"}
{"type": "scroll", "direction": "down", "amount": 500, "reasoning": "Scrolling to see more results"}
{"type": "wait", "reasoning": "Page is loading, waiting for content"}
{"type": "navigate", "url": "https://example.com", "reasoning": "Going to the target site"}
{"type": "done", "result": "Found 3 products matching the criteria: ...", "reasoning": "Task complete"}
{"type": "fail", "result": "Site shows a CAPTCHA that I cannot solve", "reasoning": "Cannot proceed"}

RULES:
- Always respond with a single JSON object. No other text.
- Click coordinates are relative to the viewport (top-left is 0,0).
- For "type": first click the input field, THEN type in the next step. One action per response.
- If the page looks the same after an action, try a different approach.
- After 3 failed attempts at the same thing, declare "fail".
- When the task is complete, use "done" with the result data.
- Include "reasoning" in every action so the user can debug your decisions.

VIEWPORT: {width}x{height} pixels."""


async def take_screenshot(page) -> str:
    """Take a screenshot and return as base64."""
    screenshot_bytes = await page.screenshot(
        type="png",
        full_page=False,  # Only the viewport — matches what the model should see
    )
    return base64.b64encode(screenshot_bytes).decode()


def ask_vision_model(
    task: str,
    screenshot_b64: str,
    history: list[dict],
) -> dict:
    """Send screenshot to vision model, get back an action."""

    messages = []

    # Include recent history (last 5 steps) for context
    for h in history[-5:]:
        messages.append({
            "role": "user",
            "content": [
                {"type": "image", "source": {
                    "type": "base64", "media_type": "image/png",
                    "data": h["screenshot"],
                }},
                {"type": "text", "text": f"Step {h['step']}: What action should I take?"},
            ],
        })
        messages.append({
            "role": "assistant",
            "content": json.dumps(h["action"]),
        })

    # Current screenshot
    messages.append({
        "role": "user",
        "content": [
            {"type": "image", "source": {
                "type": "base64", "media_type": "image/png",
                "data": screenshot_b64,
            }},
            {"type": "text", "text": f"Task: {task}\n\nWhat action should I take on this page?"},
        ],
    })

    response = model_client.messages.create(
        model=MODEL,
        max_tokens=512,
        system=SYSTEM_PROMPT.format(width=VIEWPORT_W, height=VIEWPORT_H),
        messages=messages,
    )

    # Parse the JSON action from the response
    text = response.content[0].text.strip()
    # Handle markdown code blocks
    if text.startswith("```"):
        text = text.split("```")[1]
        if text.startswith("json"):
            text = text[4:]
    return json.loads(text)


async def execute_action(page, action: dict) -> None:
    """Execute a browser action."""
    action_type = action["type"]

    if action_type == "click":
        x, y = action["x"], action["y"]
        await page.mouse.click(x, y)
        await page.wait_for_timeout(1000)  # Let the click register

    elif action_type == "type":
        await page.keyboard.type(action["text"], delay=50)
        await page.wait_for_timeout(500)

    elif action_type == "scroll":
        direction = action.get("direction", "down")
        amount = action.get("amount", 300)
        delta = amount if direction == "down" else -amount
        await page.mouse.wheel(0, delta)
        await page.wait_for_timeout(500)

    elif action_type == "wait":
        await page.wait_for_timeout(3000)

    elif action_type == "navigate":
        await page.goto(action["url"], timeout=30000)
        await page.wait_for_load_state("networkidle")

    # "done" and "fail" are handled by the caller


async def run_vision_agent(
    task: str,
    start_url: str = "",
    max_steps: int = MAX_STEPS,
) -> dict:
    """Run a vision-based browser agent to complete a task.

    Returns: {"success": bool, "result": str, "steps": int, "cost_estimate": float}
    """
    # 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.get('access_url', '?')}")

    await asyncio.sleep(20)  # Wait for container ready

    history = []
    result = {"success": False, "result": "Max steps reached", "steps": 0}

    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()
        await page.set_viewport_size({"width": VIEWPORT_W, "height": VIEWPORT_H})

        # Navigate to start URL if provided
        if start_url:
            await page.goto(start_url, timeout=30000)
            await page.wait_for_load_state("networkidle")

        # 2. Perception-action loop
        for step in range(1, max_steps + 1):
            # Perceive
            screenshot_b64 = await take_screenshot(page)

            # Reason
            action = ask_vision_model(task, screenshot_b64, history)
            action_type = action.get("type", "?")
            reasoning = action.get("reasoning", "")
            print(f"  Step {step}: {action_type} — {reasoning}")

            # Record history
            history.append({
                "step": step,
                "screenshot": screenshot_b64,
                "action": action,
            })

            # Check for terminal actions
            if action_type == "done":
                result = {
                    "success": True,
                    "result": action.get("result", "Task completed"),
                    "steps": step,
                }
                break
            elif action_type == "fail":
                result = {
                    "success": False,
                    "result": action.get("result", "Task failed"),
                    "steps": step,
                }
                break

            # Act
            try:
                await execute_action(page, action)
            except Exception as e:
                print(f"    Action failed: {e}")
                # Don't crash — let the model see the result and adapt

        # Final screenshot
        await page.screenshot(path="final-state.png", full_page=True)
        await browser.close()

    # Cleanup
    async with httpx.AsyncClient(timeout=10) as http:
        await http.delete(f"{SANDBOX_URL}/api/containers/{container_id}", headers=HEADERS)

    # Cost estimate: ~$0.003 per screenshot analysis (Sonnet 4.6)
    result["cost_estimate"] = round(result.get("steps", 0) * 0.003 + 0.004, 4)
    return result

# Example 1: Research task
result = asyncio.run(run_vision_agent(
    task="Go to news.ycombinator.com. Find the top story. Click into it. "
         "Read the first few paragraphs. Tell me what the article is about.",
    start_url="https://news.ycombinator.com",
))
print(f"Success: {result['success']}")
print(f"Result: {result['result']}")
print(f"Steps: {result['steps']}")
print(f"Cost: ${result['cost_estimate']}")

# Example 2: Price extraction
result = asyncio.run(run_vision_agent(
    task="Go to vercel.com/pricing. Extract all plan names and their monthly prices. "
         "Return them as a structured list.",
    start_url="https://vercel.com/pricing",
))

# Example 3: Form filling
result = asyncio.run(run_vision_agent(
    task="Go to the login page. Enter email 'test@example.com' and password 'demo123'. "
         "Click Sign In. Report whether login succeeded.",
    start_url="https://demo-portal.example.com/login",
))

# Extract a simplified DOM alongside the screenshot
dom_summary = await page.evaluate("""() => {
    const interesting = document.querySelectorAll(
        'button, a, input, select, textarea, [role="button"], [onclick]'
    );
    return Array.from(interesting).slice(0, 50).map(el => {
        const rect = el.getBoundingClientRect();
        return {
            tag: el.tagName.toLowerCase(),
            text: el.textContent?.trim()?.slice(0, 80),
            type: el.type || '',
            name: el.name || '',
            x: Math.round(rect.x + rect.width/2),
            y: Math.round(rect.y + rect.height/2),
            visible: rect.width > 0 && rect.height > 0,
        };
    }).filter(e => e.visible);
}""")

# Add to the prompt: "Here are the interactive elements and their positions: ..."
# This helps the model click more accurately because it has both
# visual (screenshot) and structural (DOM) information.