async playwright backend

Author: ollmer

src/agentlab/backends/browser/playwright.py

@@ -1,22 +1,29 @@
-import asyncio
 import logging
 from io import BytesIO
 from typing import Any, Callable
 
 from PIL import Image
-from playwright.async_api import Browser, Page, async_playwright
+from playwright.async_api import Page as AsyncPage
+from playwright.async_api import async_playwright
+from playwright.sync_api import Page as SyncPage
+from playwright.sync_api import sync_playwright
 
 from agentlab.actions import ToolCall, ToolSpec
-from agentlab.backends.browser.base import BrowserBackend
+from agentlab.backends.browser.base import AsyncBrowserBackend, BrowserBackend
 
 logger = logging.getLogger(__name__)
 
 
-class AsyncPlaywright(BrowserBackend):
+_pw = None  # Global Playwright instance for SyncPlaywright
+_browser = None  # Global Browser instance for SyncPlaywright
+
+
+class SyncPlaywright(BrowserBackend):
+    """Fully synchronous Playwright backend using playwright.sync_api."""
+
+    has_pw_page: bool = True
     _actions: dict[str, Callable]
-    _loop: asyncio.AbstractEventLoop
-    _browser: Browser
-    _page: Page
+    _page: SyncPage
 
     def model_post_init(self, __context: Any):
         self._actions = {
@@ -29,37 +36,139 @@ def model_post_init(self, __context: Any):
             "browser_mouse_click_xy": self.browser_mouse_click_xy,
         }
 
-    def initialize(self, loop: asyncio.AbstractEventLoop | None = None):
-        self._loop = loop or asyncio.get_event_loop()
-        self._loop.run_until_complete(self.ainitialize())
+    def initialize(self):
+        global _pw, _browser
+        if _pw is None:
+            _pw = sync_playwright().start()
+        if _browser is None:
+            _browser = _pw.chromium.launch(headless=True, chromium_sandbox=True)
+        self._page = _browser.new_page()
+
+    @property
+    def page(self) -> SyncPage:
+        return self._page
+
+    def browser_press_key(self, key: str):
+        """Press a key on the keyboard."""
+        self._page.keyboard.press(key)
+
+    def browser_type(self, selector: str, text: str):
+        """Type text into the focused element."""
+        self._page.type(selector, text)
+
+    def browser_click(self, selector: str):
+        """Click on a selector."""
+        self._page.click(selector, timeout=3000, strict=True)
+
+    def browser_drag(self, from_selector: str, to_selector: str):
+        """Drag and drop from one selector to another."""
+        from_elem = self._page.locator(from_selector)
+        from_elem.hover(timeout=500)
+        self._page.mouse.down()
+
+        to_elem = self._page.locator(to_selector)
+        to_elem.hover(timeout=500)
+        self._page.mouse.up()
+
+    def browser_hover(self, selector: str):
+        """Hover over a given element."""
+        self._page.hover(selector, timeout=3000, strict=True)
+
+    def browser_select_option(self, selector: str, value: str):
+        """Select an option from a given element."""
+        self._page.select_option(selector, value)
+
+    def browser_mouse_click_xy(self, x: int, y: int):
+        """Click at a given x, y coordinate using the mouse."""
+        self._page.mouse.click(x, y, delay=100)
+
+    def evaluate_js(self, js: str):
+        js_result = self._page.evaluate(js)
+        logger.info(f"JS result: {js_result}")
+        return js_result
+
+    def goto(self, url: str):
+        self._page.goto(url)
+
+    def page_html(self) -> str:
+        return self._page.content()
+
+    def page_screenshot(self) -> Image.Image:
+        scr_bytes = self._page.screenshot()
+        return Image.open(BytesIO(scr_bytes))
+
+    def page_axtree(self) -> str:
+        axtree = self._page.accessibility.snapshot()
+        return flatten_axtree(axtree)
+
+    def step(self, action: ToolCall) -> dict:
+        fn = self._actions[action.name]
+        try:
+            action_result = fn(**action.arguments)
+        except Exception as e:
+            action_result = f"Error executing action {action.name}: {e}"
+            logger.error(action_result)
+        html = self.page_html()
+        screenshot = self.page_screenshot()
+        axtree = self.page_axtree()
+        return {
+            "action_result": action_result,
+            "pruned_html": html,
+            "axtree_txt": axtree,
+            "screenshot": screenshot,
+        }
+
+    def actions(self) -> list[ToolSpec]:
+        return [ToolSpec.from_function(fn) for fn in self._actions.values()]
+
+    def close(self):
+        self._page.close()
+
+
+_apw = None  # Global Playwright instance for AsyncPlaywright
+_abrowser = None  # Global Browser instance for AsyncPlaywright
+
 
