Multi-Modal & Browser Agents Basics

import base64
from pathlib import Path
 
def image_data_url(path: str, mime: str = "image/png") -> str:
    raw = Path(path).read_bytes()
    b64 = base64.b64encode(raw).decode("ascii")
    return f"data:{mime};base64,{b64}"
 
url = image_data_url("fixtures/checkout-error.png")

def describe_screenshot(client, model: str, data_url: str) -> str:
    resp = client.chat.completions.create(
        model=model,  # must support vision - verify at build
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": "Describe the UI and any error in 3 bullets."},
                {"type": "image_url", "image_url": {"url": data_url}},
            ],
        }],
        max_tokens=300,
    )
    return resp.choices[0].message.content

import json
from pydantic import BaseModel, ValidationError
 
class UiFinding(BaseModel):
    page_type: str
    has_error: bool
    error_text: str | None
    primary_cta: str | None
 
def extract_ui(client, model: str, data_url: str) -> UiFinding:
    prompt = (
        "Return JSON only with keys: page_type, has_error, "
        "error_text, primary_cta. No markdown."
    )
    text = describe_with_prompt(client, model, data_url, prompt)
    return UiFinding.model_validate(json.loads(text))

def tool_screenshot(page) -> dict:
    png = page.screenshot(type="png", full_page=False)
    b64 = base64.b64encode(png).decode("ascii")
    return {
        "ok": True,
        "note": f"viewport screenshot, url={page.url}",
        "image_data_url": f"data:image/png;base64,{b64}",
    }

def run_observe_only(goal: str, page, client, model: str, max_turns: int = 3):
    messages = [{"role": "system", "content": "You inspect pages. Call screenshot when needed."}]
    messages.append({"role": "user", "content": goal})
    for _ in range(max_turns):
        out = client.chat.completions.create(model=model, messages=messages, tools=[screenshot_tool_schema])
        msg = out.choices[0].message
        if not msg.tool_calls:
            return msg.content
        for call in msg.tool_calls:
            obs = tool_screenshot(page)
            messages.append(tool_message_with_image(call.id, obs))
    return {"status": "max_turns"}

from playwright.sync_api import sync_playwright
 
def open_page(url: str):
    p = sync_playwright().start()
    browser = p.chromium.launch(headless=True)
    context = browser.new_context(viewport={"width": 1280, "height": 720})
    page = context.new_page()
    page.goto(url, wait_until="domcontentloaded", timeout=30_000)
    return p, browser, page

def observe(page, client, model: str, question: str) -> str:
    try:
        snap = page.locator("body").aria_snapshot()  # API name may vary - verify
        if len(snap) > 50:
            return client_text_only(client, model, question, snap)
    except Exception:
        pass
    obs = tool_screenshot(page)
    return describe_with_image(client, model, question, obs["image_data_url"])

def prune_images(messages: list, keep_last: int = 1) -> list:
    images = [i for i, m in enumerate(messages) if message_has_image(m)]
    drop = set(images[:-keep_last]) if keep_last else set(images)
    pruned = []
    for i, m in enumerate(messages):
        if i in drop:
            pruned.append(replace_image_with_text(m, "[prior screenshot omitted]"))
        else:
            pruned.append(m)
    return pruned

checklist = {
    "vision_model": "returns UiFinding on 5 fixtures",
    "screenshot_tool": "image + url note each call",
    "viewport": "fixed 1280x720",
    "max_turns": 5,
    "cleanup": "browser.close in finally",
    "no_secrets_in_shots": True,
}
assert all(checklist.values())