ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

refactor(tui): move browser connect/disconnect helpers into methods_browser.py

Browse Source
This commit is contained in:
Teknium
2026-09-02 11:13:52 -07:00
parent 518a19b22d
commit b1581c1985
2 changed files with 281 additions and 263 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tui_gateway/methods_browser.py
+279
View File
@@ -0,0 +1,279 @@
"""Browser connect/disconnect helpers for the browser.* RPCs (CDP probing, no network I/O on status).
Bodies are rebound onto server.py's globals at install time (see
method_ctx.bind_module), so they reference server.py globals bare.
"""
from __future__ import annotations
from .method_ctx import HandlerRegistry, bind_module
_registry = HandlerRegistry()
# ── Methods: browser / plugins / cron / skills ───────────────────────
def _resolve_browser_cdp_url() -> str:
"""Return the configured browser CDP override without network I/O.
``/browser status`` must be fast — calling
``tools.browser_tool._get_cdp_override`` would invoke
``_resolve_cdp_override``, which performs an HTTP probe to
``.../json/version`` for discovery-style URLs. That probe has
a multi-second timeout and would block the TUI on a slow or
unreachable host even though status only needs to report whether
an override is set.
Mirrors the env/config precedence of ``_get_cdp_override`` (env
var first, then ``browser.cdp_url`` from config.yaml) without the
websocket-resolution step, so the answer reflects user intent
even when the configured host is not currently reachable. The
actual WS normalization happens in ``browser_navigate`` on the
next tool call.
"""
env_url = os.environ.get("BROWSER_CDP_URL", "").strip()
if env_url:
return env_url
try:
from hermes_cli.config import read_raw_config
cfg = read_raw_config()
browser_cfg = cfg.get("browser", {}) if isinstance(cfg, dict) else {}
if isinstance(browser_cfg, dict):
return str(browser_cfg.get("cdp_url", "") or "").strip()
except Exception:
pass
return ""
def _is_default_local_cdp(parsed) -> bool:
"""Match the discovery-style local default; never the concrete WS form.
A user-supplied ``ws://127.0.0.1:9222/devtools/browser/<id>`` is a
real, connectable endpoint — collapsing it to bare ``http://...:9222``
would strip the path and break the connect.
"""
try:
port = parsed.port or 80
except ValueError:
return False
discovery_path = parsed.path in {"", "/", "/json", "/json/version"}
return (
parsed.scheme in {"http", "ws"}
and parsed.hostname in {"127.0.0.1", "localhost"}
and port == 9222
and discovery_path
)
def _http_ok(url: str, timeout: float) -> bool:
import urllib.request
try:
with urllib.request.urlopen(url, timeout=timeout) as resp:
return 200 <= getattr(resp, "status", 200) < 300
except Exception:
return False
def _probe_urls(parsed) -> list[str]:
scheme = {"ws": "http", "wss": "https"}.get(parsed.scheme, parsed.scheme)
root = f"{scheme}://{parsed.netloc}".rstrip("/")
return [f"{root}/json/version", f"{root}/json"]
def _normalize_cdp_url(parsed) -> str:
# Concrete ``/devtools/browser/<id>`` endpoints (Browserbase et al.)
# are connectable as-is. Discovery-style inputs collapse to bare
# ``scheme://host:port`` so ``_resolve_cdp_override`` can append
# ``/json/version`` later without doubling the path.
if parsed.path.startswith("/devtools/browser/"):
return parsed.geturl()
return parsed._replace(path="", params="", query="", fragment="").geturl()
def _failure_messages(url: str, port: int, system: str) -> list[str]:
from hermes_cli.browser_connect import manual_chrome_debug_command
command = manual_chrome_debug_command(port, system)
hint = (
["Start a Chromium-family browser with remote debugging, then retry /browser connect:", command]
if command
else [
"No supported Chromium-family browser executable was found in this environment.",
f"Install one or start a Chromium-family browser with --remote-debugging-port={port}, then retry /browser connect.",
]
)
return [
f"Browser CDP is not reachable at {url}.",
*hint,
"Browser not connected — start a Chromium-family browser with remote debugging and retry /browser connect",
]
def _browser_connect(rid, params: dict) -> dict:
import platform
from hermes_cli.browser_connect import DEFAULT_BROWSER_CDP_URL
from tools.browser_tool import cleanup_all_browsers
from urllib.parse import urlparse
raw_url = params.get("url")
if raw_url is not None and not isinstance(raw_url, str):
return _err(
rid, 4015, f"browser url must be a string, got {type(raw_url).__name__}"
)
url = (raw_url or "").strip() or DEFAULT_BROWSER_CDP_URL
sid = params.get("session_id") or ""
system = platform.system()
messages: list[str] = []
def announce(message: str, *, level: str = "info") -> None:
messages.append(message)
# Without a session id the TUI prints `messages` from the
# response; emitting an event would double-render. Only stream
# progress when there's a real session to scope it to.
if sid:
_emit("browser.progress", sid, {"message": message, "level": level})
parsed = urlparse(url if "://" in url else f"http://{url}")
if parsed.scheme not in {"http", "https", "ws", "wss"}:
return _err(rid, 4015, f"unsupported browser url: {url}")
if not parsed.hostname:
return _err(rid, 4015, f"missing host in browser url: {url}")
try:
port = parsed.port or (443 if parsed.scheme in {"https", "wss"} else 80)
except ValueError:
return _err(rid, 4015, f"invalid port in browser url: {url}")
# Always normalize default-local to 127.0.0.1:9222 so downstream
# comparisons + messaging match what we'll actually persist.
if _is_default_local_cdp(parsed):
url = DEFAULT_BROWSER_CDP_URL
parsed = urlparse(url)
port = parsed.port or 9222
try:
# ws[s]://.../devtools/browser/<id> endpoints (hosted CDP
# providers) don't serve the HTTP discovery path; just check
# TCP-level reachability and let browser_navigate handshake.
if parsed.scheme in {"ws", "wss"} and parsed.path.startswith(
"/devtools/browser/"
):
import socket
try:
with socket.create_connection((parsed.hostname, port), timeout=2.0):
pass
except OSError as e:
return _err(rid, 5031, f"could not reach browser CDP at {url}: {e}")
elif _is_default_local_cdp(parsed):
from hermes_cli.browser_connect import (
discover_local_cdp_url,
find_free_debug_port,
launch_chrome_debug,
local_port_in_use,
)
# Dual-stack discovery: when another app (an IDE debugger,
# a dev server) squats the IPv4 loopback on the debug port,
# a browser asked to bind that port comes up on [::1] only.
# An IPv4-only probe misses it AND hangs against squatters
# that accept TCP but never answer HTTP — the historic
# cause of `browser.manage` RPC timeouts.
discovered = discover_local_cdp_url(port, timeout=2.0)
launch_port = port
if discovered is None:
if local_port_in_use(port):
launch_port = find_free_debug_port(port)
announce(
f"Port {port} is occupied by another application that "
"isn't a CDP browser (an IDE debugger or dev server may "
f"be using it) — launching a debug browser on port "
f"{launch_port} instead..."
)
else:
announce(
"Chromium-family browser isn't running with remote debugging — attempting to launch..."
)
launch = launch_chrome_debug(launch_port, system)
if launch.launched:
# Bounded wait: the whole connect must finish well
# inside the client RPC timeout.
deadline = time.monotonic() + 10.0
while time.monotonic() < deadline:
discovered = discover_local_cdp_url(launch_port, timeout=1.0)
if discovered:
break
time.sleep(0.5)
if discovered:
announce(
f"Chromium-family browser launched and listening on port {launch_port}"
)
else:
hint = launch.hint
if hint:
announce(hint, level="error")
for line in _failure_messages(url, launch_port, system)[1:]:
announce(line, level="error")
return _ok(
rid, {"connected": False, "url": url, "messages": messages}
)
else:
announce(f"Chromium-family browser is already listening at {discovered}")
# Adopt whatever loopback/port actually answered (may be
# [::1] and/or an alternate port when 9222 was squatted).
url = discovered
parsed = urlparse(url)
else:
probes = _probe_urls(parsed)
ok = any(_http_ok(p, timeout=2.0) for p in probes)
if not ok:
return _err(rid, 5031, f"could not reach browser CDP at {url}")
normalized = _normalize_cdp_url(parsed)
# Order matters: reap sessions BEFORE publishing the new env
# so an in-flight tool call sees the old supervisor closed,
# then again AFTER so the default task's cached supervisor
# is drained against the new URL.
cleanup_all_browsers()
os.environ["BROWSER_CDP_URL"] = normalized
cleanup_all_browsers()
except Exception as e:
return _err(rid, 5031, str(e))
payload: dict[str, object] = {"connected": True, "url": normalized}
if messages:
payload["messages"] = messages
return _ok(rid, payload)
def _browser_disconnect(rid) -> dict:
# Reap, drop the env override, reap again — closes the same swap
# window covered by ``_browser_connect``.
def reap() -> None:
try:
from tools.browser_tool import cleanup_all_browsers
cleanup_all_browsers()
except Exception:
pass
reap()
os.environ.pop("BROWSER_CDP_URL", None)
reap()
return _ok(rid, {"connected": False})
def register(server) -> None:
"""Publish this module's helpers + handlers onto ``server``, rebound to its globals."""
bind_module(globals(), server, skip=("_",))
tui_gateway/server.py
+2 -263
View File
@@ -16216,269 +16216,6 @@ def _mirror_slash_side_effects(sid: str, session: dict, command: str) -> str:
# ── Methods: rollback ────────────────────────────────────────────────
# ── Methods: browser / plugins / cron / skills ───────────────────────
def _resolve_browser_cdp_url() -> str:
"""Return the configured browser CDP override without network I/O.
``/browser status`` must be fast — calling
``tools.browser_tool._get_cdp_override`` would invoke
``_resolve_cdp_override``, which performs an HTTP probe to
``.../json/version`` for discovery-style URLs. That probe has
a multi-second timeout and would block the TUI on a slow or
unreachable host even though status only needs to report whether
an override is set.
Mirrors the env/config precedence of ``_get_cdp_override`` (env
var first, then ``browser.cdp_url`` from config.yaml) without the
websocket-resolution step, so the answer reflects user intent
even when the configured host is not currently reachable. The
actual WS normalization happens in ``browser_navigate`` on the
next tool call.
"""
env_url = os.environ.get("BROWSER_CDP_URL", "").strip()
if env_url:
return env_url
try:
from hermes_cli.config import read_raw_config
cfg = read_raw_config()
browser_cfg = cfg.get("browser", {}) if isinstance(cfg, dict) else {}
if isinstance(browser_cfg, dict):
return str(browser_cfg.get("cdp_url", "") or "").strip()
except Exception:
pass
return ""
def _is_default_local_cdp(parsed) -> bool:
"""Match the discovery-style local default; never the concrete WS form.
A user-supplied ``ws://127.0.0.1:9222/devtools/browser/<id>`` is a
real, connectable endpoint — collapsing it to bare ``http://...:9222``
would strip the path and break the connect.
"""
try:
port = parsed.port or 80
except ValueError:
return False
discovery_path = parsed.path in {"", "/", "/json", "/json/version"}
return (
parsed.scheme in {"http", "ws"}
and parsed.hostname in {"127.0.0.1", "localhost"}
and port == 9222
and discovery_path
)
def _http_ok(url: str, timeout: float) -> bool:
import urllib.request
try:
with urllib.request.urlopen(url, timeout=timeout) as resp:
return 200 <= getattr(resp, "status", 200) < 300
except Exception:
return False
def _probe_urls(parsed) -> list[str]:
scheme = {"ws": "http", "wss": "https"}.get(parsed.scheme, parsed.scheme)
root = f"{scheme}://{parsed.netloc}".rstrip("/")
return [f"{root}/json/version", f"{root}/json"]
def _normalize_cdp_url(parsed) -> str:
# Concrete ``/devtools/browser/<id>`` endpoints (Browserbase et al.)
# are connectable as-is. Discovery-style inputs collapse to bare
# ``scheme://host:port`` so ``_resolve_cdp_override`` can append
# ``/json/version`` later without doubling the path.
if parsed.path.startswith("/devtools/browser/"):
return parsed.geturl()
return parsed._replace(path="", params="", query="", fragment="").geturl()
def _failure_messages(url: str, port: int, system: str) -> list[str]:
from hermes_cli.browser_connect import manual_chrome_debug_command
command = manual_chrome_debug_command(port, system)
hint = (
["Start a Chromium-family browser with remote debugging, then retry /browser connect:", command]
if command
else [
"No supported Chromium-family browser executable was found in this environment.",
f"Install one or start a Chromium-family browser with --remote-debugging-port={port}, then retry /browser connect.",
]
)
return [
f"Browser CDP is not reachable at {url}.",
*hint,
"Browser not connected — start a Chromium-family browser with remote debugging and retry /browser connect",
]
def _browser_connect(rid, params: dict) -> dict:
import platform
from hermes_cli.browser_connect import DEFAULT_BROWSER_CDP_URL
from tools.browser_tool import cleanup_all_browsers
from urllib.parse import urlparse
raw_url = params.get("url")
if raw_url is not None and not isinstance(raw_url, str):
return _err(
rid, 4015, f"browser url must be a string, got {type(raw_url).__name__}"
)
url = (raw_url or "").strip() or DEFAULT_BROWSER_CDP_URL
sid = params.get("session_id") or ""
system = platform.system()
messages: list[str] = []
def announce(message: str, *, level: str = "info") -> None:
messages.append(message)
# Without a session id the TUI prints `messages` from the
# response; emitting an event would double-render. Only stream
# progress when there's a real session to scope it to.
if sid:
_emit("browser.progress", sid, {"message": message, "level": level})
parsed = urlparse(url if "://" in url else f"http://{url}")
if parsed.scheme not in {"http", "https", "ws", "wss"}:
return _err(rid, 4015, f"unsupported browser url: {url}")
if not parsed.hostname:
return _err(rid, 4015, f"missing host in browser url: {url}")
try:
port = parsed.port or (443 if parsed.scheme in {"https", "wss"} else 80)
except ValueError:
return _err(rid, 4015, f"invalid port in browser url: {url}")
# Always normalize default-local to 127.0.0.1:9222 so downstream
# comparisons + messaging match what we'll actually persist.
if _is_default_local_cdp(parsed):
url = DEFAULT_BROWSER_CDP_URL
parsed = urlparse(url)
port = parsed.port or 9222
try:
# ws[s]://.../devtools/browser/<id> endpoints (hosted CDP
# providers) don't serve the HTTP discovery path; just check
# TCP-level reachability and let browser_navigate handshake.
if parsed.scheme in {"ws", "wss"} and parsed.path.startswith(
"/devtools/browser/"
):
import socket
try:
with socket.create_connection((parsed.hostname, port), timeout=2.0):
pass
except OSError as e:
return _err(rid, 5031, f"could not reach browser CDP at {url}: {e}")
elif _is_default_local_cdp(parsed):
from hermes_cli.browser_connect import (
discover_local_cdp_url,
find_free_debug_port,
launch_chrome_debug,
local_port_in_use,
)
# Dual-stack discovery: when another app (an IDE debugger,
# a dev server) squats the IPv4 loopback on the debug port,
# a browser asked to bind that port comes up on [::1] only.
# An IPv4-only probe misses it AND hangs against squatters
# that accept TCP but never answer HTTP — the historic
# cause of `browser.manage` RPC timeouts.
discovered = discover_local_cdp_url(port, timeout=2.0)
launch_port = port
if discovered is None:
if local_port_in_use(port):
launch_port = find_free_debug_port(port)
announce(
f"Port {port} is occupied by another application that "
"isn't a CDP browser (an IDE debugger or dev server may "
f"be using it) — launching a debug browser on port "
f"{launch_port} instead..."
)
else:
announce(
"Chromium-family browser isn't running with remote debugging — attempting to launch..."
)
launch = launch_chrome_debug(launch_port, system)
if launch.launched:
# Bounded wait: the whole connect must finish well
# inside the client RPC timeout.
deadline = time.monotonic() + 10.0
while time.monotonic() < deadline:
discovered = discover_local_cdp_url(launch_port, timeout=1.0)
if discovered:
break
time.sleep(0.5)
if discovered:
announce(
f"Chromium-family browser launched and listening on port {launch_port}"
)
else:
hint = launch.hint
if hint:
announce(hint, level="error")
for line in _failure_messages(url, launch_port, system)[1:]:
announce(line, level="error")
return _ok(
rid, {"connected": False, "url": url, "messages": messages}
)
else:
announce(f"Chromium-family browser is already listening at {discovered}")
# Adopt whatever loopback/port actually answered (may be
# [::1] and/or an alternate port when 9222 was squatted).
url = discovered
parsed = urlparse(url)
else:
probes = _probe_urls(parsed)
ok = any(_http_ok(p, timeout=2.0) for p in probes)
if not ok:
return _err(rid, 5031, f"could not reach browser CDP at {url}")
normalized = _normalize_cdp_url(parsed)
# Order matters: reap sessions BEFORE publishing the new env
# so an in-flight tool call sees the old supervisor closed,
# then again AFTER so the default task's cached supervisor
# is drained against the new URL.
cleanup_all_browsers()
os.environ["BROWSER_CDP_URL"] = normalized
cleanup_all_browsers()
except Exception as e:
return _err(rid, 5031, str(e))
payload: dict[str, object] = {"connected": True, "url": normalized}
if messages:
payload["messages"] = messages
return _ok(rid, payload)
def _browser_disconnect(rid) -> dict:
# Reap, drop the env override, reap again — closes the same swap
# window covered by ``_browser_connect``.
def reap() -> None:
try:
from tools.browser_tool import cleanup_all_browsers
cleanup_all_browsers()
except Exception:
pass
reap()
os.environ.pop("BROWSER_CDP_URL", None)
reap()
return _ok(rid, {"connected": False})
# Per-profile MCP lifecycle helpers (mcp.servers.* handlers). Defined on THIS
# namespace so the rebound handler bodies (register() below) can resolve them,
@@ -16506,6 +16243,7 @@ def _mcp_summarize_server(name, cfg): # noqa: E402
# over already exists; register() rebinds them onto this namespace.
from . import ( # noqa: E402
methods_voice as _methods_voice,
methods_browser as _methods_browser,
methods_browser_control as _methods_browser_control,
methods_bot_relay as _methods_bot_relay,
methods_complete as _methods_complete,
@@ -16520,6 +16258,7 @@ from . import ( # noqa: E402
for _m in (
_methods_voice,
_methods_browser,
_methods_browser_control,
_methods_session,
_methods_prompt,