feat: browser-use tool initial migration

This commit is contained in:
Shanyu Thibaut Juneja 2026-03-11 11:03:38 -07:00 committed by STJ
parent 38b2700553
commit b880ed454c
17 changed files with 2828 additions and 1415 deletions

View file

@ -33,7 +33,7 @@ RUN apt-get update && \
net-tools dnsutils whois \
jq parallel ripgrep grep \
less man-db procps htop \
iproute2 iputils-ping netcat-traditional \
iproute2 iputils-ping netcat-traditional socat \
nmap ncat ndiff \
sqlmap nuclei subfinder naabu ffuf \
nodejs npm pipx \
@ -215,7 +215,11 @@ RUN echo 'export PATH="/home/pentester/go/bin:/home/pentester/.local/bin:/home/p
USER root
COPY containers/docker-entrypoint.sh /usr/local/bin/docker-entrypoint.sh
RUN chmod +x /usr/local/bin/docker-entrypoint.sh
COPY containers/healthcheck.sh /usr/local/bin/healthcheck.sh
RUN chmod +x /usr/local/bin/docker-entrypoint.sh /usr/local/bin/healthcheck.sh
HEALTHCHECK --interval=15s --timeout=5s --start-period=60s --retries=3 \
CMD healthcheck.sh
USER pentester
WORKDIR /workspace

View file