-    async def ainitialize(self):
-        pw = await async_playwright().start()
-        self._browser = await pw.chromium.launch(headless=True, chromium_sandbox=True)
-        self._page = await self._browser.new_page()
+class AsyncPlaywright(AsyncBrowserBackend):
+    """Fully asynchronous Playwright backend using playwright.async_api."""
+
+    has_pw_page: bool = False
+    _actions: dict[str, Callable]
+    _page: AsyncPage
+
+    def model_post_init(self, __context: Any):
+        self._actions = {
+            "browser_press_key": self.browser_press_key,
+            "browser_type": self.browser_type,
+            "browser_click": self.browser_click,
+            "browser_drag": self.browser_drag,
+            "browser_hover": self.browser_hover,
+            "browser_select_option": self.browser_select_option,
+            "browser_mouse_click_xy": self.browser_mouse_click_xy,
+        }
+
+    async def initialize(self):
+        global _apw, _abrowser
+        if _apw is None:
+            _apw = await async_playwright().start()
+        if _abrowser is None:
+            _abrowser = await _apw.chromium.launch(headless=True, chromium_sandbox=True)
+        self._page = await _abrowser.new_page()
 
     async def browser_press_key(self, key: str):
-        """
-        Press a key on the keyboard.
-        """
+        """Press a key on the keyboard."""
         await self._page.keyboard.press(key)
 
     async def browser_type(self, selector: str, text: str):
-        """
-        Type text into the focused element.
-        """
+        """Type text into the focused element."""
         await self._page.type(selector, text)
 
     async def browser_click(self, selector: str):
-        """
-        Click on a selector.
-        """
+        """Click on a selector."""
         await self._page.click(selector, timeout=3000, strict=True)
 
     async def browser_drag(self, from_selector: str, to_selector: str):
-        """
-        Drag and drop from one selector to another.
-        """
+        """Drag and drop from one selector to another."""
         from_elem = self._page.locator(from_selector)
         await from_elem.hover(timeout=500)
         await self._page.mouse.down()
@@ -69,66 +178,59 @@ async def browser_drag(self, from_selector: str, to_selector: str):
         await self._page.mouse.up()
 
     async def browser_hover(self, selector: str):
-        """
-        Hover over a given element.
-        """
+        """Hover over a given element."""
         await self._page.hover(selector, timeout=3000, strict=True)
 
     async def browser_select_option(self, selector: str, value: str):
-        """
-        Select an option from a given element.
-        """
+        """Select an option from a given element."""
         await self._page.select_option(selector, value)
 
     async def browser_mouse_click_xy(self, x: int, y: int):
-        """
-        Click at a given x, y coordinate using the mouse.
-        """
+        """Click at a given x, y coordinate using the mouse."""
         await self._page.mouse.click(x, y, delay=100)
 
-    def run_js(self, js: str):
-        js_result = self._loop.run_until_complete(self._page.evaluate(js))
+    async def evaluate_js(self, js: str):
+        js_result = await self._page.evaluate(js)
         logger.info(f"JS result: {js_result}")
         return js_result
 
-    def goto(self, url: str):
-        self._loop.run_until_complete(self._page.goto(url))
+    async def goto(self, url: str):
+        await self._page.goto(url)
 
-    def page_html(self):
-        return self._loop.run_until_complete(self._page.content())
+    async def page_html(self) -> str:
+        return await self._page.content()
 
-    def page_screenshot(self):
-        scr_bytes = self._loop.run_until_complete(self._page.screenshot())
+    async def page_screenshot(self) -> Image.Image:
+        scr_bytes = await self._page.screenshot()
         return Image.open(BytesIO(scr_bytes))
 
-    def page_axtree(self):
-        axtree = self._loop.run_until_complete(self._page.accessibility.snapshot())
-        flat_axtree = flatten_axtree(axtree)
-        return flat_axtree
+    async def page_axtree(self) -> str:
+        axtree = await self._page.accessibility.snapshot()
+        return flatten_axtree(axtree)
 
-    def step(self, action: ToolCall):
+    async def step(self, action: ToolCall) -> dict:
         fn = self._actions[action.name]
         try:
-            action_result = self._loop.run_until_complete(fn(**action.arguments))
+            action_result = await fn(**action.arguments)
         except Exception as e:
             action_result = f"Error executing action {action.name}: {e}"
             logger.error(action_result)
-        html = self.page_html()
-        screenshot = self.page_screenshot()
-        axtree = self.page_axtree()
+        html = await self.page_html()
+        screenshot = await self.page_screenshot()
+        axtree = await self.page_axtree()
         return {
-            "tool_result": action_result,
+            "action_result": action_result,
             "pruned_html": html,
             "axtree_txt": axtree,
             "screenshot": screenshot,
         }
 
-    def actions(self) -> tuple[ToolSpec]:
-        specs = [ToolSpec.from_function(fn) for fn in self._actions.values()]
-        return tuple(specs)
+    def actions(self) -> list[ToolSpec]:
+        return [ToolSpec.from_function(fn) for fn in self._actions.values()]
 
-    def close(self):
-        self._loop.run_until_complete(self._browser.close())
+    async def close(self):
+        await self._browser.close()
+        await self._pw.stop()
 
 
 def flatten_axtree(axtree_dict: dict | None) -> str: