Spaces:
Runtime error
Runtime error
from __future__ import annotations | |
from typing import Optional, Type | |
from langchain_core.pydantic_v1 import BaseModel, Field | |
from langchain.callbacks.manager import ( | |
AsyncCallbackManagerForToolRun, | |
CallbackManagerForToolRun, | |
) | |
from langchain.tools.playwright.base import BaseBrowserTool | |
from langchain.tools.playwright.utils import ( | |
aget_current_page, | |
get_current_page, | |
) | |
class ClickToolInput(BaseModel): | |
"""Input for ClickTool.""" | |
selector: str = Field(..., description="CSS selector for the element to click") | |
class ClickTool(BaseBrowserTool): | |
"""Tool for clicking on an element with the given CSS selector.""" | |
name: str = "click_element" | |
description: str = "Click on an element with the given CSS selector" | |
args_schema: Type[BaseModel] = ClickToolInput | |
visible_only: bool = True | |
"""Whether to consider only visible elements.""" | |
playwright_strict: bool = False | |
"""Whether to employ Playwright's strict mode when clicking on elements.""" | |
playwright_timeout: float = 1_000 | |
"""Timeout (in ms) for Playwright to wait for element to be ready.""" | |
def _selector_effective(self, selector: str) -> str: | |
if not self.visible_only: | |
return selector | |
return f"{selector} >> visible=1" | |
def _run( | |
self, | |
selector: str, | |
run_manager: Optional[CallbackManagerForToolRun] = None, | |
) -> str: | |
"""Use the tool.""" | |
if self.sync_browser is None: | |
raise ValueError(f"Synchronous browser not provided to {self.name}") | |
page = get_current_page(self.sync_browser) | |
# Navigate to the desired webpage before using this tool | |
selector_effective = self._selector_effective(selector=selector) | |
from playwright.sync_api import TimeoutError as PlaywrightTimeoutError | |
try: | |
page.click( | |
selector_effective, | |
strict=self.playwright_strict, | |
timeout=self.playwright_timeout, | |
) | |
except PlaywrightTimeoutError: | |
return f"Unable to click on element '{selector}'" | |
return f"Clicked element '{selector}'" | |
async def _arun( | |
self, | |
selector: str, | |
run_manager: Optional[AsyncCallbackManagerForToolRun] = None, | |
) -> str: | |
"""Use the tool.""" | |
if self.async_browser is None: | |
raise ValueError(f"Asynchronous browser not provided to {self.name}") | |
page = await aget_current_page(self.async_browser) | |
# Navigate to the desired webpage before using this tool | |
selector_effective = self._selector_effective(selector=selector) | |
from playwright.async_api import TimeoutError as PlaywrightTimeoutError | |
try: | |
await page.click( | |
selector_effective, | |
strict=self.playwright_strict, | |
timeout=self.playwright_timeout, | |
) | |
except PlaywrightTimeoutError: | |
return f"Unable to click on element '{selector}'" | |
return f"Clicked element '{selector}'" | |