Something went wrong. Try again.
The agent that grows with you
Something went wrong. Try again.
121 kB · 2990 lines
Python
at bb/theme
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394239523962397239823992400240124022403240424052406240724082409241024112412241324142415241624172418241924202421242224232424242524262427242824292430243124322433243424352436243724382439244024412442244324442445244624472448244924502451245224532454245524562457245824592460246124622463246424652466246724682469247024712472247324742475247624772478247924802481248224832484248524862487248824892490249124922493249424952496249724982499250025012502250325042505250625072508250925102511251225132514251525162517251825192520252125222523252425252526252725282529253025312532253325342535253625372538253925402541254225432544254525462547254825492550255125522553255425552556255725582559256025612562256325642565256625672568256925702571257225732574257525762577257825792580258125822583258425852586258725882589259025912592259325942595259625972598259926002601260226032604260526062607260826092610261126122613261426152616261726182619262026212622262326242625262626272628262926302631263226332634263526362637263826392640264126422643264426452646264726482649265026512652265326542655265626572658265926602661266226632664266526662667266826692670267126722673267426752676267726782679268026812682268326842685268626872688268926902691269226932694269526962697269826992700270127022703270427052706270727082709271027112712271327142715271627172718271927202721272227232724272527262727272827292730273127322733273427352736273727382739274027412742274327442745274627472748274927502751275227532754275527562757275827592760276127622763276427652766276727682769277027712772277327742775277627772778277927802781278227832784278527862787278827892790279127922793279427952796279727982799280028012802280328042805280628072808280928102811281228132814281528162817281828192820282128222823282428252826282728282829283028312832283328342835283628372838283928402841284228432844284528462847284828492850285128522853285428552856285728582859286028612862286328642865286628672868286928702871287228732874287528762877287828792880288128822883288428852886288728882889289028912892289328942895289628972898289929002901290229032904290529062907290829092910291129122913291429152916291729182919292029212922292329242925292629272928292929302931293229332934293529362937293829392940294129422943294429452946294729482949295029512952295329542955295629572958295929602961296229632964296529662967296829692970297129722973297429752976297729782979298029812982298329842985298629872988298929902991#!/usr/bin/env python3"""Browser Tool Module
This module provides browser automation tools using agent-browser CLI. Itsupports multiple backends — **Browser Use** (cloud, default for Noussubscribers), **Browserbase** (cloud, direct credentials), and **localChromium** — with identical agent-facing behaviour. The backend isauto-detected from config and available credentials.
The tool uses agent-browser's accessibility tree (ariaSnapshot) for text-basedpage representation, making it ideal for LLM agents without vision capabilities.
Features:- **Local mode** (default): zero-cost headless Chromium via agent-browser. Works on Linux servers without a display. One-time setup: ``agent-browser install`` (downloads Chromium) or ``agent-browser install --with-deps`` (also installs system libraries for Debian/Ubuntu/Docker).- **Cloud mode**: Browserbase or Browser Use cloud execution when configured.- Session isolation per task ID- Text-based page snapshots using accessibility tree- Element interaction via ref selectors (@e1, @e2, etc.)- Task-aware content extraction using LLM summarization- Automatic cleanup of browser sessions
Environment Variables:- BROWSERBASE_API_KEY: API key for direct Browserbase cloud mode- BROWSERBASE_PROJECT_ID: Project ID for direct Browserbase cloud mode- BROWSER_USE_API_KEY: API key for direct Browser Use cloud mode- BROWSERBASE_PROXIES: Enable/disable residential proxies (default: "true")- BROWSERBASE_ADVANCED_STEALTH: Enable advanced stealth mode with custom Chromium, requires Scale Plan (default: "false")- BROWSERBASE_KEEP_ALIVE: Enable keepAlive for session reconnection after disconnects, requires paid plan (default: "true")- BROWSERBASE_SESSION_TIMEOUT: Custom session timeout in milliseconds. Set to extend beyond project default. Common values: 600000 (10min), 1800000 (30min) (default: none)
Usage: from tools.browser_tool import browser_navigate, browser_snapshot, browser_click # Navigate to a page result = browser_navigate("https://example.com", task_id="task_123") # Get page snapshot snapshot = browser_snapshot(task_id="task_123") # Click an element browser_click("@e5", task_id="task_123")"""
import atexitimport functoolsimport jsonimport loggingimport osimport reimport signalimport subprocessimport shutilimport sysimport tempfileimport threadingimport timeimport requestsfrom typing import Dict, Any, Optional, List, Tuplefrom pathlib import Pathfrom agent.auxiliary_client import call_llmfrom hermes_constants import get_hermes_homefrom utils import is_truthy_value
try: from tools.website_policy import check_website_accessexcept Exception: check_website_access = lambda url: None # noqa: E731 — fail-open if policy module unavailable
try: from tools.url_safety import is_safe_url as _is_safe_urlexcept Exception: _is_safe_url = lambda url: False # noqa: E731 — fail-closed: block all if safety module unavailablefrom tools.browser_providers.base import CloudBrowserProviderfrom tools.browser_providers.browserbase import BrowserbaseProviderfrom tools.browser_providers.browser_use import BrowserUseProviderfrom tools.browser_providers.firecrawl import FirecrawlProviderfrom tools.tool_backend_helpers import normalize_browser_cloud_provider
# Camofox local anti-detection browser backend (optional).# When CAMOFOX_URL is set, all browser operations route through the# camofox REST API instead of the agent-browser CLI.try: from tools.browser_camofox import is_camofox_mode as _is_camofox_modeexcept ImportError: _is_camofox_mode = lambda: False # noqa: E731
logger = logging.getLogger(__name__)
# Standard PATH entries for environments with minimal PATH (e.g. systemd services).# Includes Android/Termux and macOS Homebrew locations needed for agent-browser,# npx, node, and Android's glibc runner (grun)._SANE_PATH_DIRS = ( "/data/data/com.termux/files/usr/bin", "/data/data/com.termux/files/usr/sbin", "/opt/homebrew/bin", "/opt/homebrew/sbin", "/usr/local/sbin", "/usr/local/bin", "/usr/sbin", "/usr/bin", "/sbin", "/bin",)_SANE_PATH = os.pathsep.join(_SANE_PATH_DIRS)
@functools.lru_cache(maxsize=1)def _discover_homebrew_node_dirs() -> tuple[str, ...]: """Find Homebrew versioned Node.js bin directories (e.g. node@20, node@24).
When Node is installed via ``brew install node@24`` and NOT linked into /opt/homebrew/bin, agent-browser isn't discoverable on the default PATH. This function finds those directories so they can be prepended. """ dirs: list[str] = [] homebrew_opt = "/opt/homebrew/opt" if not os.path.isdir(homebrew_opt): return tuple(dirs) try: for entry in os.listdir(homebrew_opt): if entry.startswith("node") and entry != "node": bin_dir = os.path.join(homebrew_opt, entry, "bin") if os.path.isdir(bin_dir): dirs.append(bin_dir) except OSError: pass return tuple(dirs)
def _browser_candidate_path_dirs() -> list[str]: """Return ordered browser CLI PATH candidates shared by discovery and execution.""" hermes_home = get_hermes_home() hermes_node_bin = str(hermes_home / "node" / "bin") return [hermes_node_bin, *list(_discover_homebrew_node_dirs()), *_SANE_PATH_DIRS]
def _merge_browser_path(existing_path: str = "") -> str: """Prepend browser-specific PATH fallbacks without reordering existing entries.""" path_parts = [p for p in (existing_path or "").split(os.pathsep) if p] existing_parts = set(path_parts) prefix_parts: list[str] = []
for part in _browser_candidate_path_dirs(): if not part or part in existing_parts or part in prefix_parts: continue if os.path.isdir(part): prefix_parts.append(part)
return os.pathsep.join(prefix_parts + path_parts)
# Throttle screenshot cleanup to avoid repeated full directory scans._last_screenshot_cleanup_by_dir: dict[str, float] = {}
# ============================================================================# Configuration# ============================================================================
# Default timeout for browser commands (seconds)DEFAULT_COMMAND_TIMEOUT = 30
# Max tokens for snapshot content before summarizationSNAPSHOT_SUMMARIZE_THRESHOLD = 8000
# Commands that legitimately return empty stdout (e.g. close, record)._EMPTY_OK_COMMANDS: frozenset = frozenset({"close", "record"})
_cached_command_timeout: Optional[int] = None_command_timeout_resolved = False
def _get_command_timeout() -> int: """Return the configured browser command timeout from config.yaml.
Reads ``config["browser"]["command_timeout"]`` and falls back to ``DEFAULT_COMMAND_TIMEOUT`` (30s) if unset or unreadable. Result is cached after the first call and cleared by ``cleanup_all_browsers()``. """ global _cached_command_timeout, _command_timeout_resolved if _command_timeout_resolved: return _cached_command_timeout # type: ignore[return-value]
_command_timeout_resolved = True result = DEFAULT_COMMAND_TIMEOUT try: from hermes_cli.config import read_raw_config cfg = read_raw_config() val = cfg.get("browser", {}).get("command_timeout") if val is not None: result = max(int(val), 5) # Floor at 5s to avoid instant kills except Exception as e: logger.debug("Could not read command_timeout from config: %s", e) _cached_command_timeout = result return result
def _get_vision_model() -> Optional[str]: """Model for browser_vision (screenshot analysis — multimodal).""" return os.getenv("AUXILIARY_VISION_MODEL", "").strip() or None
def _get_extraction_model() -> Optional[str]: """Model for page snapshot text summarization — same as web_extract.""" return os.getenv("AUXILIARY_WEB_EXTRACT_MODEL", "").strip() or None
def _resolve_cdp_override(cdp_url: str) -> str: """Normalize a user-supplied CDP endpoint into a concrete connectable URL.
Accepts: - full websocket endpoints: ws://host:port/devtools/browser/... - HTTP discovery endpoints: http://host:port or http://host:port/json/version - bare websocket host:port values like ws://host:port
For discovery-style endpoints we fetch /json/version and return the webSocketDebuggerUrl so downstream tools always receive a concrete browser websocket instead of an ambiguous host:port URL. """ raw = (cdp_url or "").strip() if not raw: return ""
lowered = raw.lower() if "/devtools/browser/" in lowered: return raw
discovery_url = raw if lowered.startswith(("ws://", "wss://")): if raw.count(":") == 2 and raw.rstrip("/").rsplit(":", 1)[-1].isdigit() and "/" not in raw.split(":", 2)[-1]: discovery_url = ("http://" if lowered.startswith("ws://") else "https://") + raw.split("://", 1)[1] else: return raw
if discovery_url.lower().endswith("/json/version"): version_url = discovery_url else: version_url = discovery_url.rstrip("/") + "/json/version"
try: response = requests.get(version_url, timeout=10) response.raise_for_status() payload = response.json() except Exception as exc: logger.warning("Failed to resolve CDP endpoint %s via %s: %s", raw, version_url, exc) return raw
ws_url = str(payload.get("webSocketDebuggerUrl") or "").strip() if ws_url: logger.info("Resolved CDP endpoint %s -> %s", raw, ws_url) return ws_url
logger.warning("CDP discovery at %s did not return webSocketDebuggerUrl; using raw endpoint", version_url) return raw
def _get_cdp_override() -> str: """Return a normalized CDP URL override, or empty string.
Precedence is: 1. ``BROWSER_CDP_URL`` env var (live override from ``/browser connect``) 2. ``browser.cdp_url`` in config.yaml (persistent config)
When either is set, we skip both Browserbase and the local headless launcher and connect directly to the supplied Chrome DevTools Protocol endpoint. """ env_override = os.environ.get("BROWSER_CDP_URL", "").strip() if env_override: return _resolve_cdp_override(env_override)
try: from hermes_cli.config import read_raw_config
cfg = read_raw_config() browser_cfg = cfg.get("browser", {}) if isinstance(browser_cfg, dict): return _resolve_cdp_override(str(browser_cfg.get("cdp_url", "") or "")) except Exception as e: logger.debug("Could not read browser.cdp_url from config: %s", e)
return ""
def _get_dialog_policy_config() -> Tuple[str, float]: """Read ``browser.dialog_policy`` + ``browser.dialog_timeout_s`` from config.
Returns a ``(policy, timeout_s)`` tuple, falling back to the supervisor's defaults when keys are absent or invalid. """ # Defer imports so browser_tool can be imported in minimal environments. from tools.browser_supervisor import ( DEFAULT_DIALOG_POLICY, DEFAULT_DIALOG_TIMEOUT_S, _VALID_POLICIES, )
try: from hermes_cli.config import read_raw_config
cfg = read_raw_config() browser_cfg = cfg.get("browser", {}) if isinstance(cfg, dict) else {} if not isinstance(browser_cfg, dict): return DEFAULT_DIALOG_POLICY, DEFAULT_DIALOG_TIMEOUT_S policy = str(browser_cfg.get("dialog_policy") or DEFAULT_DIALOG_POLICY) if policy not in _VALID_POLICIES: logger.debug("Invalid browser.dialog_policy=%r; using default", policy) policy = DEFAULT_DIALOG_POLICY timeout_raw = browser_cfg.get("dialog_timeout_s") try: timeout_s = float(timeout_raw) if timeout_raw is not None else DEFAULT_DIALOG_TIMEOUT_S if timeout_s <= 0: timeout_s = DEFAULT_DIALOG_TIMEOUT_S except (TypeError, ValueError): timeout_s = DEFAULT_DIALOG_TIMEOUT_S return policy, timeout_s except Exception: return DEFAULT_DIALOG_POLICY, DEFAULT_DIALOG_TIMEOUT_S
def _ensure_cdp_supervisor(task_id: str) -> None: """Start a CDP supervisor for ``task_id`` if an endpoint is reachable.
Idempotent — delegates to ``SupervisorRegistry.get_or_start`` which skips when a supervisor for this ``(task_id, cdp_url)`` already exists and tears down + restarts on URL change. Safe to call on every ``browser_navigate`` / ``/browser connect`` without worrying about double-attach.
Resolves the CDP URL in this order: 1. ``BROWSER_CDP_URL`` / ``browser.cdp_url`` — covers ``/browser connect`` and config-set overrides. 2. ``_active_sessions[task_id]["cdp_url"]`` — covers Browserbase + any other cloud provider whose ``create_session`` returns a raw CDP URL.
Swallows all errors — failing to attach the supervisor must not break the browser session itself. The agent simply won't see ``pending_dialogs`` / ``frame_tree`` fields in snapshots. """ cdp_url = _get_cdp_override() if not cdp_url: # Fallback: active session may carry a per-session CDP URL from a # cloud provider (Browserbase sets this). with _cleanup_lock: session_info = _active_sessions.get(task_id, {}) maybe = str(session_info.get("cdp_url") or "") if maybe: cdp_url = _resolve_cdp_override(maybe) if not cdp_url: return try: from tools.browser_supervisor import SUPERVISOR_REGISTRY # type: ignore[import-not-found]
policy, timeout_s = _get_dialog_policy_config() SUPERVISOR_REGISTRY.get_or_start( task_id=task_id, cdp_url=cdp_url, dialog_policy=policy, dialog_timeout_s=timeout_s, ) except Exception as exc: logger.debug( "CDP supervisor attach for task=%s failed (non-fatal): %s", task_id, exc, )
def _stop_cdp_supervisor(task_id: str) -> None: """Stop the CDP supervisor for ``task_id`` if one exists. No-op otherwise.""" try: from tools.browser_supervisor import SUPERVISOR_REGISTRY # type: ignore[import-not-found]
SUPERVISOR_REGISTRY.stop(task_id) except Exception as exc: logger.debug("CDP supervisor stop for task=%s failed (non-fatal): %s", task_id, exc)
# ============================================================================# Cloud Provider Registry# ============================================================================
_PROVIDER_REGISTRY: Dict[str, type] = { "browserbase": BrowserbaseProvider, "browser-use": BrowserUseProvider, "firecrawl": FirecrawlProvider,}
_cached_cloud_provider: Optional[CloudBrowserProvider] = None_cloud_provider_resolved = False_allow_private_urls_resolved = False_cached_allow_private_urls: Optional[bool] = None_cached_agent_browser: Optional[str] = None_agent_browser_resolved = False
def _get_cloud_provider() -> Optional[CloudBrowserProvider]: """Return the configured cloud browser provider, or None for local mode.
Reads ``config["browser"]["cloud_provider"]`` once and caches the result for the process lifetime. An explicit ``local`` provider disables cloud fallback. If unset, fall back to Browserbase when direct or managed Browserbase credentials are available. """ global _cached_cloud_provider, _cloud_provider_resolved if _cloud_provider_resolved: return _cached_cloud_provider
_cloud_provider_resolved = True try: from hermes_cli.config import read_raw_config cfg = read_raw_config() browser_cfg = cfg.get("browser", {}) provider_key = None if isinstance(browser_cfg, dict) and "cloud_provider" in browser_cfg: provider_key = normalize_browser_cloud_provider( browser_cfg.get("cloud_provider") ) if provider_key == "local": _cached_cloud_provider = None return None if provider_key and provider_key in _PROVIDER_REGISTRY: _cached_cloud_provider = _PROVIDER_REGISTRY[provider_key]() except Exception as e: logger.debug("Could not read cloud_provider from config: %s", e)
if _cached_cloud_provider is None: # Prefer Browser Use (managed Nous gateway or direct API key), # fall back to Browserbase (direct credentials only). fallback_provider = BrowserUseProvider() if fallback_provider.is_configured(): _cached_cloud_provider = fallback_provider else: fallback_provider = BrowserbaseProvider() if fallback_provider.is_configured(): _cached_cloud_provider = fallback_provider
return _cached_cloud_provider
from hermes_constants import is_termux as _is_termux_environment
def _browser_install_hint() -> str: if _is_termux_environment(): return "npm install -g agent-browser && agent-browser install" return "npm install -g agent-browser && agent-browser install --with-deps"
def _requires_real_termux_browser_install(browser_cmd: str) -> bool: return _is_termux_environment() and _is_local_mode() and browser_cmd.strip() == "npx agent-browser"
def _termux_browser_install_error() -> str: return ( "Local browser automation on Termux cannot rely on the bare npx fallback. " f"Install agent-browser explicitly first: {_browser_install_hint()}" )
def _is_local_mode() -> bool: """Return True when the browser tool will use a local browser backend.""" if _get_cdp_override(): return False return _get_cloud_provider() is None
def _is_local_backend() -> bool: """Return True when the browser runs locally (no cloud provider).
SSRF protection is only meaningful for cloud backends (Browserbase, BrowserUse) where the agent could reach internal resources on a remote machine. For local backends — Camofox, or the built-in headless Chromium without a cloud provider — the user already has full terminal and network access on the same machine, so the check adds no security value. """ return _is_camofox_mode() or _get_cloud_provider() is None
_auto_local_for_private_urls_resolved = False_cached_auto_local_for_private_urls: bool = True
def _auto_local_for_private_urls() -> bool: """Return whether a cloud-configured install should auto-spawn a local Chromium for LAN/localhost URLs.
Reads ``browser.auto_local_for_private_urls`` once (default ``True``) and caches it for the process lifetime. When enabled, ``browser_navigate`` routes URLs whose host resolves to a private/loopback/LAN address to a local headless Chromium sidecar even when a cloud provider (Browserbase / Browser-Use / Firecrawl) is configured globally. Public URLs continue to use the cloud provider in the same conversation. """ global _auto_local_for_private_urls_resolved, _cached_auto_local_for_private_urls if _auto_local_for_private_urls_resolved: return _cached_auto_local_for_private_urls
_auto_local_for_private_urls_resolved = True try: from hermes_cli.config import read_raw_config cfg = read_raw_config() browser_cfg = cfg.get("browser", {}) if isinstance(browser_cfg, dict) and "auto_local_for_private_urls" in browser_cfg: _cached_auto_local_for_private_urls = bool( browser_cfg.get("auto_local_for_private_urls") ) except Exception as e: logger.debug("Could not read auto_local_for_private_urls from config: %s", e) return _cached_auto_local_for_private_urls
def _url_is_private(url: str) -> bool: """Return True when the URL's host resolves to a private/LAN/loopback address.
Reuses ``tools.url_safety.is_safe_url`` as the oracle — if the SSRF check would reject the URL, we treat it as "private" for routing purposes. DNS resolution failures are treated as NOT private (fall through to whatever backend is configured, which will surface the DNS error naturally). """ try: # is_safe_url returns False for private/loopback/link-local/CGNAT AND # for DNS failures. We only want the private-network case here, so # we parse + check the host shape as a DNS-failure sieve first. from urllib.parse import urlparse import ipaddress import socket parsed = urlparse(url) hostname = (parsed.hostname or "").strip().lower().rstrip(".") if not hostname: return False # Literal IP → check directly try: ip = ipaddress.ip_address(hostname) return ( ip.is_private or ip.is_loopback or ip.is_link_local or ip in ipaddress.ip_network("100.64.0.0/10") ) except ValueError: pass # Hostname — must resolve to confirm it's private (bare "localhost" # resolves to 127.0.0.1 via /etc/hosts). Short-circuit on obvious # names to avoid a DNS hop. if hostname in ("localhost",) or hostname.endswith(".localhost"): return True if hostname.endswith(".local") or hostname.endswith(".lan") or hostname.endswith(".internal"): return True try: addr_info = socket.getaddrinfo(hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM) except socket.gaierror: return False # DNS fail → not private, let the normal path fail for _, _, _, _, sockaddr in addr_info: try: ip = ipaddress.ip_address(sockaddr[0]) except ValueError: continue if ( ip.is_private or ip.is_loopback or ip.is_link_local or ip in ipaddress.ip_network("100.64.0.0/10") ): return True return False except Exception as exc: logger.debug("URL-privacy check failed for %s: %s", url, exc) return False
def _navigation_session_key(task_id: str, url: str) -> str: """Pick the session key that should handle ``url`` for ``task_id``.
Returns the bare task_id unless ALL of these are true: 1. A cloud provider is configured (``_get_cloud_provider()`` is not None). 2. Auto-local routing is enabled (``browser.auto_local_for_private_urls``, default True). 3. The URL resolves to a private/LAN/loopback address. 4. A CDP override is not active (that path owns the whole session). 5. Camofox mode is not active (Camofox is already local-only).
When all are true, returns ``f"{task_id}::local"`` so the hybrid-routing path spawns a local Chromium sidecar while the cloud session (if any) continues to serve public URLs. """ if task_id is None: task_id = "default" if _get_cdp_override(): return task_id if _is_camofox_mode(): return task_id if _get_cloud_provider() is None: return task_id if not _auto_local_for_private_urls(): return task_id if not _url_is_private(url): return task_id return f"{task_id}{_LOCAL_SUFFIX}"
def _is_local_sidecar_key(session_key: str) -> bool: """Return True when ``session_key`` is a hybrid-routing local sidecar.""" return session_key.endswith(_LOCAL_SUFFIX)
def _last_session_key(task_id: str) -> str: """Return the session key to use for a non-nav browser tool call.
If a previous ``browser_navigate`` on this task_id set a last-active key, use it so snapshot/click/fill/etc. hit the same session. Otherwise fall back to the bare task_id (matches original behavior for tasks that never triggered hybrid routing). """ if task_id is None: task_id = "default" return _last_active_session_key.get(task_id, task_id)
def _allow_private_urls() -> bool: """Return whether the browser is allowed to navigate to private/internal addresses.
Reads ``config["browser"]["allow_private_urls"]`` once and caches the result for the process lifetime. Defaults to ``False`` (SSRF protection active). """ global _cached_allow_private_urls, _allow_private_urls_resolved if _allow_private_urls_resolved: return _cached_allow_private_urls
_allow_private_urls_resolved = True _cached_allow_private_urls = False # safe default try: from hermes_cli.config import read_raw_config cfg = read_raw_config() browser_cfg = cfg.get("browser", {}) if isinstance(browser_cfg, dict): _cached_allow_private_urls = is_truthy_value( browser_cfg.get("allow_private_urls"), default=False ) except Exception as e: logger.debug("Could not read allow_private_urls from config: %s", e) return _cached_allow_private_urls
def _socket_safe_tmpdir() -> str: """Return a short temp directory path suitable for Unix domain sockets.
macOS sets ``TMPDIR`` to ``/var/folders/xx/.../T/`` (~51 chars). When we append ``agent-browser-hermes_…`` the resulting socket path exceeds the 104-byte macOS limit for ``AF_UNIX`` addresses, causing agent-browser to fail with "Failed to create socket directory" or silent screenshot failures.
Linux ``tempfile.gettempdir()`` already returns ``/tmp``, so this is a no-op there. On macOS we bypass ``TMPDIR`` and use ``/tmp`` directly (symlink to ``/private/tmp``, sticky-bit protected, always available). """ if sys.platform == "darwin": return "/tmp" return tempfile.gettempdir()
# Track active sessions per "session key".## A "session key" is either the bare task_id (cloud/default path) OR a composite# like f"{task_id}::local" when the hybrid-routing feature spawns a local sidecar# browser for a LAN/localhost URL while a cloud provider is configured globally.# Both forms flow through the same _active_sessions / _run_browser_command /# cleanup_browser code paths — the key is opaque to those internals.## Stores: session_name (always), bb_session_id + cdp_url (cloud mode only)_active_sessions: Dict[str, Dict[str, str]] = {} # session_key -> {session_name, ...}_recording_sessions: set = set() # session_keys with active recordings
# Tracks the most recent session_key used per task_id. Set by browser_navigate()# after it chooses a backend for a URL; read by every non-nav browser tool# (snapshot/click/fill/eval/...) so they target the session that served the last# navigation. Without this, a task that navigated to localhost on the local# sidecar would fall back to the cloud session on its next snapshot call._last_active_session_key: Dict[str, str] = {} # task_id -> session_key_LOCAL_SUFFIX = "::local"
# Flag to track if cleanup has been done_cleanup_done = False
# =============================================================================# Inactivity Timeout Configuration# =============================================================================
# Session inactivity timeout (seconds) - cleanup if no activity for this long# Default: 5 minutes. Needs headroom for LLM reasoning between browser commands,# especially when subagents are doing multi-step browser tasks.BROWSER_SESSION_INACTIVITY_TIMEOUT = int(os.environ.get("BROWSER_INACTIVITY_TIMEOUT", "300"))
# Track last activity time per session_session_last_activity: Dict[str, float] = {}
# Background cleanup thread state_cleanup_thread = None_cleanup_running = False# Protects _session_last_activity AND _active_sessions for thread safety# (subagents run concurrently via ThreadPoolExecutor)_cleanup_lock = threading.Lock()
def _emergency_cleanup_all_sessions(): """ Emergency cleanup of all active browser sessions. Called on process exit or interrupt to prevent orphaned sessions.
Also runs the orphan reaper to clean up daemons left behind by previously crashed hermes processes — this way every clean hermes exit sweeps accumulated orphans, not just ones that actively used the browser tool. """ global _cleanup_done if _cleanup_done: return _cleanup_done = True
# Clean up this process's own sessions first, so their owner_pid files # are removed before the reaper scans. if _active_sessions: logger.info("Emergency cleanup: closing %s active session(s)...", len(_active_sessions)) try: cleanup_all_browsers() except Exception as e: logger.error("Emergency cleanup error: %s", e) finally: with _cleanup_lock: _active_sessions.clear() _session_last_activity.clear() _recording_sessions.clear()
# Sweep orphans from other crashed hermes processes. Safe even if we # never used the browser — uses owner_pid liveness to avoid reaping # daemons owned by other live hermes processes. try: _reap_orphaned_browser_sessions() except Exception as e: logger.debug("Orphan reap on exit failed: %s", e)
# Register cleanup via atexit only. Previous versions installed SIGINT/SIGTERM# handlers that called sys.exit(), but this conflicts with prompt_toolkit's# async event loop — a SystemExit raised inside a key-binding callback# corrupts the coroutine state and makes the process unkillable. atexit# handlers run on any normal exit (including sys.exit), so browser sessions# are still cleaned up without hijacking signals.atexit.register(_emergency_cleanup_all_sessions)
# =============================================================================# Inactivity Cleanup Functions# =============================================================================
def _cleanup_inactive_browser_sessions(): """ Clean up browser sessions that have been inactive for longer than the timeout. This function is called periodically by the background cleanup thread to automatically close sessions that haven't been used recently, preventing orphaned sessions (local or Browserbase) from accumulating. """ current_time = time.time() sessions_to_cleanup = [] with _cleanup_lock: for task_id, last_time in list(_session_last_activity.items()): if current_time - last_time > BROWSER_SESSION_INACTIVITY_TIMEOUT: sessions_to_cleanup.append(task_id) for task_id in sessions_to_cleanup: try: elapsed = int(current_time - _session_last_activity.get(task_id, current_time)) logger.info("Cleaning up inactive session for task: %s (inactive for %ss)", task_id, elapsed) cleanup_browser(task_id) with _cleanup_lock: if task_id in _session_last_activity: del _session_last_activity[task_id] except Exception as e: logger.warning("Error cleaning up inactive session %s: %s", task_id, e)
def _write_owner_pid(socket_dir: str, session_name: str) -> None: """Record the current hermes PID as the owner of a browser socket dir.
Written atomically to ``<socket_dir>/<session_name>.owner_pid`` so the orphan reaper can distinguish daemons owned by a live hermes process (don't reap) from daemons whose owner crashed (reap). Best-effort — an OSError here just falls back to the legacy ``tracked_names`` heuristic in the reaper. """ try: path = os.path.join(socket_dir, f"{session_name}.owner_pid") with open(path, "w") as f: f.write(str(os.getpid())) except OSError as exc: logger.debug("Could not write owner_pid file for %s: %s", session_name, exc)
def _reap_orphaned_browser_sessions(): """Scan for orphaned agent-browser daemon processes from previous runs.
When the Python process that created a browser session exits uncleanly (SIGKILL, crash, gateway restart), the in-memory ``_active_sessions`` tracking is lost but the node + Chromium processes keep running.
This function scans the tmp directory for ``agent-browser-*`` socket dirs left behind by previous runs, reads the daemon PID files, and kills any daemons whose owning hermes process is no longer alive.
Ownership detection priority: 1. ``<session>.owner_pid`` file (written by current code) — if the referenced hermes PID is alive, leave the daemon alone regardless of whether it's in *this* process's ``_active_sessions``. This is cross-process safe: two concurrent hermes instances won't reap each other's daemons. 2. Fallback for daemons that predate owner_pid: check ``_active_sessions`` in the current process. If not tracked here, treat as orphan (legacy behavior).
Safe to call from any context — atexit, cleanup thread, or on demand. """ import glob
tmpdir = _socket_safe_tmpdir() pattern = os.path.join(tmpdir, "agent-browser-h_*") socket_dirs = glob.glob(pattern) # Also pick up CDP sessions socket_dirs += glob.glob(os.path.join(tmpdir, "agent-browser-cdp_*")) # Also pick up cloud-provider sessions (browser-use/browserbase/firecrawl) socket_dirs += glob.glob(os.path.join(tmpdir, "agent-browser-hermes_*"))
if not socket_dirs: return
# Build set of session_names currently tracked by this process (fallback path) with _cleanup_lock: tracked_names = { info.get("session_name") for info in _active_sessions.values() if info.get("session_name") }
reaped = 0 for socket_dir in socket_dirs: dir_name = os.path.basename(socket_dir) # dir_name is "agent-browser-{session_name}" session_name = dir_name.removeprefix("agent-browser-") if not session_name: continue
# Ownership check: prefer owner_pid file (cross-process safe). owner_pid_file = os.path.join(socket_dir, f"{session_name}.owner_pid") owner_alive: Optional[bool] = None # None = owner_pid missing/unreadable if os.path.isfile(owner_pid_file): try: owner_pid = int(Path(owner_pid_file).read_text().strip()) try: os.kill(owner_pid, 0) owner_alive = True except ProcessLookupError: owner_alive = False except PermissionError: # Owner exists but we can't signal it (different uid). # Treat as alive — don't reap someone else's session. owner_alive = True except (ValueError, OSError): owner_alive = None # corrupt file — fall through
if owner_alive is True: # Owner is alive — this session belongs to a live hermes process. continue
if owner_alive is None: # No owner_pid file (legacy daemon). Fall back to in-process # tracking: if this process knows about the session, leave alone. if session_name in tracked_names: continue
# owner_alive is False (dead owner) OR legacy daemon not tracked here. pid_file = os.path.join(socket_dir, f"{session_name}.pid") if not os.path.isfile(pid_file): # No daemon PID file — just a stale dir, remove it shutil.rmtree(socket_dir, ignore_errors=True) continue
try: daemon_pid = int(Path(pid_file).read_text().strip()) except (ValueError, OSError): shutil.rmtree(socket_dir, ignore_errors=True) continue
# Check if the daemon is still alive try: os.kill(daemon_pid, 0) # signal 0 = existence check except ProcessLookupError: # Already dead, just clean up the dir shutil.rmtree(socket_dir, ignore_errors=True) continue except PermissionError: # Alive but owned by someone else — leave it alone continue
# Daemon is alive and its owner is dead (or legacy + untracked). Reap. try: os.kill(daemon_pid, signal.SIGTERM) logger.info("Reaped orphaned browser daemon PID %d (session %s)", daemon_pid, session_name) reaped += 1 except (ProcessLookupError, PermissionError, OSError): pass
# Clean up the socket directory shutil.rmtree(socket_dir, ignore_errors=True)
if reaped: logger.info("Reaped %d orphaned browser session(s) from previous run(s)", reaped)
def _browser_cleanup_thread_worker(): """ Background thread that periodically cleans up inactive browser sessions. Runs every 30 seconds and checks for sessions that haven't been used within the BROWSER_SESSION_INACTIVITY_TIMEOUT period. On first run, also reaps orphaned sessions from previous process lifetimes. """ # One-time orphan reap on startup try: _reap_orphaned_browser_sessions() except Exception as e: logger.warning("Orphan reap error: %s", e)
while _cleanup_running: try: _cleanup_inactive_browser_sessions() except Exception as e: logger.warning("Cleanup thread error: %s", e) # Sleep in 1-second intervals so we can stop quickly if needed for _ in range(30): if not _cleanup_running: break time.sleep(1)
def _start_browser_cleanup_thread(): """Start the background cleanup thread if not already running.""" global _cleanup_thread, _cleanup_running with _cleanup_lock: if _cleanup_thread is None or not _cleanup_thread.is_alive(): _cleanup_running = True _cleanup_thread = threading.Thread( target=_browser_cleanup_thread_worker, daemon=True, name="browser-cleanup" ) _cleanup_thread.start() logger.info("Started inactivity cleanup thread (timeout: %ss)", BROWSER_SESSION_INACTIVITY_TIMEOUT)
def _stop_browser_cleanup_thread(): """Stop the background cleanup thread.""" global _cleanup_running _cleanup_running = False if _cleanup_thread is not None: _cleanup_thread.join(timeout=5)
def _update_session_activity(task_id: str): """Update the last activity timestamp for a session.""" with _cleanup_lock: _session_last_activity[task_id] = time.time()
# Register cleanup thread stop on exitatexit.register(_stop_browser_cleanup_thread)
# ============================================================================# Tool Schemas# ============================================================================
BROWSER_TOOL_SCHEMAS = [ { "name": "browser_navigate", "description": "Navigate to a URL in the browser. Initializes the session and loads the page. Must be called before other browser tools. For simple information retrieval, prefer web_search or web_extract (faster, cheaper). For plain-text endpoints — URLs ending in .md, .txt, .json, .yaml, .yml, .csv, .xml, raw.githubusercontent.com, or any documented API endpoint — prefer curl via the terminal tool or web_extract; the browser stack is overkill and much slower for these. Use browser tools when you need to interact with a page (click, fill forms, dynamic content). Returns a compact page snapshot with interactive elements and ref IDs — no need to call browser_snapshot separately after navigating.", "parameters": { "type": "object", "properties": { "url": { "type": "string", "description": "The URL to navigate to (e.g., 'https://example.com')" } }, "required": ["url"] } }, { "name": "browser_snapshot", "description": "Get a text-based snapshot of the current page's accessibility tree. Returns interactive elements with ref IDs (like @e1, @e2) for browser_click and browser_type. full=false (default): compact view with interactive elements. full=true: complete page content. Snapshots over 8000 chars are truncated or LLM-summarized. Requires browser_navigate first. Note: browser_navigate already returns a compact snapshot — use this to refresh after interactions that change the page, or with full=true for complete content.", "parameters": { "type": "object", "properties": { "full": { "type": "boolean", "description": "If true, returns complete page content. If false (default), returns compact view with interactive elements only.", "default": False } }, "required": [] } }, { "name": "browser_click", "description": "Click on an element identified by its ref ID from the snapshot (e.g., '@e5'). The ref IDs are shown in square brackets in the snapshot output. Requires browser_navigate and browser_snapshot to be called first.", "parameters": { "type": "object", "properties": { "ref": { "type": "string", "description": "The element reference from the snapshot (e.g., '@e5', '@e12')" } }, "required": ["ref"] } }, { "name": "browser_type", "description": "Type text into an input field identified by its ref ID. Clears the field first, then types the new text. Requires browser_navigate and browser_snapshot to be called first.", "parameters": { "type": "object", "properties": { "ref": { "type": "string", "description": "The element reference from the snapshot (e.g., '@e3')" }, "text": { "type": "string", "description": "The text to type into the field" } }, "required": ["ref", "text"] } }, { "name": "browser_scroll", "description": "Scroll the page in a direction. Use this to reveal more content that may be below or above the current viewport. Requires browser_navigate to be called first.", "parameters": { "type": "object", "properties": { "direction": { "type": "string", "enum": ["up", "down"], "description": "Direction to scroll" } }, "required": ["direction"] } }, { "name": "browser_back", "description": "Navigate back to the previous page in browser history. Requires browser_navigate to be called first.", "parameters": { "type": "object", "properties": {}, "required": [] } }, { "name": "browser_press", "description": "Press a keyboard key. Useful for submitting forms (Enter), navigating (Tab), or keyboard shortcuts. Requires browser_navigate to be called first.", "parameters": { "type": "object", "properties": { "key": { "type": "string", "description": "Key to press (e.g., 'Enter', 'Tab', 'Escape', 'ArrowDown')" } }, "required": ["key"] } }, { "name": "browser_get_images", "description": "Get a list of all images on the current page with their URLs and alt text. Useful for finding images to analyze with the vision tool. Requires browser_navigate to be called first.", "parameters": { "type": "object", "properties": {}, "required": [] } }, { "name": "browser_vision", "description": "Take a screenshot of the current page and analyze it with vision AI. Use this when you need to visually understand what's on the page - especially useful for CAPTCHAs, visual verification challenges, complex layouts, or when the text snapshot doesn't capture important visual information. Returns both the AI analysis and a screenshot_path that you can share with the user by including MEDIA:<screenshot_path> in your response. Requires browser_navigate to be called first.", "parameters": { "type": "object", "properties": { "question": { "type": "string", "description": "What you want to know about the page visually. Be specific about what you're looking for." }, "annotate": { "type": "boolean", "default": False, "description": "If true, overlay numbered [N] labels on interactive elements. Each [N] maps to ref @eN for subsequent browser commands. Useful for QA and spatial reasoning about page layout." } }, "required": ["question"] } }, { "name": "browser_console", "description": "Get browser console output and JavaScript errors from the current page. Returns console.log/warn/error/info messages and uncaught JS exceptions. Use this to detect silent JavaScript errors, failed API calls, and application warnings. Requires browser_navigate to be called first. When 'expression' is provided, evaluates JavaScript in the page context and returns the result — use this for DOM inspection, reading page state, or extracting data programmatically.", "parameters": { "type": "object", "properties": { "clear": { "type": "boolean", "default": False, "description": "If true, clear the message buffers after reading" }, "expression": { "type": "string", "description": "JavaScript expression to evaluate in the page context. Runs in the browser like DevTools console — full access to DOM, window, document. Return values are serialized to JSON. Example: 'document.title' or 'document.querySelectorAll(\"a\").length'" } }, "required": [] } },]
# ============================================================================# Utility Functions# ============================================================================
def _create_local_session(task_id: str) -> Dict[str, str]: import uuid session_name = f"h_{uuid.uuid4().hex[:10]}" logger.info("Created local browser session %s for task %s", session_name, task_id) return { "session_name": session_name, "bb_session_id": None, "cdp_url": None, "features": {"local": True}, }
def _create_cdp_session(task_id: str, cdp_url: str) -> Dict[str, str]: """Create a session that connects to a user-supplied CDP endpoint.""" import uuid session_name = f"cdp_{uuid.uuid4().hex[:10]}" logger.info("Created CDP browser session %s → %s for task %s", session_name, cdp_url, task_id) return { "session_name": session_name, "bb_session_id": None, "cdp_url": cdp_url, "features": {"cdp_override": True}, }
def _get_session_info(task_id: Optional[str] = None) -> Dict[str, str]: """ Get or create session info for the given session key.
In cloud mode, creates a Browserbase session with proxies enabled. In local mode, generates a session name for agent-browser --session. Also starts the inactivity cleanup thread and updates activity tracking. Thread-safe: multiple subagents can call this concurrently.
Args: task_id: Session key. Normally the task_id as-is, but may carry the ``::local`` suffix for the hybrid-routing local sidecar — in that case the cloud provider is skipped even when one is configured, and a local Chromium session is created instead.
Returns: Dict with session_name (always), bb_session_id + cdp_url (cloud only) """ if task_id is None: task_id = "default"
# Start the cleanup thread if not running (handles inactivity timeouts) _start_browser_cleanup_thread()
# Update activity timestamp for this session _update_session_activity(task_id)
with _cleanup_lock: # Check if we already have a session for this task if task_id in _active_sessions: return _active_sessions[task_id]
# Hybrid routing: session keys ending with ``::local`` force a local # Chromium regardless of the globally-configured cloud provider. Public # URLs in the same conversation continue to use the cloud session under # the bare task_id key. force_local = _is_local_sidecar_key(task_id)
# Create session outside the lock (network call in cloud mode) cdp_override = _get_cdp_override() if cdp_override and not force_local: session_info = _create_cdp_session(task_id, cdp_override) elif force_local: session_info = _create_local_session(task_id) else: provider = _get_cloud_provider() if provider is None: session_info = _create_local_session(task_id) else: try: session_info = provider.create_session(task_id) # Validate cloud provider returned a usable session if not session_info or not isinstance(session_info, dict): raise ValueError(f"Cloud provider returned invalid session: {session_info!r}") if session_info.get("cdp_url"): # Some cloud providers (including Browser-Use v3) return an HTTP # CDP discovery URL instead of a raw websocket endpoint. session_info = dict(session_info) session_info["cdp_url"] = _resolve_cdp_override(str(session_info["cdp_url"])) except Exception as e: provider_name = type(provider).__name__ logger.warning( "Cloud provider %s failed (%s); attempting fallback to local " "Chromium for task %s", provider_name, e, task_id, exc_info=True, ) try: session_info = _create_local_session(task_id) except Exception as local_error: raise RuntimeError( f"Cloud provider {provider_name} failed ({e}) and local " f"fallback also failed ({local_error})" ) from e # Mark session as degraded for observability if isinstance(session_info, dict): session_info = dict(session_info) session_info["fallback_from_cloud"] = True session_info["fallback_reason"] = str(e) session_info["fallback_provider"] = provider_name
with _cleanup_lock: # Double-check: another thread may have created a session while we # were doing the network call. Use the existing one to avoid leaking # orphan cloud sessions. if task_id in _active_sessions: return _active_sessions[task_id] _active_sessions[task_id] = session_info
# Lazy-start the CDP supervisor now that the session exists (if the # backend surfaces a CDP URL via override or session_info["cdp_url"]). # Idempotent; swallows errors. See _ensure_cdp_supervisor for details. # Skip for local sidecars — they have no CDP URL. if not force_local: _ensure_cdp_supervisor(task_id)
return session_info
def _find_agent_browser() -> str: """ Find the agent-browser CLI executable. Checks in order: current PATH, Homebrew/common bin dirs, Hermes-managed node, local node_modules/.bin/, npx fallback. Returns: Path to agent-browser executable Raises: FileNotFoundError: If agent-browser is not installed """ global _cached_agent_browser, _agent_browser_resolved if _agent_browser_resolved: if _cached_agent_browser is None: raise FileNotFoundError( "agent-browser CLI not found (cached). Install it with: " f"{_browser_install_hint()}\n" "Or run 'npm install' in the repo root to install locally.\n" "Or ensure npx is available in your PATH." ) return _cached_agent_browser
# Note: _agent_browser_resolved is set at each return site below # (not before the search) to prevent a race where a concurrent thread # sees resolved=True but _cached_agent_browser is still None.
# Check if it's in PATH (global install) which_result = shutil.which("agent-browser") if which_result: _cached_agent_browser = which_result _agent_browser_resolved = True return which_result
# Build an extended search PATH including Hermes-managed Node, macOS # versioned Homebrew installs, and fallback system dirs like Termux. extended_path = _merge_browser_path("") if extended_path: which_result = shutil.which("agent-browser", path=extended_path) if which_result: _cached_agent_browser = which_result _agent_browser_resolved = True return which_result
# Check local node_modules/.bin/ (npm install in repo root) repo_root = Path(__file__).parent.parent local_bin = repo_root / "node_modules" / ".bin" / "agent-browser" if local_bin.exists(): _cached_agent_browser = str(local_bin) _agent_browser_resolved = True return _cached_agent_browser # Check common npx locations (also search the extended fallback PATH) npx_path = shutil.which("npx") if not npx_path and extended_path: npx_path = shutil.which("npx", path=extended_path) if npx_path: _cached_agent_browser = "npx agent-browser" _agent_browser_resolved = True return _cached_agent_browser # Nothing found — cache the failure so subsequent calls don't re-scan. _agent_browser_resolved = True raise FileNotFoundError( "agent-browser CLI not found. Install it with: " f"{_browser_install_hint()}\n" "Or run 'npm install' in the repo root to install locally.\n" "Or ensure npx is available in your PATH." )
def _extract_screenshot_path_from_text(text: str) -> Optional[str]: """Extract a screenshot file path from agent-browser human-readable output.""" if not text: return None
patterns = [ r"Screenshot saved to ['\"](?P<path>/[^'\"]+?\.png)['\"]", r"Screenshot saved to (?P<path>/\S+?\.png)(?:\s|$)", r"(?P<path>/\S+?\.png)(?:\s|$)", ]
for pattern in patterns: match = re.search(pattern, text) if match: path = match.group("path").strip().strip("'\"") if path: return path
return None
def _run_browser_command( task_id: str, command: str, args: List[str] = None, timeout: Optional[int] = None,) -> Dict[str, Any]: """ Run an agent-browser CLI command using our pre-created Browserbase session. Args: task_id: Task identifier to get the right session command: The command to run (e.g., "open", "click") args: Additional arguments for the command timeout: Command timeout in seconds. ``None`` reads ``browser.command_timeout`` from config (default 30s). Returns: Parsed JSON response from agent-browser """ if timeout is None: timeout = _get_command_timeout() args = args or [] # Build the command try: browser_cmd = _find_agent_browser() except FileNotFoundError as e: logger.warning("agent-browser CLI not found: %s", e) return {"success": False, "error": str(e)}
if _requires_real_termux_browser_install(browser_cmd): error = _termux_browser_install_error() logger.warning("browser command blocked on Termux: %s", error) return {"success": False, "error": error}
# Local mode with no Chromium on disk: fail fast with an actionable # message instead of hanging for _command_timeout seconds per call. if _is_local_mode() and not _chromium_installed(): if _running_in_docker(): hint = ( "Chromium browser is missing. You're running in Docker — pull " "the latest image to get the bundled Chromium: " "docker pull ghcr.io/nousresearch/hermes-agent:latest" ) else: hint = ( "Chromium browser is missing. Install it with: " "npx agent-browser install --with-deps " "(or: npx playwright install --with-deps chromium)" ) logger.warning("browser command blocked: %s", hint) return {"success": False, "error": hint} from tools.interrupt import is_interrupted if is_interrupted(): return {"success": False, "error": "Interrupted"}
# Get session info (creates Browserbase session with proxies if needed) try: session_info = _get_session_info(task_id) except Exception as e: logger.warning("Failed to create browser session for task=%s: %s", task_id, e) return {"success": False, "error": f"Failed to create browser session: {str(e)}"} # Build the command with the appropriate backend flag. # Cloud mode: --cdp <websocket_url> connects to Browserbase. # Local mode: --session <name> launches a local headless Chromium. # The rest of the command (--json, command, args) is identical. if session_info.get("cdp_url"): # Cloud mode — connect to remote Browserbase browser via CDP # IMPORTANT: Do NOT use --session with --cdp. In agent-browser >=0.13, # --session creates a local browser instance and silently ignores --cdp. backend_args = ["--cdp", session_info["cdp_url"]] else: # Local mode — launch a headless Chromium instance backend_args = ["--session", session_info["session_name"]]
# Keep concrete executable paths intact, even when they contain spaces. # Only the synthetic npx fallback needs to expand into multiple argv items. cmd_prefix = ["npx", "agent-browser"] if browser_cmd == "npx agent-browser" else [browser_cmd]
cmd_parts = cmd_prefix + backend_args + [ "--json", command ] + args try: # Give each task its own socket directory to prevent concurrency conflicts. # Without this, parallel workers fight over the same default socket path, # causing "Failed to create socket directory: Permission denied" errors. task_socket_dir = os.path.join( _socket_safe_tmpdir(), f"agent-browser-{session_info['session_name']}" ) os.makedirs(task_socket_dir, mode=0o700, exist_ok=True) # Record this hermes PID as the session owner (cross-process safe # orphan detection — see _write_owner_pid). _write_owner_pid(task_socket_dir, session_info['session_name']) logger.debug("browser cmd=%s task=%s socket_dir=%s (%d chars)", command, task_id, task_socket_dir, len(task_socket_dir)) browser_env = {**os.environ}
# Ensure subprocesses inherit the same browser-specific PATH fallbacks # used during CLI discovery. browser_env["PATH"] = _merge_browser_path(browser_env.get("PATH", "")) browser_env["AGENT_BROWSER_SOCKET_DIR"] = task_socket_dir
# Tell the agent-browser daemon to self-terminate after being idle # for our configured inactivity timeout. This is the daemon-side # counterpart to our Python-side _cleanup_inactive_browser_sessions # — the daemon kills itself and its Chrome children when no CLI # commands arrive within the window. Added in agent-browser 0.24. if "AGENT_BROWSER_IDLE_TIMEOUT_MS" not in browser_env: idle_ms = str(BROWSER_SESSION_INACTIVITY_TIMEOUT * 1000) browser_env["AGENT_BROWSER_IDLE_TIMEOUT_MS"] = idle_ms # Use temp files for stdout/stderr instead of pipes. # agent-browser starts a background daemon that inherits file # descriptors. With capture_output=True (pipes), the daemon keeps # the pipe fds open after the CLI exits, so communicate() never # sees EOF and blocks until the timeout fires. stdout_path = os.path.join(task_socket_dir, f"_stdout_{command}") stderr_path = os.path.join(task_socket_dir, f"_stderr_{command}") stdout_fd = os.open(stdout_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) stderr_fd = os.open(stderr_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) try: proc = subprocess.Popen( cmd_parts, stdout=stdout_fd, stderr=stderr_fd, stdin=subprocess.DEVNULL, env=browser_env, ) finally: os.close(stdout_fd) os.close(stderr_fd)
try: proc.wait(timeout=timeout) except subprocess.TimeoutExpired: proc.kill() proc.wait() logger.warning("browser '%s' timed out after %ds (task=%s, socket_dir=%s)", command, timeout, task_id, task_socket_dir) return {"success": False, "error": f"Command timed out after {timeout} seconds"}
with open(stdout_path, "r") as f: stdout = f.read() with open(stderr_path, "r") as f: stderr = f.read() returncode = proc.returncode
# Clean up temp files (best-effort) for p in (stdout_path, stderr_path): try: os.unlink(p) except OSError: pass
# Log stderr for diagnostics — use warning level on failure so it's visible if stderr and stderr.strip(): level = logging.WARNING if returncode != 0 else logging.DEBUG logger.log(level, "browser '%s' stderr: %s", command, stderr.strip()[:500]) stdout_text = stdout.strip()
# Empty output with rc=0 is a broken state — treat as failure rather # than silently returning {"success": True, "data": {}}. # Some commands (close, record) legitimately return no output. if not stdout_text and returncode == 0 and command not in _EMPTY_OK_COMMANDS: logger.warning("browser '%s' returned empty output (rc=0)", command) return {"success": False, "error": f"Browser command '{command}' returned no output"}
if stdout_text: try: parsed = json.loads(stdout_text) # Warn if snapshot came back empty (common sign of daemon/CDP issues) if command == "snapshot" and parsed.get("success"): snap_data = parsed.get("data", {}) if not snap_data.get("snapshot") and not snap_data.get("refs"): logger.warning("snapshot returned empty content. " "Possible stale daemon or CDP connection issue. " "returncode=%s", returncode) return parsed except json.JSONDecodeError: raw = stdout_text[:2000] logger.warning("browser '%s' returned non-JSON output (rc=%s): %s", command, returncode, raw[:500])
if command == "screenshot": stderr_text = (stderr or "").strip() combined_text = "\n".join( part for part in [stdout_text, stderr_text] if part ) recovered_path = _extract_screenshot_path_from_text(combined_text)
if recovered_path and Path(recovered_path).exists(): logger.info( "browser 'screenshot' recovered file from non-JSON output: %s", recovered_path, ) return { "success": True, "data": { "path": recovered_path, "raw": raw, }, }
return { "success": False, "error": f"Non-JSON output from agent-browser for '{command}': {raw}" } # Check for errors if returncode != 0: error_msg = stderr.strip() if stderr else f"Command failed with code {returncode}" logger.warning("browser '%s' failed (rc=%s): %s", command, returncode, error_msg[:300]) return {"success": False, "error": error_msg} return {"success": True, "data": {}} except Exception as e: logger.warning("browser '%s' exception: %s", command, e, exc_info=True) return {"success": False, "error": str(e)}
def _extract_relevant_content( snapshot_text: str, user_task: Optional[str] = None) -> str: """Use LLM to extract relevant content from a snapshot based on the user's task.
Falls back to simple truncation when no auxiliary text model is configured. """ if user_task: extraction_prompt = ( f"You are a content extractor for a browser automation agent.\n\n" f"The user's task is: {user_task}\n\n" f"Given the following page snapshot (accessibility tree representation), " f"extract and summarize the most relevant information for completing this task. Focus on:\n" f"1. Interactive elements (buttons, links, inputs) that might be needed\n" f"2. Text content relevant to the task (prices, descriptions, headings, important info)\n" f"3. Navigation structure if relevant\n\n" f"Keep ref IDs (like [ref=e5]) for interactive elements so the agent can use them.\n\n" f"Page Snapshot:\n{snapshot_text}\n\n" f"Provide a concise summary that preserves actionable information and relevant content." ) else: extraction_prompt = ( f"Summarize this page snapshot, preserving:\n" f"1. All interactive elements with their ref IDs (like [ref=e5])\n" f"2. Key text content and headings\n" f"3. Important information visible on the page\n\n" f"Page Snapshot:\n{snapshot_text}\n\n" f"Provide a concise summary focused on interactive elements and key content." )
# Redact secrets from snapshot before sending to auxiliary LLM. # Without this, a page displaying env vars or API keys would leak # secrets to the extraction model before run_agent.py's general # redaction layer ever sees the tool result. from agent.redact import redact_sensitive_text extraction_prompt = redact_sensitive_text(extraction_prompt)
try: call_kwargs = { "task": "web_extract", "messages": [{"role": "user", "content": extraction_prompt}], "max_tokens": 4000, "temperature": 0.1, } model = _get_extraction_model() if model: call_kwargs["model"] = model response = call_llm(**call_kwargs) extracted = (response.choices[0].message.content or "").strip() or _truncate_snapshot(snapshot_text) # Redact any secrets the auxiliary LLM may have echoed back. return redact_sensitive_text(extracted) except Exception: return _truncate_snapshot(snapshot_text)
def _truncate_snapshot(snapshot_text: str, max_chars: int = 8000) -> str: """Structure-aware truncation for snapshots.
Cuts at line boundaries so that accessibility tree elements are never split mid-line, and appends a note telling the agent how much was omitted.
Args: snapshot_text: The snapshot text to truncate max_chars: Maximum characters to keep
Returns: Truncated text with indicator if truncated """ if len(snapshot_text) <= max_chars: return snapshot_text
lines = snapshot_text.split('\n') result: list[str] = [] chars = 0 for line in lines: if chars + len(line) + 1 > max_chars - 80: # reserve space for note break result.append(line) chars += len(line) + 1 remaining = len(lines) - len(result) if remaining > 0: result.append(f'\n[... {remaining} more lines truncated, use browser_snapshot for full content]') return '\n'.join(result)
# ============================================================================# Browser Tool Functions# ============================================================================
def browser_navigate(url: str, task_id: Optional[str] = None) -> str: """ Navigate to a URL in the browser. Args: url: The URL to navigate to task_id: Task identifier for session isolation Returns: JSON string with navigation result (includes stealth features info on first nav) """ # Secret exfiltration protection — block URLs that embed API keys or # tokens in query parameters. A prompt injection could trick the agent # into navigating to https://evil.com/steal?key=sk-ant-... to exfil secrets. # Also check URL-decoded form to catch %2D encoding tricks (e.g. sk%2Dant%2D...). import urllib.parse from agent.redact import _PREFIX_RE url_decoded = urllib.parse.unquote(url) if _PREFIX_RE.search(url) or _PREFIX_RE.search(url_decoded): return json.dumps({ "success": False, "error": "Blocked: URL contains what appears to be an API key or token. " "Secrets must not be sent in URLs.", })
# SSRF protection — block private/internal addresses before navigating. # Skipped for local backends (Camofox, headless Chromium without a cloud # provider) because the agent already has full local network access via # the terminal tool. Also skipped when hybrid routing will auto-spawn a # local Chromium sidecar for this URL (cloud provider configured + # private URL + ``browser.auto_local_for_private_urls`` enabled) — the # cloud provider never sees the URL in that case. Can also be opted # out globally via ``browser.allow_private_urls`` in config. effective_task_id = task_id or "default" nav_session_key = _navigation_session_key(effective_task_id, url) auto_local_this_nav = _is_local_sidecar_key(nav_session_key)
if ( not _is_local_backend() and not auto_local_this_nav and not _allow_private_urls() and not _is_safe_url(url) ): return json.dumps({ "success": False, "error": "Blocked: URL targets a private or internal address", })
# Website policy check — block before navigating blocked = check_website_access(url) if blocked: return json.dumps({ "success": False, "error": blocked["message"], "blocked_by_policy": {"host": blocked["host"], "rule": blocked["rule"], "source": blocked["source"]}, })
# Camofox backend — delegate after safety checks pass if _is_camofox_mode(): from tools.browser_camofox import camofox_navigate return camofox_navigate(url, task_id)
if auto_local_this_nav: logger.info( "browser_navigate: auto-routing %s to local Chromium sidecar " "(cloud provider %s stays on cloud for public URLs; " "set browser.auto_local_for_private_urls: false to disable)", url, type(_get_cloud_provider()).__name__ if _get_cloud_provider() else "none", )
# Get session info to check if this is a new session # (will create one with features logged if not exists) session_info = _get_session_info(nav_session_key) is_first_nav = session_info.get("_first_nav", True)
# Auto-start recording if configured and this is first navigation if is_first_nav: session_info["_first_nav"] = False _maybe_start_recording(nav_session_key)
result = _run_browser_command(nav_session_key, "open", [url], timeout=max(_get_command_timeout(), 60))
# Remember which session served this nav so snapshot/click/fill/... # on the same task_id hit it (critical when hybrid routing has both a # cloud session and a local sidecar alive concurrently). _last_active_session_key[effective_task_id] = nav_session_key if result.get("success"): data = result.get("data", {}) title = data.get("title", "") final_url = data.get("url", url)
# Post-redirect SSRF check — if the browser followed a redirect to a # private/internal address, block the result so the model can't read # internal content via subsequent browser_snapshot calls. # Skipped for local backends (same rationale as the pre-nav check), # and for the hybrid local sidecar (we're already on a local browser # hitting a private URL by design). if ( not _is_local_backend() and not auto_local_this_nav and not _allow_private_urls() and final_url and final_url != url and not _is_safe_url(final_url) ): # Navigate away to a blank page to prevent snapshot leaks _run_browser_command(nav_session_key, "open", ["about:blank"], timeout=10) return json.dumps({ "success": False, "error": "Blocked: redirect landed on a private/internal address", })
response = { "success": True, "url": final_url, "title": title } # Detect common "blocked" page patterns from title/url blocked_patterns = [ "access denied", "access to this page has been denied", "blocked", "bot detected", "verification required", "please verify", "are you a robot", "captcha", "cloudflare", "ddos protection", "checking your browser", "just a moment", "attention required" ] title_lower = title.lower() if any(pattern in title_lower for pattern in blocked_patterns): response["bot_detection_warning"] = ( f"Page title '{title}' suggests bot detection. The site may have blocked this request. " "Options: 1) Try adding delays between actions, 2) Access different pages first, " "3) Enable advanced stealth (BROWSERBASE_ADVANCED_STEALTH=true, requires Scale plan), " "4) Some sites have very aggressive bot detection that may be unavoidable." ) # Include feature info on first navigation so model knows what's active if is_first_nav and "features" in session_info: features = session_info["features"] active_features = [k for k, v in features.items() if v] if not features.get("proxies"): response["stealth_warning"] = ( "Running WITHOUT residential proxies. Bot detection may be more aggressive. " "Consider upgrading Browserbase plan for proxy support." ) response["stealth_features"] = active_features
# Auto-take a compact snapshot so the model can act immediately # without a separate browser_snapshot call. try: snap_result = _run_browser_command(nav_session_key, "snapshot", ["-c"]) if snap_result.get("success"): snap_data = snap_result.get("data", {}) snapshot_text = snap_data.get("snapshot", "") refs = snap_data.get("refs", {}) if len(snapshot_text) > SNAPSHOT_SUMMARIZE_THRESHOLD: snapshot_text = _truncate_snapshot(snapshot_text) response["snapshot"] = snapshot_text response["element_count"] = len(refs) if refs else 0 except Exception as e: logger.debug("Auto-snapshot after navigate failed: %s", e)
return json.dumps(response, ensure_ascii=False) else: return json.dumps({ "success": False, "error": result.get("error", "Navigation failed") }, ensure_ascii=False)
def browser_snapshot( full: bool = False, task_id: Optional[str] = None, user_task: Optional[str] = None) -> str: """ Get a text-based snapshot of the current page's accessibility tree. Args: full: If True, return complete snapshot. If False, return compact view. task_id: Task identifier for session isolation user_task: The user's current task (for task-aware extraction) Returns: JSON string with page snapshot """ if _is_camofox_mode(): from tools.browser_camofox import camofox_snapshot return camofox_snapshot(full, task_id, user_task)
effective_task_id = _last_session_key(task_id or "default") # Build command args based on full flag args = [] if not full: args.extend(["-c"]) # Compact mode result = _run_browser_command(effective_task_id, "snapshot", args) if result.get("success"): data = result.get("data", {}) snapshot_text = data.get("snapshot", "") refs = data.get("refs", {}) # Check if snapshot needs summarization if len(snapshot_text) > SNAPSHOT_SUMMARIZE_THRESHOLD and user_task: snapshot_text = _extract_relevant_content(snapshot_text, user_task) elif len(snapshot_text) > SNAPSHOT_SUMMARIZE_THRESHOLD: snapshot_text = _truncate_snapshot(snapshot_text) response = { "success": True, "snapshot": snapshot_text, "element_count": len(refs) if refs else 0 }
# Merge supervisor state (pending dialogs + frame tree) when a CDP # supervisor is attached to this task. No-op otherwise. See # website/docs/developer-guide/browser-supervisor.md. try: from tools.browser_supervisor import SUPERVISOR_REGISTRY # type: ignore[import-not-found] _supervisor = SUPERVISOR_REGISTRY.get(effective_task_id) if _supervisor is not None: _sv_snap = _supervisor.snapshot() if _sv_snap.active: response.update(_sv_snap.to_dict()) except Exception as _sv_exc: logger.debug("supervisor snapshot merge failed: %s", _sv_exc)
return json.dumps(response, ensure_ascii=False) else: return json.dumps({ "success": False, "error": result.get("error", "Failed to get snapshot") }, ensure_ascii=False)
def browser_click(ref: str, task_id: Optional[str] = None) -> str: """ Click on an element. Args: ref: Element reference (e.g., "@e5") task_id: Task identifier for session isolation Returns: JSON string with click result """ if _is_camofox_mode(): from tools.browser_camofox import camofox_click return camofox_click(ref, task_id)
effective_task_id = _last_session_key(task_id or "default") # Ensure ref starts with @ if not ref.startswith("@"): ref = f"@{ref}" result = _run_browser_command(effective_task_id, "click", [ref]) if result.get("success"): return json.dumps({ "success": True, "clicked": ref }, ensure_ascii=False) else: return json.dumps({ "success": False, "error": result.get("error", f"Failed to click {ref}") }, ensure_ascii=False)
def browser_type(ref: str, text: str, task_id: Optional[str] = None) -> str: """ Type text into an input field. Args: ref: Element reference (e.g., "@e3") text: Text to type task_id: Task identifier for session isolation Returns: JSON string with type result """ if _is_camofox_mode(): from tools.browser_camofox import camofox_type return camofox_type(ref, text, task_id)
effective_task_id = _last_session_key(task_id or "default") # Ensure ref starts with @ if not ref.startswith("@"): ref = f"@{ref}" # Use fill command (clears then types) result = _run_browser_command(effective_task_id, "fill", [ref, text]) if result.get("success"): return json.dumps({ "success": True, "typed": text, "element": ref }, ensure_ascii=False) else: return json.dumps({ "success": False, "error": result.get("error", f"Failed to type into {ref}") }, ensure_ascii=False)
def browser_scroll(direction: str, task_id: Optional[str] = None) -> str: """ Scroll the page. Args: direction: "up" or "down" task_id: Task identifier for session isolation Returns: JSON string with scroll result """ # Validate direction if direction not in ["up", "down"]: return json.dumps({ "success": False, "error": f"Invalid direction '{direction}'. Use 'up' or 'down'." }, ensure_ascii=False)
# Single scroll with pixel amount instead of 5x subprocess calls. # agent-browser supports: agent-browser scroll down 500 # ~500px is roughly half a viewport of travel. _SCROLL_PIXELS = 500
if _is_camofox_mode(): from tools.browser_camofox import camofox_scroll # Camofox REST API doesn't support pixel args; use repeated calls _SCROLL_REPEATS = 5 result = None for _ in range(_SCROLL_REPEATS): result = camofox_scroll(direction, task_id) return result
effective_task_id = _last_session_key(task_id or "default")
result = _run_browser_command(effective_task_id, "scroll", [direction, str(_SCROLL_PIXELS)]) if not result.get("success"): return json.dumps({ "success": False, "error": result.get("error", f"Failed to scroll {direction}") }, ensure_ascii=False)
return json.dumps({ "success": True, "scrolled": direction }, ensure_ascii=False)
def browser_back(task_id: Optional[str] = None) -> str: """ Navigate back in browser history. Args: task_id: Task identifier for session isolation Returns: JSON string with navigation result """ if _is_camofox_mode(): from tools.browser_camofox import camofox_back return camofox_back(task_id)
effective_task_id = _last_session_key(task_id or "default") result = _run_browser_command(effective_task_id, "back", []) if result.get("success"): data = result.get("data", {}) return json.dumps({ "success": True, "url": data.get("url", "") }, ensure_ascii=False) else: return json.dumps({ "success": False, "error": result.get("error", "Failed to go back") }, ensure_ascii=False)
def browser_press(key: str, task_id: Optional[str] = None) -> str: """ Press a keyboard key. Args: key: Key to press (e.g., "Enter", "Tab") task_id: Task identifier for session isolation Returns: JSON string with key press result """ if _is_camofox_mode(): from tools.browser_camofox import camofox_press return camofox_press(key, task_id)
effective_task_id = _last_session_key(task_id or "default") result = _run_browser_command(effective_task_id, "press", [key]) if result.get("success"): return json.dumps({ "success": True, "pressed": key }, ensure_ascii=False) else: return json.dumps({ "success": False, "error": result.get("error", f"Failed to press {key}") }, ensure_ascii=False)
def browser_console(clear: bool = False, expression: Optional[str] = None, task_id: Optional[str] = None) -> str: """Get browser console messages and JavaScript errors, or evaluate JS in the page. When ``expression`` is provided, evaluates JavaScript in the page context (like the DevTools console) and returns the result. Otherwise returns console output (log/warn/error/info) and uncaught exceptions. Args: clear: If True, clear the message/error buffers after reading expression: JavaScript expression to evaluate in the page context task_id: Task identifier for session isolation Returns: JSON string with console messages/errors, or eval result """ # --- JS evaluation mode --- if expression is not None: return _browser_eval(expression, task_id)
# --- Console output mode (original behaviour) --- if _is_camofox_mode(): from tools.browser_camofox import camofox_console return camofox_console(clear, task_id)
effective_task_id = _last_session_key(task_id or "default") console_args = ["--clear"] if clear else [] error_args = ["--clear"] if clear else [] console_result = _run_browser_command(effective_task_id, "console", console_args) errors_result = _run_browser_command(effective_task_id, "errors", error_args) messages = [] if console_result.get("success"): for msg in console_result.get("data", {}).get("messages", []): messages.append({ "type": msg.get("type", "log"), "text": msg.get("text", ""), "source": "console", }) errors = [] if errors_result.get("success"): for err in errors_result.get("data", {}).get("errors", []): errors.append({ "message": err.get("message", ""), "source": "exception", }) return json.dumps({ "success": True, "console_messages": messages, "js_errors": errors, "total_messages": len(messages), "total_errors": len(errors), }, ensure_ascii=False)
def _browser_eval(expression: str, task_id: Optional[str] = None) -> str: """Evaluate a JavaScript expression in the page context and return the result.""" if _is_camofox_mode(): return _camofox_eval(expression, task_id)
effective_task_id = _last_session_key(task_id or "default") result = _run_browser_command(effective_task_id, "eval", [expression])
if not result.get("success"): err = result.get("error", "eval failed") # Detect backend capability gaps and give the model a clear signal if any(hint in err.lower() for hint in ("unknown command", "not supported", "not found", "no such command")): return json.dumps({ "success": False, "error": f"JavaScript evaluation is not supported by this browser backend. {err}", }) return json.dumps({ "success": False, "error": err, })
data = result.get("data", {}) raw_result = data.get("result")
# The eval command returns the JS result as a string. If the string # is valid JSON, parse it so the model gets structured data. parsed = raw_result if isinstance(raw_result, str): try: parsed = json.loads(raw_result) except (json.JSONDecodeError, ValueError): pass # keep as string
return json.dumps({ "success": True, "result": parsed, "result_type": type(parsed).__name__, }, ensure_ascii=False, default=str)
def _camofox_eval(expression: str, task_id: Optional[str] = None) -> str: """Evaluate JS via Camofox's /tabs/{tab_id}/eval endpoint (if available).""" from tools.browser_camofox import _ensure_tab, _post try: tab_info = _ensure_tab(task_id or "default") tab_id = tab_info.get("tab_id") or tab_info.get("id") resp = _post(f"/tabs/{tab_id}/evaluate", body={"expression": expression, "userId": tab_info["user_id"]})
# Camofox returns the result in a JSON envelope raw_result = resp.get("result") if isinstance(resp, dict) else resp parsed = raw_result if isinstance(raw_result, str): try: parsed = json.loads(raw_result) except (json.JSONDecodeError, ValueError): pass
return json.dumps({ "success": True, "result": parsed, "result_type": type(parsed).__name__, }, ensure_ascii=False, default=str) except Exception as e: error_msg = str(e) # Graceful degradation — server may not support eval if any(code in error_msg for code in ("404", "405", "501")): return json.dumps({ "success": False, "error": "JavaScript evaluation is not supported by this Camofox server. " "Use browser_snapshot or browser_vision to inspect page state.", }) return tool_error(error_msg, success=False)
def _maybe_start_recording(task_id: str): """Start recording if browser.record_sessions is enabled in config.""" with _cleanup_lock: if task_id in _recording_sessions: return try: from hermes_cli.config import read_raw_config hermes_home = get_hermes_home() cfg = read_raw_config() record_enabled = cfg.get("browser", {}).get("record_sessions", False) if not record_enabled: return recordings_dir = hermes_home / "browser_recordings" recordings_dir.mkdir(parents=True, exist_ok=True) _cleanup_old_recordings(max_age_hours=72) timestamp = time.strftime("%Y%m%d_%H%M%S") recording_path = recordings_dir / f"session_{timestamp}_{task_id[:16]}.webm" result = _run_browser_command(task_id, "record", ["start", str(recording_path)]) if result.get("success"): with _cleanup_lock: _recording_sessions.add(task_id) logger.info("Auto-recording browser session %s to %s", task_id, recording_path) else: logger.debug("Could not start auto-recording: %s", result.get("error")) except Exception as e: logger.debug("Auto-recording setup failed: %s", e)
def _maybe_stop_recording(task_id: str): """Stop recording if one is active for this session.""" with _cleanup_lock: if task_id not in _recording_sessions: return try: result = _run_browser_command(task_id, "record", ["stop"]) if result.get("success"): path = result.get("data", {}).get("path", "") logger.info("Saved browser recording for session %s: %s", task_id, path) except Exception as e: logger.debug("Could not stop recording for %s: %s", task_id, e) finally: with _cleanup_lock: _recording_sessions.discard(task_id)
def browser_get_images(task_id: Optional[str] = None) -> str: """ Get all images on the current page. Args: task_id: Task identifier for session isolation Returns: JSON string with list of images (src and alt) """ if _is_camofox_mode(): from tools.browser_camofox import camofox_get_images return camofox_get_images(task_id)
effective_task_id = _last_session_key(task_id or "default") # Use eval to run JavaScript that extracts images js_code = """JSON.stringify( [...document.images].map(img => ({ src: img.src, alt: img.alt || '', width: img.naturalWidth, height: img.naturalHeight })).filter(img => img.src && !img.src.startsWith('data:')) )""" result = _run_browser_command(effective_task_id, "eval", [js_code]) if result.get("success"): data = result.get("data", {}) raw_result = data.get("result", "[]") try: # Parse the JSON string returned by JavaScript if isinstance(raw_result, str): images = json.loads(raw_result) else: images = raw_result return json.dumps({ "success": True, "images": images, "count": len(images) }, ensure_ascii=False) except json.JSONDecodeError: return json.dumps({ "success": True, "images": [], "count": 0, "warning": "Could not parse image data" }, ensure_ascii=False) else: return json.dumps({ "success": False, "error": result.get("error", "Failed to get images") }, ensure_ascii=False)
def browser_vision(question: str, annotate: bool = False, task_id: Optional[str] = None) -> str: """ Take a screenshot of the current page and analyze it with vision AI. This tool captures what's visually displayed in the browser and sends it to Gemini for analysis. Useful for understanding visual content that the text-based snapshot may not capture (CAPTCHAs, verification challenges, images, complex layouts, etc.). The screenshot is saved persistently and its file path is returned alongside the analysis, so it can be shared with users via MEDIA:<path> in the response. Args: question: What you want to know about the page visually annotate: If True, overlay numbered [N] labels on interactive elements task_id: Task identifier for session isolation Returns: JSON string with vision analysis results and screenshot_path """ if _is_camofox_mode(): from tools.browser_camofox import camofox_vision return camofox_vision(question, annotate, task_id)
import base64 import uuid as uuid_mod effective_task_id = _last_session_key(task_id or "default") # Save screenshot to persistent location so it can be shared with users from hermes_constants import get_hermes_dir screenshots_dir = get_hermes_dir("cache/screenshots", "browser_screenshots") screenshot_path = screenshots_dir / f"browser_screenshot_{uuid_mod.uuid4().hex}.png" try: screenshots_dir.mkdir(parents=True, exist_ok=True) # Prune old screenshots (older than 24 hours) to prevent unbounded disk growth _cleanup_old_screenshots(screenshots_dir, max_age_hours=24) # Take screenshot using agent-browser screenshot_args = [] if annotate: screenshot_args.append("--annotate") screenshot_args.append("--full") screenshot_args.append(str(screenshot_path)) result = _run_browser_command( effective_task_id, "screenshot", screenshot_args, ) if not result.get("success"): error_detail = result.get("error", "Unknown error") _cp = _get_cloud_provider() mode = "local" if _cp is None else f"cloud ({_cp.provider_name()})" return json.dumps({ "success": False, "error": f"Failed to take screenshot ({mode} mode): {error_detail}" }, ensure_ascii=False)
actual_screenshot_path = result.get("data", {}).get("path") if actual_screenshot_path: screenshot_path = Path(actual_screenshot_path)
# Check if screenshot file was created if not screenshot_path.exists(): _cp = _get_cloud_provider() mode = "local" if _cp is None else f"cloud ({_cp.provider_name()})" return json.dumps({ "success": False, "error": ( f"Screenshot file was not created at {screenshot_path} ({mode} mode). " f"This may indicate a socket path issue (macOS /var/folders/), " f"a missing Chromium install ('agent-browser install'), " f"or a stale daemon process." ), }, ensure_ascii=False) # Convert screenshot to base64 at full resolution. _screenshot_bytes = screenshot_path.read_bytes() _screenshot_b64 = base64.b64encode(_screenshot_bytes).decode("ascii") data_url = f"data:image/png;base64,{_screenshot_b64}" vision_prompt = ( f"You are analyzing a screenshot of a web browser.\n\n" f"User's question: {question}\n\n" f"Provide a detailed and helpful answer based on what you see in the screenshot. " f"If there are interactive elements, describe them. If there are verification challenges " f"or CAPTCHAs, describe what type they are and what action might be needed. " f"Focus on answering the user's specific question." )
# Use the centralized LLM router vision_model = _get_vision_model() logger.debug("browser_vision: analysing screenshot (%d bytes)", len(_screenshot_bytes))
# Read vision timeout/temperature from config (auxiliary.vision.*). # Local vision models (llama.cpp, ollama) can take well over 30s for # screenshot analysis, so the default timeout must be generous. vision_timeout = 120.0 vision_temperature = 0.1 try: from hermes_cli.config import load_config _cfg = load_config() _vision_cfg = _cfg.get("auxiliary", {}).get("vision", {}) _vt = _vision_cfg.get("timeout") if _vt is not None: vision_timeout = float(_vt) _vtemp = _vision_cfg.get("temperature") if _vtemp is not None: vision_temperature = float(_vtemp) except Exception: pass
call_kwargs = { "task": "vision", "messages": [ { "role": "user", "content": [ {"type": "text", "text": vision_prompt}, {"type": "image_url", "image_url": {"url": data_url}}, ], } ], "max_tokens": 2000, "temperature": vision_temperature, "timeout": vision_timeout, } if vision_model: call_kwargs["model"] = vision_model # Try full-size screenshot; on size-related rejection, downscale and retry. try: response = call_llm(**call_kwargs) except Exception as _api_err: from tools.vision_tools import ( _is_image_size_error, _resize_image_for_vision, _RESIZE_TARGET_BYTES, ) if (_is_image_size_error(_api_err) and len(data_url) > _RESIZE_TARGET_BYTES): logger.info( "Vision API rejected screenshot (%.1f MB); " "auto-resizing to ~%.0f MB and retrying...", len(data_url) / (1024 * 1024), _RESIZE_TARGET_BYTES / (1024 * 1024), ) data_url = _resize_image_for_vision( screenshot_path, mime_type="image/png") call_kwargs["messages"][0]["content"][1]["image_url"]["url"] = data_url response = call_llm(**call_kwargs) else: raise analysis = (response.choices[0].message.content or "").strip() # Redact secrets the vision LLM may have read from the screenshot. from agent.redact import redact_sensitive_text analysis = redact_sensitive_text(analysis) response_data = { "success": True, "analysis": analysis or "Vision analysis returned no content.", "screenshot_path": str(screenshot_path), } # Include annotation data if annotated screenshot was taken if annotate and result.get("data", {}).get("annotations"): response_data["annotations"] = result["data"]["annotations"] return json.dumps(response_data, ensure_ascii=False) except Exception as e: # Keep the screenshot if it was captured successfully — the failure is # in the LLM vision analysis, not the capture. Deleting a valid # screenshot loses evidence the user might need. The 24-hour cleanup # in _cleanup_old_screenshots prevents unbounded disk growth. logger.warning("browser_vision failed: %s", e, exc_info=True) error_info = {"success": False, "error": f"Error during vision analysis: {str(e)}"} if screenshot_path.exists(): error_info["screenshot_path"] = str(screenshot_path) error_info["note"] = "Screenshot was captured but vision analysis failed. You can still share it via MEDIA:<path>." return json.dumps(error_info, ensure_ascii=False)
def _cleanup_old_screenshots(screenshots_dir, max_age_hours=24): """Remove browser screenshots older than max_age_hours to prevent disk bloat.
Throttled to run at most once per hour per directory to avoid repeated scans on screenshot-heavy workflows. """ key = str(screenshots_dir) now = time.time() if now - _last_screenshot_cleanup_by_dir.get(key, 0.0) < 3600: return _last_screenshot_cleanup_by_dir[key] = now
try: cutoff = time.time() - (max_age_hours * 3600) for f in screenshots_dir.glob("browser_screenshot_*.png"): try: if f.stat().st_mtime < cutoff: f.unlink() except Exception as e: logger.debug("Failed to clean old screenshot %s: %s", f, e) except Exception as e: logger.debug("Screenshot cleanup error (non-critical): %s", e)
def _cleanup_old_recordings(max_age_hours=72): """Remove browser recordings older than max_age_hours to prevent disk bloat.""" try: hermes_home = get_hermes_home() recordings_dir = hermes_home / "browser_recordings" if not recordings_dir.exists(): return cutoff = time.time() - (max_age_hours * 3600) for f in recordings_dir.glob("session_*.webm"): try: if f.stat().st_mtime < cutoff: f.unlink() except Exception as e: logger.debug("Failed to clean old recording %s: %s", f, e) except Exception as e: logger.debug("Recording cleanup error (non-critical): %s", e)
# ============================================================================# Cleanup and Management Functions# ============================================================================
def cleanup_browser(task_id: Optional[str] = None) -> None: """ Clean up browser session(s) for a task.
Called automatically when a task completes or when inactivity timeout is reached. Closes both the agent-browser/Browserbase session and Camofox sessions.
When ``task_id`` is a bare task identifier (no ``::local`` suffix), reaps BOTH the cloud/primary session AND any hybrid-routing local sidecar that may have been spawned for LAN/localhost URLs in the same task. When ``task_id`` already carries a ``::local`` suffix (called from the inactivity cleanup loop against a specific session key), reaps only that one.
Args: task_id: Task identifier (or explicit session key) """ if task_id is None: task_id = "default"
# Expand to the full set of session keys to reap. For a bare task_id # that includes the cloud/primary key + the local sidecar if one exists. if _is_local_sidecar_key(task_id): session_keys = [task_id] bare_task_id = task_id[: -len(_LOCAL_SUFFIX)] else: session_keys = [task_id] sidecar_key = f"{task_id}{_LOCAL_SUFFIX}" with _cleanup_lock: if sidecar_key in _active_sessions: session_keys.append(sidecar_key) bare_task_id = task_id
for session_key in session_keys: _cleanup_single_browser_session(session_key)
# Drop the last-active pointer only when the bare task is being cleaned # (i.e. not when we're only reaping a sidecar mid-task). if not _is_local_sidecar_key(task_id): _last_active_session_key.pop(bare_task_id, None)
def _cleanup_single_browser_session(task_id: str) -> None: """Internal: reap a single browser session by its exact session key.""" # Stop the CDP supervisor for this task FIRST so we close our WebSocket # before the backend tears down the underlying CDP endpoint. _stop_cdp_supervisor(task_id)
# Also clean up Camofox session if running in Camofox mode. # Skip full close when managed persistence is enabled — the browser # profile (and its session cookies) must survive across agent tasks. # The inactivity reaper still frees idle resources. if _is_camofox_mode(): try: from tools.browser_camofox import camofox_close, camofox_soft_cleanup if not camofox_soft_cleanup(task_id): camofox_close(task_id) except Exception as e: logger.debug("Camofox cleanup for task %s: %s", task_id, e)
logger.debug("cleanup_browser called for task_id: %s", task_id) logger.debug("Active sessions: %s", list(_active_sessions.keys()))
# Check if session exists (under lock), but don't remove yet - # _run_browser_command needs it to build the close command. with _cleanup_lock: session_info = _active_sessions.get(task_id)
if session_info: bb_session_id = session_info.get("bb_session_id", "unknown") logger.debug("Found session for task %s: bb_session_id=%s", task_id, bb_session_id)
# Stop auto-recording before closing (saves the file) _maybe_stop_recording(task_id)
# Try to close via agent-browser first (needs session in _active_sessions) try: _run_browser_command(task_id, "close", [], timeout=10) logger.debug("agent-browser close command completed for task %s", task_id) except Exception as e: logger.warning("agent-browser close failed for task %s: %s", task_id, e)
# Now remove from tracking under lock with _cleanup_lock: _active_sessions.pop(task_id, None) _session_last_activity.pop(task_id, None)
# Cloud mode: close the cloud browser session via provider API. # Local sidecars have bb_session_id=None so this no-ops for them. if bb_session_id: provider = _get_cloud_provider() if provider is not None: try: provider.close_session(bb_session_id) except Exception as e: logger.warning("Could not close cloud browser session: %s", e) # Kill the daemon process and clean up socket directory session_name = session_info.get("session_name", "") if session_name: socket_dir = os.path.join(_socket_safe_tmpdir(), f"agent-browser-{session_name}") if os.path.exists(socket_dir): # agent-browser writes {session}.pid in the socket dir pid_file = os.path.join(socket_dir, f"{session_name}.pid") if os.path.isfile(pid_file): try: daemon_pid = int(Path(pid_file).read_text().strip()) os.kill(daemon_pid, signal.SIGTERM) logger.debug("Killed daemon pid %s for %s", daemon_pid, session_name) except (ProcessLookupError, ValueError, PermissionError, OSError): logger.debug("Could not kill daemon pid for %s (already dead or inaccessible)", session_name) shutil.rmtree(socket_dir, ignore_errors=True) logger.debug("Removed task %s from active sessions", task_id) else: logger.debug("No active session found for task_id: %s", task_id)
def cleanup_all_browsers() -> None: """ Clean up all active browser sessions. Useful for cleanup on shutdown. """ with _cleanup_lock: task_ids = list(_active_sessions.keys()) for task_id in task_ids: cleanup_browser(task_id)
# Tear down CDP supervisors for all tasks so background threads exit. try: from tools.browser_supervisor import SUPERVISOR_REGISTRY # type: ignore[import-not-found] SUPERVISOR_REGISTRY.stop_all() except Exception: pass
# Reset cached lookups so they are re-evaluated on next use. global _cached_agent_browser, _agent_browser_resolved global _cached_command_timeout, _command_timeout_resolved global _cached_chromium_installed _cached_agent_browser = None _agent_browser_resolved = False _discover_homebrew_node_dirs.cache_clear() _cached_command_timeout = None _command_timeout_resolved = False _cached_chromium_installed = None
# ============================================================================# Requirements Check# ============================================================================
# Cache for Chromium discovery. Invalidated by _reset_browser_caches._cached_chromium_installed: Optional[bool] = None
def _chromium_search_roots() -> List[str]: """Directories to scan for a Chromium / headless-shell build.
Order mirrors what agent-browser and Playwright actually probe:
1. ``PLAYWRIGHT_BROWSERS_PATH`` when set (Docker image sets this to ``/opt/hermes/.playwright``). 2. ``~/.cache/ms-playwright`` — Playwright's default on Linux/macOS. 3. ``~/Library/Caches/ms-playwright`` — Playwright's default on macOS. 4. ``%USERPROFILE%\\AppData\\Local\\ms-playwright`` — Playwright's default on Windows. """ roots: List[str] = [] env_path = os.environ.get("PLAYWRIGHT_BROWSERS_PATH", "").strip() if env_path and env_path != "0": roots.append(env_path) home = os.path.expanduser("~") roots.append(os.path.join(home, ".cache", "ms-playwright")) if sys.platform == "darwin": roots.append(os.path.join(home, "Library", "Caches", "ms-playwright")) if sys.platform == "win32": local = os.environ.get("LOCALAPPDATA") or os.path.join( home, "AppData", "Local" ) roots.append(os.path.join(local, "ms-playwright")) return roots
def _chromium_installed() -> bool: """Return True when a usable Chromium (or headless-shell) build is on disk.
agent-browser (0.26+) downloads Playwright's chromium / headless-shell builds into ``PLAYWRIGHT_BROWSERS_PATH`` and won't start without them. When the CLI is present but no browser build is, the first browser tool call hangs for the full command timeout (often ~30s each) before surfacing a useless error. Guarding the tool behind this check prevents advertising a capability that will fail at runtime. """ global _cached_chromium_installed if _cached_chromium_installed is not None: return _cached_chromium_installed
for root in _chromium_search_roots(): if not root or not os.path.isdir(root): continue try: entries = os.listdir(root) except OSError: continue # Playwright names them ``chromium-<build>`` and # ``chromium_headless_shell-<build>``; agent-browser accepts either. for entry in entries: if entry.startswith("chromium-") or entry.startswith( "chromium_headless_shell-" ): _cached_chromium_installed = True return True
_cached_chromium_installed = False return False
def _running_in_docker() -> bool: """Best-effort detection of whether we're inside a Docker container.""" if os.path.exists("/.dockerenv"): return True try: with open("/proc/1/cgroup", "rt") as fp: return "docker" in fp.read() except OSError: return False
def check_browser_requirements() -> bool: """ Check if browser tool requirements are met.
In **local mode** (no cloud provider configured): the ``agent-browser`` CLI must be findable *and* a Chromium build must be installed on disk.
In **cloud mode** (Browserbase, Browser Use, or Firecrawl): the CLI and the provider's required credentials must be present. The cloud provider hosts its own Chromium, so no local browser binary is needed.
Returns: True if all requirements are met, False otherwise """ # Camofox backend — only needs the server URL, no agent-browser CLI if _is_camofox_mode(): return True
# The agent-browser CLI is always required try: browser_cmd = _find_agent_browser() except FileNotFoundError: return False
# On Termux, the bare npx fallback is too fragile to treat as a satisfied # local browser dependency. Require a real install (global or local) so the # browser tool is not advertised as available when it will likely fail on # first use. if _requires_real_termux_browser_install(browser_cmd): return False
# In cloud mode, also require provider credentials. Cloud browsers # don't need a local Chromium binary. provider = _get_cloud_provider() if provider is not None: return provider.is_configured()
# Local mode: agent-browser needs a Chromium build on disk. Without it # the CLI hangs on first use until the command timeout fires. if not _chromium_installed(): return False
return True
# ============================================================================# Module Test# ============================================================================
if __name__ == "__main__": """ Simple test/demo when run directly """ print("🌐 Browser Tool Module") print("=" * 40)
_cp = _get_cloud_provider() mode = "local" if _cp is None else f"cloud ({_cp.provider_name()})" print(f" Mode: {mode}") # Check requirements if check_browser_requirements(): print("✅ All requirements met") else: print("❌ Missing requirements:") try: browser_cmd = _find_agent_browser() if _requires_real_termux_browser_install(browser_cmd): print(" - bare npx fallback found (insufficient on Termux local mode)") print(f" Install: {_browser_install_hint()}") elif _cp is None and not _chromium_installed(): print(" - Chromium browser binary not found") searched = ", ".join(_chromium_search_roots()) or "(no candidate paths)" print(f" Searched: {searched}") if _running_in_docker(): print( " Docker: pull the latest image — the current one " "predates the bundled Chromium install" ) print(" docker pull ghcr.io/nousresearch/hermes-agent:latest") else: print(" Install it with:") print(" npx agent-browser install --with-deps") print(" Or: npx playwright install --with-deps chromium") except FileNotFoundError: print(" - agent-browser CLI not found") print(f" Install: {_browser_install_hint()}") if _cp is not None and not _cp.is_configured(): print(f" - {_cp.provider_name()} credentials not configured") print(" Tip: set browser.cloud_provider to 'local' to use free local mode instead") print("\n📋 Available Browser Tools:") for schema in BROWSER_TOOL_SCHEMAS: print(f" 🔹 {schema['name']}: {schema['description'][:60]}...") print("\n💡 Usage:") print(" from tools.browser_tool import browser_navigate, browser_snapshot") print(" result = browser_navigate('https://example.com', task_id='my_task')") print(" snapshot = browser_snapshot(task_id='my_task')")
# ---------------------------------------------------------------------------# Registry# ---------------------------------------------------------------------------from tools.registry import registry, tool_error
_BROWSER_SCHEMA_MAP = {s["name"]: s for s in BROWSER_TOOL_SCHEMAS}
registry.register( name="browser_navigate", toolset="browser", schema=_BROWSER_SCHEMA_MAP["browser_navigate"], handler=lambda args, **kw: browser_navigate(url=args.get("url", ""), task_id=kw.get("task_id")), check_fn=check_browser_requirements, emoji="🌐",)registry.register( name="browser_snapshot", toolset="browser", schema=_BROWSER_SCHEMA_MAP["browser_snapshot"], handler=lambda args, **kw: browser_snapshot( full=args.get("full", False), task_id=kw.get("task_id"), user_task=kw.get("user_task")), check_fn=check_browser_requirements, emoji="📸",)registry.register( name="browser_click", toolset="browser", schema=_BROWSER_SCHEMA_MAP["browser_click"], handler=lambda args, **kw: browser_click(ref=args.get("ref", ""), task_id=kw.get("task_id")), check_fn=check_browser_requirements, emoji="👆",)registry.register( name="browser_type", toolset="browser", schema=_BROWSER_SCHEMA_MAP["browser_type"], handler=lambda args, **kw: browser_type(ref=args.get("ref", ""), text=args.get("text", ""), task_id=kw.get("task_id")), check_fn=check_browser_requirements, emoji="⌨️",)registry.register( name="browser_scroll", toolset="browser", schema=_BROWSER_SCHEMA_MAP["browser_scroll"], handler=lambda args, **kw: browser_scroll(direction=args.get("direction", "down"), task_id=kw.get("task_id")), check_fn=check_browser_requirements, emoji="📜",)registry.register( name="browser_back", toolset="browser", schema=_BROWSER_SCHEMA_MAP["browser_back"], handler=lambda args, **kw: browser_back(task_id=kw.get("task_id")), check_fn=check_browser_requirements, emoji="◀️",)registry.register( name="browser_press", toolset="browser", schema=_BROWSER_SCHEMA_MAP["browser_press"], handler=lambda args, **kw: browser_press(key=args.get("key", ""), task_id=kw.get("task_id")), check_fn=check_browser_requirements, emoji="⌨️",)
registry.register( name="browser_get_images", toolset="browser", schema=_BROWSER_SCHEMA_MAP["browser_get_images"], handler=lambda args, **kw: browser_get_images(task_id=kw.get("task_id")), check_fn=check_browser_requirements, emoji="🖼️",)registry.register( name="browser_vision", toolset="browser", schema=_BROWSER_SCHEMA_MAP["browser_vision"], handler=lambda args, **kw: browser_vision(question=args.get("question", ""), annotate=args.get("annotate", False), task_id=kw.get("task_id")), check_fn=check_browser_requirements, emoji="👁️",)registry.register( name="browser_console", toolset="browser", schema=_BROWSER_SCHEMA_MAP["browser_console"], handler=lambda args, **kw: browser_console(clear=args.get("clear", False), expression=args.get("expression"), task_id=kw.get("task_id")), check_fn=check_browser_requirements, emoji="🖥️",)