@ -151,11 +151,161 @@ sudo -u pentester certutil -N -d sql:/home/pentester/.pki/nssdb --empty-password
sudo -u pentester certutil -A -n "Testing Root CA" -t "C,," -i /app/certs/ca.crt -d sql:/home/pentester/.pki/nssdb
echo "✅ CA added to browser trust store"
CDP_PORT="${BROWSER_CDP_PORT:-9222}"
# Chromium always binds CDP to 127.0.0.1 regardless of --remote-debugging-address.
# We launch it on an internal port and use socat to expose it on 0.0.0.0.
CDP_INTERNAL_PORT=19222
CHROMIUM_BIN=""
CHROMIUM_RESTART_COUNT=0
CHROMIUM_MAX_RESTARTS=10
echo "Launching Chromium with CDP (internal port $CDP_INTERNAL_PORT, exposed on $CDP_PORT)..."
CHROMIUM_BIN=$(find /usr/lib/chromium* /usr/bin -name "chromium" -o -name "chromium-browser" -o -name "chrome" 2>/dev/null | head -1)
if [ -z "$CHROMIUM_BIN" ]; then
# Playwright-installed Chromium
CHROMIUM_BIN=$(find /home/pentester/.cache/ms-playwright -name "chrome" -type f 2>/dev/null | head -1)
fi
# ---------------------------------------------------------------------------
# start_chromium: launches Chromium + socat, waits for CDP readiness.
# Sets CHROMIUM_PID and SOCAT_PID on success.
# ---------------------------------------------------------------------------
start_chromium() {
if [ -z "$CHROMIUM_BIN" ]; then
echo "WARNING: Chromium binary not found, browser CDP will not be available"
return 1
fi
# Clean up stale profile lock files from previous crashes
rm -f /tmp/chromium-profile/SingletonLock /tmp/chromium-profile/SingletonCookie /tmp/chromium-profile/SingletonSocket 2>/dev/null || true
sudo -u pentester "$CHROMIUM_BIN" \
--headless \
--no-sandbox \
--disable-dev-shm-usage \
--disable-gpu \
--remote-debugging-port="$CDP_INTERNAL_PORT" \
--proxy-server="http://127.0.0.1:${CAIDO_PORT}" \
--ignore-certificate-errors \
--user-data-dir=/tmp/chromium-profile \
> /tmp/chromium.log 2>&1 &
CHROMIUM_PID=$!
echo "Started Chromium with PID $CHROMIUM_PID"
echo "Waiting for Chromium CDP to be ready..."
local cdp_ready=false
for i in {1..20}; do
if ! kill -0 $CHROMIUM_PID 2>/dev/null; then
echo "WARNING: Chromium process died during startup (iteration $i)"
echo "=== Chromium log ==="
cat /tmp/chromium.log 2>/dev/null || echo "(no log)"
return 1
fi
if curl -s "http://127.0.0.1:${CDP_INTERNAL_PORT}/json/version" | grep -q "webSocketDebuggerUrl"; then
echo "✅ Chromium CDP ready on internal port $CDP_INTERNAL_PORT (attempt $i)"
cdp_ready=true
break
fi
sleep 1
done
if [ "$cdp_ready" = false ]; then
echo "WARNING: Chromium CDP did not become ready within 20s"
return 1
fi
# Kill any leftover socat from a previous run
if [ -n "${SOCAT_PID:-}" ] && kill -0 "$SOCAT_PID" 2>/dev/null; then
kill "$SOCAT_PID" 2>/dev/null || true
wait "$SOCAT_PID" 2>/dev/null || true
fi
# Expose CDP on 0.0.0.0 so Docker port mapping can reach it from the host.
socat TCP-LISTEN:${CDP_PORT},fork,reuseaddr,bind=0.0.0.0 TCP:127.0.0.1:${CDP_INTERNAL_PORT} &
SOCAT_PID=$!
echo "Started socat CDP forwarder (PID $SOCAT_PID): 0.0.0.0:${CDP_PORT} -> 127.0.0.1:${CDP_INTERNAL_PORT}"
# Verify the socat-forwarded CDP port is reachable
local fwd_ready=false
for i in 1 2 3 4 5; do
if curl -s "http://127.0.0.1:${CDP_PORT}/json/version" | grep -q "webSocketDebuggerUrl"; then
echo "✅ CDP forwarded and reachable on port ${CDP_PORT} (attempt $i)"
fwd_ready=true
break
fi
sleep 1
done
if [ "$fwd_ready" = false ]; then
echo "WARNING: CDP not reachable via socat on port ${CDP_PORT}"
echo " socat PID $SOCAT_PID alive: $(kill -0 $SOCAT_PID 2>&1 && echo yes || echo no)"
echo " ss output: $(ss -tlnp 2>/dev/null | grep ${CDP_PORT} || echo 'port not listening')"
return 1
fi
return 0
}
# Initial Chromium launch
if ! start_chromium; then
echo "WARNING: Initial Chromium launch failed — browser_use_local may not work"
fi
# ---------------------------------------------------------------------------
# Chromium watchdog: runs in the background, checks every 10s, restarts on
# crash. Stops after CHROMIUM_MAX_RESTARTS consecutive failures.
# ---------------------------------------------------------------------------
chromium_watchdog() {
sleep 15 # let everything settle before first check
local consecutive_failures=0
while true; do
sleep 10
# If we don't have a Chromium binary, nothing to watch
[ -z "$CHROMIUM_BIN" ] && return
# Check if Chromium is still alive
if [ -n "${CHROMIUM_PID:-}" ] && kill -0 "$CHROMIUM_PID" 2>/dev/null; then
# Process alive — also verify CDP is actually responding
if curl -sf --max-time 3 "http://127.0.0.1:${CDP_INTERNAL_PORT}/json/version" | grep -q "webSocketDebuggerUrl"; then
consecutive_failures=0
continue
fi
echo "WATCHDOG: Chromium PID $CHROMIUM_PID alive but CDP not responding, killing..."
kill "$CHROMIUM_PID" 2>/dev/null || true
wait "$CHROMIUM_PID" 2>/dev/null || true
fi
CHROMIUM_RESTART_COUNT=$((CHROMIUM_RESTART_COUNT + 1))
consecutive_failures=$((consecutive_failures + 1))
if [ $consecutive_failures -gt $CHROMIUM_MAX_RESTARTS ]; then
echo "WATCHDOG: Exceeded $CHROMIUM_MAX_RESTARTS consecutive restart failures, giving up"
return
fi
echo "WATCHDOG: Chromium died — restarting (attempt $CHROMIUM_RESTART_COUNT, consecutive=$consecutive_failures)..."
if start_chromium; then
echo "WATCHDOG: Chromium restarted successfully (PID $CHROMIUM_PID)"
consecutive_failures=0
else
echo "WATCHDOG: Chromium restart failed"
fi
done
}
chromium_watchdog &
WATCHDOG_PID=$!
echo "Started Chromium watchdog with PID $WATCHDOG_PID"
echo "Starting tool server..."
cd /app
export PYTHONPATH=/app
export STRIX_SANDBOX_MODE=true
export TOOL_SERVER_TIMEOUT="${STRIX_SANDBOX_EXECUTION_TIMEOUT:-120}"
export CDP_INTERNAL_PORT
export CDP_PORT
TOOL_SERVER_LOG="/tmp/tool_server.log"
sudo -E -u pentester \
@ -165,6 +315,8 @@ sudo -E -u pentester \
--port="$TOOL_SERVER_PORT" \
--timeout="$TOOL_SERVER_TIMEOUT" > "$TOOL_SERVER_LOG" 2>&1 &
TOOL_SERVER_PID=$!
for i in {1..10}; do
if curl -s "http://127.0.0.1:$TOOL_SERVER_PORT/health" | grep -q '"status":"healthy"'; then
echo "✅ Tool server healthy on port $TOOL_SERVER_PORT"
@ -181,5 +333,22 @@ done
echo "✅ Container ready"
# ---------------------------------------------------------------------------
# Instead of exec (which orphans background processes), run the user command
# and then wait so that the entrypoint stays alive as PID 1 to reap children.
# ---------------------------------------------------------------------------
cd /workspace
exec "$@"
if [ $# -gt 0 ]; then
"$@" &
CMD_PID=$!
echo "Started user command (PID $CMD_PID): $*"
# Forward SIGTERM/SIGINT to all children for graceful shutdown
trap 'echo "Shutting down..."; kill $CMD_PID $TOOL_SERVER_PID $WATCHDOG_PID ${CHROMIUM_PID:-0} ${SOCAT_PID:-0} ${CAIDO_PID:-0} 2>/dev/null; wait' SIGTERM SIGINT
wait $CMD_PID
else
# No command: just keep running (wait on tool server)
trap 'echo "Shutting down..."; kill $TOOL_SERVER_PID $WATCHDOG_PID ${CHROMIUM_PID:-0} ${SOCAT_PID:-0} ${CAIDO_PID:-0} 2>/dev/null; wait' SIGTERM SIGINT
wait $TOOL_SERVER_PID
fi

31
containers/healthcheck.sh Executable file
View file

@ -0,0 +1,31 @@
#!/bin/bash
# Healthcheck script for the strix sandbox container.
# Checks: tool server, Caido proxy, Chromium CDP (via socat).
# Exit 0 = healthy, exit 1 = unhealthy.
set -e
TOOL_SERVER_PORT="${TOOL_SERVER_PORT:-48081}"
CAIDO_PORT=48080
CDP_PORT="${BROWSER_CDP_PORT:-9222}"
# 1. Tool server must respond healthy
if ! curl -sf --max-time 3 "http://127.0.0.1:${TOOL_SERVER_PORT}/health" | grep -q '"status":"healthy"'; then
echo "UNHEALTHY: tool server not responding on port ${TOOL_SERVER_PORT}"
exit 1
fi
# 2. Caido proxy must be reachable
if ! curl -sf --max-time 3 -o /dev/null "http://127.0.0.1:${CAIDO_PORT}/graphql/"; then
echo "UNHEALTHY: Caido proxy not responding on port ${CAIDO_PORT}"
exit 1
fi
# 3. Chromium CDP must be reachable (via socat forwarder)
if ! curl -sf --max-time 3 "http://127.0.0.1:${CDP_PORT}/json/version" | grep -q "webSocketDebuggerUrl"; then
echo "UNHEALTHY: Chromium CDP not responding on port ${CDP_PORT}"
exit 1
fi
echo "healthy"
exit 0

View file

@ -117,30 +117,38 @@ pretty = true
# Allow some flexibility for third-party libraries
[[tool.mypy.overrides]]
module = [
"litellm.*",
"tenacity.*",
"numpydoc.*",
"rich.*",
"IPython.*",
"openhands_aci.*",
"playwright.*",
"uvicorn.*",
"jinja2.*",
"pydantic_settings.*",
"jwt.*",
"httpx.*",
"gql.*",
"textual.*",
"pyte.*",
"libtmux.*",
"pytest.*",
"cvss.*",
"opentelemetry.*",
"scrubadub.*",
"traceloop.*",
"litellm.*",
"tenacity.*",
"numpydoc.*",
"rich.*",
"IPython.*",
"openhands_aci.*",
"playwright.*",
"uvicorn.*",
"jinja2.*",
"pydantic_settings.*",
"jwt.*",
"httpx.*",
"gql.*",
"textual.*",
"pyte.*",
"libtmux.*",
"pytest.*",
"cvss.*",
"opentelemetry.*",
"scrubadub.*",
"traceloop.*",
"browser_use",
"browser_use.*",
"cdp_use",
"cdp_use.*",
]
ignore_missing_imports = true
[[tool.mypy.overrides]]
module = ["strix.tools.browser.*"]
disallow_subclassing_any = false
# Relax strict rules for test files (pytest decorators are not fully typed)
[[tool.mypy.overrides]]
module = ["tests.*"]
@ -155,90 +163,101 @@ disallow_untyped_defs = false
target-version = "py312"
line-length = 100
extend-exclude = [
".git",
".mypy_cache",
".pytest_cache",
".ruff_cache",
"__pycache__",
"build",
"dist",
"migrations",
".git",
".mypy_cache",
".pytest_cache",
".ruff_cache",
"__pycache__",
"build",
"dist",
"migrations",
]
[tool.ruff.lint]
# Enable comprehensive rule sets
select = [
"E", # pycodestyle errors
"W", # pycodestyle warnings
"F", # Pyflakes
"I", # isort
"N", # pep8-naming
"UP", # pyupgrade
"YTT", # flake8-2020
"S", # flake8-bandit
"BLE", # flake8-blind-except
"FBT", # flake8-boolean-trap
"B", # flake8-bugbear
"A", # flake8-builtins
"COM", # flake8-commas
"C4", # flake8-comprehensions
"DTZ", # flake8-datetimez
"T10", # flake8-debugger
"EM", # flake8-errmsg
"FA", # flake8-future-annotations
"ISC", # flake8-implicit-str-concat
"ICN", # flake8-import-conventions
"G", # flake8-logging-format
"INP", # flake8-no-pep420
"PIE", # flake8-pie
"T20", # flake8-print
"PYI", # flake8-pyi
"PT", # flake8-pytest-style
"Q", # flake8-quotes
"RSE", # flake8-raise
"RET", # flake8-return
"SLF", # flake8-self
"SIM", # flake8-simplify
"TID", # flake8-tidy-imports
"TCH", # flake8-type-checking
"ARG", # flake8-unused-arguments
"PTH", # flake8-use-pathlib
"ERA", # eradicate
"PD", # pandas-vet
"PGH", # pygrep-hooks
"PL", # Pylint
"TRY", # tryceratops
"FLY", # flynt
"PERF", # Perflint
"RUF", # Ruff-specific rules
"E", # pycodestyle errors
"W", # pycodestyle warnings
"F", # Pyflakes
"I", # isort
"N", # pep8-naming
"UP", # pyupgrade
"YTT", # flake8-2020
"S", # flake8-bandit
"BLE", # flake8-blind-except
"FBT", # flake8-boolean-trap
"B", # flake8-bugbear
"A", # flake8-builtins
"COM", # flake8-commas
"C4", # flake8-comprehensions
"DTZ", # flake8-datetimez
"T10", # flake8-debugger
"EM", # flake8-errmsg
"FA", # flake8-future-annotations
"ISC", # flake8-implicit-str-concat
"ICN", # flake8-import-conventions
"G", # flake8-logging-format
"INP", # flake8-no-pep420
"PIE", # flake8-pie
"T20", # flake8-print
"PYI", # flake8-pyi
"PT", # flake8-pytest-style
"Q", # flake8-quotes
"RSE", # flake8-raise
"RET", # flake8-return
"SLF", # flake8-self
"SIM", # flake8-simplify
"TID", # flake8-tidy-imports
"TCH", # flake8-type-checking
"ARG", # flake8-unused-arguments
"PTH", # flake8-use-pathlib
"ERA", # eradicate
"PD", # pandas-vet
"PGH", # pygrep-hooks
"PL", # Pylint
"TRY", # tryceratops
"FLY", # flynt
"PERF", # Perflint
"RUF", # Ruff-specific rules
]
ignore = [
"S101", # Use of assert
"S104", # Possible binding to all interfaces
"S301", # Use of pickle
"COM812", # Missing trailing comma (handled by formatter)
"ISC001", # Single line implicit string concatenation (handled by formatter)
"PLR0913", # Too many arguments to function call
"TRY003", # Avoid specifying long messages outside the exception class
"EM101", # Exception must not use a string literal
"EM102", # Exception must not use an f-string literal
"FBT001", # Boolean positional arg in function definition
"FBT002", # Boolean default positional argument in function definition
"G004", # Logging statement uses f-string
"PLR2004", # Magic value used in comparison
"SLF001", # Private member accessed
"S101", # Use of assert
"S104", # Possible binding to all interfaces
"S301", # Use of pickle
"COM812", # Missing trailing comma (handled by formatter)
"ISC001", # Single line implicit string concatenation (handled by formatter)
"PLR0913", # Too many arguments to function call
"TRY003", # Avoid specifying long messages outside the exception class
"EM101", # Exception must not use a string literal
"EM102", # Exception must not use an f-string literal
"FBT001", # Boolean positional arg in function definition
"FBT002", # Boolean default positional argument in function definition
"G004", # Logging statement uses f-string
"PLR2004", # Magic value used in comparison
"SLF001", # Private member accessed
]
[tool.ruff.lint.per-file-ignores]
"tests/**/*.py" = [
"S106", # Possible hardcoded password
"S108", # Possible insecure usage of temporary file/directory
"ARG001", # Unused function argument
"PLR2004", # Magic value used in comparison
"S106", # Possible hardcoded password
"S108", # Possible insecure usage of temporary file/directory
"ARG001", # Unused function argument
"FBT003", # Boolean positional value in function call
"PLR2004", # Magic value used in comparison
]
"strix/tools/**/*.py" = [
"ARG001", # Unused function argument (tools may have unused args for interface consistency)
"ARG001", # Unused function argument (tools may have unused args for interface consistency)
]
"strix/tools/browser/**/*.py" = [
"ARG002", # Unused method argument (interface methods may not use all args)
"PLR0911", # Too many return statements (dispatchers and complex browser logic)
"PLR0912", # Too many branches (dispatch functions)
"PLR0915", # Too many statements (complex browser task handling)
]
"strix/telemetry/tracer.py" = [
"PLR0912", # Too many branches (save_run_data is legitimately complex)
"PLR0915", # Too many statements
]
[tool.ruff.lint.isort]
@ -330,12 +349,12 @@ known_third_party = ["fastapi", "pydantic", "litellm", "tenacity"]
[tool.pytest.ini_options]
minversion = "6.0"
addopts = [
"--strict-markers",
"--strict-config",
"--cov=strix",
"--cov-report=term-missing",
"--cov-report=html",
"--cov-report=xml",
"--strict-markers",
"--strict-config",
"--cov=strix",
"--cov-report=term-missing",
"--cov-report=html",
"--cov-report=xml",
]
testpaths = ["tests"]
python_files = ["test_*.py", "*_test.py"]
@ -345,24 +364,20 @@ asyncio_mode = "auto"
[tool.coverage.run]
source = ["strix"]
omit = [
"*/tests/*",
"*/migrations/*",
"*/__pycache__/*"
]
omit = ["*/tests/*", "*/migrations/*", "*/__pycache__/*"]
[tool.coverage.report]
exclude_lines = [
"pragma: no cover",
"def __repr__",
"if self.debug:",
"if settings.DEBUG",
"raise AssertionError",
"raise NotImplementedError",
"if 0:",
"if __name__ == .__main__.:",
"class .*\\bProtocol\\):",
"@(abc\\.)?abstractmethod",
"pragma: no cover",
"def __repr__",
"if self.debug:",
"if settings.DEBUG",
"raise AssertionError",
"raise NotImplementedError",
"if 0:",
"if __name__ == .__main__.:",
"class .*\\bProtocol\\):",
"@(abc\\.)?abstractmethod",
]
# ============================================================================
@ -371,5 +386,11 @@ exclude_lines = [
[tool.bandit]
exclude_dirs = ["tests", "docs", "build", "dist"]
skips = ["B101", "B601", "B404", "B603", "B607"] # Skip assert, shell injection, subprocess import and partial path checks
skips = [
"B101",
"B601",
"B404",
"B603",
"B607",
] # Skip assert, shell injection, subprocess import and partial path checks
severity = "medium"

View file

@ -23,6 +23,7 @@ HOST_GATEWAY_HOSTNAME = "host.docker.internal"
DOCKER_TIMEOUT = 60
CONTAINER_TOOL_SERVER_PORT = 48081
CONTAINER_CAIDO_PORT = 48080
CONTAINER_BROWSER_CDP_PORT = 9222
class DockerRuntime(AbstractRuntime):
@ -39,6 +40,7 @@ class DockerRuntime(AbstractRuntime):
self._tool_server_port: int | None = None
self._tool_server_token: str | None = None
self._caido_port: int | None = None
self._browser_cdp_port: int | None = None
def _find_available_port(self) -> int:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
@ -84,6 +86,10 @@ class DockerRuntime(AbstractRuntime):
if port_bindings.get(caido_port_key):
self._caido_port = int(port_bindings[caido_port_key][0]["HostPort"])
cdp_port_key = f"{CONTAINER_BROWSER_CDP_PORT}/tcp"
if port_bindings.get(cdp_port_key):
self._browser_cdp_port = int(port_bindings[cdp_port_key][0]["HostPort"])
def _wait_for_tool_server(self, max_retries: int = 30, timeout: int = 5) -> None:
host = self._resolve_docker_host()
health_url = f"http://{host}:{self._tool_server_port}/health"
@ -128,6 +134,7 @@ class DockerRuntime(AbstractRuntime):
self._tool_server_port = self._find_available_port()
self._caido_port = self._find_available_port()
self._browser_cdp_port = self._find_available_port()
self._tool_server_token = secrets.token_urlsafe(32)
execution_timeout = Config.get("strix_sandbox_execution_timeout") or "120"
@ -140,8 +147,10 @@ class DockerRuntime(AbstractRuntime):
ports={
f"{CONTAINER_TOOL_SERVER_PORT}/tcp": self._tool_server_port,
f"{CONTAINER_CAIDO_PORT}/tcp": self._caido_port,
f"{CONTAINER_BROWSER_CDP_PORT}/tcp": self._browser_cdp_port,
},
cap_add=["NET_ADMIN", "NET_RAW"],
shm_size="256m",
labels={"strix-scan-id": scan_id},
environment={
"PYTHONUNBUFFERED": "1",
@ -149,6 +158,7 @@ class DockerRuntime(AbstractRuntime):
"TOOL_SERVER_TOKEN": self._tool_server_token,
"STRIX_SANDBOX_EXECUTION_TIMEOUT": str(execution_timeout),
"HOST_GATEWAY": HOST_GATEWAY_HOSTNAME,
"BROWSER_CDP_PORT": str(CONTAINER_BROWSER_CDP_PORT),
},
extra_hosts={HOST_GATEWAY_HOSTNAME: "host-gateway"},
tty=True,
@ -163,6 +173,7 @@ class DockerRuntime(AbstractRuntime):
self._tool_server_port = None
self._tool_server_token = None
self._caido_port = None
self._browser_cdp_port = None
time.sleep(2**attempt)
else:
return container
@ -185,6 +196,7 @@ class DockerRuntime(AbstractRuntime):
self._tool_server_port = None
self._tool_server_token = None
self._caido_port = None
self._browser_cdp_port = None
try:
container = self.client.containers.get(container_name)
@ -272,9 +284,15 @@ class DockerRuntime(AbstractRuntime):
raise RuntimeError("Docker container ID is unexpectedly None")
token = existing_token or self._tool_server_token
if self._tool_server_port is None or self._caido_port is None or token is None:
if self._tool_server_port is None or token is None:
raise RuntimeError("Tool server not initialized")
if self._caido_port is None:
raise RuntimeError("Caido port not initialized")
if self._browser_cdp_port is None:
raise RuntimeError("Browser CDP port not initialized")
host = self._resolve_docker_host()
api_url = f"http://{host}:{self._tool_server_port}"
@ -286,6 +304,7 @@ class DockerRuntime(AbstractRuntime):
"auth_token": token,
"tool_server_port": self._tool_server_port,
"caido_port": self._caido_port,
"browser_cdp_port": self._browser_cdp_port,
"agent_id": agent_id,
}
@ -328,6 +347,7 @@ class DockerRuntime(AbstractRuntime):
self._tool_server_port = None
self._tool_server_token = None
self._caido_port = None
self._browser_cdp_port = None
except (NotFound, DockerException):
pass
@ -338,6 +358,7 @@ class DockerRuntime(AbstractRuntime):
self._tool_server_port = None
self._tool_server_token = None
self._caido_port = None
self._browser_cdp_port = None
if container_name is None:
return

View file

@ -8,6 +8,7 @@ class SandboxInfo(TypedDict):
auth_token: str | None
tool_server_port: int
caido_port: int
browser_cdp_port: int
agent_id: str

View file

@ -135,8 +135,30 @@ async def register_agent(
return {"status": "registered", "agent_id": agent_id}
async def _check_cdp_health() -> dict[str, Any]:
"""Probe the container-local Chromium CDP endpoint."""
cdp_port = os.getenv("CDP_INTERNAL_PORT", "19222")
cdp_url = f"http://127.0.0.1:{cdp_port}/json/version"
try:
import httpx
async with httpx.AsyncClient(timeout=3, trust_env=False) as client:
resp = await client.get(cdp_url)
if resp.status_code == 200:
data = resp.json()
return {
"status": "healthy",
"browser": data.get("Browser", "unknown"),
"ws_url": data.get("webSocketDebuggerUrl", ""),
}
return {"status": "unhealthy", "detail": f"HTTP {resp.status_code}"}
except Exception as exc: # noqa: BLE001
return {"status": "unhealthy", "detail": str(exc)}
@app.get("/health")
async def health_check() -> dict[str, Any]:
cdp_health = await _check_cdp_health()
return {
"status": "healthy",
"sandbox_mode": str(SANDBOX_MODE),
@ -144,6 +166,7 @@ async def health_check() -> dict[str, Any]:
"auth_configured": "true" if EXPECTED_TOKEN else "false",
"active_agents": len(agent_tasks),
"agents": list(agent_tasks.keys()),
"chromium_cdp": cdp_health,
}

View file

@ -31,6 +31,38 @@ from .todo import * # noqa: F403
from .web_search import * # noqa: F403
SANDBOX_MODE = os.getenv("STRIX_SANDBOX_MODE", "false").lower() == "true"
HAS_PERPLEXITY_API = bool(Config.get("perplexity_api_key"))
DISABLE_BROWSER = (Config.get("strix_disable_browser") or "false").lower() == "true"
if not SANDBOX_MODE:
from .agents_graph import * # noqa: F403
if not DISABLE_BROWSER:
from .browser import * # noqa: F403
from .file_edit import * # noqa: F403
from .finish import * # noqa: F403
from .notes import * # noqa: F403
from .proxy import * # noqa: F403
from .python import * # noqa: F403
from .reporting import * # noqa: F403
from .terminal import * # noqa: F403
from .thinking import * # noqa: F403
from .todo import * # noqa: F403
if HAS_PERPLEXITY_API:
from .web_search import * # noqa: F403
else:
from .file_edit import * # noqa: F403
from .proxy import * # noqa: F403
from .python import * # noqa: F403
from .terminal import * # noqa: F403
if not DISABLE_BROWSER:
from .browser import * # noqa: F403
__all__ = [
"ImplementedInClientSideOnlyError",
"execute_tool",

View file

@ -1,4 +1,5 @@
from .browser_actions import browser_action
from .browser_actions import browser_actions
from .browser_manager import cleanup_agent
__all__ = ["browser_action"]
__all__ = ["browser_actions", "cleanup_agent"]

View file

@ -1,240 +1,598 @@
from typing import TYPE_CHECKING, Any, Literal, NoReturn
import asyncio
import contextlib
import logging
import os
import time
from typing import Any, Literal
from strix.tools.registry import register_tool
if TYPE_CHECKING:
from .tab_manager import BrowserTabManager
from .browser_manager import (
_BrowserSession,
_close_session,
_ensure_healthy_session,
_get_session,
_launch_browser,
_launch_local_browser,
_reinitialize_after_agent,
)
BrowserAction = Literal[
logger = logging.getLogger(__name__)
BrowserUseLocalAction = Literal[
"launch",
"goto",
"run",
"close",
# Granular browser commands
"open",
"click",
"type",
"scroll_down",
"scroll_up",
"input",
"scroll",
"back",
"forward",
"new_tab",
"switch_tab",
"screenshot",
"state",
"switch",
"close_tab",
"wait",
"execute_js",
"double_click",
"keys",
"select",
"eval",
"extract",
"hover",
"press_key",
"save_pdf",
"get_console_logs",
"view_source",
"close",
"list_tabs",
"dblclick",
"rightclick",
"cookies",
"wait",
"get",
]
# Actions that are handled by browser_commands (granular CDP control).
_GRANULAR_ACTIONS = frozenset(
{
"open",
"click",
"type",
"input",
"scroll",
"back",
"screenshot",
"state",
"switch",
"close_tab",
"keys",
"select",
"eval",
"extract",
"hover",
"dblclick",
"rightclick",
"cookies",
"wait",
"get",
}
)
def _validate_url(action_name: str, url: str | None) -> None:
if not url:
raise ValueError(f"url parameter is required for {action_name} action")
# Hard timeout for a single browser task (seconds).
_TASK_TIMEOUT = 300
def _validate_coordinate(action_name: str, coordinate: str | None) -> None:
if not coordinate:
raise ValueError(f"coordinate parameter is required for {action_name} action")
def _build_llm() -> Any:
"""Build a browser-use compatible LLM from the strix LLM config.
Returns a ``ChatLiteLLM`` instance that routes to any provider via litellm.
"""
from strix.config.config import resolve_llm_config
from .llm import ChatLiteLLM
model, api_key, api_base = resolve_llm_config()
if not model:
raise ValueError("STRIX_LLM environment variable must be set")
return ChatLiteLLM(
model=model,
api_key=api_key or None,
api_base=api_base or None,
)
def _validate_text(action_name: str, text: str | None) -> None:
if not text:
raise ValueError(f"text parameter is required for {action_name} action")
# ---------------------------------------------------------------------------
# Task execution helpers
# ---------------------------------------------------------------------------
def _validate_tab_id(action_name: str, tab_id: str | None) -> None:
if not tab_id:
raise ValueError(f"tab_id parameter is required for {action_name} action")
def _is_websocket_error(exc: BaseException) -> bool:
"""Return True if the exception looks like a WebSocket / CDP disconnect."""
# Check exception type hierarchy (websockets lib, ConnectionError, etc.)
exc_type = type(exc).__name__.lower()
if any(kw in exc_type for kw in ("connectionclosed", "websocket", "connectionerror")):
return True
msg = str(exc).lower()
ws_keywords = (
"websocket",
"cdp",
"not initialized",
"not connected",
"connection closed",
"disconnected",
"broken pipe",
"connection reset",
"eof occurred",
"protocol error",
"target closed",
"session closed",
"page closed",
"browser has been closed",
"browser was closed",
"client is stopping",
"no close frame",
"close frame",
"connection was closed",
"invalid state",
)
return any(kw in msg for kw in ws_keywords)
def _validate_js_code(action_name: str, js_code: str | None) -> None:
if not js_code:
raise ValueError(f"js_code parameter is required for {action_name} action")
async def _cleanup_agent(agent: Any) -> None:
"""Best-effort cleanup of a browser-use Agent after failure.
browser-use's Agent may leave dangling asyncio tasks (CDP listeners,
DOM observers) that produce "Future exception was never retrieved"
warnings when the WebSocket is already dead. This drains them.
"""
# Agent may expose a close/stop/cleanup method.
for method_name in ("close", "stop", "cleanup"):
fn = getattr(agent, method_name, None)
if not callable(fn):
continue
try:
coro = fn()
if asyncio.iscoroutine(coro):
await asyncio.wait_for(coro, timeout=5)
except Exception: # noqa: BLE001
logger.debug("Agent %s() cleanup failed", method_name, exc_info=True)
else:
return
# Fallback: try to close the browser context the agent was using,
# which cancels its internal CDP subscriptions.
browser_ctx = getattr(agent, "browser_context", None)
if browser_ctx is not None:
close_fn = getattr(browser_ctx, "close", None)
if callable(close_fn):
with contextlib.suppress(Exception):
coro = close_fn()
if asyncio.iscoroutine(coro):
await asyncio.wait_for(coro, timeout=5)
def _validate_duration(action_name: str, duration: float | None) -> None:
if duration is None:
raise ValueError(f"duration parameter is required for {action_name} action")
async def _run_agent_task(task: str, session: _BrowserSession) -> dict[str, Any]:
"""Run a browser-use task on the session's browser with resilient reconnection.
Lifecycle:
1. Ensure the session is healthy (CDP alive, Browser fresh).
2. Run the task.
3. On WebSocket/CDP errors, invalidate the session and retry once.
4. On any terminal failure, mark the session so the next call refreshes.
"""
from browser_use import Agent
def _validate_key(action_name: str, key: str | None) -> None:
if not key:
raise ValueError(f"key parameter is required for {action_name} action")
session.task_count += 1
task_num = session.task_count
logger.info(
"Task #%d for session (cdp=%s, ws=%s, age=%.1fs, failures=%d): %.200s",
task_num,
session.cdp_url,
session.ws_url,
session.age,
session.consecutive_failures,
task,
)
def _validate_file_path(action_name: str, file_path: str | None) -> None:
if not file_path:
raise ValueError(f"file_path parameter is required for {action_name} action")
# --- Pre-flight health check (with recovery) ---
preflight_err = await _ensure_healthy_session(session, task_num)
if preflight_err:
session.consecutive_failures += 1
return {"error": preflight_err, "is_running": False}
# --- Execute (with one automatic retry on WebSocket disconnect) ---
max_attempts = 2
last_error: str = ""
def _handle_navigation_actions(
manager: "BrowserTabManager",
action: str,
url: str | None = None,
tab_id: str | None = None,
) -> dict[str, Any]:
if action == "launch":
return manager.launch_browser(url)
if action == "goto":
_validate_url(action, url)
assert url is not None
return manager.goto_url(url, tab_id)
if action == "back":
return manager.back(tab_id)
if action == "forward":
return manager.forward(tab_id)
raise ValueError(f"Unknown navigation action: {action}")
for attempt in range(1, max_attempts + 1):
llm = _build_llm()
agent = Agent(task=task, llm=llm, browser=session.browser, flash_mode=True)
def _handle_interaction_actions(
manager: "BrowserTabManager",
action: str,
coordinate: str | None = None,
text: str | None = None,
key: str | None = None,
tab_id: str | None = None,
) -> dict[str, Any]:
if action in {"click", "double_click", "hover"}:
_validate_coordinate(action, coordinate)
assert coordinate is not None
action_map = {
"click": manager.click,
"double_click": manager.double_click,
"hover": manager.hover,
}
return action_map[action](coordinate, tab_id)
if action in {"scroll_down", "scroll_up"}:
direction = "down" if action == "scroll_down" else "up"
return manager.scroll(direction, tab_id)
if action == "type":
_validate_text(action, text)
assert text is not None
return manager.type_text(text, tab_id)
if action == "press_key":
_validate_key(action, key)
assert key is not None
return manager.press_key(key, tab_id)
raise ValueError(f"Unknown interaction action: {action}")
def _raise_unknown_action(action: str) -> NoReturn:
raise ValueError(f"Unknown action: {action}")
def _handle_tab_actions(
manager: "BrowserTabManager",
action: str,
url: str | None = None,
tab_id: str | None = None,
) -> dict[str, Any]:
if action == "new_tab":
return manager.new_tab(url)
if action == "switch_tab":
_validate_tab_id(action, tab_id)
assert tab_id is not None
return manager.switch_tab(tab_id)
if action == "close_tab":
_validate_tab_id(action, tab_id)
assert tab_id is not None
return manager.close_tab(tab_id)
if action == "list_tabs":
return manager.list_tabs()
raise ValueError(f"Unknown tab action: {action}")
def _handle_utility_actions(
manager: "BrowserTabManager",
action: str,
duration: float | None = None,
js_code: str | None = None,
file_path: str | None = None,
tab_id: str | None = None,
clear: bool = False,
) -> dict[str, Any]:
if action == "wait":
_validate_duration(action, duration)
assert duration is not None
return manager.wait_browser(duration, tab_id)
if action == "execute_js":
_validate_js_code(action, js_code)
assert js_code is not None
return manager.execute_js(js_code, tab_id)
if action == "save_pdf":
_validate_file_path(action, file_path)
assert file_path is not None
return manager.save_pdf(file_path, tab_id)
if action == "get_console_logs":
return manager.get_console_logs(tab_id, clear)
if action == "view_source":
return manager.view_source(tab_id)
if action == "close":
return manager.close_browser()
raise ValueError(f"Unknown utility action: {action}")
@register_tool(requires_browser_mode=True)
def browser_action(
action: BrowserAction,
url: str | None = None,
coordinate: str | None = None,
text: str | None = None,
tab_id: str | None = None,
js_code: str | None = None,
duration: float | None = None,
key: str | None = None,
file_path: str | None = None,
clear: bool = False,
) -> dict[str, Any]:
from .tab_manager import get_browser_tab_manager
manager = get_browser_tab_manager()
try:
navigation_actions = {"launch", "goto", "back", "forward"}
interaction_actions = {
"click",
"type",
"double_click",
"hover",
"press_key",
"scroll_down",
"scroll_up",
}
tab_actions = {"new_tab", "switch_tab", "close_tab", "list_tabs"}
utility_actions = {
"wait",
"execute_js",
"save_pdf",
"get_console_logs",
"view_source",
"close",
}
if action in navigation_actions:
return _handle_navigation_actions(manager, action, url, tab_id)
if action in interaction_actions:
return _handle_interaction_actions(manager, action, coordinate, text, key, tab_id)
if action in tab_actions:
return _handle_tab_actions(manager, action, url, tab_id)
if action in utility_actions:
return _handle_utility_actions(
manager, action, duration, js_code, file_path, tab_id, clear
try:
result = await asyncio.wait_for(agent.run(), timeout=_TASK_TIMEOUT)
except TimeoutError:
logger.exception(
"Task #%d timed out after %ds (attempt %d)",
task_num,
_TASK_TIMEOUT,
attempt,
)
await _cleanup_agent(agent)
# Timeout leaves the Browser in an unknown state — invalidate.
session.invalidated = True
session.consecutive_failures += 1
return {
"error": f"Browser task timed out after {_TASK_TIMEOUT}s",
"is_running": False,
}
except Exception as exc:
logger.exception(
"Task #%d raised exception (attempt %d)",
task_num,
attempt,
)
await _cleanup_agent(agent)
last_error = str(exc)
_raise_unknown_action(action)
if _is_websocket_error(exc) and attempt < max_attempts:
logger.warning(
"Task #%d: WebSocket/CDP error — refreshing and retrying...",
task_num,
)
# Force-refresh the session and retry the task.
session.invalidated = True
refresh_err = await _ensure_healthy_session(session, task_num)
if refresh_err:
session.consecutive_failures += 1
return {"error": refresh_err, "is_running": False}
continue # retry
except (ValueError, RuntimeError) as e:
# Non-retryable error or final attempt.
session.consecutive_failures += 1
# If it smells like a connection issue, invalidate for next call.
if _is_websocket_error(exc):
session.invalidated = True
return {"error": f"Browser task failed: {exc}", "is_running": False}
# --- Success path ---
interpreted = _interpret_agent_result(result)
if "error" in interpreted:
error_msg = str(interpreted["error"])
logger.warning("Task #%d failed: %s", task_num, error_msg)
# If the agent reported a CDP/WS error, retry once.
if _is_websocket_error(Exception(error_msg)) and attempt < max_attempts:
logger.warning(
"Task #%d: agent result contains WebSocket error — refreshing and retrying...",
task_num,
)
session.invalidated = True
refresh_err = await _ensure_healthy_session(session, task_num)
if refresh_err:
session.consecutive_failures += 1
return {"error": refresh_err, "is_running": False}
continue # retry
session.consecutive_failures += 1
if _is_websocket_error(Exception(error_msg)):
session.invalidated = True
return interpreted
# Task succeeded — reset failure counter.
session.consecutive_failures = 0
result_preview = str(interpreted.get("result", ""))[:200]
logger.info("Task #%d succeeded: %s", task_num, result_preview)
return interpreted
# Should not reach here, but just in case:
session.consecutive_failures += 1
return {
"error": f"Browser task failed after {max_attempts} attempts: {last_error}",
"is_running": False,
}
def _interpret_agent_result(result: Any) -> dict[str, Any]:
"""Inspect an ``AgentHistoryList`` and return a success or error dict.
If the browser-use agent encountered errors (CDP failures, navigation
errors, etc.) this returns an ``{"error": ...}`` dict so the calling
strix agent treats the task as failed rather than succeeded.
"""
# Collect errors reported by the browser-use agent.
errors: list[str] = []
try:
if hasattr(result, "has_errors") and result.has_errors():
errors = [e for e in result.errors() if e is not None]
except Exception as exc: # noqa: BLE001
logger.debug("Error checking agent errors: %s", exc)
# Determine whether the agent considers the task successful.
success: bool | None = None
try:
if hasattr(result, "is_successful"):
success = result.is_successful()
except Exception as exc: # noqa: BLE001
logger.debug("Error checking agent success: %s", exc)
# Extract the final content the agent produced.
final_result: str | None = None
try:
if hasattr(result, "final_result"):
final_result = result.final_result()
except Exception as exc: # noqa: BLE001
logger.debug("Error extracting final_result: %s", exc)
logger.debug(
"Agent result interpretation: errors=%d, success=%s, has_final_result=%s, result_type=%s",
len(errors),
success,
final_result is not None,
type(result).__name__,
)
# If there are errors AND the agent didn't explicitly mark itself as
# successful, treat as failure and surface the errors.
if errors and success is not True:
error_summary = "; ".join(errors)
return {"error": error_summary, "is_running": False}
# If the agent explicitly reported failure (is_done=True, success=False).
if success is False:
return {
"error": str(e),
"tab_id": tab_id,
"screenshot": "",
"error": final_result or "Browser task failed (agent reported failure)",
"is_running": False,
}
return {
"message": "Task completed",
"result": final_result or str(result),
"is_running": False,
}
# ---------------------------------------------------------------------------
# CDP URL resolution
# ---------------------------------------------------------------------------
def _resolve_cdp_url(agent_state: Any) -> str:
"""Extract the CDP URL from agent_state's sandbox_info.
The sandbox container runs Chromium with ``--remote-debugging-port``
and the Docker runtime maps it to an ephemeral host port stored in
``sandbox_info["browser_cdp_port"]``.
"""
if not hasattr(agent_state, "sandbox_info") or not agent_state.sandbox_info:
raise ValueError(
"agent_state must have sandbox_info with browser_cdp_port. "
"Ensure the sandbox is initialized before launching the browser. "
f"agent_state type={type(agent_state).__name__}, "
f"has sandbox_info={hasattr(agent_state, 'sandbox_info')}"
)
sandbox_info = agent_state.sandbox_info
cdp_port = sandbox_info.get("browser_cdp_port")
if not cdp_port:
raise ValueError(
"sandbox_info is missing browser_cdp_port. "
"The sandbox container may not have Chromium CDP enabled. "
f"Available keys: {list(sandbox_info.keys())}"
)
# Resolve the Docker host (same logic used by the tool server URL).
docker_host = os.getenv("DOCKER_HOST", "")
if docker_host:
from urllib.parse import urlparse
parsed = urlparse(docker_host)
if parsed.scheme in ("tcp", "http", "https") and parsed.hostname:
host = parsed.hostname
else:
host = "127.0.0.1"
else:
host = "127.0.0.1"
cdp_url = f"http://{host}:{cdp_port}"
logger.info(
"Resolved CDP URL: %s (port=%s, host=%s, sandbox_id=%s)",
cdp_url,
cdp_port,
host,
sandbox_info.get("workspace_id", "?")[:12],
)
return cdp_url
# ---------------------------------------------------------------------------
# Tool entry-point
# ---------------------------------------------------------------------------
@register_tool(sandbox_execution=False)
async def browser_actions(
action: BrowserUseLocalAction,
task: str | None = None,
use_local: bool = False,
profile_directory: str | None = None,
# --- Granular command parameters ---
url: str | None = None,
index: int | None = None,
text: str | None = None,
value: str | None = None,
selector: str | None = None,
keys: str | None = None,
js: str | None = None,
direction: str | None = None,
amount: int | None = None,
tab: int | None = None,
x: float | None = None,
y: float | None = None,
full: bool = False,
path: str | None = None,
subcommand: str | None = None,
query: str | None = None,
name: str | None = None,
domain: str | None = None,
file: str | None = None,
timeout: int | None = None,
state: str | None = None,
secure: bool = False,
http_only: bool = False,
same_site: str | None = None,
expires: float | None = None,
*,
agent_state: Any = None,
) -> dict[str, Any]:
"""Use a browser to perform web tasks via natural language or granular commands.
This is a **blocking** tool the calling agent will wait for the browser
task to fully complete before receiving results. The browser session
persists across calls so state (cookies, auth, tabs) carries over.
Two modes:
**Sandboxed** (default): The browser runs inside the sandbox container
(with Caido proxy for traffic interception) and is controlled remotely
via CDP. Requires ``agent_state`` with ``sandbox_info``.
**Local** (``use_local=True``): Uses the system Chrome installation
directly via ``Browser.from_system_chrome()``. No sandbox required.
Preserves login sessions, cookies, and extensions. You may need to
fully close Chrome before launching. Optionally pass
``profile_directory`` (e.g. ``"Profile 1"``, ``"Default"``) to select
a specific Chrome profile.
Actions:
launch - Connect to the browser. MUST be called first.
run - Execute a natural-language browser task (requires ``task``).
close - Shut down the browser session.
open - Navigate to a URL (requires ``url``).
click - Click an element by ``index`` or coordinates (``x``, ``y``).
type - Insert text at the cursor (requires ``text``).
input - Click an element and type into it (``index`` + ``text``).
scroll - Scroll the page (``direction``, ``amount``).
back - Navigate back.
screenshot - Capture a screenshot (``full``, ``path``).
state - Get the current page DOM state.
switch - Switch to a tab by index (``tab``).
close_tab - Close a tab (``tab``).
keys - Send keyboard keys (``keys``).
select - Select a dropdown option (``index`` + ``value``).
eval - Execute JavaScript (``js``).
extract - Extract information (requires agent mode use ``run``).
hover - Hover over an element (``index``).
dblclick - Double-click an element (``index``).
rightclick - Right-click an element (``index``).
cookies - Cookie operations (``subcommand``: get/set/clear/export/import).
wait - Wait for element/text (``subcommand``: selector/text).
get - Get page info (``subcommand``: title/html/text/value/attributes/bbox).
"""
try:
from strix.tools.context import get_current_agent_id
agent_id = get_current_agent_id()
logger.info(
"browser_actions called: action=%s, agent_id=%s, has_task=%s, "
"has_agent_state=%s, use_local=%s, profile_directory=%s",
action,
agent_id,
task is not None,
agent_state is not None,
use_local,
profile_directory,
)
if action == "launch":
if use_local:
session = await _launch_local_browser(agent_id, profile_directory)
return {
"message": "Local browser launched and ready",
"mode": "local",
"profile_directory": session.profile_directory or "auto",
"is_running": True,
}
cdp_url = _resolve_cdp_url(agent_state)
session = await _launch_browser(cdp_url, agent_id)
return {
"message": "Browser launched and ready",
"mode": "sandboxed",
"cdp_url": session.cdp_url,
"ws_url": session.ws_url,
"is_running": True,
}
if action == "close":
await _close_session(agent_id)
return {"message": "Browser closed", "is_running": False}
if action == "run":
if not task:
raise ValueError("task parameter is required for run action") # noqa: TRY301
session = _get_session(agent_id)
logger.info(
"Starting task for agent %s (uptime=%.1fs, tasks=%d, local=%s): %.120s",
agent_id,
time.monotonic() - session.created_at,
session.task_count,
session.local,
task,
)
result = await _run_agent_task(task, session)
logger.info("Browser task completed for agent %s", agent_id)
# Agent.run() calls browser_session.kill() which destroys the
# CDP client and event bus. Re-create the Browser so subsequent
# granular commands (state, eval, click, …) have a working session.
try:
await _reinitialize_after_agent(session)
except Exception: # noqa: BLE001
logger.warning(
"Failed to reinitialize session after agent run for %s",
agent_id,
exc_info=True,
)
return result
if action in _GRANULAR_ACTIONS:
session = _get_session(agent_id)
from .browser_commands import handle_command
return await handle_command(
action,
session.browser,
url=url,
index=index,
text=text,
value=value,
selector=selector,
keys=keys,
js=js,
direction=direction,
amount=amount,
tab=tab,
x=x,
y=y,
full=full,
path=path,
subcommand=subcommand,
query=query,
name=name,
domain=domain,
file=file,
timeout=timeout,
state=state,
secure=secure,
http_only=http_only,
same_site=same_site,
expires=expires,
)
raise ValueError(f"Unknown action: {action}") # noqa: TRY301
except Exception as error:
logger.exception(
"browser_actions error: action=%s",
action,
)
return {"error": str(error), "is_running": False}

View file

@ -1,188 +1,276 @@
<tools>
<tool name="browser_action">
<description>Perform browser actions using a Playwright-controlled browser with multiple tabs.
The browser is PERSISTENT and remains active until explicitly closed, allowing for
multi-step workflows and long-running processes across multiple tabs.</description>
<tool name="browser_actions">
<description>Control a browser via natural language or granular commands. Supports two modes:
**Sandboxed** (default): The browser runs inside the sandbox container (with Caido proxy
intercepting all traffic) and is controlled remotely via Chrome DevTools Protocol (CDP).
**Local** (use_local=true): Uses the system Chrome installation directly. No sandbox
required. Preserves existing login sessions, cookies, and extensions. You may need to
fully close Chrome before launching. Optionally select a Chrome profile.
The browser is PERSISTENT and remains active until explicitly closed, allowing for
multi-step workflows. State (cookies, auth, tabs) carries over between calls.
Two usage styles:
- **Agent mode** (action="run"): Give a natural-language task and the browser-use agent
autonomously navigates, clicks, fills forms, and extracts information.
- **Granular mode**: Use specific actions (open, click, type, scroll, etc.) for precise,
deterministic control of the browser via CDP.</description>
<parameters>
<parameter name="action" type="string" required="true">
<description>The action to perform:
**Lifecycle:**
- launch: Start the browser. MUST be called before any other actions.
Pass use_local=true for local Chrome. Optionally pass profile_directory.
- close: Shut down the browser session.
**Agent mode:**
- run: Execute a natural-language browser task. Requires 'task' parameter.
**Navigation:**
- open: Navigate to a URL. Requires 'url'.
- back: Navigate back in history.
**Interaction:**
- click: Click an element by 'index' or coordinates ('x', 'y').
- type: Insert text at the current cursor position. Requires 'text'.
- input: Click an element then type into it. Requires 'index' and 'text'.
- hover: Hover over an element. Requires 'index'.
- dblclick: Double-click an element. Requires 'index'.
- rightclick: Right-click an element. Requires 'index'.
- keys: Send keyboard keys (e.g. "Enter", "Ctrl+a"). Requires 'keys'.
- select: Select a dropdown option. Requires 'index' and 'value'.
- scroll: Scroll the page. Optional 'direction' (up/down/left/right) and 'amount'.
**Information:**
- screenshot: Capture a screenshot. Optional 'full' (full page) and 'path' (save to file).
- state: Get the current page DOM state with element indices.
- eval: Execute JavaScript. Requires 'js'.
- get: Get page info. Requires 'subcommand': title, html, text, value, attributes, bbox.
**Cookies:**
- cookies: Cookie operations. Requires 'subcommand': get, set, clear, export, import.
**Waiting:**
- wait: Wait for conditions. Requires 'subcommand': selector, text.
**Tabs:**
- switch: Switch to a tab by index. Requires 'tab'.
- close_tab: Close a tab. Optional 'tab' (defaults to focused tab).</description>
</parameter>
<parameter name="task" type="string" required="false">
<description>Required for 'run' action. A natural-language description of what to do in
the browser, e.g. "Go to example.com and find the contact email".</description>
</parameter>
<parameter name="use_local" type="boolean" required="false">
<description>Only used with action='launch'. When true, uses the local system Chrome
installation instead of connecting to the sandbox container via CDP. Default: false.</description>
</parameter>
<parameter name="profile_directory" type="string" required="false">
<description>Only used with action='launch' and use_local=true. The Chrome profile
directory to use, e.g. "Default", "Profile 1". If not specified, auto-selected.</description>
</parameter>
<parameter name="url" type="string" required="false">
<description>Required for 'launch', 'goto', and optionally for 'new_tab' actions. The URL to launch the browser at, navigate to, or load in new tab. Must include appropriate protocol (e.g., http://, https://, file://).</description>
<description>URL to navigate to. Required for 'open' action. Also used by cookies get/clear/export.</description>
</parameter>
<parameter name="coordinate" type="string" required="false">
<description>Required for 'click', 'double_click', and 'hover' actions. Format: "x,y" (e.g., "432,321"). Coordinates should target the center of elements (buttons, links, etc.). Must be within the browser viewport resolution. Be very careful to calculate the coordinates correctly based on the previous screenshot.</description>
<parameter name="index" type="integer" required="false">
<description>Element index from the DOM state. Used by: click, input, select, hover, dblclick,
rightclick, and get (text/value/attributes/bbox).</description>
</parameter>
<parameter name="text" type="string" required="false">
<description>Required for 'type' action. The text to type in the field.</description>
<description>Text content. Required for 'type' and 'input' actions. Also used by wait text.</description>
</parameter>
<parameter name="tab_id" type="string" required="false">
<description>Required for 'switch_tab' and 'close_tab' actions. Optional for other actions to specify which tab to operate on. The ID of the tab to operate on. The first tab created during 'launch' has ID "tab_1". If not provided, actions will operate on the currently active tab.</description>
<parameter name="value" type="string" required="false">
<description>Value to select. Required for 'select' action. Also used for cookie value in cookies set.</description>
</parameter>
<parameter name="js_code" type="string" required="false">
<description>Required for 'execute_js' action. JavaScript code to execute in the page context. The code runs in the context of the current page and has access to the DOM and all page-defined variables and functions. The last evaluated expression's value is returned in the response.</description>
<parameter name="selector" type="string" required="false">
<description>CSS selector. Used by wait selector and get html.</description>
</parameter>
<parameter name="duration" type="string" required="false">
<description>Required for 'wait' action. Number of seconds to pause execution. Can be fractional (e.g., 0.5 for half a second).</description>
<parameter name="keys" type="string" required="false">
<description>Keyboard keys to send. Required for 'keys' action. Examples: "Enter", "Ctrl+a", "Escape".</description>
</parameter>
<parameter name="key" type="string" required="false">
<description>Required for 'press_key' action. The key to press. Valid values include: - Single characters: 'a'-'z', 'A'-'Z', '0'-'9' - Special keys: 'Enter', 'Escape', 'ArrowLeft', 'ArrowRight', etc. - Modifier keys: 'Shift', 'Control', 'Alt', 'Meta' - Function keys: 'F1'-'F12'</description>
<parameter name="js" type="string" required="false">
<description>JavaScript code to execute. Required for 'eval' action.</description>
</parameter>
<parameter name="file_path" type="string" required="false">
<description>Required for 'save_pdf' action. The file path where to save the PDF.</description>
<parameter name="direction" type="string" required="false">
<description>Scroll direction: "up", "down", "left", "right". Default: "down".</description>
</parameter>
<parameter name="clear" type="boolean" required="false">
<description>For 'get_console_logs' action: whether to clear console logs after retrieving them. Default is False (keep logs).</description>
<parameter name="amount" type="integer" required="false">
<description>Scroll amount in pixels. Default: 500.</description>
</parameter>
<parameter name="tab" type="integer" required="false">
<description>Tab index. Required for 'switch', optional for 'close_tab'.</description>
</parameter>
<parameter name="x" type="number" required="false">
<description>X coordinate for click action (use with 'y' for coordinate-based clicking).</description>
</parameter>
<parameter name="y" type="number" required="false">
<description>Y coordinate for click action (use with 'x' for coordinate-based clicking).</description>
</parameter>
<parameter name="full" type="boolean" required="false">
<description>For 'screenshot': capture the full page. Default: false.</description>
</parameter>
<parameter name="path" type="string" required="false">
<description>For 'screenshot': save to this file path instead of returning base64.</description>
</parameter>
<parameter name="subcommand" type="string" required="false">
<description>Sub-action for compound commands:
- cookies: "get", "set", "clear", "export", "import"
- wait: "selector", "text"
- get: "title", "html", "text", "value", "attributes", "bbox"</description>
</parameter>
<parameter name="query" type="string" required="false">
<description>Query for 'extract' action (requires agent mode).</description>
</parameter>
<parameter name="name" type="string" required="false">
<description>Cookie name. Used by cookies set.</description>
</parameter>
<parameter name="domain" type="string" required="false">
<description>Cookie domain. Used by cookies set.</description>
</parameter>
<parameter name="file" type="string" required="false">
<description>File path for cookies export/import.</description>
</parameter>
<parameter name="timeout" type="integer" required="false">
<description>Timeout in milliseconds for wait actions. Default: 30000.</description>
</parameter>
<parameter name="state" type="string" required="false">
<description>Wait state for wait selector: "visible", "hidden", "attached", "detached". Default: "visible".</description>
</parameter>
<parameter name="secure" type="boolean" required="false">
<description>Cookie secure flag. Used by cookies set. Default: false.</description>
</parameter>
<parameter name="http_only" type="boolean" required="false">
<description>Cookie httpOnly flag. Used by cookies set. Default: false.</description>
</parameter>
<parameter name="same_site" type="string" required="false">
<description>Cookie SameSite policy. Used by cookies set. Values: "Strict", "Lax", "None".</description>
</parameter>
<parameter name="expires" type="number" required="false">
<description>Cookie expiration timestamp. Used by cookies set.</description>
</parameter>
</parameters>
<returns type="Dict[str, Any]">
<description>Response containing: - screenshot: Base64 encoded PNG of the current page state - url: Current page URL - title: Current page title - viewport: Current browser viewport dimensions - tab_id: ID of the current active tab - all_tabs: Dict of all open tab IDs and their URLs - message: Status message about the action performed - js_result: Result of JavaScript execution (for execute_js action) - pdf_saved: File path of saved PDF (for save_pdf action) - console_logs: Array of console messages (for get_console_logs action) Limited to 50KB total and 200 most recent logs. Individual messages truncated at 1KB. - page_source: HTML source code (for view_source action) Large pages are truncated to 100KB (keeping beginning and end sections).</description>
<description>Returns a dict with action-specific results. On error, returns an error field.
For 'run': message + result. For granular actions: action-specific keys.</description>
</returns>
<notes>
Important usage rules:
1. PERSISTENCE: The browser remains active and maintains its state until
explicitly closed with the 'close' action. This allows for multi-step workflows
across multiple tool calls and tabs.
2. Browser interaction MUST start with 'launch' and end with 'close'.
3. Only one action can be performed per call.
4. To visit a new URL not reachable from current page, either:
- Use 'goto' action
- Open a new tab with the URL
- Close browser and relaunch
5. Click coordinates must be derived from the most recent screenshot.
6. You MUST click on the center of the element, not the edge. You MUST calculate
the coordinates correctly based on the previous screenshot, otherwise the click
will fail. After clicking, check the new screenshot to verify the click was
successful.
7. Tab management:
- First tab from 'launch' is "tab_1"
- New tabs are numbered sequentially ("tab_2", "tab_3", etc.)
- Must have at least one tab open at all times
- Actions affect the currently active tab unless tab_id is specified
8. JavaScript execution (following Playwright evaluation patterns):
- Code runs in the browser page context, not the tool context
- Has access to DOM (document, window, etc.) and page variables/functions
- The LAST EVALUATED EXPRESSION is automatically returned - no return statement needed
- For simple values: document.title (returns the title)
- For objects: {title: document.title, url: location.href} (returns the object)
- For async operations: Use await and the promise result will be returned
- AVOID explicit return statements - they can break evaluation
- object literals must be wrapped in paranthesis when they are the final expression
- Variables from tool context are NOT available - pass data as parameters if needed
- Examples of correct patterns:
* Single value: document.querySelectorAll('img').length
* Object result: {images: document.images.length, links: document.links.length}
* Async operation: await fetch(location.href).then(r => r.status)
* DOM manipulation: document.body.style.backgroundColor = 'red'; 'background changed'
9. Wait action:
- Time is specified in seconds
- Can be used to wait for page loads, animations, etc.
- Can be fractional (e.g., 0.5 seconds)
- Screenshot is captured after the wait
10. The browser can operate concurrently with other tools. You may invoke
terminal, python, or other tools (in separate assistant messages) while maintaining
the active browser session, enabling sophisticated multi-tool workflows.
11. Keyboard actions:
- Use press_key for individual key presses
- Use type for typing regular text
- Some keys have special names based on Playwright's key documentation
12. All code in the js_code parameter is executed as-is - there's no need to
escape special characters or worry about formatting. Just write your JavaScript
code normally. It can be single line or multi-line.
13. For form filling, click on the field first, then use 'type' to enter text.
14. The browser runs in headless mode using Chrome engine for security and performance.
15. RESOURCE MANAGEMENT:
- ALWAYS close tabs you no longer need using 'close_tab' action.
- ALWAYS close the browser with 'close' action when you have completely finished
all browser-related tasks. Do not leave the browser running if you're done with it.
- If you opened multiple tabs, close them as soon as you've extracted the needed
information from each one.
Important usage rules:
1. PERSISTENCE: The browser remains active and maintains its state (cookies, auth, tabs)
until explicitly closed with the 'close' action.
2. Browser interaction MUST start with 'launch' and end with 'close'.
3. You MUST call 'launch' before any other actions. Calling actions without launching
first will return an error.
4. For 'run' action: the browser-use agent autonomously handles the task. You MUST wait
for the result — it is returned synchronously.
5. For granular actions: use 'state' first to get element indices, then use those indices
with click, input, select, hover, etc.
6. The browser persists across tasks. Do NOT close it between actions.
7. ALWAYS close the browser with 'close' when completely finished.
8. There is a 5-minute timeout per 'run' task. Break large tasks into smaller steps.
9. Granular actions are faster and more deterministic than 'run' for simple operations.
Use 'run' for complex multi-step tasks; use granular actions for precise control.
</notes>
<examples>
# Launch browser at URL (creates tab_1)
<function=browser_action>
<parameter=action>launch</parameter>
<parameter=url>https://example.com</parameter>
</function>
# Launch the sandbox browser (default, must be done first)
<function=browser_actions>
<parameter=action>launch</parameter>
</function>
# Navigate to different URL
<function=browser_action>
<parameter=action>goto</parameter>
<parameter=url>https://github.com</parameter>
</function>
# Launch the local system Chrome instead
<function=browser_actions>
<parameter=action>launch</parameter>
<parameter=use_local>true</parameter>
</function>
# Open new tab with different URL
<function=browser_action>
<parameter=action>new_tab</parameter>
<parameter=url>https://another-site.com</parameter>
</function>
# Navigate to a URL
<function=browser_actions>
<parameter=action>open</parameter>
<parameter=url>https://example.com</parameter>
</function>
# Wait for page load
<function=browser_action>
<parameter=action>wait</parameter>
<parameter=duration>2.5</parameter>
</function>
# Get current page state with element indices
<function=browser_actions>
<parameter=action>state</parameter>
</function>
# Click login button at coordinates from screenshot
<function=browser_action>
<parameter=action>click</parameter>
<parameter=coordinate>450,300</parameter>
</function>
# Click element by index (from state output)
<function=browser_actions>
<parameter=action>click</parameter>
<parameter=index>5</parameter>
</function>
# Click username field and type
<function=browser_action>
<parameter=action>click</parameter>
<parameter=coordinate>400,200</parameter>
</function>
# Click by coordinates
<function=browser_actions>
<parameter=action>click</parameter>
<parameter=x>100</parameter>
<parameter=y>200</parameter>
</function>
<function=browser_action>
<parameter=action>type</parameter>
<parameter=text>user@example.com</parameter>
</function>
# Type into an input field (click + type)
<function=browser_actions>
<parameter=action>input</parameter>
<parameter=index>3</parameter>
<parameter=text>admin@example.com</parameter>
</function>
# Click password field and type
<function=browser_action>
<parameter=action>click</parameter>
<parameter=coordinate>400,250</parameter>
</function>
# Send keyboard keys
<function=browser_actions>
<parameter=action>keys</parameter>
<parameter=keys>Enter</parameter>
</function>
<function=browser_action>
<parameter=action>type</parameter>
<parameter=text>mypassword123</parameter>
</function>
# Scroll down
<function=browser_actions>
<parameter=action>scroll</parameter>
<parameter=direction>down</parameter>
<parameter=amount>500</parameter>
</function>
# Press Enter key
<function=browser_action>
<parameter=action>press_key</parameter>
<parameter=key>Enter</parameter>
</function>
# Take a screenshot
<function=browser_actions>
<parameter=action>screenshot</parameter>
<parameter=full>true</parameter>
<parameter=path>/tmp/page.png</parameter>
</function>
# Execute JavaScript to get page stats (correct pattern - no return statement)
<function=browser_action>
<parameter=action>execute_js</parameter>
<parameter=js_code>const images = document.querySelectorAll('img');
const links = document.querySelectorAll('a');
{
images: images.length,
links: links.length,
title: document.title
}</parameter>
</function>
# Get page title
<function=browser_actions>
<parameter=action>get</parameter>
<parameter=subcommand>title</parameter>
</function>
# Scroll down
<function=browser_action>
<parameter=action>scroll_down</parameter>
</function>
# Wait for an element to appear
<function=browser_actions>
<parameter=action>wait</parameter>
<parameter=subcommand>selector</parameter>
<parameter=selector>#login-form</parameter>
<parameter=timeout>10000</parameter>
</function>
# Get console logs
<function=browser_action>
<parameter=action>get_console_logs</parameter>
</function>
# Get cookies for a URL
<function=browser_actions>
<parameter=action>cookies</parameter>
<parameter=subcommand>get</parameter>
<parameter=url>https://example.com</parameter>
</function>
# View page source
<function=browser_action>
<parameter=action>view_source</parameter>
</function>
# Execute JavaScript
<function=browser_actions>
<parameter=action>eval</parameter>
<parameter=js>document.title</parameter>
</function>
# Run a natural-language browser task
<function=browser_actions>
<parameter=action>run</parameter>
<parameter=task>Go to https://example.com/login, fill in username "admin" and password "secret", then click the login button</parameter>
</function>
# Close the browser when done
<function=browser_actions>
<parameter=action>close</parameter>
</function>
</examples>
</tool>
</tools>

View file

@ -0,0 +1,725 @@
from __future__ import annotations
import asyncio
import base64
import json
import logging
from pathlib import Path
from typing import Any
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
async def _ensure_browser_started(bs: Any) -> None:
"""Ensure the BrowserSession has an active CDP connection and watchdogs.
Checks both the root CDP client and session manager, which are required
by ``get_or_create_cdp_session()``. Calls ``start()`` if either is
missing ``start()`` is idempotent and safe to call when already connected.
"""
has_cdp = getattr(bs, "_cdp_client_root", None) is not None
has_manager = getattr(bs, "session_manager", None) is not None
if has_cdp and has_manager:
return
await bs.start()
async def _execute_js(bs: Any, js: str) -> Any:
"""Execute JavaScript in the browser via CDP."""
cdp_session = await bs.get_or_create_cdp_session(target_id=None, focus=False)
if not cdp_session:
raise RuntimeError("No active browser session")
result = await cdp_session.cdp_client.send.Runtime.evaluate(
params={"expression": js, "returnByValue": True},
session_id=cdp_session.session_id,
)
return result.get("result", {}).get("value")
async def _get_element_center(bs: Any, node: Any) -> tuple[float, float] | None:
"""Get the center coordinates of an element."""
try:
cdp_session = await bs.cdp_client_for_node(node)
session_id = cdp_session.session_id
backend_node_id = node.backend_node_id
try:
await cdp_session.cdp_client.send.DOM.scrollIntoViewIfNeeded(
params={"backendNodeId": backend_node_id}, session_id=session_id
)
await asyncio.sleep(0.05)
except Exception: # noqa: BLE001
logger.debug("scrollIntoViewIfNeeded failed", exc_info=True)
element_rect = await bs.get_element_coordinates(backend_node_id, cdp_session)
except Exception:
logger.exception("Failed to get element center")
return None
else:
if element_rect:
center_x = element_rect.x + element_rect.width / 2
center_y = element_rect.y + element_rect.height / 2
return center_x, center_y
return None
async def _get_node_or_error(
bs: Any, index: int | None
) -> tuple[Any | None, dict[str, Any] | None]:
"""Look up a DOM node by index, returning (node, None) or (None, error_dict)."""
if index is None:
return None, {"error": "index parameter is required"}
node = await bs.get_element_by_index(index)
if node is None:
return None, {"error": f"Element index {index} not found - page may have changed"}
return node, None
# ---------------------------------------------------------------------------
# Command dispatcher
# ---------------------------------------------------------------------------
async def handle_command(action: str, bs: Any, **params: Any) -> dict[str, Any]:
"""Dispatch a granular browser command.
Parameters are passed as keyword args each action uses only the keys it
needs and ignores the rest.
"""
await _ensure_browser_started(bs)
if action == "open":
return await _cmd_open(bs, **params)
if action == "click":
return await _cmd_click(bs, **params)
if action == "type":
return await _cmd_type(bs, **params)
if action == "input":
return await _cmd_input(bs, **params)
if action == "scroll":
return await _cmd_scroll(bs, **params)
if action == "back":
return await _cmd_back(bs)
if action == "screenshot":
return await _cmd_screenshot(bs, **params)
if action == "state":
return await _cmd_state(bs)
if action == "switch":
return await _cmd_switch(bs, **params)
if action in ("close_tab", "close-tab"):
return await _cmd_close_tab(bs, **params)
if action == "keys":
return await _cmd_keys(bs, **params)
if action == "select":
return await _cmd_select(bs, **params)
if action == "eval":
return await _cmd_eval(bs, **params)
if action == "extract":
return await _cmd_extract(**params)
if action == "hover":
return await _cmd_hover(bs, **params)
if action == "dblclick":
return await _cmd_dblclick(bs, **params)
if action == "rightclick":
return await _cmd_rightclick(bs, **params)
if action == "cookies":
return await _cmd_cookies(bs, **params)
if action == "wait":
return await _cmd_wait(bs, **params)
if action == "get":
return await _cmd_get(bs, **params)
raise ValueError(f"Unknown browser command: {action}")
# ---------------------------------------------------------------------------
# Individual commands
# ---------------------------------------------------------------------------
async def _cmd_open(bs: Any, *, url: str | None = None, **_: Any) -> dict[str, Any]:
if not url:
return {"error": "url parameter is required for open action"}
if not url.startswith(("http://", "https://", "file://")):
url = "https://" + url
from browser_use.browser.events import NavigateToUrlEvent
await bs.event_bus.dispatch(NavigateToUrlEvent(url=url))
result: dict[str, Any] = {"url": url}
if getattr(bs, "browser_profile", None) and bs.browser_profile.use_cloud and bs.cdp_url:
from urllib.parse import quote
result["live_url"] = f"https://live.browser-use.com/?wss={quote(bs.cdp_url, safe='')}"
return result
async def _cmd_click(
bs: Any,
*,
index: int | None = None,
x: float | None = None,
y: float | None = None,
**_: Any,
) -> dict[str, Any]:
if x is not None and y is not None:
from browser_use.browser.events import ClickCoordinateEvent
await bs.event_bus.dispatch(ClickCoordinateEvent(coordinate_x=x, coordinate_y=y))
return {"clicked_coordinate": {"x": x, "y": y}}
if index is not None:
from browser_use.browser.events import ClickElementEvent
node, err = await _get_node_or_error(bs, index)
if err:
return err
await bs.event_bus.dispatch(ClickElementEvent(node=node))
return {"clicked": index}
return {"error": "Provide index or (x, y) coordinates"}
async def _cmd_type(bs: Any, *, text: str | None = None, **_: Any) -> dict[str, Any]:
if not text:
return {"error": "text parameter is required for type action"}
cdp_session = await bs.get_or_create_cdp_session(target_id=None, focus=False)
if not cdp_session:
return {"error": "No active browser session"}
await cdp_session.cdp_client.send.Input.insertText(
params={"text": text}, session_id=cdp_session.session_id
)
return {"typed": text}
async def _cmd_input(
bs: Any, *, index: int | None = None, text: str | None = None, **_: Any
) -> dict[str, Any]:
if not text:
return {"error": "text parameter is required for input action"}
from browser_use.browser.events import ClickElementEvent, TypeTextEvent
node, err = await _get_node_or_error(bs, index)
if err:
return err
await bs.event_bus.dispatch(ClickElementEvent(node=node))
await bs.event_bus.dispatch(TypeTextEvent(node=node, text=text))
return {"input": text, "element": index}
async def _cmd_scroll(
bs: Any,
*,
direction: str | None = None,
amount: int | None = None,
**_: Any,
) -> dict[str, Any]:
from browser_use.browser.events import ScrollEvent
d = direction or "down"
a = amount or 500
await bs.event_bus.dispatch(ScrollEvent(direction=d, amount=a))
return {"scrolled": d, "amount": a}
async def _cmd_back(bs: Any) -> dict[str, Any]:
from browser_use.browser.events import GoBackEvent
await bs.event_bus.dispatch(GoBackEvent())
return {"back": True}
async def _cmd_screenshot(
bs: Any, *, full: bool = False, path: str | None = None, **_: Any
) -> dict[str, Any]:
data = await bs.take_screenshot(full_page=full)
if path:
p = Path(path)
p.write_bytes(data)
return {"saved": str(p), "size": len(data)}
return {"screenshot": base64.b64encode(data).decode(), "size": len(data)}
async def _cmd_state(bs: Any) -> dict[str, Any]:
state = await bs.get_browser_state_summary()
assert state.dom_state is not None
state_text = state.dom_state.llm_representation()
if state.page_info:
pi = state.page_info
header = (
f"viewport: {pi.viewport_width}x{pi.viewport_height}\n"
f"page: {pi.page_width}x{pi.page_height}\n"
f"scroll: ({pi.scroll_x}, {pi.scroll_y})\n"
)
state_text = header + state_text
return {"_raw_text": state_text}
async def _cmd_switch(bs: Any, *, tab: int | None = None, **_: Any) -> dict[str, Any]:
if tab is None:
return {"error": "tab parameter is required for switch action"}
from browser_use.browser.events import SwitchTabEvent
page_targets = bs.session_manager.get_all_page_targets() if bs.session_manager else []
if tab < 0 or tab >= len(page_targets):
return {"error": f"Invalid tab index {tab}. Available: 0-{len(page_targets) - 1}"}
target_id = page_targets[tab].target_id
await bs.event_bus.dispatch(SwitchTabEvent(target_id=target_id))
return {"switched": tab}
async def _cmd_close_tab(bs: Any, *, tab: int | None = None, **_: Any) -> dict[str, Any]:
from browser_use.browser.events import CloseTabEvent
page_targets = bs.session_manager.get_all_page_targets() if bs.session_manager else []
if tab is not None:
if tab < 0 or tab >= len(page_targets):
return {"error": f"Invalid tab index {tab}. Available: 0-{len(page_targets) - 1}"}
target_id = page_targets[tab].target_id
else:
focused = bs.session_manager.get_focused_target() if bs.session_manager else None
if not focused:
return {"error": "No focused tab to close"}
target_id = focused.target_id
await bs.event_bus.dispatch(CloseTabEvent(target_id=target_id))
return {"closed": tab}
async def _cmd_keys(bs: Any, *, keys: str | None = None, **_: Any) -> dict[str, Any]:
if not keys:
return {"error": "keys parameter is required"}
from browser_use.browser.events import SendKeysEvent
await bs.event_bus.dispatch(SendKeysEvent(keys=keys))
return {"sent": keys}
async def _cmd_select(
bs: Any, *, index: int | None = None, value: str | None = None, **_: Any
) -> dict[str, Any]:
if not value:
return {"error": "value parameter is required for select action"}
from browser_use.browser.events import SelectDropdownOptionEvent
node, err = await _get_node_or_error(bs, index)
if err:
return err
await bs.event_bus.dispatch(SelectDropdownOptionEvent(node=node, text=value))
return {"selected": value, "element": index}
async def _cmd_eval(bs: Any, *, js: str | None = None, **_: Any) -> dict[str, Any]:
if not js:
return {"error": "js parameter is required for eval action"}
result = await _execute_js(bs, js)
return {"result": result}
async def _cmd_extract(**params: Any) -> dict[str, Any]:
query = params.get("query")
if not query:
return {"error": "query parameter is required for extract action"}
return {"query": query, "error": "extract requires agent mode — use action='run'"}
async def _cmd_hover(bs: Any, *, index: int | None = None, **_: Any) -> dict[str, Any]:
node, err = await _get_node_or_error(bs, index)
if err:
return err
coords = await _get_element_center(bs, node)
if not coords:
return {"error": "Could not get element coordinates for hover"}
cx, cy = coords
cdp_session = await bs.cdp_client_for_node(node)
await cdp_session.cdp_client.send.Input.dispatchMouseEvent(
params={"type": "mouseMoved", "x": cx, "y": cy},
session_id=cdp_session.session_id,
)
return {"hovered": index}
async def _cmd_dblclick(bs: Any, *, index: int | None = None, **_: Any) -> dict[str, Any]:
node, err = await _get_node_or_error(bs, index)
if err:
return err
coords = await _get_element_center(bs, node)
if not coords:
return {"error": "Could not get element coordinates for double-click"}
cx, cy = coords
cdp_session = await bs.cdp_client_for_node(node)
sid = cdp_session.session_id
send = cdp_session.cdp_client.send.Input.dispatchMouseEvent
await send(params={"type": "mouseMoved", "x": cx, "y": cy}, session_id=sid)
await asyncio.sleep(0.05)
await send(
params={"type": "mousePressed", "x": cx, "y": cy, "button": "left", "clickCount": 2},
session_id=sid,
)
await asyncio.sleep(0.05)
await send(
params={"type": "mouseReleased", "x": cx, "y": cy, "button": "left", "clickCount": 2},
session_id=sid,
)
return {"double_clicked": index}
async def _cmd_rightclick(bs: Any, *, index: int | None = None, **_: Any) -> dict[str, Any]:
node, err = await _get_node_or_error(bs, index)
if err:
return err
coords = await _get_element_center(bs, node)
if not coords:
return {"error": "Could not get element coordinates for right-click"}
cx, cy = coords
cdp_session = await bs.cdp_client_for_node(node)
sid = cdp_session.session_id
send = cdp_session.cdp_client.send.Input.dispatchMouseEvent
await send(params={"type": "mouseMoved", "x": cx, "y": cy}, session_id=sid)
await asyncio.sleep(0.05)
await send(
params={"type": "mousePressed", "x": cx, "y": cy, "button": "right", "clickCount": 1},
session_id=sid,
)
await asyncio.sleep(0.05)
await send(
params={"type": "mouseReleased", "x": cx, "y": cy, "button": "right", "clickCount": 1},
session_id=sid,
)
return {"right_clicked": index}
# ---------------------------------------------------------------------------
# Cookies
# ---------------------------------------------------------------------------
async def _cmd_cookies(bs: Any, **params: Any) -> dict[str, Any]:
sub = params.get("subcommand")
if sub == "get":
return await _cookies_get(bs, **params)
if sub == "set":
return await _cookies_set(bs, **params)
if sub == "clear":
return await _cookies_clear(bs, **params)
if sub == "export":
return await _cookies_export(bs, **params)
if sub == "import":
return await _cookies_import(bs, **params)
return {"error": "subcommand required: get, set, clear, export, import"}
def _filter_cookies_by_url(cookie_list: list[dict[str, Any]], url: str) -> list[dict[str, Any]]:
from urllib.parse import urlparse
domain = urlparse(url).netloc
return [
c
for c in cookie_list
if domain.endswith(str(c.get("domain", "")).lstrip("."))
or str(c.get("domain", "")).lstrip(".").endswith(domain)
]
def _cookie_to_dict(c: Any) -> dict[str, Any]:
d: dict[str, Any] = {
"name": c.get("name", ""),
"value": c.get("value", ""),
"domain": c.get("domain", ""),
"path": c.get("path", "/"),
"secure": c.get("secure", False),
"httpOnly": c.get("httpOnly", False),
}
if "sameSite" in c:
d["sameSite"] = c.get("sameSite")
if "expires" in c:
d["expires"] = c.get("expires")
return d
async def _cookies_get(bs: Any, **params: Any) -> dict[str, Any]:
cookies = await bs._cdp_get_cookies()
cookie_list = [_cookie_to_dict(c) for c in cookies]
url = params.get("url")
if url:
cookie_list = _filter_cookies_by_url(cookie_list, url)
return {"cookies": cookie_list}
async def _cookies_set(bs: Any, **params: Any) -> dict[str, Any]:
from cdp_use.cdp.network import Cookie
cookie_dict: dict[str, Any] = {
"name": params.get("name", ""),
"value": params.get("value", ""),
"path": params.get("path", "/"),
"secure": params.get("secure", False),
"httpOnly": params.get("http_only", False),
}
if params.get("domain"):
cookie_dict["domain"] = params["domain"]
if params.get("same_site"):
cookie_dict["sameSite"] = params["same_site"]
if params.get("expires"):
cookie_dict["expires"] = params["expires"]
if not params.get("domain"):
hostname = await _execute_js(bs, "window.location.hostname")
if hostname:
cookie_dict["domain"] = hostname
try:
cookie_obj = Cookie(**cookie_dict)
await bs._cdp_set_cookies([cookie_obj])
return {"set": params.get("name"), "success": True}
except Exception as e:
logger.exception("Failed to set cookie")
return {"set": params.get("name"), "success": False, "error": str(e)}
async def _cookies_clear(bs: Any, **params: Any) -> dict[str, Any]:
url = params.get("url")
if url:
from urllib.parse import urlparse
cookies = await bs._cdp_get_cookies()
domain = urlparse(url).netloc
cdp_session = await bs.get_or_create_cdp_session(target_id=None, focus=False)
if cdp_session:
for cookie in cookies:
cookie_domain = str(cookie.get("domain", "")).lstrip(".")
if domain.endswith(cookie_domain) or cookie_domain.endswith(domain):
await cdp_session.cdp_client.send.Network.deleteCookies(
params={
"name": cookie.get("name", ""),
"domain": cookie.get("domain"),
"path": cookie.get("path", "/"),
},
session_id=cdp_session.session_id,
)
else:
await bs._cdp_clear_cookies()
return {"cleared": True, "url": url}
async def _cookies_export(bs: Any, **params: Any) -> dict[str, Any]:
file_path = params.get("file")
if not file_path:
return {"error": "file parameter is required for cookies export"}
cookies = await bs._cdp_get_cookies()
cookie_list = [_cookie_to_dict(c) for c in cookies]
url = params.get("url")
if url:
cookie_list = _filter_cookies_by_url(cookie_list, url)
p = Path(file_path)
p.write_text(json.dumps(cookie_list, indent=2, ensure_ascii=False), encoding="utf-8")
return {"exported": len(cookie_list), "file": str(p)}
async def _cookies_import(bs: Any, **params: Any) -> dict[str, Any]:
file_path = params.get("file")
if not file_path:
return {"error": "file parameter is required for cookies import"}
p = Path(file_path)
if not p.exists():
return {"error": f"File not found: {p}"}
cookies = json.loads(p.read_text())
cdp_session = await bs.get_or_create_cdp_session(target_id=None, focus=False)
if not cdp_session:
return {"error": "No active browser session"}
cookie_list = []
for c in cookies:
cookie_params: dict[str, Any] = {
"name": c["name"],
"value": c["value"],
"domain": c.get("domain"),
"path": c.get("path", "/"),
"secure": c.get("secure", False),
"httpOnly": c.get("httpOnly", False),
}
if c.get("sameSite"):
cookie_params["sameSite"] = c["sameSite"]
if c.get("expires"):
cookie_params["expires"] = c["expires"]
cookie_list.append(cookie_params)
try:
await cdp_session.cdp_client.send.Network.setCookies(
params={"cookies": cookie_list},
session_id=cdp_session.session_id,
)
return {"imported": len(cookie_list), "file": str(p)}
except Exception as e: # noqa: BLE001
return {"error": f"Failed to import cookies: {e}"}
# ---------------------------------------------------------------------------
# Wait
# ---------------------------------------------------------------------------
async def _cmd_wait(bs: Any, **params: Any) -> dict[str, Any]:
sub = params.get("subcommand")
if sub == "selector":
return await _wait_selector(bs, **params)
if sub == "text":
return await _wait_text(bs, **params)
return {"error": "subcommand required: selector, text"}
async def _wait_selector(bs: Any, **params: Any) -> dict[str, Any]:
selector = params.get("selector")
if not selector:
return {"error": "selector parameter is required"}
timeout_ms = params.get("timeout", 30000)
timeout_s = timeout_ms / 1000.0
wait_state = params.get("state", "visible")
poll = 0.1
elapsed = 0.0
js_map = {
"attached": f"document.querySelector({json.dumps(selector)}) !== null",
"detached": f"document.querySelector({json.dumps(selector)}) === null",
"visible": (
f"(function(){{ const el = document.querySelector({json.dumps(selector)});"
" if (!el) return false; const s = window.getComputedStyle(el);"
" const r = el.getBoundingClientRect();"
" return s.display !== 'none' && s.visibility !== 'hidden'"
" && s.opacity !== '0' && r.width > 0 && r.height > 0; }})()"
),
"hidden": (
f"(function(){{ const el = document.querySelector({json.dumps(selector)});"
" if (!el) return true; const s = window.getComputedStyle(el);"
" const r = el.getBoundingClientRect();"
" return s.display === 'none' || s.visibility === 'hidden'"
" || s.opacity === '0' || r.width === 0 || r.height === 0; }})()"
),
}
check_js = js_map.get(wait_state, js_map["attached"])
while elapsed < timeout_s:
if await _execute_js(bs, check_js):
return {"selector": selector, "found": True}
await asyncio.sleep(poll)
elapsed += poll
return {"selector": selector, "found": False}
async def _wait_text(bs: Any, **params: Any) -> dict[str, Any]:
text = params.get("text")
if not text:
return {"error": "text parameter is required"}
timeout_s = params.get("timeout", 30000) / 1000.0
poll = 0.1
elapsed = 0.0
check_js = f"document.body.innerText.includes({json.dumps(text)})"
while elapsed < timeout_s:
if await _execute_js(bs, check_js):
return {"text": text, "found": True}
await asyncio.sleep(poll)
elapsed += poll
return {"text": text, "found": False}
# ---------------------------------------------------------------------------
# Get
# ---------------------------------------------------------------------------
async def _cmd_get(bs: Any, **params: Any) -> dict[str, Any]:
sub = params.get("subcommand")
if sub == "title":
title = await _execute_js(bs, "document.title")
return {"title": title or ""}
if sub == "html":
selector = params.get("selector")
if selector:
js = (
f"(function(){{ const el = document.querySelector({json.dumps(selector)});"
" return el ? el.outerHTML : null; }})()"
)
else:
js = "document.documentElement.outerHTML"
html = await _execute_js(bs, js)
return {"html": html or ""}
if sub == "text":
return await _get_text(bs, **params)
if sub == "value":
return await _get_value(bs, **params)
if sub == "attributes":
return await _get_attributes(bs, **params)
if sub == "bbox":
return await _get_bbox(bs, **params)
return {"error": "subcommand required: title, html, text, value, attributes, bbox"}
async def _get_text(bs: Any, **params: Any) -> dict[str, Any]:
index = params.get("index")
node, err = await _get_node_or_error(bs, index)
if err:
return err
text = node.get_all_children_text(max_depth=10) if node else ""
return {"index": index, "text": text}
async def _get_value(bs: Any, **params: Any) -> dict[str, Any]:
index = params.get("index")
node, err = await _get_node_or_error(bs, index)
if err:
return err
assert node is not None
try:
cdp_session = await bs.cdp_client_for_node(node)
resolve_result = await cdp_session.cdp_client.send.DOM.resolveNode(
params={"backendNodeId": node.backend_node_id},
session_id=cdp_session.session_id,
)
object_id = resolve_result["object"].get("objectId")
if object_id:
value_result = await cdp_session.cdp_client.send.Runtime.callFunctionOn(
params={
"objectId": object_id,
"functionDeclaration": "function() { return this.value; }",
"returnByValue": True,
},
session_id=cdp_session.session_id,
)
val = value_result.get("result", {}).get("value")
return {"index": index, "value": val or ""}
except Exception:
logger.exception("Failed to get element value")
return {"index": index, "value": ""}
async def _get_attributes(bs: Any, **params: Any) -> dict[str, Any]:
index = params.get("index")
node, err = await _get_node_or_error(bs, index)
if err:
return err
assert node is not None
attrs = node.attributes or {}
return {"index": index, "attributes": dict(attrs)}
async def _get_bbox(bs: Any, **params: Any) -> dict[str, Any]:
index = params.get("index")
node, err = await _get_node_or_error(bs, index)
if err:
return err
assert node is not None
try:
cdp_session = await bs.cdp_client_for_node(node)
box_result = await cdp_session.cdp_client.send.DOM.getBoxModel(
params={"backendNodeId": node.backend_node_id},
session_id=cdp_session.session_id,
)
model = box_result["model"]
content = model.get("content", [])
if len(content) >= 8:
x = min(content[0], content[2], content[4], content[6])
y = min(content[1], content[3], content[5], content[7])
w = max(content[0], content[2], content[4], content[6]) - x
h = max(content[1], content[3], content[5], content[7]) - y
return {"index": index, "bbox": {"x": x, "y": y, "width": w, "height": h}}
except Exception:
logger.exception("Failed to get element bbox")
return {"index": index, "bbox": {}}

View file

@ -1,581 +0,0 @@
import asyncio
import base64
import contextlib
import logging
import threading
from pathlib import Path
from typing import Any, cast
from playwright.async_api import Browser, BrowserContext, Page, Playwright, async_playwright
logger = logging.getLogger(__name__)
MAX_PAGE_SOURCE_LENGTH = 20_000
MAX_CONSOLE_LOG_LENGTH = 30_000
MAX_INDIVIDUAL_LOG_LENGTH = 1_000
MAX_CONSOLE_LOGS_COUNT = 200
MAX_JS_RESULT_LENGTH = 5_000
class _BrowserState:
"""Singleton state for the shared browser instance."""
lock = threading.Lock()
event_loop: asyncio.AbstractEventLoop | None = None
event_loop_thread: threading.Thread | None = None
playwright: Playwright | None = None
browser: Browser | None = None
_state = _BrowserState()
def _ensure_event_loop() -> None:
if _state.event_loop is not None:
return
def run_loop() -> None:
_state.event_loop = asyncio.new_event_loop()
asyncio.set_event_loop(_state.event_loop)
_state.event_loop.run_forever()
_state.event_loop_thread = threading.Thread(target=run_loop, daemon=True)
_state.event_loop_thread.start()
while _state.event_loop is None:
threading.Event().wait(0.01)
async def _create_browser() -> Browser:
if _state.browser is not None and _state.browser.is_connected():
return _state.browser
if _state.browser is not None:
with contextlib.suppress(Exception):
await _state.browser.close()
_state.browser = None
if _state.playwright is not None:
with contextlib.suppress(Exception):
await _state.playwright.stop()
_state.playwright = None
_state.playwright = await async_playwright().start()
_state.browser = await _state.playwright.chromium.launch(
headless=True,
args=[
"--no-sandbox",
"--disable-dev-shm-usage",
"--disable-gpu",
"--disable-web-security",
],
)
return _state.browser
def _get_browser() -> tuple[asyncio.AbstractEventLoop, Browser]:
with _state.lock:
_ensure_event_loop()
assert _state.event_loop is not None
if _state.browser is None or not _state.browser.is_connected():
future = asyncio.run_coroutine_threadsafe(_create_browser(), _state.event_loop)
future.result(timeout=30)
assert _state.browser is not None
return _state.event_loop, _state.browser
class BrowserInstance:
def __init__(self) -> None:
self.is_running = True
self._execution_lock = threading.Lock()
self._loop: asyncio.AbstractEventLoop | None = None
self._browser: Browser | None = None
self.context: BrowserContext | None = None
self.pages: dict[str, Page] = {}
self.current_page_id: str | None = None
self._next_tab_id = 1
self.console_logs: dict[str, list[dict[str, Any]]] = {}
def _run_async(self, coro: Any) -> dict[str, Any]:
if not self._loop or not self.is_running:
raise RuntimeError("Browser instance is not running")
future = asyncio.run_coroutine_threadsafe(coro, self._loop)
return cast("dict[str, Any]", future.result(timeout=30)) # 30 second timeout
async def _setup_console_logging(self, page: Page, tab_id: str) -> None:
self.console_logs[tab_id] = []
def handle_console(msg: Any) -> None:
text = msg.text
if len(text) > MAX_INDIVIDUAL_LOG_LENGTH:
text = text[:MAX_INDIVIDUAL_LOG_LENGTH] + "... [TRUNCATED]"
log_entry = {
"type": msg.type,
"text": text,
"location": msg.location,
"timestamp": asyncio.get_event_loop().time(),
}
self.console_logs[tab_id].append(log_entry)
if len(self.console_logs[tab_id]) > MAX_CONSOLE_LOGS_COUNT:
self.console_logs[tab_id] = self.console_logs[tab_id][-MAX_CONSOLE_LOGS_COUNT:]
page.on("console", handle_console)
async def _create_context(self, url: str | None = None) -> dict[str, Any]:
assert self._browser is not None
self.context = await self._browser.new_context(
viewport={"width": 1280, "height": 720},
user_agent=(
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
),
)
page = await self.context.new_page()
tab_id = f"tab_{self._next_tab_id}"
self._next_tab_id += 1
self.pages[tab_id] = page
self.current_page_id = tab_id
await self._setup_console_logging(page, tab_id)
if url:
await page.goto(url, wait_until="domcontentloaded")
return await self._get_page_state(tab_id)
async def _get_page_state(self, tab_id: str | None = None) -> dict[str, Any]:
if not tab_id:
tab_id = self.current_page_id
if not tab_id or tab_id not in self.pages:
raise ValueError(f"Tab '{tab_id}' not found")
page = self.pages[tab_id]
await asyncio.sleep(2)
screenshot_bytes = await page.screenshot(type="png", full_page=False)
screenshot_b64 = base64.b64encode(screenshot_bytes).decode("utf-8")
url = page.url
title = await page.title()
viewport = page.viewport_size
all_tabs = {}
for tid, tab_page in self.pages.items():
all_tabs[tid] = {
"url": tab_page.url,
"title": await tab_page.title() if not tab_page.is_closed() else "Closed",
}
return {
"screenshot": screenshot_b64,
"url": url,
"title": title,
"viewport": viewport,
"tab_id": tab_id,
"all_tabs": all_tabs,
}
def launch(self, url: str | None = None) -> dict[str, Any]:
with self._execution_lock:
if self.context is not None:
raise ValueError("Browser is already launched")
self._loop, self._browser = _get_browser()
return self._run_async(self._create_context(url))
def goto(self, url: str, tab_id: str | None = None) -> dict[str, Any]:
with self._execution_lock:
return self._run_async(self._goto(url, tab_id))
async def _goto(self, url: str, tab_id: str | None = None) -> dict[str, Any]:
if not tab_id:
tab_id = self.current_page_id
if not tab_id or tab_id not in self.pages:
raise ValueError(f"Tab '{tab_id}' not found")
page = self.pages[tab_id]
await page.goto(url, wait_until="domcontentloaded")
return await self._get_page_state(tab_id)
def click(self, coordinate: str, tab_id: str | None = None) -> dict[str, Any]:
with self._execution_lock:
return self._run_async(self._click(coordinate, tab_id))
async def _click(self, coordinate: str, tab_id: str | None = None) -> dict[str, Any]:
if not tab_id:
tab_id = self.current_page_id
if not tab_id or tab_id not in self.pages:
raise ValueError(f"Tab '{tab_id}' not found")
try:
x, y = map(int, coordinate.split(","))
except ValueError as e:
raise ValueError(f"Invalid coordinate format: {coordinate}. Use 'x,y'") from e
page = self.pages[tab_id]
await page.mouse.click(x, y)
return await self._get_page_state(tab_id)
def type_text(self, text: str, tab_id: str | None = None) -> dict[str, Any]:
with self._execution_lock:
return self._run_async(self._type_text(text, tab_id))
async def _type_text(self, text: str, tab_id: str | None = None) -> dict[str, Any]:
if not tab_id:
tab_id = self.current_page_id
if not tab_id or tab_id not in self.pages:
raise ValueError(f"Tab '{tab_id}' not found")
page = self.pages[tab_id]
await page.keyboard.type(text)
return await self._get_page_state(tab_id)
def scroll(self, direction: str, tab_id: str | None = None) -> dict[str, Any]:
with self._execution_lock:
return self._run_async(self._scroll(direction, tab_id))
async def _scroll(self, direction: str, tab_id: str | None = None) -> dict[str, Any]:
if not tab_id:
tab_id = self.current_page_id
if not tab_id or tab_id not in self.pages:
raise ValueError(f"Tab '{tab_id}' not found")
page = self.pages[tab_id]
if direction == "down":
await page.keyboard.press("PageDown")
elif direction == "up":
await page.keyboard.press("PageUp")
else:
raise ValueError(f"Invalid scroll direction: {direction}")
return await self._get_page_state(tab_id)
def back(self, tab_id: str | None = None) -> dict[str, Any]:
with self._execution_lock:
return self._run_async(self._back(tab_id))
async def _back(self, tab_id: str | None = None) -> dict[str, Any]:
if not tab_id:
tab_id = self.current_page_id
if not tab_id or tab_id not in self.pages:
raise ValueError(f"Tab '{tab_id}' not found")
page = self.pages[tab_id]
await page.go_back(wait_until="domcontentloaded")
return await self._get_page_state(tab_id)
def forward(self, tab_id: str | None = None) -> dict[str, Any]:
with self._execution_lock:
return self._run_async(self._forward(tab_id))
async def _forward(self, tab_id: str | None = None) -> dict[str, Any]:
if not tab_id:
tab_id = self.current_page_id
if not tab_id or tab_id not in self.pages:
raise ValueError(f"Tab '{tab_id}' not found")
page = self.pages[tab_id]
await page.go_forward(wait_until="domcontentloaded")
return await self._get_page_state(tab_id)
def new_tab(self, url: str | None = None) -> dict[str, Any]:
with self._execution_lock:
return self._run_async(self._new_tab(url))
async def _new_tab(self, url: str | None = None) -> dict[str, Any]:
if not self.context:
raise ValueError("Browser not launched")
page = await self.context.new_page()
tab_id = f"tab_{self._next_tab_id}"
self._next_tab_id += 1
self.pages[tab_id] = page
self.current_page_id = tab_id
await self._setup_console_logging(page, tab_id)
if url:
await page.goto(url, wait_until="domcontentloaded")
return await self._get_page_state(tab_id)
def switch_tab(self, tab_id: str) -> dict[str, Any]:
with self._execution_lock:
return self._run_async(self._switch_tab(tab_id))
async def _switch_tab(self, tab_id: str) -> dict[str, Any]:
if tab_id not in self.pages:
raise ValueError(f"Tab '{tab_id}' not found")
self.current_page_id = tab_id
return await self._get_page_state(tab_id)
def close_tab(self, tab_id: str) -> dict[str, Any]:
with self._execution_lock:
return self._run_async(self._close_tab(tab_id))
async def _close_tab(self, tab_id: str) -> dict[str, Any]:
if tab_id not in self.pages:
raise ValueError(f"Tab '{tab_id}' not found")
if len(self.pages) == 1:
raise ValueError("Cannot close the last tab")
page = self.pages.pop(tab_id)
await page.close()
if tab_id in self.console_logs:
del self.console_logs[tab_id]
if self.current_page_id == tab_id:
self.current_page_id = next(iter(self.pages.keys()))
return await self._get_page_state(self.current_page_id)
def wait(self, duration: float, tab_id: str | None = None) -> dict[str, Any]:
with self._execution_lock:
return self._run_async(self._wait(duration, tab_id))
async def _wait(self, duration: float, tab_id: str | None = None) -> dict[str, Any]:
await asyncio.sleep(duration)
return await self._get_page_state(tab_id)
def execute_js(self, js_code: str, tab_id: str | None = None) -> dict[str, Any]:
with self._execution_lock:
return self._run_async(self._execute_js(js_code, tab_id))
async def _execute_js(self, js_code: str, tab_id: str | None = None) -> dict[str, Any]:
if not tab_id:
tab_id = self.current_page_id
if not tab_id or tab_id not in self.pages:
raise ValueError(f"Tab '{tab_id}' not found")
page = self.pages[tab_id]
try:
result = await page.evaluate(js_code)
except Exception as e: # noqa: BLE001
result = {
"error": True,
"error_type": type(e).__name__,
"error_message": str(e),
}
result_str = str(result)
if len(result_str) > MAX_JS_RESULT_LENGTH:
result = result_str[:MAX_JS_RESULT_LENGTH] + "... [JS result truncated at 5k chars]"
state = await self._get_page_state(tab_id)
state["js_result"] = result
return state
def get_console_logs(self, tab_id: str | None = None, clear: bool = False) -> dict[str, Any]:
with self._execution_lock:
return self._run_async(self._get_console_logs(tab_id, clear))
async def _get_console_logs(
self, tab_id: str | None = None, clear: bool = False
) -> dict[str, Any]:
if not tab_id:
tab_id = self.current_page_id
if not tab_id or tab_id not in self.pages:
raise ValueError(f"Tab '{tab_id}' not found")
logs = self.console_logs.get(tab_id, [])
total_length = sum(len(str(log)) for log in logs)
if total_length > MAX_CONSOLE_LOG_LENGTH:
truncated_logs: list[dict[str, Any]] = []
current_length = 0
for log in reversed(logs):
log_length = len(str(log))
if current_length + log_length <= MAX_CONSOLE_LOG_LENGTH:
truncated_logs.insert(0, log)
current_length += log_length
else:
break
if len(truncated_logs) < len(logs):
truncation_notice = {
"type": "info",
"text": (
f"[TRUNCATED: {len(logs) - len(truncated_logs)} older logs "
f"removed to stay within {MAX_CONSOLE_LOG_LENGTH} character limit]"
),
"location": {},
"timestamp": 0,
}
truncated_logs.insert(0, truncation_notice)
logs = truncated_logs
if clear:
self.console_logs[tab_id] = []
state = await self._get_page_state(tab_id)
state["console_logs"] = logs
return state
def view_source(self, tab_id: str | None = None) -> dict[str, Any]:
with self._execution_lock:
return self._run_async(self._view_source(tab_id))
async def _view_source(self, tab_id: str | None = None) -> dict[str, Any]:
if not tab_id:
tab_id = self.current_page_id
if not tab_id or tab_id not in self.pages:
raise ValueError(f"Tab '{tab_id}' not found")
page = self.pages[tab_id]
source = await page.content()
original_length = len(source)
if original_length > MAX_PAGE_SOURCE_LENGTH:
truncation_message = (
f"\n\n<!-- [TRUNCATED: {original_length - MAX_PAGE_SOURCE_LENGTH} "
"characters removed] -->\n\n"
)
available_space = MAX_PAGE_SOURCE_LENGTH - len(truncation_message)
truncate_point = available_space // 2
source = source[:truncate_point] + truncation_message + source[-truncate_point:]
state = await self._get_page_state(tab_id)
state["page_source"] = source
return state
def double_click(self, coordinate: str, tab_id: str | None = None) -> dict[str, Any]:
with self._execution_lock:
return self._run_async(self._double_click(coordinate, tab_id))
async def _double_click(self, coordinate: str, tab_id: str | None = None) -> dict[str, Any]:
if not tab_id:
tab_id = self.current_page_id
if not tab_id or tab_id not in self.pages:
raise ValueError(f"Tab '{tab_id}' not found")
try:
x, y = map(int, coordinate.split(","))
except ValueError as e:
raise ValueError(f"Invalid coordinate format: {coordinate}. Use 'x,y'") from e
page = self.pages[tab_id]
await page.mouse.dblclick(x, y)
return await self._get_page_state(tab_id)
def hover(self, coordinate: str, tab_id: str | None = None) -> dict[str, Any]:
with self._execution_lock:
return self._run_async(self._hover(coordinate, tab_id))
async def _hover(self, coordinate: str, tab_id: str | None = None) -> dict[str, Any]:
if not tab_id:
tab_id = self.current_page_id
if not tab_id or tab_id not in self.pages:
raise ValueError(f"Tab '{tab_id}' not found")
try:
x, y = map(int, coordinate.split(","))
except ValueError as e:
raise ValueError(f"Invalid coordinate format: {coordinate}. Use 'x,y'") from e
page = self.pages[tab_id]
await page.mouse.move(x, y)
return await self._get_page_state(tab_id)
def press_key(self, key: str, tab_id: str | None = None) -> dict[str, Any]:
with self._execution_lock:
return self._run_async(self._press_key(key, tab_id))
async def _press_key(self, key: str, tab_id: str | None = None) -> dict[str, Any]:
if not tab_id:
tab_id = self.current_page_id
if not tab_id or tab_id not in self.pages:
raise ValueError(f"Tab '{tab_id}' not found")
page = self.pages[tab_id]
await page.keyboard.press(key)
return await self._get_page_state(tab_id)
def save_pdf(self, file_path: str, tab_id: str | None = None) -> dict[str, Any]:
with self._execution_lock:
return self._run_async(self._save_pdf(file_path, tab_id))
async def _save_pdf(self, file_path: str, tab_id: str | None = None) -> dict[str, Any]:
if not tab_id:
tab_id = self.current_page_id
if not tab_id or tab_id not in self.pages:
raise ValueError(f"Tab '{tab_id}' not found")
if not Path(file_path).is_absolute():
file_path = str(Path("/workspace") / file_path)
page = self.pages[tab_id]
await page.pdf(path=file_path)
state = await self._get_page_state(tab_id)
state["pdf_saved"] = file_path
return state
def close(self) -> None:
with self._execution_lock:
self.is_running = False
if self._loop and self.context:
future = asyncio.run_coroutine_threadsafe(self._close_context(), self._loop)
with contextlib.suppress(Exception):
future.result(timeout=5)
self.pages.clear()
self.console_logs.clear()
self.current_page_id = None
self.context = None
async def _close_context(self) -> None:
try:
if self.context:
await self.context.close()
except (OSError, RuntimeError) as e:
logger.warning(f"Error closing context: {e}")
def is_alive(self) -> bool:
return (
self.is_running
and self.context is not None
and self._browser is not None
and self._browser.is_connected()
)

View file

@ -0,0 +1,542 @@
import asyncio
import atexit
import contextlib
import logging
import threading
import time
from typing import Any
logger = logging.getLogger(__name__)
# Maximum consecutive task failures before the session is forcibly invalidated.
_MAX_CONSECUTIVE_FAILURES = 3
# Maximum session age (seconds) before forcing a fresh Browser object on next task.
_MAX_SESSION_AGE = 1800 # 30 minutes
# CDP recovery: how long to wait for watchdog restart (seconds) and poll interval.
_CDP_RECOVERY_TIMEOUT = 60
_CDP_RECOVERY_INTERVAL = 2
# ---------------------------------------------------------------------------
# Persistent browser session
# ---------------------------------------------------------------------------
# browser-use's Agent.run() calls browser_session.start() internally, which
# is idempotent — it only connects to CDP if _cdp_client_root is None.
# We therefore do NOT call browser.start() during launch; we just create the
# Browser object with the CDP URL and let Agent.run() manage the connection
# on the caller's event loop.
# ---------------------------------------------------------------------------
class _BrowserSession:
__slots__ = (
"browser",
"cdp_url",
"consecutive_failures",
"created_at",
"invalidated",
"local",
"profile_directory",
"task_count",
"ws_url",
)
def __init__(
self,
browser: Any,
cdp_url: str,
ws_url: str,
*,
local: bool = False,
profile_directory: str | None = None,
) -> None:
self.browser = browser
self.cdp_url = cdp_url
self.ws_url = ws_url
self.created_at = time.monotonic()
self.task_count = 0
self.consecutive_failures = 0
self.invalidated = False
self.local = local
self.profile_directory = profile_directory
@property
def age(self) -> float:
return time.monotonic() - self.created_at
@property
def needs_refresh(self) -> bool:
"""Session should get a fresh Browser object before the next task."""
return (
self.invalidated
or self.consecutive_failures >= _MAX_CONSECUTIVE_FAILURES
or self.age > _MAX_SESSION_AGE
)
_sessions: dict[str, _BrowserSession] = {}
_lock = threading.Lock()
async def _safe_close_browser(browser: Any, label: str = "") -> None:
"""Best-effort close/stop of a Browser object, swallowing all errors."""
if browser is None:
return
tag = f" ({label})" if label else ""
for method_name in ("close", "stop"):
fn = getattr(browser, method_name, None)
if not callable(fn):
continue
try:
coro = fn()
if asyncio.iscoroutine(coro):
await asyncio.wait_for(coro, timeout=10)
except TimeoutError:
logger.warning("Browser%s %s() timed out after 10s", tag, method_name)
except Exception as exc: # noqa: BLE001
logger.debug("Browser%s %s() failed: %s", tag, method_name, exc)
else:
logger.debug("Browser%s closed via %s()", tag, method_name)
return
async def _refresh_browser(session: _BrowserSession) -> None:
"""Close the existing Browser and create a fresh one.
For sandboxed sessions: reconnects via CDP (the primary recovery path when
the WebSocket is stale after a crash, timeout, or too many failures).
For local sessions: creates a fresh ``Browser.from_system_chrome()`` with
the same profile directory.
"""
old_browser = session.browser
session.browser = None
# Close the old browser in the background — don't let it block recovery.
await _safe_close_browser(old_browser, label="stale")
from browser_use import Browser
if session.local:
kwargs: dict[str, Any] = {}
if session.profile_directory:
kwargs["profile_directory"] = session.profile_directory
session.browser = Browser.from_system_chrome(**kwargs)
session.invalidated = False
session.consecutive_failures = 0
logger.info(
"Local session refreshed: new Browser.from_system_chrome(profile=%s)",
session.profile_directory or "auto",
)
return
ws_url, _ = await asyncio.to_thread(
_wait_for_cdp,
session.cdp_url,
max_attempts=10,
interval=2.0,
)
session.browser = Browser(cdp_url=ws_url)
session.ws_url = ws_url
session.invalidated = False
session.consecutive_failures = 0
logger.info(
"Session refreshed: new Browser with ws=%s (old ws was stale)",
ws_url,
)
def _wait_for_cdp(
cdp_url: str, max_attempts: int = 30, interval: float = 1.0
) -> tuple[str, dict[str, Any]]:
"""Block until the CDP endpoint at *cdp_url* responds to ``/json/version``.
Returns ``(ws_url, version_info)`` where *ws_url* is the WebSocket
debugger URL rewritten to be reachable from the host. Chromium inside
Docker reports ``ws://127.0.0.1:<internal_port>/...`` which is not
reachable from the host we replace the host:port with the values from
*cdp_url* (the Docker-mapped endpoint).
The Chromium process inside the sandbox container may still be starting
when the tool server is already healthy. We poll the CDP ``/json/version``
endpoint before handing the URL to browser-use.
"""
from urllib.parse import urlparse
import httpx
version_url = cdp_url.rstrip("/") + "/json/version"
last_error: str = ""
for attempt in range(1, max_attempts + 1):
try:
with httpx.Client(trust_env=False, timeout=5) as client:
resp = client.get(version_url)
if resp.status_code == 200 and "webSocketDebuggerUrl" in resp.text:
version_info = resp.json()
raw_ws = version_info.get("webSocketDebuggerUrl", "")
# Rewrite the WebSocket URL: Chromium reports the container-
# internal address (e.g. ws://127.0.0.1:19222/devtools/...)
# but we need it on the Docker-mapped host:port.
parsed_cdp = urlparse(cdp_url)
parsed_ws = urlparse(raw_ws)
ws_url = raw_ws.replace(
f"{parsed_ws.hostname}:{parsed_ws.port}",
f"{parsed_cdp.hostname}:{parsed_cdp.port}",
)
logger.info(
"CDP ready at %s (attempt %d): browser=%s, ws_raw=%s, ws_rewritten=%s",
cdp_url,
attempt,
version_info.get("Browser", "unknown"),
raw_ws,
ws_url,
)
return ws_url, version_info
last_error = f"HTTP {resp.status_code}, body={resp.text[:200]}"
logger.debug("CDP not ready at %s (attempt %d): %s", cdp_url, attempt, last_error)
except httpx.ConnectError as e:
last_error = f"ConnectError: {e}"
logger.debug("CDP connect failed (attempt %d): %s", attempt, last_error)
except httpx.TimeoutException as e:
last_error = f"Timeout: {e}"
logger.debug("CDP timeout (attempt %d): %s", attempt, last_error)
except httpx.RequestError as e:
last_error = f"RequestError({type(e).__name__}): {e}"
logger.debug("CDP request error (attempt %d): %s", attempt, last_error)
time.sleep(interval)
raise ConnectionError(
f"Chromium CDP at {cdp_url} did not become ready after {max_attempts}s. "
f"Last error: {last_error}. "
"The sandbox browser may have failed to start — check container logs."
)
async def _launch_browser(cdp_url: str, agent_id: str) -> _BrowserSession:
"""Create a new browser session connected to the sandbox browser via CDP.
The *cdp_url* points to the Chromium instance running inside the Docker
sandbox (exposed via port mapping). browser-use's ``Browser`` connects
over the Chrome DevTools Protocol rather than launching a local process.
We do NOT call ``browser.start()`` here ``Agent.run()`` does that
internally (it's idempotent), which ensures the CDP connection is
established on the correct event loop.
"""
with _lock:
if agent_id in _sessions:
logger.info("Reusing existing browser session for agent %s", agent_id)
return _sessions[agent_id]
logger.info("Launching browser for agent %s: cdp_url=%s", agent_id, cdp_url)
# Wait for the container's Chromium to be ready and get the rewritten WS URL.
# We pass the ws:// URL directly so browser-use skips its own /json/version
# fetch (which would get the container-internal address).
ws_url, version_info = await asyncio.to_thread(_wait_for_cdp, cdp_url)
logger.info(
"Chromium version: %s, protocol: %s, user-agent: %s",
version_info.get("Browser", "?"),
version_info.get("Protocol-Version", "?"),
version_info.get("User-Agent", "?")[:80],
)
from browser_use import Browser
browser = Browser(cdp_url=ws_url)
logger.info(
"Browser object created for agent %s (ws=%s) — CDP connect deferred to Agent.run()",
agent_id,
ws_url,
)
session = _BrowserSession(browser=browser, cdp_url=cdp_url, ws_url=ws_url)
with _lock:
if agent_id in _sessions:
# Another thread raced us; discard ours.
logger.info("Race: another session appeared for agent %s, discarding ours", agent_id)
return _sessions[agent_id]
_sessions[agent_id] = session
return session
async def _launch_local_browser(
agent_id: str, profile_directory: str | None = None
) -> _BrowserSession:
"""Create a browser session using the local system Chrome.
Uses ``Browser.from_system_chrome()`` which auto-detects the Chrome
executable and user data directory. An optional *profile_directory*
(e.g. ``"Profile 1"``, ``"Default"``) selects a specific Chrome profile.
"""
with _lock:
if agent_id in _sessions:
logger.info("Reusing existing browser session for agent %s", agent_id)
return _sessions[agent_id]
logger.info(
"Launching local browser for agent %s (profile=%s)",
agent_id,
profile_directory or "auto",
)
from browser_use import Browser
kwargs: dict[str, Any] = {}
if profile_directory:
kwargs["profile_directory"] = profile_directory
browser = Browser.from_system_chrome(headless=False, **kwargs)
logger.info(
"Local Browser object created for agent %s (profile=%s)",
agent_id,
profile_directory or "auto",
)
session = _BrowserSession(
browser=browser,
cdp_url="",
ws_url="",
local=True,
profile_directory=profile_directory,
)
with _lock:
if agent_id in _sessions:
logger.info("Race: another session appeared for agent %s, discarding ours", agent_id)
return _sessions[agent_id]
_sessions[agent_id] = session
return session
def _get_session(agent_id: str) -> _BrowserSession:
"""Return the existing session for *agent_id*.
Raises ``ValueError`` if no session exists (i.e. ``launch`` was not called).
"""
with _lock:
session = _sessions.get(agent_id)
if session is None:
with _lock:
active = list(_sessions.keys())
raise ValueError(
"Browser not launched. You must call browser_use_local_action "
f"with action='launch' before running tasks. "
f"Active sessions: {active}, requested agent_id: {agent_id}"
)
return session
async def _shutdown_session(session: _BrowserSession) -> None:
"""Tear down a session's browser."""
logger.info("Shutting down browser session (cdp_url=%s)", session.cdp_url)
session.invalidated = True
await _safe_close_browser(session.browser, label="shutdown")
session.browser = None
logger.info("Browser session shut down")
async def _close_session(agent_id: str) -> None:
"""Tear down the browser session for *agent_id*."""
with _lock:
session = _sessions.pop(agent_id, None)
if session is None:
logger.debug("close_session called but no session for agent %s", agent_id)
return
await _shutdown_session(session)
def cleanup_agent(agent_id: str) -> None:
"""Best-effort sync cleanup, called from non-async contexts."""
with _lock:
session = _sessions.pop(agent_id, None)
if session is None:
return
try:
loop = asyncio.get_event_loop()
if loop.is_running():
asyncio.run_coroutine_threadsafe(_shutdown_session(session), loop)
else:
loop.run_until_complete(_shutdown_session(session))
except Exception: # noqa: BLE001
logger.debug("cleanup_agent: best-effort shutdown failed for %s", agent_id)
def _close_all() -> None:
with _lock:
agent_ids = list(_sessions.keys())
for aid in agent_ids:
with contextlib.suppress(Exception):
with _lock:
session = _sessions.pop(aid, None)
if session is not None:
# At atexit there's no running loop, so create one.
with contextlib.suppress(Exception):
asyncio.run(_shutdown_session(session))
atexit.register(_close_all)
async def _reinitialize_after_agent(session: _BrowserSession) -> None:
"""Re-create and start the Browser after Agent.run() kills it.
browser-use's ``Agent.run()`` calls ``browser_session.kill()`` on
completion, which destroys the CDP client (``_cdp_client_root = None``),
clears the event bus (removing all handlers), and creates a fresh empty
``EventBus``. Core handlers registered in ``model_post_init()`` are lost
and ``start()`` on the dead session would hang.
The fix: create a brand-new ``Browser`` object (which runs
``model_post_init()`` registers fresh handlers) pointing at the same
CDP endpoint, then call ``start()`` to connect and attach watchdogs.
This ensures subsequent granular commands (state, eval, click, etc.)
have a fully functional session.
"""
from browser_use import Browser
if session.local:
kwargs: dict[str, Any] = {}
if session.profile_directory:
kwargs["profile_directory"] = session.profile_directory
session.browser = Browser.from_system_chrome(**kwargs)
else:
session.browser = Browser(cdp_url=session.ws_url)
await session.browser.start()
logger.info("Session reinitialized after agent run (local=%s)", session.local)
# ---------------------------------------------------------------------------
# CDP health checks & recovery
# ---------------------------------------------------------------------------
def _check_cdp_alive(cdp_url: str) -> bool:
"""Quick check that the CDP endpoint is reachable (non-blocking from sync)."""
import httpx
version_url = cdp_url.rstrip("/") + "/json/version"
try:
with httpx.Client(trust_env=False, timeout=5) as client:
resp = client.get(version_url)
return resp.status_code == 200 and "webSocketDebuggerUrl" in resp.text
except Exception: # noqa: BLE001
return False
async def _wait_for_cdp_recovery(session: _BrowserSession, task_num: int) -> bool:
"""Wait for the CDP endpoint to come back after a crash/restart.
Returns True if CDP recovered, False if it's still dead after the timeout.
"""
max_attempts = int(_CDP_RECOVERY_TIMEOUT / _CDP_RECOVERY_INTERVAL)
for attempt in range(1, max_attempts + 1):
await asyncio.sleep(_CDP_RECOVERY_INTERVAL)
if await asyncio.to_thread(_check_cdp_alive, session.cdp_url):
logger.info(
"Task #%d: CDP recovered after %ds",
task_num,
attempt * _CDP_RECOVERY_INTERVAL,
)
return True
return False
async def _ensure_healthy_session(session: _BrowserSession, task_num: int) -> str | None:
"""Ensure the session has a working Browser before running a task.
Returns ``None`` on success or an error message string on failure.
"""
# --- Local sessions: only refresh on invalidation / too many failures / age ---
if session.local:
if session.needs_refresh:
reason = (
"invalidated"
if session.invalidated
else f"{session.consecutive_failures} consecutive failures"
if session.consecutive_failures >= _MAX_CONSECUTIVE_FAILURES
else f"age {session.age:.0f}s > {_MAX_SESSION_AGE}s"
)
logger.warning(
"Task #%d: local session needs refresh (%s) — rebuilding Browser",
task_num,
reason,
)
try:
await _refresh_browser(session)
except Exception as exc:
logger.exception("Task #%d: local session refresh failed", task_num)
return f"Failed to refresh local browser session: {exc}"
return None
# --- Sandboxed sessions: CDP health checks ---
# 1) If the session was invalidated (prior crash/timeout/too many failures)
# or is too old, proactively refresh the Browser object.
if session.needs_refresh:
reason = (
"invalidated"
if session.invalidated
else f"{session.consecutive_failures} consecutive failures"
if session.consecutive_failures >= _MAX_CONSECUTIVE_FAILURES
else f"age {session.age:.0f}s > {_MAX_SESSION_AGE}s"
)
logger.warning(
"Task #%d: session needs refresh (%s) — rebuilding Browser",
task_num,
reason,
)
# Make sure CDP is alive first (may need watchdog restart).
cdp_alive = await asyncio.to_thread(_check_cdp_alive, session.cdp_url)
if not cdp_alive and not await _wait_for_cdp_recovery(session, task_num):
return (
f"Chromium CDP at {session.cdp_url} is not responding after "
f"{_CDP_RECOVERY_TIMEOUT}s. The sandbox browser may have "
"crashed. Check container logs."
)
try:
await _refresh_browser(session)
except Exception as exc:
logger.exception("Task #%d: session refresh failed", task_num)
return f"Failed to refresh browser session: {exc}"
return None
# 2) Normal pre-flight: verify CDP is alive.
if await asyncio.to_thread(_check_cdp_alive, session.cdp_url):
return None # all good
# 3) CDP down — wait for watchdog, then refresh.
logger.warning(
"Task #%d: CDP not responding at %s — waiting for watchdog restart...",
task_num,
session.cdp_url,
)
if not await _wait_for_cdp_recovery(session, task_num):
session.invalidated = True
return (
f"Chromium CDP at {session.cdp_url} is not responding after "
f"{_CDP_RECOVERY_TIMEOUT}s. The sandbox browser may have crashed. "
"Check container logs."
)
try:
await _refresh_browser(session)
except Exception as exc:
session.invalidated = True
logger.exception(
"Task #%d: reconnection after CDP recovery failed",
task_num,
)
return f"Chromium restarted but reconnection failed: {exc}"
return None

331
strix/tools/browser/llm.py Normal file
View file

@ -0,0 +1,331 @@
"""
NOTE: This is a temporary workaround.
TODO: Migrate this to a standalone library and add regression tests/coverage to
ensure it works with all providers and models. I'd hate for this to fail on
some edge cases etc, and realistically, it's something tiny that the OS
community could make use of. LiteLLM is really cool.
"""
import logging
from dataclasses import dataclass, field
from typing import Any, TypeVar, overload
from browser_use.llm.base import BaseChatModel
from browser_use.llm.exceptions import ModelProviderError, ModelRateLimitError
from browser_use.llm.messages import (
AssistantMessage,
BaseMessage,
SystemMessage,
UserMessage,
)
from browser_use.llm.schema import SchemaOptimizer
from browser_use.llm.views import ChatInvokeCompletion, ChatInvokeUsage
from pydantic import BaseModel
logger = logging.getLogger(__name__)
T = TypeVar("T", bound=BaseModel)
def _serialize_messages(messages: list[BaseMessage]) -> list[dict[str, Any]]:
"""Convert browser-use messages to litellm-compatible dicts (OpenAI format).
LiteLLM accepts OpenAI-format message dicts for all providers, handling
provider-specific conversion (e.g. image blocks for Anthropic) internally.
"""
result: list[dict[str, Any]] = []
for msg in messages:
if isinstance(msg, UserMessage):
d: dict[str, Any] = {"role": "user"}
if isinstance(msg.content, str):
d["content"] = msg.content
else:
parts: list[dict[str, Any]] = []
for part in msg.content:
if part.type == "text":
parts.append({"type": "text", "text": part.text})
elif part.type == "image_url":
parts.append(
{
"type": "image_url",
"image_url": {
"url": part.image_url.url,
"detail": part.image_url.detail,
},
}
)
d["content"] = parts
if msg.name is not None:
d["name"] = msg.name
result.append(d)
elif isinstance(msg, SystemMessage):
d = {"role": "system"}
if isinstance(msg.content, str):
d["content"] = msg.content
else:
d["content"] = [{"type": "text", "text": p.text} for p in msg.content]
if msg.name is not None:
d["name"] = msg.name
result.append(d)
elif isinstance(msg, AssistantMessage):
d = {"role": "assistant"}
if msg.content is not None:
if isinstance(msg.content, str):
d["content"] = msg.content
else:
parts = []
for part in msg.content:
if part.type == "text":
parts.append({"type": "text", "text": part.text})
elif part.type == "refusal":
parts.append(
{
"type": "text",
"text": f"[Refusal] {part.refusal}",
}
)
d["content"] = parts
else:
d["content"] = None
if msg.name is not None:
d["name"] = msg.name
if msg.tool_calls:
d["tool_calls"] = [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.function.name,
"arguments": tc.function.arguments,
},
}
for tc in msg.tool_calls
]
result.append(d)
return result
@dataclass
class ChatLiteLLM(BaseChatModel):
"""Chat model that routes to any provider via LiteLLM.
Uses litellm's unified ``acompletion`` API to support all providers
(OpenAI, Anthropic, Google, Ollama, OpenRouter, DeepSeek, etc.)
through a single interface.
The ``model`` parameter uses litellm's model format, e.g.::
"gpt-4o"
"anthropic/claude-sonnet-4-20250514"
"openrouter/google/gemini-2.0-flash-001"
"ollama/llama3"
Structured output (``output_format``) is handled via litellm's
``response_format`` parameter which translates across providers.
"""
model: str
api_key: str | None = None
api_base: str | None = None
temperature: float | None = 0.0
max_tokens: int | None = 4096
max_retries: int = 3
# Resolved lazily in __post_init__
_provider_name: str = field(default="", init=False, repr=False)
_clean_model: str = field(default="", init=False, repr=False)
def __post_init__(self) -> None:
"""Resolve provider info from the model string via litellm."""
try:
from litellm import get_llm_provider
self._clean_model, self._provider_name, _, _ = get_llm_provider(self.model)
except Exception: # noqa: BLE001
if "/" in self.model:
self._provider_name, self._clean_model = self.model.split("/", 1)
else:
self._provider_name = "openai"
self._clean_model = self.model
logger.debug(
"ChatLiteLLM initialized: model=%s, provider=%s, clean=%s, api_base=%s",
self.model,
self._provider_name,
self._clean_model,
self.api_base or "(default)",
)
@property
def provider(self) -> str:
return self._provider_name or "litellm"
@property
def name(self) -> str:
return self._clean_model or self.model
# ------------------------------------------------------------------
# Usage parsing
# ------------------------------------------------------------------
@staticmethod
def _parse_usage(response: Any) -> ChatInvokeUsage | None:
"""Extract token usage from a litellm response."""
usage = getattr(response, "usage", None)
if usage is None:
return None
prompt_tokens = getattr(usage, "prompt_tokens", 0) or 0
completion_tokens = getattr(usage, "completion_tokens", 0) or 0
# Cache info — litellm exposes these at the top level for Anthropic/OpenAI
prompt_cached = getattr(usage, "cache_read_input_tokens", None)
cache_creation = getattr(usage, "cache_creation_input_tokens", None)
# Fallback: nested prompt_tokens_details (OpenAI style)
if prompt_cached is None:
details = getattr(usage, "prompt_tokens_details", None)
if details:
prompt_cached = getattr(details, "cached_tokens", None)
return ChatInvokeUsage(
prompt_tokens=prompt_tokens,
prompt_cached_tokens=int(prompt_cached) if prompt_cached is not None else None,
prompt_cache_creation_tokens=int(cache_creation)
if cache_creation is not None
else None,
prompt_image_tokens=None,
completion_tokens=completion_tokens,
total_tokens=prompt_tokens + completion_tokens,
)
# ------------------------------------------------------------------
# ainvoke — the single method browser-use calls
# ------------------------------------------------------------------
@overload
async def ainvoke(
self,
messages: list[BaseMessage],
output_format: None = None,
**kwargs: Any,
) -> ChatInvokeCompletion[str]: ...
@overload
async def ainvoke(
self,
messages: list[BaseMessage],
output_format: type[T],
**kwargs: Any,
) -> ChatInvokeCompletion[T]: ...
async def ainvoke(
self,
messages: list[BaseMessage],
output_format: type[T] | None = None,
**kwargs: Any,
) -> ChatInvokeCompletion[T] | ChatInvokeCompletion[str]:
"""Invoke the model via litellm.
Args:
messages: List of browser-use chat messages.
output_format: Optional Pydantic model class for structured output.
**kwargs: Extra keyword args (``session_id`` etc.) ignored.
Returns:
``ChatInvokeCompletion`` with either a string or parsed Pydantic model.
"""
import litellm
litellm_messages = _serialize_messages(messages)
params: dict[str, Any] = {
"model": self.model,
"messages": litellm_messages,
"num_retries": self.max_retries,
}
if self.temperature is not None:
params["temperature"] = self.temperature
if self.max_tokens is not None:
params["max_tokens"] = self.max_tokens
if self.api_key:
params["api_key"] = self.api_key
if self.api_base:
params["api_base"] = self.api_base
# Structured output via JSON schema response format.
# LiteLLM translates this across providers (OpenAI native, Anthropic
# via tool use, etc.).
if output_format is not None:
schema = SchemaOptimizer.create_optimized_json_schema(output_format)
params["response_format"] = {
"type": "json_schema",
"json_schema": {
"name": "agent_output",
"strict": True,
"schema": schema,
},
}
try:
response = await litellm.acompletion(**params)
except litellm.RateLimitError as e:
raise ModelRateLimitError(message=str(e), model=self.name) from e
except litellm.Timeout as e:
raise ModelProviderError(message=f"Request timed out: {e}", model=self.name) from e
except litellm.APIConnectionError as e:
raise ModelProviderError(message=str(e), model=self.name) from e
except litellm.APIError as e:
status = getattr(e, "status_code", 502) or 502
raise ModelProviderError(message=str(e), status_code=status, model=self.name) from e
except ModelProviderError:
raise
except Exception as e:
raise ModelProviderError(message=str(e), model=self.name) from e
# --- Parse response ---
choice = response.choices[0] if response.choices else None
if choice is None:
raise ModelProviderError(
message="Empty response: no choices returned by the model",
status_code=502,
model=self.name,
)
content = choice.message.content or ""
usage = self._parse_usage(response)
stop_reason = choice.finish_reason
# Extract thinking/reasoning content (Anthropic extended thinking,
# DeepSeek reasoning, etc.) if the provider surfaces it.
thinking: str | None = None
msg_obj = choice.message
reasoning = getattr(msg_obj, "reasoning_content", None)
if reasoning:
thinking = str(reasoning)
if output_format is not None:
if not content:
raise ModelProviderError(
message="Model returned empty content for structured output request",
status_code=500,
model=self.name,
)
parsed = output_format.model_validate_json(content)
return ChatInvokeCompletion(
completion=parsed,
thinking=thinking,
usage=usage,
stop_reason=stop_reason,
)
return ChatInvokeCompletion(
completion=content,
thinking=thinking,
usage=usage,
stop_reason=stop_reason,
)

View file

@ -1,361 +0,0 @@
import atexit
import contextlib
import threading
from typing import Any
from strix.tools.context import get_current_agent_id
from .browser_instance import BrowserInstance
class BrowserTabManager:
def __init__(self) -> None:
self._browsers_by_agent: dict[str, BrowserInstance] = {}
self._lock = threading.Lock()
self._register_cleanup_handlers()
def _get_agent_browser(self) -> BrowserInstance | None:
agent_id = get_current_agent_id()
with self._lock:
return self._browsers_by_agent.get(agent_id)
def _set_agent_browser(self, browser: BrowserInstance | None) -> None:
agent_id = get_current_agent_id()
with self._lock:
if browser is None:
self._browsers_by_agent.pop(agent_id, None)
else:
self._browsers_by_agent[agent_id] = browser
def launch_browser(self, url: str | None = None) -> dict[str, Any]:
with self._lock:
agent_id = get_current_agent_id()
if agent_id in self._browsers_by_agent:
raise ValueError("Browser is already launched")
try:
browser = BrowserInstance()
result = browser.launch(url)
self._browsers_by_agent[agent_id] = browser
result["message"] = "Browser launched successfully"
except (OSError, ValueError, RuntimeError) as e:
raise RuntimeError(f"Failed to launch browser: {e}") from e
else:
return result
def goto_url(self, url: str, tab_id: str | None = None) -> dict[str, Any]:
browser = self._get_agent_browser()
if browser is None:
raise ValueError("Browser not launched")
try:
result = browser.goto(url, tab_id)
result["message"] = f"Navigated to {url}"
except (OSError, ValueError, RuntimeError) as e:
raise RuntimeError(f"Failed to navigate to URL: {e}") from e
else:
return result
def click(self, coordinate: str, tab_id: str | None = None) -> dict[str, Any]:
browser = self._get_agent_browser()
if browser is None:
raise ValueError("Browser not launched")
try:
result = browser.click(coordinate, tab_id)
result["message"] = f"Clicked at {coordinate}"
except (OSError, ValueError, RuntimeError) as e:
raise RuntimeError(f"Failed to click: {e}") from e
else:
return result
def type_text(self, text: str, tab_id: str | None = None) -> dict[str, Any]:
browser = self._get_agent_browser()
if browser is None:
raise ValueError("Browser not launched")
try:
result = browser.type_text(text, tab_id)
result["message"] = f"Typed text: {text[:50]}{'...' if len(text) > 50 else ''}"
except (OSError, ValueError, RuntimeError) as e:
raise RuntimeError(f"Failed to type text: {e}") from e
else:
return result
def scroll(self, direction: str, tab_id: str | None = None) -> dict[str, Any]:
browser = self._get_agent_browser()
if browser is None:
raise ValueError("Browser not launched")
try:
result = browser.scroll(direction, tab_id)
result["message"] = f"Scrolled {direction}"
except (OSError, ValueError, RuntimeError) as e:
raise RuntimeError(f"Failed to scroll: {e}") from e
else:
return result
def back(self, tab_id: str | None = None) -> dict[str, Any]:
browser = self._get_agent_browser()
if browser is None:
raise ValueError("Browser not launched")
try:
result = browser.back(tab_id)
result["message"] = "Navigated back"
except (OSError, ValueError, RuntimeError) as e:
raise RuntimeError(f"Failed to go back: {e}") from e
else:
return result
def forward(self, tab_id: str | None = None) -> dict[str, Any]:
browser = self._get_agent_browser()
if browser is None:
raise ValueError("Browser not launched")
try:
result = browser.forward(tab_id)
result["message"] = "Navigated forward"
except (OSError, ValueError, RuntimeError) as e:
raise RuntimeError(f"Failed to go forward: {e}") from e
else:
return result
def new_tab(self, url: str | None = None) -> dict[str, Any]:
browser = self._get_agent_browser()
if browser is None:
raise ValueError("Browser not launched")
try:
result = browser.new_tab(url)
result["message"] = f"Created new tab {result.get('tab_id', '')}"
except (OSError, ValueError, RuntimeError) as e:
raise RuntimeError(f"Failed to create new tab: {e}") from e
else:
return result
def switch_tab(self, tab_id: str) -> dict[str, Any]:
browser = self._get_agent_browser()
if browser is None:
raise ValueError("Browser not launched")
try:
result = browser.switch_tab(tab_id)
result["message"] = f"Switched to tab {tab_id}"
except (OSError, ValueError, RuntimeError) as e:
raise RuntimeError(f"Failed to switch tab: {e}") from e
else:
return result
def close_tab(self, tab_id: str) -> dict[str, Any]:
browser = self._get_agent_browser()
if browser is None:
raise ValueError("Browser not launched")
try:
result = browser.close_tab(tab_id)
result["message"] = f"Closed tab {tab_id}"
except (OSError, ValueError, RuntimeError) as e:
raise RuntimeError(f"Failed to close tab: {e}") from e
else:
return result
def wait_browser(self, duration: float, tab_id: str | None = None) -> dict[str, Any]:
browser = self._get_agent_browser()
if browser is None:
raise ValueError("Browser not launched")
try:
result = browser.wait(duration, tab_id)
result["message"] = f"Waited {duration}s"
except (OSError, ValueError, RuntimeError) as e:
raise RuntimeError(f"Failed to wait: {e}") from e
else:
return result
def execute_js(self, js_code: str, tab_id: str | None = None) -> dict[str, Any]:
browser = self._get_agent_browser()
if browser is None:
raise ValueError("Browser not launched")
try:
result = browser.execute_js(js_code, tab_id)
result["message"] = "JavaScript executed successfully"
except (OSError, ValueError, RuntimeError) as e:
raise RuntimeError(f"Failed to execute JavaScript: {e}") from e
else:
return result
def double_click(self, coordinate: str, tab_id: str | None = None) -> dict[str, Any]:
browser = self._get_agent_browser()
if browser is None:
raise ValueError("Browser not launched")
try:
result = browser.double_click(coordinate, tab_id)
result["message"] = f"Double clicked at {coordinate}"
except (OSError, ValueError, RuntimeError) as e:
raise RuntimeError(f"Failed to double click: {e}") from e
else:
return result
def hover(self, coordinate: str, tab_id: str | None = None) -> dict[str, Any]:
browser = self._get_agent_browser()
if browser is None:
raise ValueError("Browser not launched")
try:
result = browser.hover(coordinate, tab_id)
result["message"] = f"Hovered at {coordinate}"
except (OSError, ValueError, RuntimeError) as e:
raise RuntimeError(f"Failed to hover: {e}") from e
else:
return result
def press_key(self, key: str, tab_id: str | None = None) -> dict[str, Any]:
browser = self._get_agent_browser()
if browser is None:
raise ValueError("Browser not launched")
try:
result = browser.press_key(key, tab_id)
result["message"] = f"Pressed key {key}"
except (OSError, ValueError, RuntimeError) as e:
raise RuntimeError(f"Failed to press key: {e}") from e
else:
return result
def save_pdf(self, file_path: str, tab_id: str | None = None) -> dict[str, Any]:
browser = self._get_agent_browser()
if browser is None:
raise ValueError("Browser not launched")
try:
result = browser.save_pdf(file_path, tab_id)
result["message"] = f"Page saved as PDF: {file_path}"
except (OSError, ValueError, RuntimeError) as e:
raise RuntimeError(f"Failed to save PDF: {e}") from e
else:
return result
def get_console_logs(self, tab_id: str | None = None, clear: bool = False) -> dict[str, Any]:
browser = self._get_agent_browser()
if browser is None:
raise ValueError("Browser not launched")
try:
result = browser.get_console_logs(tab_id, clear)
action_text = "cleared and retrieved" if clear else "retrieved"
logs = result.get("console_logs", [])
truncated = any(log.get("text", "").startswith("[TRUNCATED:") for log in logs)
truncated_text = " (truncated)" if truncated else ""
result["message"] = (
f"Console logs {action_text} for tab "
f"{result.get('tab_id', 'current')}{truncated_text}"
)
except (OSError, ValueError, RuntimeError) as e:
raise RuntimeError(f"Failed to get console logs: {e}") from e
else:
return result
def view_source(self, tab_id: str | None = None) -> dict[str, Any]:
browser = self._get_agent_browser()
if browser is None:
raise ValueError("Browser not launched")
try:
result = browser.view_source(tab_id)
result["message"] = "Page source retrieved"
except (OSError, ValueError, RuntimeError) as e:
raise RuntimeError(f"Failed to get page source: {e}") from e
else:
return result
def list_tabs(self) -> dict[str, Any]:
browser = self._get_agent_browser()
if browser is None:
return {"tabs": {}, "total_count": 0, "current_tab": None}
try:
tab_info = {}
for tid, tab_page in browser.pages.items():
try:
tab_info[tid] = {
"url": tab_page.url,
"title": "Unknown" if tab_page.is_closed() else "Active",
"is_current": tid == browser.current_page_id,
}
except (AttributeError, RuntimeError):
tab_info[tid] = {
"url": "Unknown",
"title": "Closed",
"is_current": False,
}
return {
"tabs": tab_info,
"total_count": len(tab_info),
"current_tab": browser.current_page_id,
}
except (OSError, ValueError, RuntimeError) as e:
raise RuntimeError(f"Failed to list tabs: {e}") from e
def close_browser(self) -> dict[str, Any]:
agent_id = get_current_agent_id()
with self._lock:
browser = self._browsers_by_agent.pop(agent_id, None)
if browser is None:
raise ValueError("Browser not launched")
try:
browser.close()
except (OSError, ValueError, RuntimeError) as e:
raise RuntimeError(f"Failed to close browser: {e}") from e
else:
return {
"message": "Browser closed successfully",
"screenshot": "",
"is_running": False,
}
def cleanup_agent(self, agent_id: str) -> None:
with self._lock:
browser = self._browsers_by_agent.pop(agent_id, None)
if browser:
with contextlib.suppress(Exception):
browser.close()
def cleanup_dead_browser(self) -> None:
with self._lock:
dead_agents = []
for agent_id, browser in self._browsers_by_agent.items():
if not browser.is_alive():
dead_agents.append(agent_id)
for agent_id in dead_agents:
browser = self._browsers_by_agent.pop(agent_id)
with contextlib.suppress(Exception):
browser.close()
def close_all(self) -> None:
with self._lock:
browsers = list(self._browsers_by_agent.values())
self._browsers_by_agent.clear()
for browser in browsers:
with contextlib.suppress(Exception):
browser.close()
def _register_cleanup_handlers(self) -> None:
atexit.register(self.close_all)
_browser_tab_manager = BrowserTabManager()
def get_browser_tab_manager() -> BrowserTabManager:
return _browser_tab_manager

View file

@ -99,10 +99,18 @@ async def _execute_tool_in_sandbox(tool_name: str, agent_state: Any, **kwargs: A
async def _execute_tool_locally(tool_name: str, agent_state: Any | None, **kwargs: Any) -> Any:
from strix.tools.context import set_current_agent_id
tool_func = get_tool_by_name(tool_name)
if not tool_func:
raise ValueError(f"Tool '{tool_name}' not found")
# Propagate agent_id so tools can scope per-agent resources (e.g. browser instances).
# NOTE: This is needed for browser_use tools to work correctly.
agent_id = getattr(agent_state, "agent_id", None) if agent_state else None
if agent_id:
set_current_agent_id(agent_id)
converted_kwargs = convert_arguments(tool_func, kwargs)
if needs_agent_state(tool_name):