411 lines
15 KiB
Python
411 lines
15 KiB
Python
"""Use the Browser Use CLI 3.0 (https://browser-use.com) for browser automation
|
|
|
|
When browser.backend is "browser-use", the model gets ``browser_exec`` tool
|
|
instead of default browser tools
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import time
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from utils import is_truthy_value
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_BACKEND_KEY = "browser-use"
|
|
BACKEND_DISABLED = "off"
|
|
|
|
# Cloud daemon names become the BU_NAME env var
|
|
_SESSION_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$")
|
|
|
|
_DEFAULT_TIMEOUT_S = 300
|
|
_MIN_TIMEOUT_S = 5
|
|
_MAX_TIMEOUT_S = 1800
|
|
_STDERR_CAP_CHARS = 4000
|
|
|
|
# Filesystem-safe task ids for per-task workspace dirs.
|
|
_TASK_ID_SAFE_RE = re.compile(r"[^A-Za-z0-9._-]+")
|
|
|
|
# Screenshot paths printed by capture_screenshot() in the exec output
|
|
_IMAGE_PATH_RE = re.compile(r"(/[^\s\"']+?\.(?:png|jpe?g|webp))", re.IGNORECASE)
|
|
|
|
|
|
def _base_subprocess_env() -> dict:
|
|
from tools.browser_tool import _build_browser_env
|
|
|
|
return _build_browser_env()
|
|
|
|
|
|
def _read_browser_cfg() -> dict:
|
|
"""Return the ``browser:`` config section, or {} on any failure."""
|
|
try:
|
|
from hermes_cli.config import cfg_get, read_raw_config
|
|
|
|
cfg = cfg_get(read_raw_config(), "browser", default={})
|
|
return cfg if isinstance(cfg, dict) else {}
|
|
except Exception as e:
|
|
logger.debug("Could not read browser config section: %s", e)
|
|
return {}
|
|
|
|
|
|
def get_browser_backend() -> str:
|
|
"""Return the configured browser backend key ("" = legacy stack)."""
|
|
return str(_read_browser_cfg().get("backend") or "").strip().lower()
|
|
|
|
|
|
def is_legacy_browser_use_cloud_config(browser_cfg: dict) -> bool:
|
|
"""True for pre-CLI direct-API Browser Use cloud configs"""
|
|
if not isinstance(browser_cfg, dict):
|
|
return False
|
|
if browser_cfg.get("backend"):
|
|
return False # an explicit backend choice wins
|
|
provider = str(browser_cfg.get("cloud_provider") or "").strip().lower()
|
|
if provider not in {"browser-use", ""}:
|
|
return False # explicit local/Browserbase/… choices win
|
|
if is_truthy_value(browser_cfg.get("use_gateway"), default=False):
|
|
return False
|
|
return bool(os.getenv("BROWSER_USE_API_KEY"))
|
|
|
|
|
|
def is_browser_use_cli_mode() -> bool:
|
|
"""True when the Browser Use CLI replaces the built-in browser stack"""
|
|
backend = get_browser_backend()
|
|
if backend:
|
|
return backend == _BACKEND_KEY
|
|
return is_legacy_browser_use_cloud_config(_read_browser_cfg())
|
|
|
|
|
|
def _find_cli() -> Optional[List[str]]:
|
|
"""Locate the browser-use CLI, or None when it can't be run.
|
|
|
|
Prefers an installed browser-use binary; falls back to running it
|
|
through uvx
|
|
"""
|
|
direct = shutil.which("browser-use")
|
|
if direct:
|
|
return [direct]
|
|
uvx = shutil.which("uvx")
|
|
if uvx:
|
|
return [uvx, "browser-use"]
|
|
return None
|
|
|
|
|
|
def _workspace_dir(task_id: Optional[str]) -> Optional[str]:
|
|
"""Stable per-task scratch dir that persists across browser_exec calls"""
|
|
existing = os.environ.get("BH_AGENT_WORKSPACE")
|
|
if existing:
|
|
return existing
|
|
try:
|
|
from pathlib import Path
|
|
|
|
from hermes_constants import get_hermes_home
|
|
|
|
safe = _TASK_ID_SAFE_RE.sub("_", str(task_id or "default"))[:80] or "default"
|
|
path = Path(get_hermes_home()) / "cache" / "browser-use" / "workspace" / safe
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
return str(path)
|
|
except Exception as e:
|
|
logger.debug("browser_exec workspace unavailable: %s", e)
|
|
return None
|
|
|
|
|
|
def _find_screenshot(stdout: str, since: float) -> Optional[str]:
|
|
"""Return the last screenshot path printed during this exec, or None.
|
|
|
|
Only accepts files that exist and were written after the exec started
|
|
"""
|
|
for path in reversed(_IMAGE_PATH_RE.findall(stdout or "")):
|
|
try:
|
|
if os.path.isfile(path) and os.path.getmtime(path) >= since - 1:
|
|
return path
|
|
except OSError:
|
|
continue
|
|
return None
|
|
|
|
|
|
def _native_screenshot_result(result: Dict[str, Any], path: str) -> Optional[Dict[str, Any]]:
|
|
"""Build a multimodal tool result attaching path for vision models"""
|
|
try:
|
|
from pathlib import Path
|
|
|
|
from tools.vision_tools import (
|
|
_resize_image_for_vision,
|
|
_should_use_native_vision_fast_path,
|
|
)
|
|
|
|
if not _should_use_native_vision_fast_path():
|
|
return None
|
|
data_url = _resize_image_for_vision(Path(path))
|
|
text = json.dumps(result, ensure_ascii=False)
|
|
return {
|
|
"_multimodal": True,
|
|
"content": [
|
|
{
|
|
"type": "text",
|
|
"text": (
|
|
text
|
|
+ "\n\nThe screenshot from this call is attached — "
|
|
"inspect it with your native vision."
|
|
),
|
|
},
|
|
{"type": "image_url", "image_url": {"url": data_url}},
|
|
],
|
|
"text_summary": text,
|
|
"meta": {"screenshot_path": path, "native_vision": True},
|
|
}
|
|
except Exception as e:
|
|
logger.debug("Native screenshot attach failed (falling back to text): %s", e)
|
|
return None
|
|
|
|
|
|
def browser_exec(
|
|
code: str,
|
|
session: str = "",
|
|
timeout_s: int = _DEFAULT_TIMEOUT_S,
|
|
task_id: Optional[str] = None,
|
|
):
|
|
"""Run Python code through the browser-use CLI, and return its output"""
|
|
from tools.registry import tool_error, tool_result
|
|
|
|
if not code or not code.strip():
|
|
return tool_error("No code provided. Pass Python that uses the pre-imported helpers, e.g. new_tab(\"https://example.com\") then print(page_info()).")
|
|
|
|
cmd = _find_cli()
|
|
if not cmd:
|
|
return tool_error(
|
|
"browser-use CLI not found on PATH, and uvx is unavailable for a "
|
|
"zero-install run. Install it with `uv tool install browser-use` "
|
|
"(or `pipx install browser-use`), then run `browser-use --doctor` "
|
|
"to verify the setup."
|
|
)
|
|
|
|
env = _base_subprocess_env()
|
|
if session:
|
|
if not _SESSION_RE.match(session):
|
|
return tool_error(
|
|
f"Invalid session name {session!r}: use 1-64 letters, digits, "
|
|
"dashes, or underscores (e.g. 'r7k2')."
|
|
)
|
|
env["BU_NAME"] = session
|
|
|
|
workspace = _workspace_dir(task_id)
|
|
if workspace:
|
|
env["BH_AGENT_WORKSPACE"] = workspace
|
|
|
|
# BU_AUTOSPAWN makes the CLI start a Browser Use cloud browser when no
|
|
# local Chrome/CDP endpoint is reachable (their API key authenticates it)
|
|
if "BU_AUTOSPAWN" not in env and is_legacy_browser_use_cloud_config(_read_browser_cfg()):
|
|
env["BU_AUTOSPAWN"] = "1"
|
|
|
|
try:
|
|
timeout = max(_MIN_TIMEOUT_S, min(int(timeout_s), _MAX_TIMEOUT_S))
|
|
except (TypeError, ValueError):
|
|
timeout = _DEFAULT_TIMEOUT_S
|
|
|
|
# Windows: hide the console the .cmd shim would flash (as browser_tool does)
|
|
popen_extra: dict = {}
|
|
if os.name == "nt":
|
|
try:
|
|
from hermes_cli._subprocess_compat import windows_hide_flags
|
|
|
|
popen_extra["creationflags"] = windows_hide_flags()
|
|
_si = subprocess.STARTUPINFO()
|
|
_si.dwFlags |= subprocess.STARTF_USESHOWWINDOW
|
|
popen_extra["startupinfo"] = _si
|
|
except Exception as e:
|
|
logger.debug("Windows hide-flags unavailable: %s", e)
|
|
|
|
started = time.time()
|
|
try:
|
|
proc = subprocess.run(
|
|
cmd,
|
|
input=code,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout,
|
|
env=env,
|
|
**popen_extra,
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
return tool_error(
|
|
f"browser-use exec timed out after {timeout}s. The daemon may "
|
|
"still be working; retry with a larger timeout_s (max "
|
|
f"{_MAX_TIMEOUT_S}), or split the work into several calls that "
|
|
"append to workspace files — anything already written to the "
|
|
"workspace is preserved."
|
|
)
|
|
except OSError as e:
|
|
return tool_error(f"Failed to launch browser-use CLI: {e}")
|
|
|
|
result = {
|
|
"success": proc.returncode == 0,
|
|
"exit_code": proc.returncode,
|
|
"output": proc.stdout,
|
|
}
|
|
if workspace:
|
|
result["workspace"] = workspace
|
|
if session:
|
|
result["session"] = session
|
|
stderr = (proc.stderr or "").strip()
|
|
if stderr:
|
|
if len(stderr) > _STDERR_CAP_CHARS:
|
|
stderr = stderr[:_STDERR_CAP_CHARS] + "\n… (stderr truncated)"
|
|
result["stderr"] = stderr
|
|
|
|
screenshot = _find_screenshot(proc.stdout, started)
|
|
if screenshot:
|
|
result["screenshot_path"] = screenshot
|
|
native = _native_screenshot_result(result, screenshot)
|
|
if native is not None:
|
|
return native
|
|
return tool_result(result)
|
|
|
|
|
|
# The tool description is the CLI's skill, fetched from browser-use skill
|
|
_HEADER_BASE = (
|
|
"Drive a real web browser via the Browser Use CLI. The `code` argument "
|
|
"is piped verbatim to the `browser-use` CLI on stdin and executed as "
|
|
"full Python (standard library available) with the CLI's pre-imported "
|
|
"browser helpers; stdout comes back in the result. Start `code` with a "
|
|
"one-line comment describing the step for the user in plain, "
|
|
"non-technical language, max 60 chars (e.g. `# Searching Amazon for "
|
|
"paper towels`) — the UI displays it as the step label.\n\n"
|
|
"STATE: the browser session and the workspace persist across calls; "
|
|
"Python variables do NOT (each call is a fresh interpreter). The "
|
|
"workspace is a stable directory — path in $BH_AGENT_WORKSPACE and "
|
|
"returned as `workspace` in every result. For multi-item tasks "
|
|
"('collect all N products / every entry / the full table'), append each "
|
|
"batch to a JSON/CSV file in the workspace as you go, then read it back "
|
|
"to assemble the final answer; define reusable functions in "
|
|
"agent_helpers.py there — the harness auto-imports it into every call. "
|
|
"Do aggregation in code, not in your head: dedupe, count, sort, and "
|
|
"format with Python inside the exec. Before giving a final answer on a "
|
|
"multi-item task, verify the collected count against what was asked "
|
|
"and go back for anything missing.\n\n"
|
|
"Batch each sub-procedure (navigate, wait, extract, act) into one call "
|
|
"— do not spend a call per action — but for long extractions prefer "
|
|
"several medium calls that append to workspace files over one giant "
|
|
"call, so progress survives timeouts. js() takes a JS expression: "
|
|
"js('document.title') or js('(() => {...})()') — a bare '() => {...}' "
|
|
"returns the function itself, uncalled. The CLI's own documentation "
|
|
"follows and is complete (no need to read separate browser-use skill "
|
|
"files) — where it shows shell heredocs (browser-use <<'PY' … PY), pass "
|
|
"the Python as `code` instead; where it shows BU_NAME=<name>, pass "
|
|
"session=<name> instead."
|
|
)
|
|
|
|
_HEADER_VISION = (
|
|
" Screenshots are attached to your context automatically: when the exec "
|
|
"output contains a capture_screenshot() path, the image arrives with "
|
|
"this tool's result and you inspect it directly with your own vision — "
|
|
"never send browser screenshots to a separate vision tool."
|
|
)
|
|
|
|
_HEADER_TEXT_ONLY = (
|
|
" Your model cannot view images, so work text-first: page_info() for "
|
|
"state, js() for reading/extracting DOM text, fill_input(selector, "
|
|
"text) for inputs, and js(\"document.querySelector('…').click()\") for "
|
|
"clicks — skip the screenshot-driven workflow described below."
|
|
)
|
|
|
|
_DESCRIPTION_HEADER = _HEADER_BASE
|
|
|
|
|
|
def _description_header() -> str:
|
|
"""Header tailored to whether the active model can see images natively"""
|
|
try:
|
|
from tools.vision_tools import _should_use_native_vision_fast_path
|
|
|
|
if _should_use_native_vision_fast_path():
|
|
return _HEADER_BASE + _HEADER_VISION
|
|
except Exception:
|
|
pass
|
|
return _HEADER_BASE + _HEADER_TEXT_ONLY
|
|
|
|
_skill_text_cache: Optional[str] = None
|
|
_skill_text_fetched = False
|
|
|
|
|
|
def _cli_skill_text() -> str:
|
|
"""Return the installed CLI's skill"""
|
|
global _skill_text_cache, _skill_text_fetched
|
|
if _skill_text_fetched:
|
|
return _skill_text_cache or ""
|
|
_skill_text_fetched = True
|
|
cmd = _find_cli()
|
|
if not cmd:
|
|
return ""
|
|
try:
|
|
proc = subprocess.run(
|
|
[*cmd, "skill"], capture_output=True, text=True, timeout=30
|
|
)
|
|
if proc.returncode == 0 and proc.stdout.strip():
|
|
_skill_text_cache = proc.stdout.strip()
|
|
except Exception as e:
|
|
logger.debug("Could not fetch browser-use skill text: %s", e)
|
|
return _skill_text_cache or ""
|
|
|
|
|
|
def _dynamic_schema_overrides() -> dict:
|
|
skill = _cli_skill_text()
|
|
if not skill:
|
|
return {}
|
|
return {"description": _description_header() + "\n\n---\n\n" + skill}
|
|
|
|
|
|
BROWSER_EXEC_SCHEMA = {
|
|
"name": "browser_exec",
|
|
# Static fallback, used only when the CLI (and uvx) is unavailable
|
|
"description": (
|
|
_HEADER_BASE
|
|
+ "\n\n(The browser-use CLI is not installed yet, so its full skill "
|
|
"documentation could not be loaded. Install it with "
|
|
"`uv tool install browser-use`.)"
|
|
),
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"code": {
|
|
"type": "string",
|
|
"description": "Python code to execute using the pre-imported browser helpers. Use print(...) for any data you need back.",
|
|
},
|
|
"session": {
|
|
"type": "string",
|
|
"description": "Named cloud browser session (sets BU_NAME). Omit for the local default daemon. Use the same name you passed to start_remote_daemon().",
|
|
},
|
|
"timeout_s": {
|
|
"type": "integer",
|
|
"description": f"Max seconds to wait for the code to finish (default {_DEFAULT_TIMEOUT_S}, max {_MAX_TIMEOUT_S}).",
|
|
"default": _DEFAULT_TIMEOUT_S,
|
|
},
|
|
},
|
|
"required": ["code"],
|
|
},
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Registry
|
|
# ---------------------------------------------------------------------------
|
|
from tools.registry import registry
|
|
|
|
registry.register(
|
|
name="browser_exec",
|
|
toolset="browser-use",
|
|
schema=BROWSER_EXEC_SCHEMA,
|
|
handler=lambda args, **kw: browser_exec(
|
|
code=args.get("code", ""),
|
|
session=args.get("session", "") or "",
|
|
timeout_s=args.get("timeout_s", _DEFAULT_TIMEOUT_S),
|
|
task_id=kw.get("task_id"),
|
|
),
|
|
check_fn=is_browser_use_cli_mode,
|
|
dynamic_schema_overrides=_dynamic_schema_overrides,
|
|
emoji="🌐",
|
|
)
|