| 1 | from __future__ import annotations |
| 2 | |
| 3 | import json |
| 4 | import re |
| 5 | import time |
| 6 | import uuid |
| 7 | from pathlib import Path |
| 8 | from typing import Any |
| 9 | |
| 10 | from helpers import files |
| 11 | from helpers.print_style import PrintStyle |
| 12 | from helpers.tool import Response, Tool |
| 13 | from plugins._browser.helpers.config import activate_browser_model |
| 14 | from plugins._browser.helpers.selector import get_tool_runtime |
| 15 | |
| 16 | |
| 17 | HISTORY_SCREENSHOT_QUALITY = 62 |
| 18 | HISTORY_SCREENSHOT_ACTION_DENYLIST = {"close", "close_all"} |
| 19 | |
| 20 | |
| 21 | async def get_runtime(context_id: str, create: bool = True, agent: Any | None = None): |
| 22 | if agent is not None: |
| 23 | return await get_tool_runtime(agent) |
| 24 | from plugins._browser.helpers.runtime import get_runtime as get_container_runtime |
| 25 | |
| 26 | return await get_container_runtime(context_id, create=create) |
| 27 | |
| 28 | |
| 29 | class Browser(Tool): |
| 30 | async def execute( |
| 31 | self, |
| 32 | action: str = "", |
| 33 | browser_id: int | str | None = None, |
| 34 | url: str = "", |
| 35 | ref: int | str | None = None, |
| 36 | target_ref: int | str | None = None, |
| 37 | text: str = "", |
| 38 | selector: str = "", |
| 39 | selectors: list[str] | None = None, |
| 40 | script: str = "", |
| 41 | modifiers: list[str] | str | None = None, |
| 42 | keys: list[str] | None = None, |
| 43 | key: str = "", |
| 44 | include_content: bool = False, |
| 45 | focus_popup: bool | None = None, |
| 46 | event_type: str = "", |
| 47 | x: float = 0.0, |
| 48 | y: float = 0.0, |
| 49 | to_x: float = 0.0, |
| 50 | to_y: float = 0.0, |
| 51 | offset_x: float = 0.0, |
| 52 | offset_y: float = 0.0, |
| 53 | target_offset_x: float = 0.0, |
| 54 | target_offset_y: float = 0.0, |
| 55 | delta_x: float = 0.0, |
| 56 | delta_y: float = 0.0, |
| 57 | button: str = "left", |
| 58 | quality: int = 80, |
| 59 | full_page: bool = False, |
| 60 | path: str = "", |
| 61 | paths: list[str] | None = None, |
| 62 | value: str = "", |
| 63 | values: list[str] | None = None, |
| 64 | checked: bool | None = None, |
| 65 | width: int = 0, |
| 66 | height: int = 0, |
| 67 | calls: list[dict[str, Any]] | None = None, |
| 68 | **kwargs: Any, |
| 69 | ) -> Response: |
| 70 | method_action = str(self.method or "").strip().lower().replace("-", "_") |
| 71 | requested_action = str(action or "").strip().lower().replace("-", "_") |
| 72 | clipboard_action = "" |
| 73 | if method_action == "clipboard" and requested_action in {"copy", "cut", "paste"}: |
| 74 | clipboard_action = requested_action |
| 75 | action = "clipboard" |
| 76 | else: |
| 77 | action = str(action or self.method or "state").strip().lower().replace("-", "_") |
| 78 | try: |
| 79 | activate_browser_model(self.agent) |
| 80 | except Exception as exc: |
| 81 | PrintStyle.warning(f"Browser model preset could not be activated: {exc}") |
| 82 | try: |
| 83 | runtime = await get_runtime(self.agent.context.id, agent=self.agent) |
| 84 | except Exception as exc: |
| 85 | return Response(message=f"Browser runtime unavailable: {exc}", break_loop=False) |
| 86 | |
| 87 | if isinstance(modifiers, str): |
| 88 | modifiers = [modifiers] if modifiers else None |
| 89 | elif isinstance(modifiers, list) and not modifiers: |
| 90 | modifiers = None |
| 91 | keys = self._normalize_keys(keys) |
| 92 | |
| 93 | try: |
| 94 | if action == "open": |
| 95 | result = await runtime.call("open", url or "") |
| 96 | elif action == "screenshot": |
| 97 | result = await runtime.call( |
| 98 | "screenshot_file", |
| 99 | browser_id, |
| 100 | quality=quality, |
| 101 | full_page=full_page, |
| 102 | path=path, |
| 103 | ) |
| 104 | elif action == "list": |
| 105 | result = await runtime.call("list", include_content=bool(include_content)) |
| 106 | elif action == "state": |
| 107 | result = await runtime.call("state", browser_id) |
| 108 | elif action in {"set_active", "setactive", "activate", "focus"}: |
| 109 | result = await runtime.call("set_active", browser_id) |
| 110 | elif action == "navigate": |
| 111 | result = await runtime.call("navigate", browser_id, url) |
| 112 | elif action == "back": |
| 113 | result = await runtime.call("back", browser_id) |
| 114 | elif action == "forward": |
| 115 | result = await runtime.call("forward", browser_id) |
| 116 | elif action == "reload": |
| 117 | result = await runtime.call("reload", browser_id) |
| 118 | elif action == "content": |
| 119 | payload = self._selector_payload(selector, selectors) |
| 120 | result = await runtime.call("content", browser_id, payload) |
| 121 | elif action == "detail": |
| 122 | result = await runtime.call( |
| 123 | "detail", |
| 124 | browser_id, |
| 125 | await self._resolve_ref(runtime, browser_id, ref, selector, action), |
| 126 | ) |
| 127 | elif action == "click": |
| 128 | resolved_ref = await self._resolve_ref( |
| 129 | runtime, |
| 130 | browser_id, |
| 131 | ref, |
| 132 | selector, |
| 133 | action, |
| 134 | required=not self._has_coordinates(x, y), |
| 135 | ) |
| 136 | if resolved_ref is None and self._has_coordinates(x, y): |
| 137 | result = await runtime.call( |
| 138 | "mouse", browser_id, "click", x, y, |
| 139 | button=button or "left", modifiers=modifiers, |
| 140 | ) |
| 141 | elif modifiers: |
| 142 | result = await runtime.call( |
| 143 | "click", browser_id, resolved_ref, |
| 144 | modifiers=modifiers, focus_popup=focus_popup, |
| 145 | ) |
| 146 | else: |
| 147 | result = await runtime.call("click", browser_id, resolved_ref) |
| 148 | elif action == "type": |
| 149 | resolved_ref = await self._resolve_ref( |
| 150 | runtime, |
| 151 | browser_id, |
| 152 | ref, |
| 153 | selector, |
| 154 | action, |
| 155 | required=False, |
| 156 | ) |
| 157 | if resolved_ref is None: |
| 158 | result = await runtime.call("keyboard", browser_id, key="", text=text) |
| 159 | else: |
| 160 | result = await runtime.call("type", browser_id, resolved_ref, text) |
| 161 | elif action == "submit": |
| 162 | result = await runtime.call( |
| 163 | "submit", |
| 164 | browser_id, |
| 165 | await self._resolve_ref(runtime, browser_id, ref, selector, action), |
| 166 | ) |
| 167 | elif action in {"type_submit", "typesubmit"}: |
| 168 | result = await runtime.call( |
| 169 | "type_submit", |
| 170 | browser_id, |
| 171 | await self._resolve_ref(runtime, browser_id, ref, selector, action), |
| 172 | text, |
| 173 | ) |
| 174 | elif action == "scroll": |
| 175 | result = await runtime.call( |
| 176 | "scroll", |
| 177 | browser_id, |
| 178 | await self._resolve_ref(runtime, browser_id, ref, selector, action), |
| 179 | ) |
| 180 | elif action == "evaluate": |
| 181 | result = await runtime.call("evaluate", browser_id, script) |
| 182 | elif action in {"key_chord", "keychord"}: |
| 183 | if not keys: |
| 184 | raise ValueError("key_chord requires non-empty 'keys' list") |
| 185 | result = await runtime.call("key_chord", browser_id, keys) |
| 186 | elif action == "hover": |
| 187 | result = await runtime.call( |
| 188 | "hover", |
| 189 | browser_id, |
| 190 | ref=ref, |
| 191 | x=x, |
| 192 | y=y, |
| 193 | offset_x=offset_x, |
| 194 | offset_y=offset_y, |
| 195 | ) |
| 196 | elif action == "double_click": |
| 197 | result = await runtime.call( |
| 198 | "double_click", |
| 199 | browser_id, |
| 200 | ref=ref, |
| 201 | x=x, |
| 202 | y=y, |
| 203 | button=button or "left", |
| 204 | modifiers=modifiers, |
| 205 | offset_x=offset_x, |
| 206 | offset_y=offset_y, |
| 207 | ) |
| 208 | elif action == "right_click": |
| 209 | result = await runtime.call( |
| 210 | "right_click", |
| 211 | browser_id, |
| 212 | ref=ref, |
| 213 | x=x, |
| 214 | y=y, |
| 215 | modifiers=modifiers, |
| 216 | offset_x=offset_x, |
| 217 | offset_y=offset_y, |
| 218 | ) |
| 219 | elif action == "drag": |
| 220 | result = await runtime.call( |
| 221 | "drag", |
| 222 | browser_id, |
| 223 | ref=ref, |
| 224 | target_ref=target_ref, |
| 225 | x=x, |
| 226 | y=y, |
| 227 | to_x=to_x, |
| 228 | to_y=to_y, |
| 229 | offset_x=offset_x, |
| 230 | offset_y=offset_y, |
| 231 | target_offset_x=target_offset_x, |
| 232 | target_offset_y=target_offset_y, |
| 233 | ) |
| 234 | elif action == "wheel": |
| 235 | result = await runtime.call( |
| 236 | "wheel", |
| 237 | browser_id, |
| 238 | x, |
| 239 | y, |
| 240 | delta_x, |
| 241 | delta_y, |
| 242 | ) |
| 243 | elif action == "keyboard": |
| 244 | result = await runtime.call( |
| 245 | "keyboard", |
| 246 | browser_id, |
| 247 | key=key, |
| 248 | text=text, |
| 249 | ) |
| 250 | elif action == "clipboard": |
| 251 | normalized_clipboard_action = clipboard_action or str( |
| 252 | kwargs.get("clipboard_action") |
| 253 | or kwargs.get("operation") |
| 254 | or event_type |
| 255 | or "" |
| 256 | ).strip().lower() |
| 257 | result = await runtime.call( |
| 258 | "clipboard", |
| 259 | browser_id, |
| 260 | action=normalized_clipboard_action, |
| 261 | text=text, |
| 262 | ) |
| 263 | elif action in {"copy", "cut", "paste"}: |
| 264 | result = await runtime.call( |
| 265 | "clipboard", |
| 266 | browser_id, |
| 267 | action=action, |
| 268 | text=text, |
| 269 | ) |
| 270 | elif action == "set_viewport": |
| 271 | result = await runtime.call("set_viewport", browser_id, width, height) |
| 272 | elif action == "select_option": |
| 273 | result = await runtime.call( |
| 274 | "select_option", |
| 275 | browser_id, |
| 276 | await self._resolve_ref(runtime, browser_id, ref, selector, action), |
| 277 | value=value, |
| 278 | values=values, |
| 279 | ) |
| 280 | elif action == "set_checked": |
| 281 | result = await runtime.call( |
| 282 | "set_checked", |
| 283 | browser_id, |
| 284 | await self._resolve_ref(runtime, browser_id, ref, selector, action), |
| 285 | checked=True if checked is None else bool(checked), |
| 286 | ) |
| 287 | elif action == "upload_file": |
| 288 | result = await runtime.call( |
| 289 | "upload_file", |
| 290 | browser_id, |
| 291 | await self._resolve_ref(runtime, browser_id, ref, selector, action), |
| 292 | path=path, |
| 293 | paths=paths, |
| 294 | ) |
| 295 | elif action == "mouse": |
| 296 | result = await runtime.call( |
| 297 | "mouse", browser_id, event_type or "click", x, y, |
| 298 | button=button or "left", modifiers=modifiers, |
| 299 | ) |
| 300 | elif action == "multi": |
| 301 | if not calls: |
| 302 | raise ValueError("multi requires non-empty 'calls' list") |
| 303 | result = await runtime.call("multi", list(calls)) |
| 304 | elif action == "close": |
| 305 | result = await runtime.call("close_browser", browser_id) |
| 306 | elif action == "close_all": |
| 307 | result = await runtime.call("close_all_browsers") |
| 308 | else: |
| 309 | return Response( |
| 310 | message=f"Unknown browser action: {action}", |
| 311 | break_loop=False, |
| 312 | ) |
| 313 | await self._record_history_screenshot(runtime, action, result, browser_id) |
| 314 | except Exception as exc: |
| 315 | return Response(message=f"Browser {action} failed: {exc}", break_loop=False) |
| 316 | |
| 317 | return Response(message=self._format_result(action, result), break_loop=False) |
| 318 | |
| 319 | def get_log_object(self): |
| 320 | return self.agent.context.log.log( |
| 321 | type="tool", |
| 322 | heading=f"icon://captive_portal {self.agent.agent_name}: Using browser", |
| 323 | content="", |
| 324 | kvps=self.args, |
| 325 | _tool_name=self.name, |
| 326 | ) |
| 327 | |
| 328 | @staticmethod |
| 329 | def _require_ref(ref: int | str | None) -> int | str: |
| 330 | if ref is None or str(ref).strip() == "": |
| 331 | raise ValueError("ref is required for this browser action") |
| 332 | return ref |
| 333 | |
| 334 | @staticmethod |
| 335 | def _has_ref(ref: int | str | None) -> bool: |
| 336 | return ref is not None and str(ref).strip() != "" |
| 337 | |
| 338 | @staticmethod |
| 339 | def _has_coordinates(x: float, y: float) -> bool: |
| 340 | return bool(float(x or 0) or float(y or 0)) |
| 341 | |
| 342 | @classmethod |
| 343 | async def _resolve_ref( |
| 344 | cls, |
| 345 | runtime: Any, |
| 346 | browser_id: int | str | None, |
| 347 | ref: int | str | None, |
| 348 | selector: str = "", |
| 349 | action: str = "action", |
| 350 | *, |
| 351 | required: bool = True, |
| 352 | ) -> int | str | None: |
| 353 | if cls._has_ref(ref): |
| 354 | return ref |
| 355 | |
| 356 | selector = str(selector or "").strip() |
| 357 | if selector: |
| 358 | content = await runtime.call("content", browser_id, {"selector": selector}) |
| 359 | resolved = cls._first_ref_from_content(content, selector) |
| 360 | if resolved is not None: |
| 361 | return resolved |
| 362 | raise ValueError( |
| 363 | f"{action} could not resolve selector {selector!r} to a browser ref" |
| 364 | ) |
| 365 | |
| 366 | if required: |
| 367 | return cls._require_ref(ref) |
| 368 | return None |
| 369 | |
| 370 | @staticmethod |
| 371 | def _first_ref_from_content(content: Any, selector: str = "") -> str | None: |
| 372 | if isinstance(content, dict): |
| 373 | values: list[Any] = [] |
| 374 | if selector and selector in content: |
| 375 | values.append(content.get(selector)) |
| 376 | values.extend(value for key, value in content.items() if key != selector) |
| 377 | text = "\n".join(str(value or "") for value in values) |
| 378 | else: |
| 379 | text = str(content or "") |
| 380 | match = re.search(r"\[[^\]\n]*?\b(\d+)\]", text) |
| 381 | return match.group(1) if match else None |
| 382 | |
| 383 | @staticmethod |
| 384 | def _normalize_keys(keys: list[str] | str | None) -> list[str]: |
| 385 | if keys is None: |
| 386 | return [] |
| 387 | if isinstance(keys, str): |
| 388 | raw = re.split(r"\s*\+\s*|\s*,\s*", keys.strip()) |
| 389 | elif isinstance(keys, list): |
| 390 | raw = keys |
| 391 | else: |
| 392 | raw = [str(keys)] |
| 393 | aliases = { |
| 394 | "cmd": "Meta", |
| 395 | "command": "Meta", |
| 396 | "control": "Control", |
| 397 | "ctrl": "Control", |
| 398 | "escape": "Escape", |
| 399 | "esc": "Escape", |
| 400 | "meta": "Meta", |
| 401 | "option": "Alt", |
| 402 | "return": "Enter", |
| 403 | "space": "Space", |
| 404 | } |
| 405 | normalized: list[str] = [] |
| 406 | for key in raw: |
| 407 | value = str(key or "").strip() |
| 408 | if not value: |
| 409 | continue |
| 410 | normalized.append(aliases.get(value.lower(), value.upper() if len(value) == 1 and value.isalpha() else value)) |
| 411 | return normalized |
| 412 | |
| 413 | @staticmethod |
| 414 | def _selector_payload(selector: str = "", selectors: list[str] | None = None) -> dict | None: |
| 415 | if selectors: |
| 416 | return {"selectors": selectors} |
| 417 | if selector: |
| 418 | return {"selector": selector} |
| 419 | return None |
| 420 | |
| 421 | async def _record_history_screenshot( |
| 422 | self, |
| 423 | runtime: Any, |
| 424 | action: str, |
| 425 | result: Any, |
| 426 | requested_browser_id: int | str | None = None, |
| 427 | ) -> None: |
| 428 | if not getattr(self, "log", None): |
| 429 | return |
| 430 | if action in HISTORY_SCREENSHOT_ACTION_DENYLIST: |
| 431 | return |
| 432 | |
| 433 | screenshot = result if action == "screenshot" and isinstance(result, dict) else None |
| 434 | if not self._screenshot_has_reference(screenshot): |
| 435 | target_browser_id = self._browser_id_from_result(result) or requested_browser_id |
| 436 | try: |
| 437 | screenshot = await runtime.call( |
| 438 | "screenshot_file", |
| 439 | target_browser_id, |
| 440 | quality=HISTORY_SCREENSHOT_QUALITY, |
| 441 | full_page=False, |
| 442 | path="", |
| 443 | ) |
| 444 | except Exception as exc: |
| 445 | PrintStyle.debug( |
| 446 | "Browser history screenshot capture failed:", |
| 447 | f"browser_id={target_browser_id}", |
| 448 | f"quality={HISTORY_SCREENSHOT_QUALITY}", |
| 449 | f"error={exc}", |
| 450 | ) |
| 451 | return |
| 452 | |
| 453 | if not self._screenshot_has_reference(screenshot): |
| 454 | return |
| 455 | |
| 456 | a0_path = str(screenshot.get("a0_path") or "").strip() |
| 457 | local_path = str(screenshot.get("path") or (files.fix_dev_path(a0_path) if a0_path else "")) |
| 458 | state = screenshot.get("state") if isinstance(screenshot.get("state"), dict) else {} |
| 459 | chat_context_id = self._agent_context_id() |
| 460 | browser_context_id = str(screenshot.get("context_id") or state.get("context_id") or "").strip() |
| 461 | snapshot = { |
| 462 | "mime": screenshot.get("mime") or "image/jpeg", |
| 463 | "browser_id": screenshot.get("browser_id") or state.get("id") or requested_browser_id, |
| 464 | "context_id": chat_context_id or browser_context_id, |
| 465 | "browser_context_id": browser_context_id, |
| 466 | } |
| 467 | update_payload: dict[str, Any] = {"browser_snapshot": snapshot} |
| 468 | if local_path: |
| 469 | uri = f"img://{local_path}&t={time.time()}" |
| 470 | snapshot.update( |
| 471 | { |
| 472 | "uri": uri, |
| 473 | "path": local_path, |
| 474 | "a0_path": screenshot.get("a0_path") or files.normalize_a0_path(local_path), |
| 475 | "ephemeral": False, |
| 476 | } |
| 477 | ) |
| 478 | update_payload["Screenshot"] = uri |
| 479 | else: |
| 480 | ephemeral_ref = self._screenshot_ephemeral_ref(screenshot) |
| 481 | snapshot.update( |
| 482 | { |
| 483 | "ephemeral": bool(ephemeral_ref), |
| 484 | "ephemeral_ref": ephemeral_ref, |
| 485 | } |
| 486 | ) |
| 487 | self.log.update(**update_payload) |
| 488 | |
| 489 | def _history_screenshot_path(self, action: str) -> str: |
| 490 | if not getattr(self, "agent", None) or not getattr(self.agent, "context", None): |
| 491 | return "" |
| 492 | context_id = self._agent_context_id() |
| 493 | if not context_id: |
| 494 | return "" |
| 495 | from helpers import persist_chat |
| 496 | |
| 497 | token = str(getattr(getattr(self, "log", None), "id", "") or uuid.uuid4()) |
| 498 | safe_action = files.safe_file_name(str(action or "browser")) |
| 499 | safe_token = files.safe_file_name(token) |
| 500 | timestamp = time.strftime("%Y%m%d-%H%M%S") |
| 501 | return str( |
| 502 | Path(persist_chat.get_chat_folder_path(context_id)) |
| 503 | / "browser" |
| 504 | / "screenshots" |
| 505 | / f"{timestamp}-{safe_action}-{safe_token}.jpg" |
| 506 | ) |
| 507 | |
| 508 | @staticmethod |
| 509 | def _browser_id_from_result(result: Any) -> Any: |
| 510 | if not isinstance(result, dict): |
| 511 | return None |
| 512 | browsers = result.get("browsers") if isinstance(result.get("browsers"), list) else [] |
| 513 | last_interacted_id = result.get("last_interacted_browser_id") |
| 514 | listed_browser = None |
| 515 | if last_interacted_id is not None: |
| 516 | listed_browser = next( |
| 517 | ( |
| 518 | browser |
| 519 | for browser in browsers |
| 520 | if isinstance(browser, dict) and str(browser.get("id")) == str(last_interacted_id) |
| 521 | ), |
| 522 | None, |
| 523 | ) |
| 524 | if listed_browser is None and browsers: |
| 525 | listed_browser = next((browser for browser in browsers if isinstance(browser, dict)), None) |
| 526 | state = result.get("state") if isinstance(result.get("state"), dict) else {} |
| 527 | return ( |
| 528 | result.get("id") |
| 529 | or result.get("browser_id") |
| 530 | or state.get("id") |
| 531 | or last_interacted_id |
| 532 | or (listed_browser or {}).get("id") |
| 533 | ) |
| 534 | |
| 535 | @staticmethod |
| 536 | def _screenshot_has_path(screenshot: Any) -> bool: |
| 537 | return isinstance(screenshot, dict) and bool(screenshot.get("path") or screenshot.get("a0_path")) |
| 538 | |
| 539 | @classmethod |
| 540 | def _screenshot_has_reference(cls, screenshot: Any) -> bool: |
| 541 | return cls._screenshot_has_path(screenshot) or bool(cls._screenshot_ephemeral_ref(screenshot)) |
| 542 | |
| 543 | @staticmethod |
| 544 | def _screenshot_ephemeral_ref(screenshot: Any) -> str: |
| 545 | if not isinstance(screenshot, dict): |
| 546 | return "" |
| 547 | ref = str(screenshot.get("ephemeral_ref") or "").strip() |
| 548 | if ref: |
| 549 | return ref |
| 550 | vision_load = screenshot.get("vision_load") |
| 551 | if isinstance(vision_load, dict): |
| 552 | tool_args = vision_load.get("tool_args") |
| 553 | if isinstance(tool_args, dict): |
| 554 | paths = tool_args.get("paths") |
| 555 | if isinstance(paths, list) and paths: |
| 556 | first = str(paths[0] or "").strip() |
| 557 | if first.startswith("a0-ephemeral-image://"): |
| 558 | return first |
| 559 | return "" |
| 560 | |
| 561 | def _agent_context_id(self) -> str: |
| 562 | return str(getattr(getattr(self.agent, "context", None), "id", "") or "").strip() |
| 563 | |
| 564 | @staticmethod |
| 565 | def _format_result(action: str, result: Any) -> str: |
| 566 | if action == "content" and isinstance(result, dict): |
| 567 | if set(result.keys()) == {"document"}: |
| 568 | return str(result.get("document") or "") |
| 569 | return json.dumps(result, indent=2, ensure_ascii=False) |
| 570 | |
| 571 | return json.dumps(result, indent=2, ensure_ascii=False, default=str) |