Building a Browser-Controlling Agent with Playwright

import base64
import json
from urllib.parse import urlparse
from playwright.sync_api import sync_playwright
 
ALLOWED_HOSTS = {"example.com", "www.example.com"}
 
class BrowserSession:
    def __init__(self):
        self._p = sync_playwright().start()
        self.browser = self._p.chromium.launch(headless=True)
        self.context = self.browser.new_context(viewport={"width": 1280, "height": 720})
        self.page = self.context.new_page()
        self.page.set_default_timeout(15_000)
 
    def close(self):
        self.context.close()
        self.browser.close()
        self._p.stop()
 
    def goto(self, url: str) -> dict:
        host = urlparse(url).hostname or ""
        if host not in ALLOWED_HOSTS:
            return {"ok": False, "error": f"host not allowed: {host}"}
        self.page.goto(url, wait_until="domcontentloaded")
        return {"ok": True, "url": self.page.url, "title": self.page.title()}
 
    def click(self, selector: str) -> dict:
        self.page.locator(selector).first.click()
        return {"ok": True, "url": self.page.url}
 
    def type_text(self, selector: str, text: str) -> dict:
        self.page.locator(selector).first.fill(text)
        return {"ok": True}
 
    def read_text(self, selector: str = "body", limit: int = 4000) -> dict:
        text = self.page.locator(selector).inner_text()[:limit]
        return {"ok": True, "text": text, "url": self.page.url}
 
    def screenshot(self) -> dict:
        png = self.page.screenshot(type="png", full_page=False)
        b64 = base64.b64encode(png).decode("ascii")
        return {
            "ok": True,
            "url": self.page.url,
            "image_data_url": f"data:image/png;base64,{b64}",
        }
 
TOOLS = [
    {"type": "function", "function": {
        "name": "goto",
        "description": "Navigate to an allowlisted http(s) URL.",
        "parameters": {"type": "object", "properties": {"url": {"type": "string"}}, "required": ["url"]},
    }},
    {"type": "function", "function": {
        "name": "click",
        "description": "Click the first element matching a CSS selector.",
        "parameters": {"type": "object", "properties": {"selector": {"type": "string"}}, "required": ["selector"]},
    }},
    {"type": "function", "function": {
        "name": "type_text",
        "description": "Fill a field matching a CSS selector.",
        "parameters": {
            "type": "object",
            "properties": {"selector": {"type": "string"}, "text": {"type": "string"}},
            "required": ["selector", "text"],
        },
    }},
    {"type": "function", "function": {
        "name": "read_text",
        "description": "Read visible text for a selector (truncated).",
        "parameters": {"type": "object", "properties": {"selector": {"type": "string"}, "limit": {"type": "integer"}}},
    }},
    {"type": "function", "function": {
        "name": "screenshot",
        "description": "Capture the current viewport as PNG for vision models.",
        "parameters": {"type": "object", "properties": {}},
    }},
]
 
def run_tool(session: BrowserSession, name: str, args: dict) -> dict:
    if name == "goto":
        return session.goto(args["url"])
    if name == "click":
        return session.click(args["selector"])
    if name == "type_text":
        return session.type_text(args["selector"], args["text"])
    if name == "read_text":
        return session.read_text(args.get("selector", "body"), args.get("limit", 4000))
    if name == "screenshot":
        return session.screenshot()
    return {"ok": False, "error": f"unknown tool {name}"}

def run_browser_agent(client, model: str, goal: str, max_turns: int = 12):
    session = BrowserSession()
    messages = [
        {"role": "system", "content": "Use browser tools to complete the goal. Prefer read_text; screenshot only when needed."},
        {"role": "user", "content": goal},
    ]
    try:
        for _ in range(max_turns):
            out = client.chat.completions.create(
                model=model, messages=messages, tools=TOOLS, tool_choice="auto"
            )
            msg = out.choices[0].message
            messages.append(msg)
            if not msg.tool_calls:
                return msg.content
            for call in msg.tool_calls:
                args = json.loads(call.function.arguments or "{}")
                result = run_tool(session, call.function.name, args)
                # Host: if image_data_url present, attach as multi-modal tool result
                messages.append({
                    "role": "tool",
                    "tool_call_id": call.id,
                    "content": json.dumps({k: v for k, v in result.items() if k != "image_data_url"}),
                })
        return {"status": "max_turns"}
    finally:
        session.close()