"""Headless Firefox browser automation via Playwright.
Provides navigation, interaction, content extraction, waiting,
screenshots, JavaScript execution, cookie management, network
interception, session persistence, download handling, and console/dialog
tools -- all exposed through the v3 multi-tool format.
"""
import asyncio
import fnmatch
import hashlib
import jsonutil as json
import logging
import os
import re
import shutil
import time
import uuid
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional
from urllib.parse import urlparse
from tools._safe_http import assert_safe_socks_proxy_url
from tools.alter_privileges import has_privilege, PRIVILEGES
try:
from playwright.async_api import (
async_playwright,
Browser,
BrowserContext,
Page,
Playwright,
TimeoutError as PlaywrightTimeout,
Error as PlaywrightError,
)
PLAYWRIGHT_AVAILABLE = True
except ImportError as e:
logging.error(f"Playwright not installed: {e}")
PLAYWRIGHT_AVAILABLE = False
async_playwright = None
Browser = None
BrowserContext = None
Page = None
Playwright = None
PlaywrightTimeout = Exception
PlaywrightError = Exception
try:
import html2text
HTML2TEXT_AVAILABLE = True
except ImportError:
HTML2TEXT_AVAILABLE = False
html2text = None
logger = logging.getLogger(__name__)
OUTPUT_DIR = Path(
os.environ.get("BROWSER_OUTPUT_DIR", "/home/star/large_files/browser_outputs")
)
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
SESSIONS_DIR = Path(
os.environ.get("BROWSER_SESSIONS_DIR", "/home/star/large_files/browser_sessions")
)
SESSIONS_DIR.mkdir(parents=True, exist_ok=True)
DOWNLOADS_DIR = OUTPUT_DIR / "downloads"
DOWNLOADS_DIR.mkdir(parents=True, exist_ok=True)
DEFAULT_TIMEOUT = 30000
MAX_CONTENT_LENGTH = 1000000
MAX_CONSOLE_LOGS = 1000
MAX_INTERCEPTED_RESPONSES = 100
MAX_RESPONSE_BODY_SIZE = 10 * 1024 * 1024
_JSON_PARSE_OFFLOAD_THRESHOLD = 64 * 1024
def _proxy_derived_context_id(validated_proxy_url: str) -> str:
"""Derive a stable browser-context id for a validated SOCKS proxy URL.
Hashes the proxy URL (SHA-256, truncated) into a deterministic
``proxy_<digest>`` id so that every request using the same proxy reuses one
browser context instead of spawning a fresh one each time. Determinism is the
point: the same proxy always maps to the same context key.
Called by ``HeadlessBrowserManager.ensure_proxied_context`` in this module;
there are no external callers.
Args:
validated_proxy_url: An already-validated SOCKS proxy URL.
Returns:
str: A ``proxy_<16-hex-digest>`` context identifier.
"""
digest = hashlib.sha256(validated_proxy_url.encode()).hexdigest()[:16]
return f"proxy_{digest}"
async def _json_loads_network_body(body_bytes: bytes) -> Any:
"""Parse JSON from a captured HTTP body without blocking the event loop.
Decodes intercepted network response bytes and parses them as JSON, offloading
the parse to a worker thread (``asyncio.to_thread``) once the payload reaches
``_JSON_PARSE_OFFLOAD_THRESHOLD`` so a large captured body cannot stall the
async browser pipeline; small bodies are parsed inline.
Called by the page response handler set up in
``HeadlessBrowserManager._setup_page_handlers`` and by
``browser_wait_for_response`` in this module; there are no external callers.
Args:
body_bytes: The raw UTF-8-encoded response body to parse.
Returns:
The decoded JSON value (dict, list, or scalar).
"""
text = body_bytes.decode("utf-8")
if len(body_bytes) >= _JSON_PARSE_OFFLOAD_THRESHOLD:
return await asyncio.to_thread(json.loads, text)
return json.loads(text)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _json_response(
status: str,
message: str = "",
data: Optional[Dict] = None,
error: Optional[str] = None,
) -> str:
"""Build the standard JSON envelope every browser tool returns.
Centralizes the response shape so all ~50 ``browser_*`` handlers emit a
consistent ``{"status": ..., "message"/"data"/"error": ...}`` string (pretty
printed, with non-serializable values coerced via ``default=str``). Optional
fields are only included when provided.
Called by essentially every handler and helper in this module to return both
success and error payloads; there are no external callers.
Args:
status (str): Outcome status, typically ``"success"``, ``"error"``, or
``"partial"``.
message (str): Optional human-readable message.
data (Optional[Dict]): Optional result payload.
error (Optional[str]): Optional error description.
Returns:
str: The serialized JSON envelope.
"""
response = {"status": status}
if message:
response["message"] = message
if data:
response["data"] = data
if error:
response["error"] = error
return json.dumps(response, indent=2, default=str)
def _validate_url(url: str) -> tuple:
"""Validate a navigation URL and block obvious SSRF targets.
Parses the URL and enforces the basic safety rules before the browser is
pointed at it: the scheme must be ``http`` or ``https``, a host must be
present, and loopback/all-interfaces hosts (``localhost``, ``127.0.0.1``,
``0.0.0.0``, ``::1``) are rejected. This is a lightweight guard layered ahead
of the deeper proxy/redirect checks elsewhere.
Called by ``browser_navigate`` in this module; there are no external callers.
Args:
url (str): The URL to validate.
Returns:
tuple: ``(is_valid, error_message)`` where ``error_message`` is empty on
success and describes the rejection otherwise.
"""
try:
parsed = urlparse(url)
if parsed.scheme not in ["http", "https"]:
return (
False,
f"Invalid URL scheme: {parsed.scheme}. Only http and https are allowed.",
)
if not parsed.netloc:
return False, "Invalid URL: missing domain"
blocked_hosts = ["localhost", "127.0.0.1", "0.0.0.0", "::1"]
if parsed.hostname and parsed.hostname.lower() in blocked_hosts:
return False, f"Blocked URL: {parsed.hostname} is not allowed"
return True, ""
except Exception as e:
return False, f"URL validation error: {str(e)}"
def _truncate_content(content: str, max_length: int = MAX_CONTENT_LENGTH) -> str:
"""Truncate extracted page content to a maximum length.
Caps the size of text/HTML/markdown pulled from a page so a huge document
cannot blow up the tool result returned to the model, appending a visible
truncation marker when content is cut.
Called by ``browser_get_content`` in this module; there are no external
callers.
Args:
content (str): The extracted content to bound.
max_length (int): Maximum length to keep; defaults to
``MAX_CONTENT_LENGTH``.
Returns:
str: The original content if within the limit, otherwise the truncated
prefix with an appended notice.
"""
if len(content) <= max_length:
return content
return content[:max_length] + "\n\n[Content truncated - exceeded maximum length]"
# ---------------------------------------------------------------------------
# Authorization
# ---------------------------------------------------------------------------
async def _check_unsandboxed_exec(ctx) -> str | None:
"""Gate side-effecting browser actions on the ``UNSANDBOXED_EXEC`` privilege.
Authorization guard for the browser tools that actually drive a real Firefox
instance (navigation, clicks, typing, screenshots, etc.). It reads the caller
and config off the context and consults
``tools.alter_privileges.has_privilege`` (Redis-backed) so only sufficiently
trusted users can reach the network through the headless browser.
Called by ``browser_navigate``, ``browser_reload``, ``browser_screenshot``,
``browser_close_context``, ``browser_click``, ``browser_type``, and
``browser_fill`` in this module; there are no external callers.
Args:
ctx: Tool execution context; supplies ``user_id``, ``redis``, and
``config`` for the privilege check.
Returns:
A JSON error envelope when the user lacks ``UNSANDBOXED_EXEC``, or
``None`` when the action is permitted.
"""
user_id = getattr(ctx, "user_id", "") or ""
redis = getattr(ctx, "redis", None)
config = getattr(ctx, "config", None)
if not await has_privilege(redis, user_id, PRIVILEGES["UNSANDBOXED_EXEC"], config):
return _json_response(
"error",
error="The user does not have the UNSANDBOXED_EXEC privilege. Ask an admin to grant it with the alter_privileges tool.",
)
return None
# ---------------------------------------------------------------------------
# Browser manager
# ---------------------------------------------------------------------------
[docs]
class HeadlessBrowserManager:
"""Process-wide owner of the shared headless Firefox instance.
Lazily launches a single Playwright Firefox browser and multiplexes it across
named contexts (each with its own page, console-log buffer, response-pattern
list, intercepted responses, downloads, and dialog handler), so every browser
tool and other consumers share one warm browser rather than spawning their
own. It tracks usage/error counts and self-heals: too many errors or a
disconnected browser trigger a restart. A module-level singleton ``_bm`` is
the instance everything uses, and it is imported directly by
``tools/web_scraper.py`` and ``bot/voice/puter.py`` to reuse the same browser.
Attributes:
default_context_id: The default context id.
is_initialized: The is initialized.
last_used: The last used.
usage_count: The usage count.
error_count: The error count.
max_errors_before_restart: The max errors before restart.
idle_timeout: The idle timeout.
_lock: The lock.
default_viewport: The default viewport.
default_user_agent: The default user agent.
"""
[docs]
def __init__(self):
"""Set up manager bookkeeping without launching a browser.
Initializes all per-instance state (context/page maps, console-log,
interception, download, dialog, and persistent-session registries, the
async lock, default viewport/user-agent, and error/usage counters) to
empty defaults. The actual Playwright/Firefox launch is deferred to
``initialize`` so the module can be imported cheaply.
Called once at import time to build the module-level ``_bm`` singleton.
"""
self.playwright: Optional[Any] = None
self.browser: Optional[Any] = None
self.contexts: Dict[str, Any] = {}
self.pages: Dict[str, Any] = {}
self.default_context_id = "default"
self.is_initialized = False
self.last_used = None
self.usage_count = 0
self.error_count = 0
self.max_errors_before_restart = 5
self.idle_timeout = 300
self._lock = asyncio.Lock()
self.default_viewport = {"width": 1920, "height": 1080}
self.default_user_agent = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:120.0) "
"Gecko/20100101 Firefox/120.0"
)
self.intercepted_responses: Dict[str, Dict] = {}
self.response_patterns: Dict[str, List[str]] = {}
self.console_logs: Dict[str, List[Dict]] = {}
self.downloads: Dict[str, Dict] = {}
self.pending_downloads: Dict[str, asyncio.Future] = {}
self.persistent_contexts: Dict[str, str] = {}
self.dialog_handlers: Dict[str, str] = {}
self.pending_dialogs: Dict[str, Dict] = {}
[docs]
async def initialize(self) -> bool:
"""Launch the Firefox browser and create the default context.
Starts Playwright and launches a headless Firefox instance (with autoplay
and desktop-notification preferences disabled), then creates the
``default`` context so a page is ready to use. Guarded by the async lock
and idempotent: if an initialized, live browser already exists it returns
immediately. On failure it cleans up and reports ``False``.
Called by ``ensure_ready`` and ``restart`` in this module, and by
``browser_create_persistent_context`` when no Playwright instance exists
yet.
Returns:
bool: ``True`` if the browser is ready, ``False`` if Playwright is
missing or the launch failed.
"""
if not PLAYWRIGHT_AVAILABLE:
logger.error("Playwright is not installed.")
return False
async with self._lock:
if self.is_initialized and self.browser:
return True
logger.info("Initializing headless Firefox browser...")
try:
self.playwright = await async_playwright().start()
self.browser = await self.playwright.firefox.launch(
headless=True,
firefox_user_prefs={
"dom.webnotifications.enabled": False,
"media.autoplay.default": 5,
"permissions.default.desktop-notification": 2,
},
)
await self._create_context(self.default_context_id)
self.is_initialized = True
self.last_used = time.time()
self.error_count = 0
logger.info("Headless Firefox browser initialized successfully")
return True
except Exception as e:
logger.error(f"Failed to initialize browser: {e}")
await self.cleanup()
return False
async def _create_context(self, context_id: str, **options):
"""Create a fresh (non-persistent) browser context and its first page.
Builds a new Playwright ``BrowserContext`` under ``context_id`` with the
requested viewport/user-agent/geolocation/device/proxy options (a string
proxy is validated through ``assert_safe_socks_proxy_url``), registers
its console-log, response-pattern, and dialog-handler state, opens an
initial page, and wires up event listeners via ``_setup_page_handlers``.
Called by ``initialize``, ``ensure_proxied_context``, ``get_page``,
``get_context``, and ``create_new_context`` in this module.
Args:
context_id (str): Key under which to register the new context.
**options: Optional ``viewport``, ``user_agent``,
``ignore_https_errors``, ``geolocation``, ``device``, and
``proxy`` settings.
Returns:
The created Playwright browser context.
Raises:
RuntimeError: If the browser has not been initialized.
"""
if not self.browser:
raise RuntimeError("Browser not initialized")
context_options = {
"viewport": options.get("viewport", self.default_viewport),
"user_agent": options.get("user_agent", self.default_user_agent),
"accept_downloads": True,
"ignore_https_errors": options.get("ignore_https_errors", False),
}
if "geolocation" in options:
context_options["geolocation"] = options["geolocation"]
context_options["permissions"] = ["geolocation"]
if "device" in options:
device = self.playwright.devices.get(options["device"])
if device:
context_options.update(device)
p = options.get("proxy")
if p:
if isinstance(p, str):
context_options["proxy"] = {"server": assert_safe_socks_proxy_url(p)}
elif isinstance(p, dict):
context_options["proxy"] = p
context = await self.browser.new_context(**context_options)
self.contexts[context_id] = context
self.console_logs[context_id] = []
self.response_patterns[context_id] = []
self.dialog_handlers[context_id] = "accept"
page = await context.new_page()
self.pages[context_id] = page
await self._setup_page_handlers(page, context_id)
return context
async def _create_persistent_context(
self, session_name: str, context_id: str, **options
):
"""Create a context backed by an on-disk persistent profile.
Like ``_create_context`` but launches a ``launch_persistent_context``
rooted at a sanitized ``SESSIONS_DIR/<session>`` user-data directory, so
cookies and local storage survive across runs. It records the
context-to-session mapping in ``persistent_contexts`` and wires the first
page's event listeners via ``_setup_page_handlers``. A string proxy is
validated through ``assert_safe_socks_proxy_url``.
Called by ``browser_create_persistent_context`` in this module.
Args:
session_name (str): Logical session name; non-word characters are
replaced to form the on-disk directory.
context_id (str): Key under which to register the new context.
**options: Optional ``viewport``, ``user_agent``,
``ignore_https_errors``, ``geolocation``, and ``proxy`` settings.
Returns:
The created persistent Playwright browser context.
Raises:
RuntimeError: If Playwright has not been initialized.
"""
if not self.playwright:
raise RuntimeError("Playwright not initialized")
safe_session_name = re.sub(r"[^\w\-]", "_", session_name)
user_data_dir = SESSIONS_DIR / safe_session_name
user_data_dir.mkdir(parents=True, exist_ok=True)
context_options = {
"viewport": options.get("viewport", self.default_viewport),
"user_agent": options.get("user_agent", self.default_user_agent),
"accept_downloads": True,
"ignore_https_errors": options.get("ignore_https_errors", False),
}
if "geolocation" in options:
context_options["geolocation"] = options["geolocation"]
context_options["permissions"] = ["geolocation"]
p = options.get("proxy")
if p:
if isinstance(p, str):
context_options["proxy"] = {"server": assert_safe_socks_proxy_url(p)}
elif isinstance(p, dict):
context_options["proxy"] = p
context = await self.playwright.firefox.launch_persistent_context(
str(user_data_dir), headless=True, **context_options
)
self.contexts[context_id] = context
self.persistent_contexts[context_id] = safe_session_name
self.console_logs[context_id] = []
self.response_patterns[context_id] = []
self.dialog_handlers[context_id] = "accept"
pages = context.pages
page = pages[0] if pages else await context.new_page()
self.pages[context_id] = page
await self._setup_page_handlers(page, context_id)
return context
async def _setup_page_handlers(self, page, context_id: str):
"""Attach response, console, download, and dialog listeners to a page.
Registers the per-page event handlers that populate this manager's
capture buffers for the given context: pattern-matched response
interception (``intercepted_responses``), console-log collection
(``console_logs``), automatic download saving to ``DOWNLOADS_DIR``
(``downloads`` plus any awaiting future in ``pending_downloads``), and
dialog auto-accept/dismiss per ``dialog_handlers``. Closures capture
``context_id`` so captures are attributed to the right context.
Called by ``_create_context`` and ``_create_persistent_context`` in this
module right after a page is opened.
Args:
page: The Playwright page to attach listeners to.
context_id (str): Context the page belongs to; used to route captures.
"""
async def on_response(response):
"""Capture a network response when it matches a watched pattern.
Stores a trimmed record (status, headers, decoded body, method,
timestamp) into ``intercepted_responses`` for responses whose URL
matches a glob in this context's ``response_patterns``, decoding the
body with ``_json_loads_network_body`` for JSON, evicting the oldest
entry past ``MAX_INTERCEPTED_RESPONSES``, and skipping oversized
bodies. Bound to the page ``"response"`` event by the enclosing
``_setup_page_handlers``.
Args:
response: The Playwright response that fired the event.
"""
try:
patterns = self.response_patterns.get(context_id, [])
if not patterns:
return
url = response.url
if not any(fnmatch.fnmatch(url, p) for p in patterns):
return
if len(self.intercepted_responses) >= MAX_INTERCEPTED_RESPONSES:
oldest_id = next(iter(self.intercepted_responses))
del self.intercepted_responses[oldest_id]
response_id = f"resp_{uuid.uuid4().hex[:12]}"
body = None
content_type = response.headers.get("content-type", "")
try:
body_bytes = await response.body()
if len(body_bytes) <= MAX_RESPONSE_BODY_SIZE:
if "application/json" in content_type:
body = await _json_loads_network_body(body_bytes)
elif "text/" in content_type:
body = body_bytes.decode("utf-8")
else:
body = f"[Binary data: {len(body_bytes)} bytes]"
else:
body = f"[Response too large: {len(body_bytes)} bytes]"
except Exception as e:
body = f"[Could not read body: {str(e)}]"
self.intercepted_responses[response_id] = {
"response_id": response_id,
"context_id": context_id,
"url": url,
"status": response.status,
"status_text": response.status_text,
"headers": dict(response.headers),
"body": body,
"method": response.request.method,
"timestamp": datetime.now().isoformat(),
}
except Exception as e:
logger.warning(f"Error in response handler: {e}")
page.on("response", on_response)
def on_console(msg):
"""Record a browser console message for this context.
Appends the message type, text, timestamp, and (when available)
source URL/line to this context's ``console_logs`` ring buffer,
dropping the oldest entry once ``MAX_CONSOLE_LOGS`` is reached. Bound
to the page ``"console"`` event by the enclosing
``_setup_page_handlers``.
Args:
msg: The Playwright console message that fired the event.
"""
try:
logs = self.console_logs.get(context_id, [])
if len(logs) >= MAX_CONSOLE_LOGS:
logs.pop(0)
log_entry = {
"type": msg.type,
"text": msg.text,
"timestamp": datetime.now().isoformat(),
}
try:
location = msg.location
if location:
log_entry["url"] = location.get("url", "")
log_entry["line"] = location.get("lineNumber", 0)
except Exception:
pass
logs.append(log_entry)
self.console_logs[context_id] = logs
except Exception as e:
logger.warning(f"Error in console handler: {e}")
page.on("console", on_console)
async def on_download(download):
"""Save a triggered download and record its metadata.
Assigns a unique download id, writes the file under ``DOWNLOADS_DIR``
with a sanitized name, records state/size/error into the ``downloads``
map, and resolves any future waiting in ``pending_downloads`` for this
context (so ``browser_wait_for_download`` can return). Bound to the
page ``"download"`` event by the enclosing ``_setup_page_handlers``.
Args:
download: The Playwright download that fired the event.
"""
try:
download_id = f"dl_{uuid.uuid4().hex[:12]}"
suggested_filename = download.suggested_filename
safe_filename = re.sub(r"[^\w\-\.]", "_", suggested_filename)
save_path = DOWNLOADS_DIR / f"{download_id}_{safe_filename}"
download_info = {
"download_id": download_id,
"context_id": context_id,
"url": download.url,
"suggested_filename": suggested_filename,
"path": str(save_path),
"state": "pending",
"timestamp": datetime.now().isoformat(),
}
self.downloads[download_id] = download_info
try:
await download.save_as(str(save_path))
download_info["state"] = "completed"
download_info["size"] = (
save_path.stat().st_size if save_path.exists() else 0
)
except Exception as e:
download_info["state"] = "failed"
download_info["error"] = str(e)
if context_id in self.pending_downloads:
future = self.pending_downloads.pop(context_id)
if not future.done():
future.set_result(download_info)
except Exception as e:
logger.warning(f"Error in download handler: {e}")
page.on("download", on_download)
async def on_dialog(dialog):
"""Auto-handle a JavaScript dialog per the context policy.
Records the dialog details into ``pending_dialogs`` (so
``browser_get_pending_dialog`` can report them) and accepts or
dismisses it according to this context's entry in ``dialog_handlers``
(defaulting to accept), preventing alerts/confirms/prompts from
blocking automation. Bound to the page ``"dialog"`` event by the
enclosing ``_setup_page_handlers``.
Args:
dialog: The Playwright dialog that fired the event.
"""
try:
dialog_info = {
"type": dialog.type,
"message": dialog.message,
"default_value": dialog.default_value,
"timestamp": datetime.now().isoformat(),
}
self.pending_dialogs[context_id] = dialog_info
action = self.dialog_handlers.get(context_id, "accept")
if action == "dismiss":
await dialog.dismiss()
dialog_info["handled"] = "dismissed"
else:
await dialog.accept()
dialog_info["handled"] = (
"accepted"
if action == "accept"
else "auto-accepted (manual mode)"
)
except Exception as e:
logger.warning(f"Error in dialog handler: {e}")
page.on("dialog", on_dialog)
[docs]
async def ensure_ready(self) -> bool:
"""Guarantee a live browser, restarting or initializing as needed.
The health gate every page/context accessor calls first. It restarts the
browser if the error count has crossed ``max_errors_before_restart``,
initializes it if not yet started, and restarts it if the existing
browser has lost its connection; on success it refreshes ``last_used``.
Called by ``ensure_proxied_context``, ``get_page``, ``get_context``,
``create_new_context``, ``browser_emulate_device``, and
``browser_set_geolocation`` in this module.
Returns:
bool: ``True`` when a connected browser is available, ``False``
otherwise.
"""
current_time = time.time()
if self.error_count >= self.max_errors_before_restart:
await self.restart()
return self.is_initialized
if not self.is_initialized:
return await self.initialize()
try:
if self.browser and self.browser.is_connected():
self.last_used = current_time
return True
await self.restart()
return self.is_initialized
except Exception:
self.error_count += 1
await self.restart()
return self.is_initialized
[docs]
async def restart(self) -> bool:
"""Tear down and relaunch the browser from scratch.
Recovery path used after repeated errors or a lost connection: it runs
``cleanup`` to release all pages/contexts/browser/Playwright resources,
then ``initialize`` to launch a fresh browser and default context.
Called by ``ensure_ready`` in this module and exposed through the
``browser_restart`` tool handler.
Returns:
bool: ``True`` if the relaunch succeeded, ``False`` otherwise.
"""
await self.cleanup()
return await self.initialize()
[docs]
async def cleanup(self):
"""Close all pages, contexts, the browser, and Playwright.
Best-effort teardown that swallows per-resource errors while closing
every open page and context, the browser, and the Playwright driver,
clearing the page/context maps and marking the manager uninitialized.
Holds the async lock so it cannot race a concurrent initialize.
Called by ``restart`` in this module; safe to invoke during shutdown.
"""
async with self._lock:
try:
for page in self.pages.values():
try:
await page.close()
except Exception:
pass
self.pages.clear()
for context in self.contexts.values():
try:
await context.close()
except Exception:
pass
self.contexts.clear()
if self.browser:
try:
await self.browser.close()
except Exception:
pass
self.browser = None
if self.playwright:
try:
await self.playwright.stop()
except Exception:
pass
self.playwright = None
self.is_initialized = False
except Exception as e:
logger.error(f"Error during browser cleanup: {e}")
[docs]
async def resolve_context_id(
self,
context_id: Optional[str],
proxy: Optional[str],
) -> str:
"""Choose the context to use, preferring a proxy-derived one.
Resolves the effective browser context for a request: when ``proxy`` is
set it routes to the stable SOCKS-backed context from
``ensure_proxied_context`` (the proxy deliberately overrides any explicit
``context_id``); otherwise it falls back to ``context_id`` or the default
context. This is why setting a proxy is enough to isolate a session.
Called by ``_ctx_or_err`` in this module (which all proxy-aware browser
tools route through).
Args:
context_id (Optional[str]): Explicit context id, used only when no
proxy is supplied.
proxy (Optional[str]): Optional SOCKS proxy URL; when present, wins.
Returns:
str: The resolved context id to operate on.
"""
if proxy and str(proxy).strip():
return await self.ensure_proxied_context(proxy.strip())
return context_id or self.default_context_id
[docs]
async def ensure_proxied_context(self, proxy_url: str) -> str:
"""Return (creating if needed) the context bound to a SOCKS proxy.
Validates the proxy with ``assert_safe_socks_proxy_url``, derives a stable
context id via ``_proxy_derived_context_id``, ensures the browser is
ready, and lazily creates a proxied context the first time that proxy is
seen so subsequent requests through the same proxy reuse one context.
Called by ``resolve_context_id`` in this module.
Args:
proxy_url: The SOCKS proxy URL to route the context through.
Returns:
str: The context id bound to this proxy.
Raises:
ValueError: If the proxy URL fails validation.
RuntimeError: If the browser could not be made ready.
"""
validated = assert_safe_socks_proxy_url(proxy_url.strip())
ctx_id = _proxy_derived_context_id(validated)
if not await self.ensure_ready():
raise RuntimeError("Browser not available")
if ctx_id in self.contexts:
return ctx_id
await self._create_context(ctx_id, proxy=validated)
return ctx_id
[docs]
async def get_page(self, context_id: str = None):
"""Return the active page for a context, creating it if absent.
The workhorse accessor nearly every browser handler calls to obtain a
page to drive. It ensures the browser is ready, creates the context (or
just a new page within an existing context) when one does not yet exist,
and bumps ``usage_count``/``last_used``.
Called by most ``browser_*`` handlers in this module (navigation,
interaction, content, waiting, screenshots, JS, etc.).
Args:
context_id (str): Context to fetch the page from; defaults to the
default context.
Returns:
The Playwright page for the context.
Raises:
RuntimeError: If the browser is not available.
"""
context_id = context_id or self.default_context_id
if not await self.ensure_ready():
raise RuntimeError("Browser not available")
if context_id not in self.pages:
if context_id not in self.contexts:
await self._create_context(context_id)
else:
page = await self.contexts[context_id].new_page()
self.pages[context_id] = page
self.usage_count += 1
self.last_used = time.time()
return self.pages[context_id]
[docs]
async def get_context(self, context_id: str = None):
"""Return the browser context for an id, creating it if absent.
Ensures the browser is ready and lazily creates the named context when it
does not already exist, then returns it. Used by the cookie tools, which
operate at the context level rather than on a page.
Called by ``browser_get_cookies``, ``browser_set_cookies``, and
``browser_clear_cookies`` in this module.
Args:
context_id (str): Context to fetch; defaults to the default context.
Returns:
The Playwright browser context.
Raises:
RuntimeError: If the browser is not available.
"""
context_id = context_id or self.default_context_id
if not await self.ensure_ready():
raise RuntimeError("Browser not available")
if context_id not in self.contexts:
await self._create_context(context_id)
return self.contexts[context_id]
[docs]
async def create_new_context(self, context_id: str = None, **options) -> str:
"""Create a brand-new isolated context and return its id.
Ensures the browser is ready, generates a random ``context_<hex>`` id when
none is given, and delegates to ``_create_context`` with the supplied
options (device, geolocation, proxy, etc.). Used to spin up incognito-like
or specially-configured contexts.
Called by ``browser_emulate_device``, ``browser_set_geolocation``, and
``browser_new_context`` in this module.
Args:
context_id (str): Desired context id; auto-generated when omitted.
**options: Context options forwarded to ``_create_context``.
Returns:
str: The id of the newly created context.
Raises:
RuntimeError: If the browser is not available.
"""
if not await self.ensure_ready():
raise RuntimeError("Browser not available")
context_id = context_id or f"context_{uuid.uuid4().hex[:8]}"
await self._create_context(context_id, **options)
return context_id
[docs]
async def close_context(self, context_id: str):
"""Close a context's page and context, ignoring errors.
Best-effort teardown of a single context: closes its page and the context
itself (swallowing close errors) and removes both from the manager's
maps, freeing the resources held by that browsing session.
Called by ``browser_close_context`` in this module.
Args:
context_id (str): The context to close.
"""
if context_id in self.pages:
try:
await self.pages[context_id].close()
except Exception:
pass
del self.pages[context_id]
if context_id in self.contexts:
try:
await self.contexts[context_id].close()
except Exception:
pass
del self.contexts[context_id]
[docs]
def record_error(self):
"""Increment the failure counter that drives auto-restart.
Bumps ``error_count``; once it reaches ``max_errors_before_restart``, the
next ``ensure_ready`` call recycles the browser. Handlers call this on
unexpected Playwright failures so a flaky browser self-heals.
Called from the exception paths of most ``browser_*`` handlers in this
module.
"""
self.error_count += 1
_bm = HeadlessBrowserManager()
async def _ctx_or_err(
context_id: Optional[str],
proxy: Optional[str],
) -> tuple[Optional[str], Optional[str]]:
"""Resolve a context id for a handler, capturing validation errors.
Thin wrapper over ``_bm.resolve_context_id`` that converts the proxy/context
inputs into a usable context id while turning the proxy-validation and
browser-availability exceptions into a return value, so the ~40 browser
handlers can branch on an error string instead of wrapping every call in
try/except.
Called by nearly every proxy-aware ``browser_*`` handler in this module.
Args:
context_id (Optional[str]): Explicit context id (ignored when a proxy is
given).
proxy (Optional[str]): Optional SOCKS proxy URL.
Returns:
tuple: ``(context_id, None)`` on success, or ``(None, error_message)``
when proxy validation fails or the browser is unavailable.
"""
try:
return await _bm.resolve_context_id(context_id, proxy), None
except (ValueError, RuntimeError) as e:
return None, str(e)
# ===================================================================
# Tool handler functions
# ===================================================================
# Navigation --------------------------------------------------------
[docs]
async def browser_navigate(
url: str,
wait_until: str = "load",
timeout: int = DEFAULT_TIMEOUT,
context_id: str = None,
proxy: str = None,
ctx=None,
) -> str:
"""Navigate the headless browser to a URL.
Backs the ``browser_navigate`` tool. After requiring ``UNSANDBOXED_EXEC``
(``_check_unsandboxed_exec``) and validating the URL and wait state
(``_validate_url``), it resolves the context (``_ctx_or_err``), grabs the page
(``_bm.get_page``), and drives ``page.goto``, returning the final URL, title,
and HTTP status. Playwright errors are recorded via ``_bm.record_error`` so a
flaky browser self-heals.
Dispatched by the tool runner in ``tools/__init__.py`` for the registered
``browser_navigate`` tool (``tool_def.handler(**arguments, ctx=ctx)``); also
invoked directly in ``tests/test_unsandboxed_exec_gating.py``.
Args:
url (str): URL to navigate to.
wait_until (str): When navigation is considered complete (``load``,
``domcontentloaded``, ``networkidle``, or ``commit``).
timeout (int): Navigation timeout in milliseconds.
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL; routes through a proxied context.
ctx (ToolContext): Tool execution context; required for the privilege
check.
Returns:
str: JSON envelope; on success carries the resulting url, title, status,
``ok`` flag, and context id; on failure an error message.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
auth_err = await _check_unsandboxed_exec(ctx)
if auth_err:
return auth_err
valid, error_msg = _validate_url(url)
if not valid:
return _json_response("error", error=error_msg)
valid_wait_states = ["load", "domcontentloaded", "networkidle", "commit"]
if wait_until not in valid_wait_states:
return _json_response(
"error",
error=f"Invalid wait_until value. Must be one of: {valid_wait_states}",
)
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
response = await page.goto(url, wait_until=wait_until, timeout=timeout)
return _json_response(
"success",
message="Navigation complete",
data={
"url": page.url,
"title": await page.title(),
"status": response.status if response else None,
"ok": response.ok if response else None,
"context_id": ctx_id,
},
)
except PlaywrightTimeout:
_bm.record_error()
return _json_response("error", error=f"Navigation timed out after {timeout}ms")
except PlaywrightError as e:
_bm.record_error()
return _json_response("error", error=f"Navigation error: {str(e)}")
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Unexpected error: {str(e)}")
[docs]
async def browser_go_back(
timeout: int = DEFAULT_TIMEOUT, context_id: str = None, proxy: str = None
) -> str:
"""Navigate back in the browser history.
Backs the ``browser_go_back`` tool. Resolves the context (``_ctx_or_err``),
gets the page (``_bm.get_page``), and calls ``page.go_back``, returning the
resulting url, title, and status. Requires no extra privilege.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_go_back``
tool; there are no direct internal callers.
Args:
timeout (int): Navigation timeout in milliseconds.
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope with the resulting url, title, and status, or an error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
response = await page.go_back(timeout=timeout)
return _json_response(
"success",
message="Navigated back",
data={
"url": page.url,
"title": await page.title(),
"status": response.status if response else None,
},
)
except PlaywrightTimeout:
return _json_response("error", error="Back navigation timed out")
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Error navigating back: {str(e)}")
[docs]
async def browser_go_forward(
timeout: int = DEFAULT_TIMEOUT, context_id: str = None, proxy: str = None
) -> str:
"""Navigate forward in the browser history.
Backs the ``browser_go_forward`` tool. Resolves the context (``_ctx_or_err``),
gets the page (``_bm.get_page``), and calls ``page.go_forward``, returning the
resulting url, title, and status. Requires no extra privilege.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_go_forward``
tool; there are no direct internal callers.
Args:
timeout (int): Navigation timeout in milliseconds.
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope with the resulting url, title, and status, or an error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
response = await page.go_forward(timeout=timeout)
return _json_response(
"success",
message="Navigated forward",
data={
"url": page.url,
"title": await page.title(),
"status": response.status if response else None,
},
)
except PlaywrightTimeout:
return _json_response("error", error="Forward navigation timed out")
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Error navigating forward: {str(e)}")
[docs]
async def browser_reload(
wait_until: str = "load",
timeout: int = DEFAULT_TIMEOUT,
context_id: str = None,
proxy: str = None,
ctx=None,
) -> str:
"""Reload the current page.
Backs the ``browser_reload`` tool. After requiring ``UNSANDBOXED_EXEC``
(``_check_unsandboxed_exec``), it resolves the context (``_ctx_or_err``),
gets the page (``_bm.get_page``), and calls ``page.reload``, returning the
resulting url, title, and status.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments, ctx=ctx)`` for the registered
``browser_reload`` tool; there are no direct internal callers.
Args:
wait_until (str): When the reload is considered complete.
timeout (int): Timeout in milliseconds.
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
ctx (ToolContext): Tool execution context; required for the privilege
check.
Returns:
str: JSON envelope with the resulting url, title, and status, or an error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
auth_err = await _check_unsandboxed_exec(ctx)
if auth_err:
return auth_err
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
response = await page.reload(wait_until=wait_until, timeout=timeout)
return _json_response(
"success",
message="Page reloaded",
data={
"url": page.url,
"title": await page.title(),
"status": response.status if response else None,
},
)
except PlaywrightTimeout:
return _json_response("error", error="Page reload timed out")
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Error reloading page: {str(e)}")
async def browser_screenshot(
selector: str = None,
full_page: bool = False,
format: str = "png",
quality: int = None,
timeout: int = DEFAULT_TIMEOUT,
context_id: str = None,
proxy: str = None,
ctx=None,
) -> str:
"""Capture a screenshot and return it as base64 (privileged variant).
Earlier ``UNSANDBOXED_EXEC``-gated screenshot implementation that returns the
image inline as base64. It is shadowed later in this module by a second
``browser_screenshot`` definition that instead saves to a file, so this
version is effectively dead code at import time and is not the handler the
``browser_screenshot`` tool registers. Documented for completeness; do not
rely on it being reachable.
Args:
selector (str): Optional CSS selector to screenshot instead of the page.
full_page (bool): Whether to capture the full scrollable page.
format (str): Image format, ``png`` or ``jpeg``.
quality (int): JPEG quality 0-100 (ignored for png).
timeout (int): Timeout in milliseconds.
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
ctx (ToolContext): Tool execution context; required for the privilege
check.
Returns:
str: JSON envelope carrying the base64-encoded image, or an error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
auth_err = await _check_unsandboxed_exec(ctx)
if auth_err:
return auth_err
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
params = {"type": format, "full_page": full_page, "timeout": timeout}
if selector:
params["selector"] = selector
if format == "jpeg" and quality is not None:
params["quality"] = quality
screenshot_bytes = await page.screenshot(**params)
return _json_response(
"success",
message="Screenshot captured",
data={
"base64": base64.b64encode(screenshot_bytes).decode("utf-8"),
},
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Screenshot error: {str(e)}")
[docs]
async def browser_close_context(
context_id: str = None, proxy: str = None, ctx=None
) -> str:
"""Close a browser context (window), except the default one.
Backs the ``browser_close_context`` tool. After requiring ``UNSANDBOXED_EXEC``
(``_check_unsandboxed_exec``) and resolving the context (``_ctx_or_err``), it
refuses to close the default context and otherwise delegates to
``_bm.close_context`` to free that browsing session's page and context.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments, ctx=ctx)`` for the registered
``browser_close_context`` tool; there are no direct internal callers.
Args:
context_id (str): Context to close (optional if ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL (same one used to create it).
ctx (ToolContext): Tool execution context; required for the privilege
check.
Returns:
str: JSON envelope confirming closure, or an error (including refusal to
close the default context).
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
auth_err = await _check_unsandboxed_exec(ctx)
if auth_err:
return auth_err
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
if ctx_id == _bm.default_context_id:
return _json_response("error", error="Cannot close the default context")
try:
await _bm.close_context(ctx_id)
return _json_response("success", message=f"Context '{ctx_id}' closed")
except Exception as e:
return _json_response("error", error=f"Error closing context: {str(e)}")
# Interaction -------------------------------------------------------
[docs]
async def browser_click(
selector: str,
button: str = "left",
click_count: int = 1,
timeout: int = DEFAULT_TIMEOUT,
context_id: str = None,
proxy: str = None,
ctx=None,
) -> str:
"""Click an element on the page.
Backs the ``browser_click`` tool. After requiring ``UNSANDBOXED_EXEC``
(``_check_unsandboxed_exec``) and validating the mouse button, it resolves
the context (``_ctx_or_err``), gets the page (``_bm.get_page``), and calls
``page.click`` with the requested button and click count.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments, ctx=ctx)`` for the registered
``browser_click`` tool; there are no direct internal callers.
Args:
selector (str): Element selector (CSS, XPath, ``text=...``).
button (str): Mouse button: ``left``, ``right``, or ``middle``.
click_count (int): Number of clicks (e.g. 2 for double-click).
timeout (int): Element wait timeout in milliseconds.
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
ctx (ToolContext): Tool execution context; required for the privilege
check.
Returns:
str: JSON envelope confirming the click, or an error (e.g. element not
found within the timeout).
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
auth_err = await _check_unsandboxed_exec(ctx)
if auth_err:
return auth_err
if button not in ("left", "right", "middle"):
return _json_response(
"error", error="Invalid button. Must be left, right, or middle"
)
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
await page.click(
selector, button=button, click_count=click_count, timeout=timeout
)
return _json_response(
"success",
message=f"Clicked element: {selector}",
data={
"selector": selector,
"button": button,
"click_count": click_count,
},
)
except PlaywrightTimeout:
return _json_response(
"error", error=f"Element not found within {timeout}ms: {selector}"
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Click error: {str(e)}")
[docs]
async def browser_type(
selector: str,
text: str,
delay: int = 0,
clear_first: bool = False,
timeout: int = DEFAULT_TIMEOUT,
context_id: str = None,
proxy: str = None,
ctx=None,
) -> str:
"""Type text into an input element, simulating keystrokes.
Backs the ``browser_type`` tool. After requiring ``UNSANDBOXED_EXEC``
(``_check_unsandboxed_exec``), it resolves the context (``_ctx_or_err``),
gets the page (``_bm.get_page``), optionally clears the field first, then
drives ``page.type`` with the requested per-key ``delay`` so the input looks
human-typed (useful for fields that react to keystrokes).
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments, ctx=ctx)`` for the registered
``browser_type`` tool; there are no direct internal callers.
Args:
selector (str): Target input element selector.
text (str): Text to type.
delay (int): Delay between keystrokes in milliseconds.
clear_first (bool): Whether to clear the field before typing.
timeout (int): Element wait timeout in milliseconds.
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
ctx (ToolContext): Tool execution context; required for the privilege
check.
Returns:
str: JSON envelope confirming the typed text length, or an error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
auth_err = await _check_unsandboxed_exec(ctx)
if auth_err:
return auth_err
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
if clear_first:
await page.fill(selector, "", timeout=timeout)
await page.type(selector, text, delay=delay, timeout=timeout)
return _json_response(
"success",
message=f"Typed text into: {selector}",
data={
"selector": selector,
"text_length": len(text),
"delay": delay,
},
)
except PlaywrightTimeout:
return _json_response(
"error", error=f"Element not found within {timeout}ms: {selector}"
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Type error: {str(e)}")
[docs]
async def browser_fill(
selector: str,
value: str,
timeout: int = DEFAULT_TIMEOUT,
context_id: str = None,
proxy: str = None,
ctx=None,
) -> str:
"""Fill an input field instantly (no per-key typing).
Backs the ``browser_fill`` tool. After requiring ``UNSANDBOXED_EXEC``
(``_check_unsandboxed_exec``), it resolves the context (``_ctx_or_err``),
gets the page (``_bm.get_page``), and calls ``page.fill``, which sets the
value in one shot. Faster than ``browser_type`` when keystroke simulation is
unnecessary.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments, ctx=ctx)`` for the registered
``browser_fill`` tool; there are no direct internal callers.
Args:
selector (str): Target input element selector.
value (str): Value to set.
timeout (int): Element wait timeout in milliseconds.
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
ctx (ToolContext): Tool execution context; required for the privilege
check.
Returns:
str: JSON envelope confirming the filled value length, or an error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
auth_err = await _check_unsandboxed_exec(ctx)
if auth_err:
return auth_err
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
await page.fill(selector, value, timeout=timeout)
return _json_response(
"success",
message=f"Filled element: {selector}",
data={
"selector": selector,
"value_length": len(value),
},
)
except PlaywrightTimeout:
return _json_response(
"error", error=f"Element not found within {timeout}ms: {selector}"
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Fill error: {str(e)}")
[docs]
async def browser_select_option(
selector: str,
value: str = None,
label: str = None,
index: int = None,
timeout: int = DEFAULT_TIMEOUT,
context_id: str = None,
proxy: str = None,
) -> str:
"""Select an option in a dropdown/``<select>`` element.
Backs the ``browser_select_option`` tool. Requires at least one of
``value``/``label``/``index``, resolves the context (``_ctx_or_err``), gets
the page (``_bm.get_page``), and calls ``page.select_option`` with whichever
matchers were supplied, returning the selected values.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered
``browser_select_option`` tool; there are no direct internal callers.
Args:
selector (str): The ``<select>`` element selector.
value (str): Option value to select.
label (str): Option visible label to select.
index (int): Option index to select.
timeout (int): Element wait timeout in milliseconds.
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope with the selected values, or an error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
if value is None and label is None and index is None:
return _json_response(
"error", error="Must provide at least one of: value, label, or index"
)
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
opts = {}
if value is not None:
opts["value"] = value
if label is not None:
opts["label"] = label
if index is not None:
opts["index"] = index
selected = await page.select_option(selector, timeout=timeout, **opts)
return _json_response(
"success",
message=f"Selected option in: {selector}",
data={
"selector": selector,
"selected_values": selected,
},
)
except PlaywrightTimeout:
return _json_response(
"error", error=f"Element not found within {timeout}ms: {selector}"
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Select error: {str(e)}")
[docs]
async def browser_press_key(
key: str,
selector: str = None,
timeout: int = DEFAULT_TIMEOUT,
context_id: str = None,
proxy: str = None,
) -> str:
"""Press a keyboard key, optionally focused on an element.
Backs the ``browser_press_key`` tool. Resolves the context (``_ctx_or_err``),
gets the page (``_bm.get_page``), and either presses the key on a specific
element (``page.press``) when a selector is given or on the page keyboard
(``page.keyboard.press``) otherwise. Supports chords like ``Control+a``.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_press_key``
tool; there are no direct internal callers.
Args:
key (str): Key (or chord) to press, e.g. ``Enter``, ``Tab``, ``Control+a``.
selector (str): Optional element to focus before pressing.
timeout (int): Element wait timeout in milliseconds.
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope confirming the key press, or an error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
if selector:
await page.press(selector, key, timeout=timeout)
else:
await page.keyboard.press(key)
return _json_response(
"success",
message=f"Pressed key: {key}",
data={"key": key, "selector": selector},
)
except PlaywrightTimeout:
return _json_response(
"error", error=f"Element not found within {timeout}ms: {selector}"
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Key press error: {str(e)}")
[docs]
async def browser_hover(
selector: str,
timeout: int = DEFAULT_TIMEOUT,
context_id: str = None,
proxy: str = None,
) -> str:
"""Hover the mouse over an element.
Backs the ``browser_hover`` tool. Resolves the context (``_ctx_or_err``),
gets the page (``_bm.get_page``), and calls ``page.hover`` to move the
pointer over the element, useful for triggering hover menus or tooltips.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_hover`` tool;
there are no direct internal callers.
Args:
selector (str): Element to hover over.
timeout (int): Element wait timeout in milliseconds.
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope confirming the hover, or an error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
await page.hover(selector, timeout=timeout)
return _json_response(
"success", message=f"Hovering over: {selector}", data={"selector": selector}
)
except PlaywrightTimeout:
return _json_response(
"error", error=f"Element not found within {timeout}ms: {selector}"
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Hover error: {str(e)}")
# Content Extraction ------------------------------------------------
[docs]
async def browser_get_content(
format: str = "text",
selector: str = None,
context_id: str = None,
proxy: str = None,
) -> str:
"""Extract page content as HTML, plain text, or markdown.
Backs the ``browser_get_content`` tool. Resolves the context (``_ctx_or_err``)
and page (``_bm.get_page``), then reads either the whole page or a selected
element. For ``markdown`` it converts the HTML through ``html2text`` (when
available) and collapses blank runs; all output is bounded by
``_truncate_content`` so a huge page cannot overflow the tool result.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_get_content``
tool; there are no direct internal callers.
Args:
format (str): Output format: ``html``, ``text``, or ``markdown``.
selector (str): Optional element to extract instead of the whole page.
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope with the (possibly truncated) content, its length, the
page url, and title, or an error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
valid_formats = ["html", "text", "markdown"]
if format not in valid_formats:
return _json_response(
"error", error=f"Invalid format. Must be one of: {valid_formats}"
)
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
if selector:
element = await page.query_selector(selector)
if not element:
return _json_response("error", error=f"Element not found: {selector}")
content = (
await element.inner_html()
if format == "html"
else await element.inner_text()
)
else:
content = (
await page.content()
if format == "html"
else await page.inner_text("body")
)
if format == "markdown":
if not HTML2TEXT_AVAILABLE:
return _json_response(
"error", error="html2text not installed for markdown conversion"
)
h = html2text.HTML2Text()
h.ignore_links = False
h.ignore_images = False
h.body_width = 0
html_content = content if selector else await page.content()
content = h.handle(html_content)
content = re.sub(r"\n\s*\n\s*\n+", "\n\n", content).strip()
content = _truncate_content(content)
return _json_response(
"success",
message=f"Retrieved {format} content",
data={
"format": format,
"content": content,
"length": len(content),
"url": page.url,
"title": await page.title(),
},
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Content extraction error: {str(e)}")
[docs]
async def browser_get_text(
selector: str,
timeout: int = DEFAULT_TIMEOUT,
context_id: str = None,
proxy: str = None,
) -> str:
"""Get the inner text of a single element.
Backs the ``browser_get_text`` tool. Resolves the context (``_ctx_or_err``)
and page (``_bm.get_page``), then returns ``page.inner_text`` for the matched
selector, waiting up to the timeout for it to appear.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_get_text``
tool; there are no direct internal callers.
Args:
selector (str): Element whose text to read.
timeout (int): Element wait timeout in milliseconds.
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope with the element text, or an error (e.g. not found).
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
text = await page.inner_text(selector, timeout=timeout)
return _json_response(
"success",
message="Retrieved element text",
data={"selector": selector, "text": text},
)
except PlaywrightTimeout:
return _json_response(
"error", error=f"Element not found within {timeout}ms: {selector}"
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Get text error: {str(e)}")
[docs]
async def browser_get_attribute(
selector: str,
attribute: str,
timeout: int = DEFAULT_TIMEOUT,
context_id: str = None,
proxy: str = None,
) -> str:
"""Read a named attribute from an element.
Backs the ``browser_get_attribute`` tool. Resolves the context
(``_ctx_or_err``) and page (``_bm.get_page``), then returns
``page.get_attribute`` for the selector/attribute pair (e.g. ``href``,
``src``, ``class``).
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_get_attribute``
tool; there are no direct internal callers.
Args:
selector (str): Element to inspect.
attribute (str): Attribute name to read.
timeout (int): Element wait timeout in milliseconds.
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope with the attribute value (``None`` if absent), or an
error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
value = await page.get_attribute(selector, attribute, timeout=timeout)
return _json_response(
"success",
message="Retrieved attribute value",
data={
"selector": selector,
"attribute": attribute,
"value": value,
},
)
except PlaywrightTimeout:
return _json_response(
"error", error=f"Element not found within {timeout}ms: {selector}"
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Get attribute error: {str(e)}")
[docs]
async def browser_get_page_info(context_id: str = None, proxy: str = None) -> str:
"""Report the current page's URL, title, and viewport.
Backs the ``browser_get_page_info`` tool. Resolves the context
(``_ctx_or_err``) and page (``_bm.get_page``) and returns lightweight page
metadata without extracting content.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_get_page_info``
tool; there are no direct internal callers.
Args:
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope with the url, title, and viewport size, or an error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
return _json_response(
"success",
message="Retrieved page info",
data={
"url": page.url,
"title": await page.title(),
"viewport": page.viewport_size,
},
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Get page info error: {str(e)}")
[docs]
async def browser_query_selector_all(
selector: str,
attributes: str = None,
timeout: int = DEFAULT_TIMEOUT,
context_id: str = None,
proxy: str = None,
) -> str:
"""Find all elements matching a selector and return their data.
Backs the ``browser_query_selector_all`` tool. Optionally parses
``attributes`` (a JSON array of attribute names), resolves the context
(``_ctx_or_err``) and page (``_bm.get_page``), runs
``page.query_selector_all``, and for each match collects its inner text plus
any requested attributes. Results are capped at 100 elements.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered
``browser_query_selector_all`` tool; there are no direct internal callers.
Args:
selector (str): Selector to match.
attributes (str): Optional JSON array of attribute names to also collect.
timeout (int): Reserved wait timeout in milliseconds.
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope with the match count and the per-element data (first
100), or an error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
try:
parsed_attrs = None
if attributes:
try:
parsed_attrs = (
json.loads(attributes)
if isinstance(attributes, str)
else attributes
)
except (json.JSONDecodeError, TypeError):
parsed_attrs = None
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
elements = await page.query_selector_all(selector)
results = []
for i, element in enumerate(elements):
item = {"index": i}
try:
item["text"] = await element.inner_text()
except Exception:
item["text"] = ""
if parsed_attrs:
for attr in parsed_attrs:
if attr != "text":
try:
item[attr] = await element.get_attribute(attr)
except Exception:
item[attr] = None
results.append(item)
return _json_response(
"success",
message=f"Found {len(results)} elements",
data={
"selector": selector,
"count": len(results),
"elements": results[:100],
},
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Query selector all error: {str(e)}")
# Waiting -----------------------------------------------------------
[docs]
async def browser_wait_for_selector(
selector: str,
state: str = "visible",
timeout: int = DEFAULT_TIMEOUT,
context_id: str = None,
proxy: str = None,
) -> str:
"""Wait for an element to reach a given state.
Backs the ``browser_wait_for_selector`` tool. Validates the requested state,
resolves the context (``_ctx_or_err``) and page (``_bm.get_page``), and calls
``page.wait_for_selector`` to block until the element is visible, hidden,
attached, or detached (or the timeout elapses).
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered
``browser_wait_for_selector`` tool; there are no direct internal callers.
Args:
selector (str): Element to wait on.
state (str): Target state: ``visible``, ``hidden``, ``attached``, or
``detached``.
timeout (int): Wait timeout in milliseconds.
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope confirming the state was reached, or a timeout/error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
valid_states = ["visible", "hidden", "attached", "detached"]
if state not in valid_states:
return _json_response(
"error", error=f"Invalid state. Must be one of: {valid_states}"
)
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
await page.wait_for_selector(selector, state=state, timeout=timeout)
return _json_response(
"success",
message=f"Element reached state: {state}",
data={
"selector": selector,
"state": state,
},
)
except PlaywrightTimeout:
return _json_response(
"error", error=f"Timeout waiting for {selector} to be {state}"
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Wait error: {str(e)}")
[docs]
async def browser_wait_for_load_state(
state: str = "load",
timeout: int = DEFAULT_TIMEOUT,
context_id: str = None,
proxy: str = None,
) -> str:
"""Wait for the page to reach a load lifecycle state.
Backs the ``browser_wait_for_load_state`` tool. Validates the state, resolves
the context (``_ctx_or_err``) and page (``_bm.get_page``), and calls
``page.wait_for_load_state`` to block until the page is loaded, DOM-content
loaded, or network-idle.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered
``browser_wait_for_load_state`` tool; there are no direct internal callers.
Args:
state (str): ``load``, ``domcontentloaded``, or ``networkidle``.
timeout (int): Wait timeout in milliseconds.
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope confirming the load state and url, or a timeout/error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
valid_states = ["load", "domcontentloaded", "networkidle"]
if state not in valid_states:
return _json_response(
"error", error=f"Invalid state. Must be one of: {valid_states}"
)
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
await page.wait_for_load_state(state, timeout=timeout)
return _json_response(
"success",
message=f"Page reached load state: {state}",
data={
"state": state,
"url": page.url,
},
)
except PlaywrightTimeout:
return _json_response("error", error=f"Timeout waiting for {state} state")
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Wait error: {str(e)}")
[docs]
async def browser_wait_for_url(
url_pattern: str,
timeout: int = DEFAULT_TIMEOUT,
context_id: str = None,
proxy: str = None,
) -> str:
"""Wait until the page URL matches a pattern.
Backs the ``browser_wait_for_url`` tool. Resolves the context (``_ctx_or_err``)
and page (``_bm.get_page``) and calls ``page.wait_for_url``; a pattern
anchored with ``^``/``$`` is compiled as a regex, otherwise it is treated as a
glob substring. Useful for awaiting redirects after a click or login.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_wait_for_url``
tool; there are no direct internal callers.
Args:
url_pattern (str): Regex (when anchored) or glob substring to match.
timeout (int): Wait timeout in milliseconds.
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope with the matched url, or a timeout/error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
if url_pattern.startswith("^") or url_pattern.endswith("$"):
pattern = re.compile(url_pattern)
await page.wait_for_url(pattern, timeout=timeout)
else:
await page.wait_for_url(f"**{url_pattern}**", timeout=timeout)
return _json_response(
"success",
message="URL matched pattern",
data={
"pattern": url_pattern,
"url": page.url,
},
)
except PlaywrightTimeout:
return _json_response(
"error", error=f"Timeout waiting for URL to match: {url_pattern}"
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Wait error: {str(e)}")
[docs]
async def browser_wait(
milliseconds: int, context_id: str = None, proxy: str = None
) -> str:
"""Sleep for a fixed number of milliseconds.
Backs the ``browser_wait`` tool. A simple ``asyncio.sleep`` (bounded to
0-60000 ms) for pacing automation between steps; it does not touch a page or
context.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_wait`` tool;
there are no direct internal callers.
Args:
milliseconds (int): Time to wait, clamped to the 0-60000 range.
context_id (str): Accepted for interface uniformity; unused.
proxy (str): Accepted for interface uniformity; unused.
Returns:
str: JSON envelope confirming the wait, or an error if out of range.
"""
if milliseconds < 0 or milliseconds > 60000:
return _json_response("error", error="Milliseconds must be between 0 and 60000")
try:
await asyncio.sleep(milliseconds / 1000)
return _json_response(
"success",
message=f"Waited {milliseconds}ms",
data={"milliseconds": milliseconds},
)
except Exception as e:
return _json_response("error", error=f"Wait error: {str(e)}")
# Screenshots & PDF -------------------------------------------------
[docs]
async def browser_screenshot(
full_page: bool = True,
selector: str = None,
filename: str = None,
context_id: str = None,
proxy: str = None,
) -> str:
"""Capture a screenshot to a file under the output directory.
Backs the registered ``browser_screenshot`` tool (this definition shadows the
earlier base64 variant). Resolves the context (``_ctx_or_err``) and page
(``_bm.get_page``), then writes a PNG to ``OUTPUT_DIR`` (auto-named with a
timestamp when no filename is given) for either a selected element or the
full/visible page, returning the saved path.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_screenshot``
tool; there are no direct internal callers.
Args:
full_page (bool): Whether to capture the full scrollable page.
selector (str): Optional element to screenshot instead of the page.
filename (str): Optional base filename (``.png`` is appended).
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope with the saved file path and metadata, or an error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
if not filename:
filename = f"screenshot_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
filepath = OUTPUT_DIR / f"{filename}.png"
if selector:
element = await page.query_selector(selector)
if not element:
return _json_response("error", error=f"Element not found: {selector}")
await element.screenshot(path=str(filepath))
else:
await page.screenshot(path=str(filepath), full_page=full_page)
return _json_response(
"success",
message="Screenshot saved",
data={
"path": str(filepath),
"full_page": full_page,
"selector": selector,
"url": page.url,
},
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Screenshot error: {str(e)}")
[docs]
async def browser_pdf(
filename: str = None,
format: str = "A4",
landscape: bool = False,
print_background: bool = True,
context_id: str = None,
proxy: str = None,
) -> str:
"""Render the current page to a PDF file.
Backs the ``browser_pdf`` tool. Validates the paper format, resolves the
context (``_ctx_or_err``) and page (``_bm.get_page``), then writes a PDF to
``OUTPUT_DIR`` (auto-named with a timestamp when no filename is given) via
``page.pdf``, honoring orientation and background-printing options.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_pdf`` tool;
there are no direct internal callers.
Args:
filename (str): Optional base filename (``.pdf`` is appended).
format (str): Paper size: ``A4``, ``Letter``, ``Legal``, ``Tabloid``,
``A3``, or ``A5``.
landscape (bool): Whether to use landscape orientation.
print_background (bool): Whether to print background graphics.
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope with the saved PDF path and metadata, or an error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
valid_formats = ["A4", "Letter", "Legal", "Tabloid", "A3", "A5"]
if format not in valid_formats:
return _json_response(
"error", error=f"Invalid format. Must be one of: {valid_formats}"
)
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
if not filename:
filename = f"page_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
filepath = OUTPUT_DIR / f"{filename}.pdf"
await page.pdf(
path=str(filepath),
format=format,
landscape=landscape,
print_background=print_background,
)
return _json_response(
"success",
message="PDF saved",
data={
"path": str(filepath),
"format": format,
"landscape": landscape,
"url": page.url,
},
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"PDF generation error: {str(e)}")
# JavaScript --------------------------------------------------------
[docs]
async def browser_evaluate(
expression: str, context_id: str = None, proxy: str = None
) -> str:
"""Evaluate a JavaScript expression and return its result.
Backs the ``browser_evaluate`` tool. Resolves the context (``_ctx_or_err``)
and page (``_bm.get_page``), runs ``page.evaluate(expression)``, and
serializes the return value (coercing non-JSON types via ``default=str``),
truncating the payload past ``MAX_CONTENT_LENGTH``.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_evaluate``
tool; there are no direct internal callers.
Args:
expression (str): JavaScript expression to evaluate in the page.
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope with the evaluated result (and its type), truncated
when too large, or an error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
result = await page.evaluate(expression)
result_str = json.dumps(result, default=str)
if len(result_str) > MAX_CONTENT_LENGTH:
return _json_response(
"success",
message="JavaScript evaluated (result truncated)",
data={
"result": "[Result too large - truncated]",
"result_type": type(result).__name__,
"result_length": len(result_str),
},
)
return _json_response(
"success",
message="JavaScript evaluated",
data={
"result": result,
"result_type": type(result).__name__,
},
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"JavaScript evaluation error: {str(e)}")
[docs]
async def browser_execute(
script: str, context_id: str = None, proxy: str = None
) -> str:
"""Execute a JavaScript snippet for its side effects.
Backs the ``browser_execute`` tool. Resolves the context (``_ctx_or_err``)
and page (``_bm.get_page``) and runs ``page.evaluate(script)`` without
returning the result, for scripts run purely for their effect on the page.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_execute``
tool; there are no direct internal callers.
Args:
script (str): JavaScript code to run in the page.
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope confirming execution (script length), or an error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
await page.evaluate(script)
return _json_response(
"success",
message="JavaScript executed",
data={"script_length": len(script)},
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"JavaScript execution error: {str(e)}")
# Cookies -----------------------------------------------------------
[docs]
async def browser_get_cookies(
urls: str = None, context_id: str = None, proxy: str = None
) -> str:
"""List the context's cookies with values redacted.
Backs the ``browser_get_cookies`` tool. Optionally parses ``urls`` (a JSON
array to filter by), fetches cookies from the context (``_bm.get_context``,
then ``context.cookies``), and returns sanitized metadata (name, domain,
path, flags, and value length) without exposing the secret cookie values.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_get_cookies``
tool; there are no direct internal callers.
Args:
urls (str): Optional JSON array of URLs to filter cookies by.
context_id (str): Browser context whose cookies to read.
proxy (str): Accepted for interface uniformity; this handler reads the
context directly.
Returns:
str: JSON envelope with the sanitized cookie list and count, or an error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
try:
parsed_urls = None
if urls:
try:
parsed_urls = json.loads(urls) if isinstance(urls, str) else urls
except (json.JSONDecodeError, TypeError):
parsed_urls = None
context = await _bm.get_context(context_id)
cookies = (
await context.cookies(parsed_urls)
if parsed_urls
else await context.cookies()
)
sanitized = [
{
"name": c.get("name"),
"domain": c.get("domain"),
"path": c.get("path"),
"expires": c.get("expires"),
"httpOnly": c.get("httpOnly"),
"secure": c.get("secure"),
"sameSite": c.get("sameSite"),
"value_length": len(c.get("value", "")),
}
for c in cookies
]
return _json_response(
"success",
message=f"Retrieved {len(cookies)} cookies",
data={
"cookies": sanitized,
"count": len(cookies),
},
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Get cookies error: {str(e)}")
[docs]
async def browser_set_cookies(
cookies: str, context_id: str = None, proxy: str = None
) -> str:
"""Add cookies to the browser context.
Backs the ``browser_set_cookies`` tool. Parses ``cookies`` (a JSON array of
cookie objects), validates that each has a name/value and a ``url`` or
``domain``, then adds them via ``_bm.get_context`` plus
``context.add_cookies``. Lets automation pre-seed an authenticated session.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_set_cookies``
tool; there are no direct internal callers.
Args:
cookies (str): JSON array of cookie objects to add.
context_id (str): Browser context to add the cookies to.
proxy (str): Accepted for interface uniformity; this handler uses the
context directly.
Returns:
str: JSON envelope with the count and names set, or a validation error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
try:
parsed = json.loads(cookies) if isinstance(cookies, str) else cookies
except (json.JSONDecodeError, TypeError):
return _json_response(
"error", error="cookies must be a JSON array of cookie objects"
)
if not parsed or not isinstance(parsed, list):
return _json_response("error", error="Cookies must be a non-empty list")
for cookie in parsed:
if "name" not in cookie or "value" not in cookie:
return _json_response(
"error", error="Each cookie must have 'name' and 'value'"
)
if "url" not in cookie and "domain" not in cookie:
return _json_response(
"error", error="Each cookie must have 'domain' or 'url'"
)
try:
context = await _bm.get_context(context_id)
await context.add_cookies(parsed)
return _json_response(
"success",
message=f"Set {len(parsed)} cookies",
data={
"count": len(parsed),
"names": [c["name"] for c in parsed],
},
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Set cookies error: {str(e)}")
[docs]
async def browser_clear_cookies(context_id: str = None, proxy: str = None) -> str:
"""Clear all cookies in the browser context.
Backs the ``browser_clear_cookies`` tool. Fetches the context
(``_bm.get_context``) and calls ``context.clear_cookies`` to drop every
cookie, effectively logging the session out.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_clear_cookies``
tool; there are no direct internal callers.
Args:
context_id (str): Browser context whose cookies to clear.
proxy (str): Accepted for interface uniformity; this handler uses the
context directly.
Returns:
str: JSON envelope confirming cookies were cleared, or an error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
try:
context = await _bm.get_context(context_id)
await context.clear_cookies()
return _json_response("success", message="All cookies cleared")
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Clear cookies error: {str(e)}")
# Advanced ----------------------------------------------------------
[docs]
async def browser_emulate_device(
device_name: str, context_id: str = None, proxy: str = None
) -> str:
"""Create a context emulating a named mobile device or tablet.
Backs the ``browser_emulate_device`` tool. Ensures the browser is ready
(``_bm.ensure_ready``), looks the device up in Playwright's device registry,
and spins up a new context configured for it via ``_bm.create_new_context``
(passing ``device``), returning the new context id and the device viewport.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered
``browser_emulate_device`` tool; there are no direct internal callers.
Args:
device_name (str): Playwright device name (e.g. ``iPhone 12``,
``Pixel 5``).
context_id (str): Optional id for the new context; auto-named otherwise.
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope with the new context id and device info, or an error
(e.g. unknown device, with a sample of valid names).
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
try:
if not await _bm.ensure_ready():
return _json_response("error", error="Browser not available")
device = _bm.playwright.devices.get(device_name)
if not device:
available = list(_bm.playwright.devices.keys())[:20]
return _json_response(
"error",
error=f"Unknown device: {device_name}. Some available: {available}",
)
new_ctx_id = context_id or f"device_{device_name.replace(' ', '_').lower()}"
await _bm.create_new_context(new_ctx_id, device=device_name)
return _json_response(
"success",
message=f"Device emulation enabled: {device_name}",
data={
"context_id": new_ctx_id,
"device": device_name,
"viewport": device.get("viewport"),
"user_agent": device.get("user_agent", "")[:100] + "...",
},
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Device emulation error: {str(e)}")
[docs]
async def browser_set_geolocation(
latitude: float,
longitude: float,
accuracy: float = 100,
context_id: str = None,
proxy: str = None,
) -> str:
"""Create a context with a spoofed geolocation.
Backs the ``browser_set_geolocation`` tool. Validates the latitude/longitude
ranges, ensures the browser is ready (``_bm.ensure_ready``), and creates a new
context with the geolocation (and the geolocation permission granted) via
``_bm.create_new_context``, so pages that read ``navigator.geolocation`` see
the supplied coordinates.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered
``browser_set_geolocation`` tool; there are no direct internal callers.
Args:
latitude (float): Latitude in ``-90..90``.
longitude (float): Longitude in ``-180..180``.
accuracy (float): Reported accuracy in meters.
context_id (str): Optional id for the new context; auto-named otherwise.
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope with the new context id and coordinates, or an error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
if not (-90 <= latitude <= 90):
return _json_response("error", error="Latitude must be between -90 and 90")
if not (-180 <= longitude <= 180):
return _json_response("error", error="Longitude must be between -180 and 180")
try:
if not await _bm.ensure_ready():
return _json_response("error", error="Browser not available")
new_ctx_id = context_id or f"geo_{latitude:.2f}_{longitude:.2f}"
await _bm.create_new_context(
new_ctx_id,
geolocation={
"latitude": latitude,
"longitude": longitude,
"accuracy": accuracy,
},
)
return _json_response(
"success",
message="Geolocation set",
data={
"context_id": new_ctx_id,
"latitude": latitude,
"longitude": longitude,
"accuracy": accuracy,
},
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Geolocation error: {str(e)}")
[docs]
async def browser_set_viewport(
width: int, height: int, context_id: str = None, proxy: str = None
) -> str:
"""Resize the page viewport.
Backs the ``browser_set_viewport`` tool. Validates the dimensions
(100-4000 px each), resolves the context (``_ctx_or_err``) and page
(``_bm.get_page``), and calls ``page.set_viewport_size`` to change the
rendered window size.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_set_viewport``
tool; there are no direct internal callers.
Args:
width (int): Viewport width in pixels (100-4000).
height (int): Viewport height in pixels (100-4000).
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope confirming the new size, or an error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
if not (100 <= width <= 4000):
return _json_response("error", error="Width must be between 100 and 4000")
if not (100 <= height <= 4000):
return _json_response("error", error="Height must be between 100 and 4000")
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
await page.set_viewport_size({"width": width, "height": height})
return _json_response(
"success", message="Viewport set", data={"width": width, "height": height}
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Set viewport error: {str(e)}")
[docs]
async def browser_block_resources(
resource_types: str, context_id: str = None, proxy: str = None
) -> str:
"""Block selected resource types from loading.
Backs the ``browser_block_resources`` tool. Parses and validates
``resource_types`` (a JSON array), resolves the context (``_ctx_or_err``) and
page (``_bm.get_page``), and installs a ``page.route`` handler that aborts
requests of the blocked types and continues the rest, speeding up loads and
saving bandwidth (e.g. blocking images and fonts).
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered
``browser_block_resources`` tool; there are no direct internal callers.
Args:
resource_types (str): JSON array of types to block, drawn from
``image``, ``stylesheet``, ``font``, ``script``, ``media``,
``websocket``.
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope listing the blocked types, or a validation error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
try:
parsed_types = (
json.loads(resource_types)
if isinstance(resource_types, str)
else resource_types
)
except (json.JSONDecodeError, TypeError):
return _json_response(
"error", error="resource_types must be a JSON array of strings"
)
valid_types = ["image", "stylesheet", "font", "script", "media", "websocket"]
invalid = [t for t in parsed_types if t not in valid_types]
if invalid:
return _json_response(
"error", error=f"Invalid resource types: {invalid}. Valid: {valid_types}"
)
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
async def route_handler(route):
"""Abort blocked resource requests, continue the rest.
Per-request callback registered via ``page.route`` by the enclosing
``browser_block_resources``: aborts the request when its resource
type is in the captured ``parsed_types`` set and otherwise lets it
proceed.
Args:
route: The Playwright route for the intercepted request.
"""
if route.request.resource_type in parsed_types:
await route.abort()
else:
await route.continue_()
await page.route("**/*", route_handler)
return _json_response(
"success",
message="Resource blocking enabled",
data={"blocked_types": parsed_types},
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Block resources error: {str(e)}")
[docs]
async def browser_get_status() -> str:
"""Report the browser manager's runtime status.
Backs the ``browser_get_status`` tool. Reads diagnostic fields off the shared
``_bm`` singleton (whether Playwright is available, init/connection state,
usage and error counts, and the list of active contexts/pages) without
touching any page.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_get_status``
tool; there are no direct internal callers.
Returns:
str: JSON envelope with the manager status snapshot.
"""
status = {
"playwright_available": PLAYWRIGHT_AVAILABLE,
"is_initialized": _bm.is_initialized,
"browser_connected": _bm.browser.is_connected() if _bm.browser else False,
"usage_count": _bm.usage_count,
"error_count": _bm.error_count,
"active_contexts": list(_bm.contexts.keys()),
"active_pages": list(_bm.pages.keys()),
"last_used": _bm.last_used,
}
return _json_response("success", message="Browser status retrieved", data=status)
[docs]
async def browser_restart() -> str:
"""Restart the shared browser instance.
Backs the ``browser_restart`` tool. Delegates to ``_bm.restart`` to tear down
and relaunch Firefox, used to recover from a wedged or misbehaving browser.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_restart``
tool; there are no direct internal callers.
Returns:
str: JSON envelope reporting success or failure of the restart.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
try:
success = await _bm.restart()
if success:
return _json_response("success", message="Browser restarted successfully")
return _json_response("error", error="Failed to restart browser")
except Exception as e:
return _json_response("error", error=f"Restart error: {str(e)}")
[docs]
async def browser_new_context(
context_id: str = None, incognito: bool = True, proxy: str = None
) -> str:
"""Create a new isolated browser context (incognito-like window).
Backs the ``browser_new_context`` tool. Optionally validates a SOCKS proxy
(``assert_safe_socks_proxy_url``) and creates a fresh context via
``_bm.create_new_context``, returning its id. Each context has its own
cookies, storage, and cache, so concurrent isolated sessions are possible.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_new_context``
tool; there are no direct internal callers.
Args:
context_id (str): Optional id for the new context; auto-named otherwise.
incognito (bool): Echoed back in the result; contexts are isolated by
default.
proxy (str): Optional SOCKS proxy URL to route the context through.
Returns:
str: JSON envelope with the new context id, or an error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
try:
kwargs = {}
if proxy and str(proxy).strip():
kwargs["proxy"] = assert_safe_socks_proxy_url(proxy.strip())
new_id = await _bm.create_new_context(context_id, **kwargs)
return _json_response(
"success",
message="New context created",
data={
"context_id": new_id,
"incognito": incognito,
},
)
except ValueError as e:
return _json_response("error", error=str(e))
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Create context error: {str(e)}")
# Network Interception ----------------------------------------------
[docs]
async def browser_enable_response_interception(
url_patterns: str, context_id: str = None, proxy: str = None
) -> str:
"""Start capturing network responses matching URL patterns.
Backs the ``browser_enable_response_interception`` tool. Parses
``url_patterns`` (a JSON array of globs), resolves the context
(``_ctx_or_err``), ensures a page exists (``_bm.get_page``), and appends the
patterns to this context's ``_bm.response_patterns``. The ``on_response``
handler installed by ``_setup_page_handlers`` then stores matching responses
in ``_bm.intercepted_responses`` for later retrieval.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered
``browser_enable_response_interception`` tool; there are no direct internal
callers.
Args:
url_patterns (str): JSON array of glob patterns to capture.
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope with the active pattern list and context id, or an
error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
try:
parsed_patterns = (
json.loads(url_patterns) if isinstance(url_patterns, str) else url_patterns
)
except (json.JSONDecodeError, TypeError):
return _json_response(
"error", error="url_patterns must be a JSON array of strings"
)
if not parsed_patterns or not isinstance(parsed_patterns, list):
return _json_response("error", error="url_patterns must be a non-empty list")
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
await _bm.get_page(ctx_id)
existing = _bm.response_patterns.get(ctx_id, [])
for pattern in parsed_patterns:
if pattern not in existing:
existing.append(pattern)
_bm.response_patterns[ctx_id] = existing
return _json_response(
"success",
message="Response interception enabled",
data={
"patterns": existing,
"context_id": ctx_id,
},
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Enable interception error: {str(e)}")
[docs]
async def browser_wait_for_response(
url_pattern: str,
timeout: int = DEFAULT_TIMEOUT,
context_id: str = None,
proxy: str = None,
) -> str:
"""Wait for one network response matching a pattern and return its body.
Backs the ``browser_wait_for_response`` tool. Resolves the context
(``_ctx_or_err``) and page (``_bm.get_page``), then uses
``page.expect_response`` to block until a response URL matches the glob,
decoding the body (JSON via ``_json_loads_network_body``, text inline, large
bodies summarized). Unlike interception, this captures a single awaited
response inline.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered
``browser_wait_for_response`` tool; there are no direct internal callers.
Args:
url_pattern (str): Glob pattern the response URL must match.
timeout (int): Wait timeout in milliseconds.
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope with the response url, status, headers, body, and
method, or a timeout/error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
page = await _bm.get_page(ctx_id)
async with page.expect_response(
lambda resp: fnmatch.fnmatch(resp.url, url_pattern),
timeout=timeout,
) as response_info:
pass
response = await response_info.value
body = None
content_type = response.headers.get("content-type", "")
try:
body_bytes = await response.body()
if len(body_bytes) <= MAX_RESPONSE_BODY_SIZE:
if "application/json" in content_type:
body = await _json_loads_network_body(body_bytes)
elif "text/" in content_type:
body = body_bytes.decode("utf-8")
else:
body = f"[Binary data: {len(body_bytes)} bytes]"
else:
body = f"[Response too large: {len(body_bytes)} bytes]"
except Exception as e:
body = f"[Could not read body: {str(e)}]"
return _json_response(
"success",
message="Response captured",
data={
"url": response.url,
"status": response.status,
"status_text": response.status_text,
"headers": dict(response.headers),
"body": body,
"method": response.request.method,
},
)
except PlaywrightTimeout:
return _json_response(
"error", error=f"Timeout waiting for response matching: {url_pattern}"
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Wait for response error: {str(e)}")
[docs]
async def browser_get_intercepted_responses(
context_id: str = None,
proxy: str = None,
url_filter: str = None,
) -> str:
"""List captured network responses as lightweight summaries.
Backs the ``browser_get_intercepted_responses`` tool. Reads from
``_bm.intercepted_responses``, optionally narrowing to a context
(``_ctx_or_err``) and/or a URL glob, and returns one summary per response
(id, url, status, method, timestamp, and whether a body was captured) so the
caller can then fetch a specific body via ``browser_get_response_body``.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered
``browser_get_intercepted_responses`` tool; there are no direct internal
callers.
Args:
context_id (str): Optional context to filter by.
proxy (str): Optional SOCKS proxy URL used to derive the context filter.
url_filter (str): Optional glob to filter response URLs.
Returns:
str: JSON envelope with the count and per-response summaries, or an error.
"""
try:
responses = list(_bm.intercepted_responses.values())
if context_id or (proxy and str(proxy).strip()):
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
responses = [r for r in responses if r.get("context_id") == ctx_id]
if url_filter:
responses = [
r for r in responses if fnmatch.fnmatch(r.get("url", ""), url_filter)
]
summaries = [
{
"response_id": r["response_id"],
"url": r["url"],
"status": r["status"],
"method": r["method"],
"timestamp": r["timestamp"],
"has_body": r.get("body") is not None,
}
for r in responses
]
return _json_response(
"success",
message=f"Found {len(summaries)} intercepted responses",
data={
"count": len(summaries),
"responses": summaries,
},
)
except Exception as e:
return _json_response(
"error", error=f"Get intercepted responses error: {str(e)}"
)
[docs]
async def browser_get_response_body(response_id: str) -> str:
"""Return the full stored record for one intercepted response.
Backs the ``browser_get_response_body`` tool. Looks the id up in
``_bm.intercepted_responses`` and returns the complete captured record
(headers and decoded body included), the companion to
``browser_get_intercepted_responses`` which only returns summaries.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered
``browser_get_response_body`` tool; there are no direct internal callers.
Args:
response_id (str): Id of a previously intercepted response.
Returns:
str: JSON envelope with the full response record, or an error if the id
is unknown.
"""
try:
if response_id not in _bm.intercepted_responses:
return _json_response("error", error=f"Response not found: {response_id}")
return _json_response(
"success",
message="Response body retrieved",
data=_bm.intercepted_responses[response_id],
)
except Exception as e:
return _json_response("error", error=f"Get response body error: {str(e)}")
[docs]
async def browser_clear_intercepted_responses(
context_id: str = None, proxy: str = None
) -> str:
"""Drop stored intercepted responses, optionally scoped to a context.
Backs the ``browser_clear_intercepted_responses`` tool. When a context (or
proxy) is given it resolves it (``_ctx_or_err``) and removes only that
context's entries from ``_bm.intercepted_responses``; otherwise it clears the
whole capture buffer, freeing memory.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered
``browser_clear_intercepted_responses`` tool; there are no direct internal
callers.
Args:
context_id (str): Optional context to scope the clear to.
proxy (str): Optional SOCKS proxy URL used to derive the context.
Returns:
str: JSON envelope reporting how many responses were cleared, or an error.
"""
try:
if context_id or (proxy and str(proxy).strip()):
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
to_remove = [
rid
for rid, r in _bm.intercepted_responses.items()
if r.get("context_id") == ctx_id
]
for rid in to_remove:
del _bm.intercepted_responses[rid]
count = len(to_remove)
else:
count = len(_bm.intercepted_responses)
_bm.intercepted_responses.clear()
return _json_response(
"success", message=f"Cleared {count} intercepted responses"
)
except Exception as e:
return _json_response("error", error=f"Clear responses error: {str(e)}")
# Session Persistence -----------------------------------------------
[docs]
async def browser_create_persistent_context(
session_name: str, context_id: str = None, proxy: str = None
) -> str:
"""Create a context whose cookies and storage persist on disk.
Backs the ``browser_create_persistent_context`` tool. Rejects a duplicate
context id, ensures Playwright is initialized, optionally validates a SOCKS
proxy, then calls ``_bm._create_persistent_context`` to launch a context
rooted at a ``SESSIONS_DIR`` profile so a logged-in session can be reused
across runs.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered
``browser_create_persistent_context`` tool; there are no direct internal
callers.
Args:
session_name (str): Logical session name (also the on-disk profile name).
context_id (str): Optional context id; defaults to ``session_name``.
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope with the context id, sanitized session name, and
session path, or an error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
if not session_name or not isinstance(session_name, str):
return _json_response("error", error="session_name must be a non-empty string")
try:
ctx_id = context_id or session_name
if ctx_id in _bm.contexts:
return _json_response(
"error", error=f"Context '{ctx_id}' already exists. Close it first."
)
if not _bm.playwright:
await _bm.initialize()
kwargs = {}
if proxy and str(proxy).strip():
kwargs["proxy"] = assert_safe_socks_proxy_url(proxy.strip())
await _bm._create_persistent_context(session_name, ctx_id, **kwargs)
safe_session_name = re.sub(r"[^\w\-]", "_", session_name)
return _json_response(
"success",
message="Persistent context created",
data={
"context_id": ctx_id,
"session_name": safe_session_name,
"session_path": str(SESSIONS_DIR / safe_session_name),
},
)
except ValueError as e:
return _json_response("error", error=str(e))
except Exception as e:
_bm.record_error()
return _json_response(
"error", error=f"Create persistent context error: {str(e)}"
)
[docs]
async def browser_list_sessions() -> str:
"""List saved persistent browser sessions on disk.
Backs the ``browser_list_sessions`` tool. Walks ``SESSIONS_DIR`` for saved
profile directories, computes each one's on-disk size, and notes whether a
live context (from ``_bm.persistent_contexts``) is currently bound to it.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_list_sessions``
tool; there are no direct internal callers.
Returns:
str: JSON envelope listing each session's name, path, size, and active
context (if any), or an error.
"""
try:
sessions = []
if SESSIONS_DIR.exists():
for session_dir in SESSIONS_DIR.iterdir():
if session_dir.is_dir():
size = sum(
f.stat().st_size for f in session_dir.rglob("*") if f.is_file()
)
active_context = None
for ctx_id, sess_name in _bm.persistent_contexts.items():
if sess_name == session_dir.name:
active_context = ctx_id
break
sessions.append(
{
"session_name": session_dir.name,
"path": str(session_dir),
"size_bytes": size,
"size_mb": round(size / (1024 * 1024), 2),
"active_context": active_context,
}
)
return _json_response(
"success",
message=f"Found {len(sessions)} sessions",
data={
"count": len(sessions),
"sessions": sessions,
},
)
except Exception as e:
return _json_response("error", error=f"List sessions error: {str(e)}")
[docs]
async def browser_delete_session(session_name: str) -> str:
"""Delete a saved persistent session directory from disk.
Backs the ``browser_delete_session`` tool. Resolves the sanitized session
path under ``SESSIONS_DIR``, refuses to delete a session still bound to a
live context (checked against ``_bm.persistent_contexts``), and otherwise
removes the directory tree off-thread via ``asyncio.to_thread(shutil.rmtree)``.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_delete_session``
tool; there are no direct internal callers.
Args:
session_name (str): Name of the saved session to delete.
Returns:
str: JSON envelope confirming deletion, or an error (missing session or
session in use).
"""
if not session_name:
return _json_response("error", error="session_name is required")
try:
safe_session_name = re.sub(r"[^\w\-]", "_", session_name)
session_path = SESSIONS_DIR / safe_session_name
if not session_path.exists():
return _json_response("error", error=f"Session not found: {session_name}")
for ctx_id, sess_name in _bm.persistent_contexts.items():
if sess_name == safe_session_name:
return _json_response(
"error",
error=f"Session is in use by context '{ctx_id}'. Close it first.",
)
await asyncio.to_thread(shutil.rmtree, session_path)
return _json_response(
"success",
message=f"Session '{session_name}' deleted",
data={
"session_name": safe_session_name,
"path": str(session_path),
},
)
except Exception as e:
return _json_response("error", error=f"Delete session error: {str(e)}")
# Downloads ---------------------------------------------------------
[docs]
async def browser_wait_for_download(
timeout: int = DEFAULT_TIMEOUT, context_id: str = None, proxy: str = None
) -> str:
"""Wait for the next file download in a context to complete.
Backs the ``browser_wait_for_download`` tool. Resolves the context
(``_ctx_or_err``), ensures a page exists (``_bm.get_page``), and registers a
future in ``_bm.pending_downloads`` that the ``on_download`` handler resolves
once the file is saved to ``DOWNLOADS_DIR``; the future is awaited with a
timeout.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered
``browser_wait_for_download`` tool; there are no direct internal callers.
Args:
timeout (int): Wait timeout in milliseconds.
context_id (str): Browser context to use (ignored when ``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope with the completed download info, or a timeout/error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
await _bm.get_page(ctx_id)
future = asyncio.get_event_loop().create_future()
_bm.pending_downloads[ctx_id] = future
try:
download_info = await asyncio.wait_for(future, timeout=timeout / 1000)
return _json_response(
"success", message="Download completed", data=download_info
)
except asyncio.TimeoutError:
_bm.pending_downloads.pop(ctx_id, None)
return _json_response(
"error", error=f"Timeout waiting for download after {timeout}ms"
)
except Exception as e:
_bm.record_error()
return _json_response("error", error=f"Wait for download error: {str(e)}")
[docs]
async def browser_get_downloads(
context_id: str = None, proxy: str = None, state: str = None
) -> str:
"""List captured downloads, optionally filtered.
Backs the ``browser_get_downloads`` tool. Reads ``_bm.downloads``, optionally
narrowing to a context (``_ctx_or_err``) and/or a state, and returns the
matching download records (id, url, filename, saved path, state, size).
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_get_downloads``
tool; there are no direct internal callers.
Args:
context_id (str): Optional context to filter by.
proxy (str): Optional SOCKS proxy URL used to derive the context filter.
state (str): Optional state filter (``completed``, ``pending``,
``failed``).
Returns:
str: JSON envelope with the count and matching download records, or an
error.
"""
try:
downloads = list(_bm.downloads.values())
if context_id or (proxy and str(proxy).strip()):
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
downloads = [d for d in downloads if d.get("context_id") == ctx_id]
if state:
downloads = [d for d in downloads if d.get("state") == state]
return _json_response(
"success",
message=f"Found {len(downloads)} downloads",
data={
"count": len(downloads),
"downloads": downloads,
},
)
except Exception as e:
return _json_response("error", error=f"Get downloads error: {str(e)}")
[docs]
async def browser_save_download(download_id: str, filename: str = None) -> str:
"""Return a download's info, optionally renaming the saved file.
Backs the ``browser_save_download`` tool. Looks the id up in
``_bm.downloads``; when a ``filename`` is given it renames the file on disk
(within ``DOWNLOADS_DIR``, sanitizing the name) and updates the record.
Returns the (possibly updated) download info.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_save_download``
tool; there are no direct internal callers.
Args:
download_id (str): Id of a captured download.
filename (str): Optional new filename to rename the saved file to.
Returns:
str: JSON envelope with the download info, or an error if the id is
unknown.
"""
try:
if download_id not in _bm.downloads:
return _json_response("error", error=f"Download not found: {download_id}")
download_info = _bm.downloads[download_id]
if filename:
old_path = Path(download_info["path"])
if old_path.exists():
safe_filename = re.sub(r"[^\w\-\.]", "_", filename)
new_path = DOWNLOADS_DIR / safe_filename
old_path.rename(new_path)
download_info["path"] = str(new_path)
download_info["renamed_to"] = safe_filename
return _json_response(
"success", message="Download info retrieved", data=download_info
)
except Exception as e:
return _json_response("error", error=f"Save download error: {str(e)}")
# Console & Dialog --------------------------------------------------
[docs]
async def browser_get_console_logs(
context_id: str = None,
proxy: str = None,
level: str = None,
limit: int = 100,
) -> str:
"""Return captured browser console messages for a context.
Backs the ``browser_get_console_logs`` tool. Resolves the context
(``_ctx_or_err``) and reads its buffer from ``_bm.console_logs`` (populated by
the ``on_console`` page handler), optionally filtering by log level and
capping to the most recent ``limit`` entries.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered
``browser_get_console_logs`` tool; there are no direct internal callers.
Args:
context_id (str): Browser context whose logs to read (ignored when
``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
level (str): Optional level filter (``log``, ``warn``, ``error``,
``info``, ``debug``, ``warning``).
limit (int): Maximum number of most-recent entries to return.
Returns:
str: JSON envelope with the matching console logs and count, or an error.
"""
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
logs = _bm.console_logs.get(ctx_id, [])
if level:
valid_levels = ["log", "warn", "error", "info", "debug", "warning"]
if level not in valid_levels:
return _json_response(
"error", error=f"Invalid level. Must be one of: {valid_levels}"
)
logs = [log for log in logs if log.get("type") == level]
logs = logs[-limit:]
return _json_response(
"success",
message=f"Retrieved {len(logs)} console logs",
data={
"count": len(logs),
"logs": logs,
},
)
except Exception as e:
return _json_response("error", error=f"Get console logs error: {str(e)}")
[docs]
async def browser_clear_console_logs(context_id: str = None, proxy: str = None) -> str:
"""Clear a context's captured console logs.
Backs the ``browser_clear_console_logs`` tool. Resolves the context
(``_ctx_or_err``) and empties its ``_bm.console_logs`` buffer, reporting how
many entries were dropped.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered
``browser_clear_console_logs`` tool; there are no direct internal callers.
Args:
context_id (str): Browser context whose logs to clear (ignored when
``proxy`` is set).
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope reporting the number of logs cleared, or an error.
"""
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
count = len(_bm.console_logs.get(ctx_id, []))
_bm.console_logs[ctx_id] = []
return _json_response("success", message=f"Cleared {count} console logs")
except Exception as e:
return _json_response("error", error=f"Clear console logs error: {str(e)}")
[docs]
async def browser_handle_dialog(
action: str, prompt_text: str = None, context_id: str = None, proxy: str = None
) -> str:
"""Set how JavaScript dialogs are auto-handled for a context.
Backs the ``browser_handle_dialog`` tool. Validates the action, resolves the
context (``_ctx_or_err``), and stores the policy in ``_bm.dialog_handlers`` so
the ``on_dialog`` page handler accepts, dismisses, or auto-accepts future
alerts/confirms/prompts; it also echoes back the last seen dialog from
``_bm.pending_dialogs``.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered ``browser_handle_dialog``
tool; there are no direct internal callers.
Args:
action (str): Policy: ``accept``, ``dismiss``, or ``manual``.
prompt_text (str): Optional text echoed back (for prompt dialogs).
context_id (str): Browser context to configure (ignored when ``proxy`` is
set).
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope confirming the policy and the last dialog seen, or a
validation error.
"""
if not PLAYWRIGHT_AVAILABLE:
return _json_response("error", error="Playwright not installed")
valid_actions = ["accept", "dismiss", "manual"]
if action not in valid_actions:
return _json_response(
"error", error=f"Invalid action. Must be one of: {valid_actions}"
)
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
_bm.dialog_handlers[ctx_id] = action
last_dialog = _bm.pending_dialogs.get(ctx_id)
return _json_response(
"success",
message=f"Dialog handler set to: {action}",
data={
"action": action,
"prompt_text": prompt_text,
"last_dialog": last_dialog,
},
)
except Exception as e:
return _json_response("error", error=f"Handle dialog error: {str(e)}")
[docs]
async def browser_get_pending_dialog(context_id: str = None, proxy: str = None) -> str:
"""Report the last JavaScript dialog seen in a context.
Backs the ``browser_get_pending_dialog`` tool. Resolves the context
(``_ctx_or_err``) and returns its entry from ``_bm.pending_dialogs`` (recorded
by the ``on_dialog`` page handler), so the caller can inspect the type,
message, and how it was handled.
Dispatched by the tool runner in ``tools/__init__.py``, which calls
``tool_def.handler(**arguments)`` for the registered
``browser_get_pending_dialog`` tool; there are no direct internal callers.
Args:
context_id (str): Browser context to inspect (ignored when ``proxy`` is
set).
proxy (str): Optional SOCKS proxy URL.
Returns:
str: JSON envelope with the last dialog info, or a ``None`` data payload
when no dialog has appeared, or an error.
"""
try:
ctx_id, err = await _ctx_or_err(context_id, proxy)
if err:
return _json_response("error", error=err)
dialog_info = _bm.pending_dialogs.get(ctx_id)
if dialog_info:
return _json_response(
"success", message="Dialog info retrieved", data=dialog_info
)
return _json_response("success", message="No dialog has appeared", data=None)
except Exception as e:
return _json_response("error", error=f"Get pending dialog error: {str(e)}")
# ===================================================================
# v3 multi-tool registration
# ===================================================================
def _p(props: dict, required: list = None):
"""Build a JSON-Schema ``parameters`` object for a tool entry.
Tiny convenience used throughout the ``TOOLS`` list below to assemble each
handler's ``object`` parameter schema from a properties map and an optional
required list, keeping the registration block compact.
Called only by the ``TOOLS`` definitions in this module.
Args:
props (dict): Mapping of parameter name to its JSON-Schema fragment.
required (list): Optional list of required parameter names.
Returns:
dict: A JSON-Schema object with ``type``, ``properties``, and (when
given) ``required``.
"""
schema = {"type": "object", "properties": props}
if required:
schema["required"] = required
return schema
_S = {"type": "string"}
_I = {"type": "integer"}
_N = {"type": "number"}
_B = {"type": "boolean"}
_PROXY = {
"type": "string",
"description": (
"Optional SOCKS5 proxy URL (e.g. socks5h://127.0.0.1:9050 for Tor). "
"Omit for direct connection. When set, uses a stable proxied context and overrides context_id."
),
}
TOOLS = [
# --- Navigation ---
{
"name": "browser_navigate",
"description": "Navigate the headless Firefox browser to a URL.",
"parameters": _p(
{
"url": {**_S, "description": "URL to navigate to."},
"wait_until": {
**_S,
"description": "When to consider navigation complete (load|domcontentloaded|networkidle|commit).",
},
"timeout": {**_I, "description": "Timeout in ms (default 30000)."},
"context_id": {**_S, "description": "Browser context ID (optional)."},
"proxy": _PROXY,
},
["url"],
),
"handler": browser_navigate,
},
{
"name": "browser_go_back",
"description": "Navigate back in browser history.",
"parameters": _p(
{
"timeout": {**_I, "description": "Timeout in ms."},
"context_id": _S,
"proxy": _PROXY,
}
),
"handler": browser_go_back,
},
{
"name": "browser_go_forward",
"description": "Navigate forward in browser history.",
"parameters": _p(
{
"timeout": {**_I, "description": "Timeout in ms."},
"context_id": _S,
"proxy": _PROXY,
}
),
"handler": browser_go_forward,
},
{
"name": "browser_reload",
"description": "Reload the current page.",
"parameters": _p(
{"wait_until": _S, "timeout": _I, "context_id": _S, "proxy": _PROXY}
),
"handler": browser_reload,
},
{
"name": "browser_close_context",
"description": "Close a browser context (window).",
"parameters": _p(
{
"context_id": {
**_S,
"description": "ID of the context to close (optional if proxy is set).",
},
"proxy": _PROXY,
}
),
"handler": browser_close_context,
},
# --- Interaction ---
{
"name": "browser_click",
"description": "Click on an element on the page.",
"parameters": _p(
{
"selector": {
**_S,
"description": "Element selector (CSS, XPath, text=...).",
},
"button": _S,
"click_count": _I,
"timeout": _I,
"context_id": _S,
"proxy": _PROXY,
},
["selector"],
),
"handler": browser_click,
},
{
"name": "browser_type",
"description": "Type text into an input element with optional delay.",
"parameters": _p(
{
"selector": _S,
"text": _S,
"delay": _I,
"clear_first": _B,
"timeout": _I,
"context_id": _S,
"proxy": _PROXY,
},
["selector", "text"],
),
"handler": browser_type,
},
{
"name": "browser_fill",
"description": "Fill an input field instantly (no typing simulation).",
"parameters": _p(
{
"selector": _S,
"value": _S,
"timeout": _I,
"context_id": _S,
"proxy": _PROXY,
},
["selector", "value"],
),
"handler": browser_fill,
},
{
"name": "browser_fill_form",
"description": "Fill multiple form fields at once.",
"parameters": _p(
{
"fields": {
**_S,
"description": "JSON object mapping selectors to values.",
},
"timeout": _I,
"context_id": _S,
"proxy": _PROXY,
},
["fields"],
),
"handler": browser_fill_form,
},
{
"name": "browser_select_option",
"description": "Select an option from a dropdown/select element.",
"parameters": _p(
{
"selector": _S,
"value": _S,
"label": _S,
"index": _I,
"timeout": _I,
"context_id": _S,
"proxy": _PROXY,
},
["selector"],
),
"handler": browser_select_option,
},
{
"name": "browser_press_key",
"description": "Press a keyboard key (Enter, Tab, Escape, etc.).",
"parameters": _p(
{
"key": {
**_S,
"description": "Key to press (e.g. Enter, Tab, Control+a).",
},
"selector": _S,
"timeout": _I,
"context_id": _S,
"proxy": _PROXY,
},
["key"],
),
"handler": browser_press_key,
},
{
"name": "browser_hover",
"description": "Hover over an element.",
"parameters": _p(
{"selector": _S, "timeout": _I, "context_id": _S, "proxy": _PROXY},
["selector"],
),
"handler": browser_hover,
},
{
"name": "browser_scroll",
"description": "Scroll the page or an element.",
"parameters": _p(
{
"direction": {**_S, "description": "up|down|left|right."},
"amount": _I,
"selector": _S,
"context_id": _S,
"proxy": _PROXY,
}
),
"handler": browser_scroll,
},
# --- Content Extraction ---
{
"name": "browser_get_content",
"description": "Get page content in HTML, text, or markdown format.",
"parameters": _p(
{
"format": {**_S, "description": "html|text|markdown."},
"selector": _S,
"context_id": _S,
"proxy": _PROXY,
}
),
"handler": browser_get_content,
},
{
"name": "browser_get_text",
"description": "Get the text content of an element.",
"parameters": _p(
{"selector": _S, "timeout": _I, "context_id": _S, "proxy": _PROXY},
["selector"],
),
"handler": browser_get_text,
},
{
"name": "browser_get_attribute",
"description": "Get an attribute value from an element.",
"parameters": _p(
{
"selector": _S,
"attribute": {
**_S,
"description": "Attribute name (href, src, class, etc.).",
},
"timeout": _I,
"context_id": _S,
"proxy": _PROXY,
},
["selector", "attribute"],
),
"handler": browser_get_attribute,
},
{
"name": "browser_get_page_info",
"description": "Get current page URL, title, and viewport info.",
"parameters": _p({"context_id": _S, "proxy": _PROXY}),
"handler": browser_get_page_info,
},
{
"name": "browser_query_selector_all",
"description": "Find all elements matching a selector.",
"parameters": _p(
{
"selector": _S,
"attributes": {
**_S,
"description": "JSON array of attribute names to retrieve.",
},
"timeout": _I,
"context_id": _S,
"proxy": _PROXY,
},
["selector"],
),
"handler": browser_query_selector_all,
},
# --- Waiting ---
{
"name": "browser_wait_for_selector",
"description": "Wait for an element to appear, disappear, or change state.",
"parameters": _p(
{
"selector": _S,
"state": {**_S, "description": "visible|hidden|attached|detached."},
"timeout": _I,
"context_id": _S,
"proxy": _PROXY,
},
["selector"],
),
"handler": browser_wait_for_selector,
},
{
"name": "browser_wait_for_load_state",
"description": "Wait for page load state (load, domcontentloaded, networkidle).",
"parameters": _p(
{"state": _S, "timeout": _I, "context_id": _S, "proxy": _PROXY}
),
"handler": browser_wait_for_load_state,
},
{
"name": "browser_wait_for_url",
"description": "Wait for the URL to match a pattern.",
"parameters": _p(
{"url_pattern": _S, "timeout": _I, "context_id": _S, "proxy": _PROXY},
["url_pattern"],
),
"handler": browser_wait_for_url,
},
{
"name": "browser_wait",
"description": "Wait for a specified number of milliseconds.",
"parameters": _p(
{
"milliseconds": {**_I, "description": "Time to wait (0-60000)."},
"context_id": _S,
"proxy": _PROXY,
},
["milliseconds"],
),
"handler": browser_wait,
},
# --- Screenshots & PDF ---
{
"name": "browser_screenshot",
"description": "Take a screenshot of the page or element.",
"parameters": _p(
{
"full_page": _B,
"selector": _S,
"filename": _S,
"context_id": _S,
"proxy": _PROXY,
}
),
"handler": browser_screenshot,
},
{
"name": "browser_pdf",
"description": "Generate a PDF of the current page.",
"parameters": _p(
{
"filename": _S,
"format": {**_S, "description": "A4|Letter|Legal|Tabloid|A3|A5."},
"landscape": _B,
"print_background": _B,
"context_id": _S,
"proxy": _PROXY,
}
),
"handler": browser_pdf,
},
# --- JavaScript ---
{
"name": "browser_evaluate",
"description": "Evaluate JavaScript and return the result.",
"parameters": _p(
{"expression": _S, "context_id": _S, "proxy": _PROXY}, ["expression"]
),
"handler": browser_evaluate,
},
{
"name": "browser_execute",
"description": "Execute JavaScript code on the page.",
"parameters": _p({"script": _S, "context_id": _S, "proxy": _PROXY}, ["script"]),
"handler": browser_execute,
},
# --- Cookies ---
{
"name": "browser_get_cookies",
"description": "Get browser cookies.",
"parameters": _p(
{
"urls": {**_S, "description": "JSON array of URLs to filter cookies."},
"context_id": _S,
"proxy": _PROXY,
}
),
"handler": browser_get_cookies,
},
{
"name": "browser_set_cookies",
"description": "Set browser cookies.",
"parameters": _p(
{
"cookies": {**_S, "description": "JSON array of cookie objects."},
"context_id": _S,
"proxy": _PROXY,
},
["cookies"],
),
"handler": browser_set_cookies,
},
{
"name": "browser_clear_cookies",
"description": "Clear all browser cookies.",
"parameters": _p({"context_id": _S, "proxy": _PROXY}),
"handler": browser_clear_cookies,
},
# --- Advanced ---
{
"name": "browser_emulate_device",
"description": "Emulate a mobile device or tablet.",
"parameters": _p(
{
"device_name": {
**_S,
"description": "Device name (e.g. iPhone 12, Pixel 5).",
},
"context_id": _S,
"proxy": _PROXY,
},
["device_name"],
),
"handler": browser_emulate_device,
},
{
"name": "browser_set_geolocation",
"description": "Set the browser's geolocation.",
"parameters": _p(
{
"latitude": _N,
"longitude": _N,
"accuracy": _N,
"context_id": _S,
"proxy": _PROXY,
},
["latitude", "longitude"],
),
"handler": browser_set_geolocation,
},
{
"name": "browser_set_viewport",
"description": "Set the browser viewport size.",
"parameters": _p(
{"width": _I, "height": _I, "context_id": _S, "proxy": _PROXY},
["width", "height"],
),
"handler": browser_set_viewport,
},
{
"name": "browser_block_resources",
"description": "Block images, CSS, fonts, or other resources.",
"parameters": _p(
{
"resource_types": {
**_S,
"description": "JSON array of types: image|stylesheet|font|script|media|websocket.",
},
"context_id": _S,
"proxy": _PROXY,
},
["resource_types"],
),
"handler": browser_block_resources,
},
{
"name": "browser_new_context",
"description": "Create a new browser context (like incognito window).",
"parameters": _p({"context_id": _S, "incognito": _B, "proxy": _PROXY}),
"handler": browser_new_context,
},
# --- Status & Management ---
{
"name": "browser_get_status",
"description": "Get the browser manager status.",
"parameters": _p({}),
"handler": browser_get_status,
},
{
"name": "browser_restart",
"description": "Restart the browser instance.",
"parameters": _p({}),
"handler": browser_restart,
},
# --- Network Interception ---
{
"name": "browser_enable_response_interception",
"description": "Enable interception of network responses matching URL patterns.",
"parameters": _p(
{
"url_patterns": {**_S, "description": "JSON array of glob patterns."},
"context_id": _S,
"proxy": _PROXY,
},
["url_patterns"],
),
"handler": browser_enable_response_interception,
},
{
"name": "browser_wait_for_response",
"description": "Wait for a specific network response and capture its body.",
"parameters": _p(
{"url_pattern": _S, "timeout": _I, "context_id": _S, "proxy": _PROXY},
["url_pattern"],
),
"handler": browser_wait_for_response,
},
{
"name": "browser_get_intercepted_responses",
"description": "Get all captured network responses.",
"parameters": _p({"context_id": _S, "proxy": _PROXY, "url_filter": _S}),
"handler": browser_get_intercepted_responses,
},
{
"name": "browser_get_response_body",
"description": "Get the full body of an intercepted response.",
"parameters": _p({"response_id": _S}, ["response_id"]),
"handler": browser_get_response_body,
},
{
"name": "browser_clear_intercepted_responses",
"description": "Clear stored intercepted responses.",
"parameters": _p({"context_id": _S, "proxy": _PROXY}),
"handler": browser_clear_intercepted_responses,
},
# --- Session Persistence ---
{
"name": "browser_create_persistent_context",
"description": "Create a browser context with persistent session storage.",
"parameters": _p(
{"session_name": _S, "context_id": _S, "proxy": _PROXY}, ["session_name"]
),
"handler": browser_create_persistent_context,
},
{
"name": "browser_list_sessions",
"description": "List all saved browser sessions.",
"parameters": _p({}),
"handler": browser_list_sessions,
},
{
"name": "browser_delete_session",
"description": "Delete a saved browser session.",
"parameters": _p({"session_name": _S}, ["session_name"]),
"handler": browser_delete_session,
},
# --- Downloads ---
{
"name": "browser_wait_for_download",
"description": "Wait for a file download to complete.",
"parameters": _p({"timeout": _I, "context_id": _S, "proxy": _PROXY}),
"handler": browser_wait_for_download,
},
{
"name": "browser_get_downloads",
"description": "Get list of all captured downloads.",
"parameters": _p(
{
"context_id": _S,
"proxy": _PROXY,
"state": {**_S, "description": "Filter: completed|pending|failed."},
}
),
"handler": browser_get_downloads,
},
{
"name": "browser_save_download",
"description": "Get download info or rename a downloaded file.",
"parameters": _p({"download_id": _S, "filename": _S}, ["download_id"]),
"handler": browser_save_download,
},
# --- Console & Dialog ---
{
"name": "browser_get_console_logs",
"description": "Get captured console log messages from the browser.",
"parameters": _p(
{
"context_id": _S,
"proxy": _PROXY,
"level": {**_S, "description": "Filter: log|warn|error|info|debug."},
"limit": _I,
}
),
"handler": browser_get_console_logs,
},
{
"name": "browser_clear_console_logs",
"description": "Clear captured console logs.",
"parameters": _p({"context_id": _S, "proxy": _PROXY}),
"handler": browser_clear_console_logs,
},
{
"name": "browser_handle_dialog",
"description": "Configure how to handle JavaScript alerts, confirms, and prompts.",
"parameters": _p(
{
"action": {**_S, "description": "accept|dismiss|manual."},
"prompt_text": _S,
"context_id": _S,
"proxy": _PROXY,
},
["action"],
),
"handler": browser_handle_dialog,
},
{
"name": "browser_get_pending_dialog",
"description": "Get information about the last JavaScript dialog.",
"parameters": _p({"context_id": _S, "proxy": _PROXY}),
"handler": browser_get_pending_dialog,
},
]