host
fafc0734c7
Page.startScreencast everyNthFrame 2 -> 1，画质 q60、尺寸 1280x800 不动。静态页不产生额外成本（帧只在重绘时来， 且 #187 起桥只写变化的帧字节）；收益全部落在真正需要跟 手的时候：滚动、拖拽、动画期间帧率翻倍。测试断言同步。
591 lines
22 KiB
Python
591 lines
22 KiB
Python
"""Headless Chromium automation for Codex tasks, via Playwright.
|
|
|
|
One Browser object owns one page for the lifetime of an MCP
|
|
process. It is deliberately small: navigate, snapshot (page text),
|
|
click, fill, evaluate, screenshot, close. Anything fancier the
|
|
agent can do with the shell tool, which has the same sandbox
|
|
privileges as this browser (tenant Docker container, egress
|
|
proxy whitelist).
|
|
|
|
Browser launch facts that matter in production (verified
|
|
2026-10-30 in a runsc + user 1003 + docker-codex AppArmor
|
|
container, the same shape as a production Docker task):
|
|
- Chromium must run --no-sandbox (rootless container user, no
|
|
userns sandbox) and --disable-dev-shm-usage (small /dev/shm).
|
|
- Browsers live at PLAYWRIGHT_BROWSERS_PATH=/opt/ms-playwright
|
|
in the image (the build installs them there, NOT in a home
|
|
directory, because the container user has no home and the
|
|
build ran as root).
|
|
- Network goes through the egress proxy when HTTP_PROXY /
|
|
HTTPS_PROXY is set in the MCP env (worker writes it from
|
|
proxy_env, same as websearch): Playwright gets an explicit
|
|
proxy server, because Chromium does not reliably pick up env
|
|
proxies on Linux. Domains outside the Tinyproxy whitelist
|
|
then fail with a proxy 403, which the tool reports plainly
|
|
instead of bypassing.
|
|
- Screenshots and downloads land in the working directory
|
|
(the task worktree), never outside it.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import time
|
|
|
|
BROWSERS_PATH = "/opt/ms-playwright"
|
|
SNAPSHOT_LIMIT = 8000
|
|
NAV_TIMEOUT_MS = 60000
|
|
|
|
|
|
class BrowserError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def _validate_basename(name, what):
|
|
if (not name or os.path.basename(name) != name or name in (".", "..")
|
|
or "/" in name or "\\" in name):
|
|
raise BrowserError(
|
|
"%s must be a plain basename, got %r" % (what, name))
|
|
return name
|
|
|
|
|
|
class Browser:
|
|
"""A single Chromium page. ``browser_factory`` is injectable:
|
|
tests pass a fake returning a fake sync_playwright object."""
|
|
|
|
def __init__(self, cwd=None, proxy=None, browser_factory=None):
|
|
self.cwd = os.path.abspath(cwd or os.getcwd())
|
|
self.proxy = (proxy or os.environ.get("HTTPS_PROXY")
|
|
or os.environ.get("HTTP_PROXY") or "")
|
|
self._factory = browser_factory
|
|
self._pw = None
|
|
self._browser = None
|
|
self._page = None
|
|
self._current_url = ""
|
|
self._last_title = ""
|
|
# ref snapshot state (playwright-mcp style): ref -> css path
|
|
self._ref_paths = {}
|
|
# screencast sink: a queue.Queue set by the remote service.
|
|
# The CDP frame callback only enqueues; the MCP main thread
|
|
# drains it (pump_screencast) because Playwright sync calls
|
|
# must run on the thread that created the page.
|
|
self._frame_sink = None
|
|
self._cdp = None
|
|
self._cdp_page = None
|
|
|
|
def _ensure(self):
|
|
if self._page is not None:
|
|
return
|
|
try:
|
|
if self._factory is not None:
|
|
pw = self._factory()
|
|
if hasattr(pw, "start"):
|
|
pw = pw.start()
|
|
else:
|
|
from playwright.sync_api import sync_playwright
|
|
pw = sync_playwright().start()
|
|
launch = {"args": ["--no-sandbox",
|
|
"--disable-dev-shm-usage",
|
|
"--disable-gpu"]}
|
|
if self.proxy:
|
|
launch["proxy"] = {"server": self.proxy}
|
|
browser = pw.chromium.launch(**launch)
|
|
page = browser.new_page(
|
|
viewport={"width": 1280, "height": 800})
|
|
except Exception as exc: # noqa: BLE001 - surfaced as tool error
|
|
raise BrowserError("browser launch failed: %s" % exc)
|
|
self._pw = pw
|
|
self._browser = browser
|
|
self._page = page
|
|
|
|
def _info(self):
|
|
try:
|
|
title = self._page.title()
|
|
except Exception: # noqa: BLE001 - title is best-effort
|
|
title = ""
|
|
self._last_title = title
|
|
return {"url": self._current_url, "title": title}
|
|
|
|
def cached_info(self):
|
|
"""Plain-attribute state, safe to read from any thread
|
|
(the remote HTTP thread must never call page.title())."""
|
|
return {"url": self._current_url, "title": self._last_title,
|
|
"started": self._page is not None}
|
|
|
|
def navigate(self, url):
|
|
if not url or not str(url).strip():
|
|
raise BrowserError("url is required")
|
|
self._ensure()
|
|
try:
|
|
self._page.goto(str(url), timeout=NAV_TIMEOUT_MS,
|
|
wait_until="domcontentloaded")
|
|
self._current_url = self._page.url
|
|
self._start_screencast()
|
|
self._eager_frame()
|
|
except Exception as exc: # noqa: BLE001
|
|
raise BrowserError("navigate failed: %s" % exc)
|
|
result = self._info()
|
|
result["ok"] = True
|
|
return result
|
|
|
|
def snapshot(self):
|
|
self._ensure()
|
|
try:
|
|
text = self._page.locator("body").inner_text(timeout=10000)
|
|
except Exception as exc: # noqa: BLE001
|
|
raise BrowserError("snapshot failed: %s" % exc)
|
|
truncated = len(text) > SNAPSHOT_LIMIT
|
|
if truncated:
|
|
text = text[:SNAPSHOT_LIMIT]
|
|
result = self._info()
|
|
result["text"] = text
|
|
result["truncated"] = truncated
|
|
return result
|
|
|
|
def click(self, selector):
|
|
if not selector or not str(selector).strip():
|
|
raise BrowserError("selector is required")
|
|
self._ensure()
|
|
try:
|
|
self._page.click(str(selector), timeout=10000)
|
|
self._current_url = self._page.url
|
|
except Exception as exc: # noqa: BLE001
|
|
raise BrowserError("click failed: %s" % exc)
|
|
result = self._info()
|
|
result["ok"] = True
|
|
return result
|
|
|
|
def fill(self, selector, text):
|
|
if not selector or not str(selector).strip():
|
|
raise BrowserError("selector is required")
|
|
self._ensure()
|
|
try:
|
|
self._page.fill(str(selector), "" if text is None else str(text),
|
|
timeout=10000)
|
|
except Exception as exc: # noqa: BLE001
|
|
raise BrowserError("fill failed: %s" % exc)
|
|
result = self._info()
|
|
result["ok"] = True
|
|
return result
|
|
|
|
def mouse_move(self, x, y):
|
|
try:
|
|
fx = float(x)
|
|
fy = float(y)
|
|
except (TypeError, ValueError):
|
|
raise BrowserError("mouse_move needs numeric x/y, got %r/%r"
|
|
% (x, y))
|
|
if not (0 <= fx <= 1280 and 0 <= fy <= 800):
|
|
raise BrowserError(
|
|
"mouse_move out of viewport 1280x800: %s,%s" % (x, y))
|
|
self._ensure()
|
|
try:
|
|
self._page.mouse.move(fx, fy)
|
|
except Exception as exc: # noqa: BLE001
|
|
raise BrowserError("mouse_move failed: %s" % exc)
|
|
result = self._info()
|
|
result["ok"] = True
|
|
result["x"] = fx
|
|
result["y"] = fy
|
|
return result
|
|
|
|
def key_press(self, key):
|
|
if not key or not str(key).strip():
|
|
raise BrowserError("key is required")
|
|
self._ensure()
|
|
try:
|
|
self._page.keyboard.press(str(key))
|
|
except Exception as exc: # noqa: BLE001
|
|
raise BrowserError("key_press failed: %s" % exc)
|
|
result = self._info()
|
|
result["ok"] = True
|
|
result["key"] = str(key)
|
|
return result
|
|
|
|
def type_text(self, text):
|
|
if text is None:
|
|
raise BrowserError("text is required")
|
|
self._ensure()
|
|
try:
|
|
self._page.keyboard.type(str(text))
|
|
except Exception as exc: # noqa: BLE001
|
|
raise BrowserError("type_text failed: %s" % exc)
|
|
result = self._info()
|
|
result["ok"] = True
|
|
return result
|
|
|
|
def snapshot_refs(self):
|
|
"""Accessibility-style snapshot with refs (playwright-mcp
|
|
style): interactive elements get a stable ref valid until
|
|
the next snapshot_refs call; click_ref(ref) clicks one.
|
|
The ref maps to a computed CSS path, so it survives
|
|
re-renders better than an index into a stale list."""
|
|
self._ensure()
|
|
script = """
|
|
() => {
|
|
const path = (el) => {
|
|
const parts = [];
|
|
while (el && el.nodeType === 1 && el !== document.body) {
|
|
let tag = el.tagName.toLowerCase();
|
|
const parent = el.parentElement;
|
|
if (parent) {
|
|
const sibs = Array.from(parent.children).filter(
|
|
c => c.tagName === el.tagName);
|
|
if (sibs.length > 1)
|
|
tag += ':nth-of-type(' + (sibs.indexOf(el) + 1) + ')';
|
|
}
|
|
parts.unshift(tag);
|
|
el = parent;
|
|
}
|
|
return parts.join(' > ');
|
|
};
|
|
const out = [];
|
|
const els = document.querySelectorAll(
|
|
'a, button, input, select, textarea, [role="button"], ' +
|
|
'[role="link"], [role="textbox"], [role="checkbox"], ' +
|
|
'[role="radio"], [role="combobox"]');
|
|
els.forEach((el) => {
|
|
const r = el.getBoundingClientRect();
|
|
if (r.width === 0 && r.height === 0) return;
|
|
const name = (el.getAttribute('aria-label') ||
|
|
el.textContent || el.value || el.placeholder || '')
|
|
.trim().replace(/\\s+/g, ' ').slice(0, 80);
|
|
out.push({tag: el.tagName.toLowerCase(),
|
|
role: el.getAttribute('role') || '',
|
|
name: name, path: path(el)});
|
|
});
|
|
return out;
|
|
}
|
|
"""
|
|
try:
|
|
elements = self._page.evaluate(script)
|
|
except Exception as exc: # noqa: BLE001
|
|
raise BrowserError("snapshot_refs failed: %s" % exc)
|
|
self._ref_paths = {}
|
|
items = []
|
|
for i, el in enumerate(elements or []):
|
|
ref = "e%d" % i
|
|
self._ref_paths[ref] = el.get("path", "")
|
|
items.append({"ref": ref, "tag": el.get("tag", ""),
|
|
"role": el.get("role", ""),
|
|
"name": el.get("name", "")})
|
|
result = self._info()
|
|
result["elements"] = items
|
|
result["count"] = len(items)
|
|
return result
|
|
|
|
def click_ref(self, ref):
|
|
if not ref or ref not in self._ref_paths:
|
|
raise BrowserError(
|
|
"unknown ref %r: run browser_snapshot_refs first" % (ref,))
|
|
self._ensure()
|
|
path = self._ref_paths[ref]
|
|
try:
|
|
self._page.locator(path).first.click(timeout=10000)
|
|
self._current_url = self._page.url
|
|
except Exception as exc: # noqa: BLE001
|
|
raise BrowserError("click_ref failed: %s" % exc)
|
|
result = self._info()
|
|
result["ok"] = True
|
|
result["ref"] = ref
|
|
return result
|
|
|
|
def set_frame_sink(self, sink):
|
|
"""Remote service: frames arrive here via CDP callback."""
|
|
self._frame_sink = sink
|
|
if self._page is not None:
|
|
self._start_screencast()
|
|
|
|
def _start_screencast(self):
|
|
if self._frame_sink is None or self._page is None:
|
|
return
|
|
if self._cdp is not None and self._cdp_page is self._page:
|
|
return
|
|
try:
|
|
cdp = self._page.context.new_cdp_session(self._page)
|
|
|
|
def _on_frame(params):
|
|
try:
|
|
self._frame_sink.put(
|
|
(params.get("sessionId", ""),
|
|
params.get("data", "")))
|
|
except Exception: # noqa: BLE001 - callback must not die
|
|
pass
|
|
|
|
cdp.on("Page.screencastFrame", _on_frame)
|
|
# everyNthFrame 1: full frame rate during motion.
|
|
# Static pages cost nothing extra (frames only arrive
|
|
# on repaint, and the bridge writes only changed
|
|
# bytes), so dropping the frame-skip buys
|
|
# responsiveness exactly when it matters -- scrolling,
|
|
# dragging, animation -- without a static-page tax.
|
|
cdp.send("Page.startScreencast",
|
|
{"format": "jpeg", "quality": 60,
|
|
"maxWidth": 1280, "maxHeight": 800,
|
|
"everyNthFrame": 1})
|
|
self._cdp = cdp
|
|
self._cdp_page = self._page
|
|
except Exception: # noqa: BLE001 - screencast is best-effort
|
|
self._cdp = None
|
|
self._cdp_page = None
|
|
|
|
def _eager_frame(self):
|
|
"""Push one current-viewport frame into the sink, best-effort.
|
|
|
|
Screencast frames only arrive on repaint, and the page may
|
|
already have painted before startScreencast ran -- so a
|
|
static page would otherwise show nothing until its next
|
|
repaint. Capturing the viewport once right after navigate
|
|
guarantees the bridge file (and any open stream) has a
|
|
picture within milliseconds of the page loading.
|
|
"""
|
|
if self._cdp is None or self._frame_sink is None:
|
|
return
|
|
try:
|
|
result = self._cdp.send(
|
|
"Page.captureScreenshot",
|
|
{"format": "jpeg", "quality": 60})
|
|
data = result.get("data") if isinstance(result, dict) \
|
|
else None
|
|
if data:
|
|
# Empty session id: pump stores it but must not ack
|
|
# it (there is no screencast frame to acknowledge).
|
|
self._frame_sink.put(("", data))
|
|
except Exception: # noqa: BLE001 - screencast still flows
|
|
pass
|
|
|
|
def pump_screencast(self):
|
|
"""Main-thread only: store the latest frame and ack every
|
|
real screencast frame (no ack = CDP stops sending after the
|
|
first). Eager frames carry no session id and skip ack.
|
|
Returns the latest frame data (base64 jpeg) or ''."""
|
|
if self._frame_sink is None:
|
|
return ""
|
|
latest = ""
|
|
while True:
|
|
try:
|
|
sid, data = self._frame_sink.get_nowait()
|
|
except Exception: # noqa: BLE001 - queue empty
|
|
break
|
|
if data:
|
|
latest = data
|
|
if self._cdp is not None and sid:
|
|
try:
|
|
self._cdp.send("Page.screencastFrameAck",
|
|
{"sessionId": sid})
|
|
except Exception: # noqa: BLE001 - ack best-effort
|
|
pass
|
|
if latest:
|
|
self._latest_frame = latest
|
|
return getattr(self, "_latest_frame", "")
|
|
|
|
def click_at(self, x, y):
|
|
"""Click by viewport coordinates (CSS pixels, origin at the
|
|
top-left of the 1280x800 viewport). This is the tool the
|
|
console live view drives: the user clicks on a screenshot
|
|
and the coordinates arrive here. Bounds-checked so a
|
|
mis-scaled click cannot scroll the page into nowhere."""
|
|
try:
|
|
fx = float(x)
|
|
fy = float(y)
|
|
except (TypeError, ValueError):
|
|
raise BrowserError("click_at needs numeric x/y, got %r/%r"
|
|
% (x, y))
|
|
if not (0 <= fx <= 1280 and 0 <= fy <= 800):
|
|
raise BrowserError(
|
|
"click_at out of viewport 1280x800: %s,%s" % (x, y))
|
|
self._ensure()
|
|
try:
|
|
self._page.mouse.click(fx, fy)
|
|
self._current_url = self._page.url
|
|
except Exception as exc: # noqa: BLE001
|
|
raise BrowserError("click_at failed: %s" % exc)
|
|
result = self._info()
|
|
result["ok"] = True
|
|
result["x"] = fx
|
|
result["y"] = fy
|
|
return result
|
|
|
|
def live_shot(self, output_filename=None):
|
|
"""Screenshot to a FIXED live-view file plus a JSON status
|
|
sidecar, so a watcher (console, user) can poll one stable
|
|
path instead of chasing per-shot filenames. The agent
|
|
should call this after every navigate/click/fill while a
|
|
live view is wanted. Returns path, bytes, url, title and
|
|
a unix timestamp."""
|
|
self._ensure()
|
|
name = _validate_basename(
|
|
output_filename or "browser-live.png", "output_filename")
|
|
path = os.path.join(self.cwd, name)
|
|
try:
|
|
self._page.screenshot(path=path)
|
|
except Exception as exc: # noqa: BLE001
|
|
raise BrowserError("live_shot failed: %s" % exc)
|
|
info = self._info()
|
|
status = {"image": name, "path": path,
|
|
"bytes": os.path.getsize(path),
|
|
"url": info["url"], "title": info["title"],
|
|
"timestamp": int(time.time())}
|
|
status_path = os.path.join(self.cwd, "browser-live.json")
|
|
with open(status_path, "w", encoding="utf-8") as fh:
|
|
json.dump(status, fh, ensure_ascii=False)
|
|
status["status_path"] = status_path
|
|
return status
|
|
|
|
def evaluate(self, expression):
|
|
if not expression or not str(expression).strip():
|
|
raise BrowserError("expression is required")
|
|
self._ensure()
|
|
try:
|
|
value = self._page.evaluate(str(expression))
|
|
except Exception as exc: # noqa: BLE001
|
|
raise BrowserError("evaluate failed: %s" % exc)
|
|
return {"result": str(value)[:4000]}
|
|
|
|
def screenshot(self, output_filename=None):
|
|
self._ensure()
|
|
name = _validate_basename(
|
|
output_filename or ("shot-%d.png" % int(time.time())),
|
|
"output_filename")
|
|
path = os.path.join(self.cwd, name)
|
|
try:
|
|
self._page.screenshot(path=path)
|
|
except Exception as exc: # noqa: BLE001
|
|
raise BrowserError("screenshot failed: %s" % exc)
|
|
return {"path": path, "filename": name,
|
|
"bytes": os.path.getsize(path)}
|
|
|
|
def close(self):
|
|
errors = []
|
|
if self._cdp is not None:
|
|
try:
|
|
self._cdp.send("Page.stopScreencast")
|
|
except Exception: # noqa: BLE001 - best-effort
|
|
pass
|
|
self._cdp = None
|
|
self._cdp_page = None
|
|
for closer in (lambda: self._page and self._page.close(),
|
|
lambda: self._browser and self._browser.close(),
|
|
lambda: self._pw and self._pw.stop()):
|
|
try:
|
|
closer()
|
|
except Exception as exc: # noqa: BLE001 - close is best-effort
|
|
errors.append(str(exc))
|
|
self._page = None
|
|
self._browser = None
|
|
self._pw = None
|
|
return {"closed": True, "errors": errors}
|
|
|
|
|
|
# ------------------------------------------------------- codex-home glue
|
|
|
|
def ensure_codex_home_playwright(codex_home_dir, command_args=None,
|
|
script_dir=None):
|
|
"""Seed playwright MCP files + config into a tenant codex-home.
|
|
|
|
MCP scripts are SYSTEM files, not tenant data: a tenant never
|
|
edits them (user files live in the worktree, not codex-home),
|
|
so unlike the old never-overwrite rule, a differing script is
|
|
REFRESHED to the deployed version and reported in
|
|
``script_refreshed``. Without this, every MCP feature ships
|
|
dead for existing projects: they keep the first-seeded script
|
|
forever and only brand-new projects get the update.
|
|
Appends ``[mcp_servers.playwright]`` only when absent. No
|
|
secrets live here -- proxy vars arrive via
|
|
``ensure_playwright_mcp_env`` from worker proxy_env, and the
|
|
browsers path is a fixed image path.
|
|
"""
|
|
import shutil
|
|
|
|
home = os.path.abspath(codex_home_dir)
|
|
os.makedirs(home, exist_ok=True)
|
|
here = os.path.abspath(script_dir or os.path.dirname(__file__))
|
|
script_changed = False
|
|
script_differs = []
|
|
script_refreshed = []
|
|
for name in ("playwright_browser.py", "playwright_mcp.py",
|
|
"playwright_remote.py"):
|
|
src = os.path.join(here, name)
|
|
dst = os.path.join(home, name)
|
|
if not os.path.exists(src):
|
|
continue
|
|
if not os.path.exists(dst):
|
|
shutil.copyfile(src, dst)
|
|
script_changed = True
|
|
else:
|
|
with open(src, "rb") as fh:
|
|
src_bytes = fh.read()
|
|
with open(dst, "rb") as fh:
|
|
dst_bytes = fh.read()
|
|
if src_bytes != dst_bytes:
|
|
shutil.copyfile(src, dst)
|
|
script_refreshed.append(name)
|
|
script_changed = True
|
|
continue
|
|
config_path = os.path.join(home, "config.toml")
|
|
if command_args is None:
|
|
command_args = ["python3",
|
|
os.path.join(home, "playwright_mcp.py")]
|
|
section = ("\n[mcp_servers.playwright]\n"
|
|
"command = %s\n"
|
|
"args = [%s]\n"
|
|
% (json.dumps(command_args[0]),
|
|
", ".join(json.dumps(a) for a in command_args[1:])))
|
|
config_changed = False
|
|
existing = ""
|
|
if os.path.exists(config_path):
|
|
with open(config_path, encoding="utf-8") as fh:
|
|
existing = fh.read()
|
|
if "[mcp_servers.playwright]" not in existing:
|
|
with open(config_path, "a", encoding="utf-8") as fh:
|
|
if existing and not existing.endswith("\n"):
|
|
fh.write("\n")
|
|
fh.write(section)
|
|
config_changed = True
|
|
return {"config_path": config_path,
|
|
"script_path": os.path.join(home, "playwright_mcp.py"),
|
|
"config_changed": config_changed,
|
|
"script_changed": script_changed,
|
|
"script_differs": script_differs,
|
|
"script_refreshed": script_refreshed}
|
|
|
|
|
|
def ensure_playwright_mcp_env(codex_home_dir, env):
|
|
"""Write [mcp_servers.playwright.env] into a tenant config.toml.
|
|
|
|
Same rationale as the websearch/imagegen env writers: Codex's
|
|
MCP subprocess does not inherit the Codex process env, so the
|
|
proxy vars and PLAYWRIGHT_BROWSERS_PATH must live in
|
|
config.toml. Idempotent in-place replacement of the env
|
|
subsection; other sections untouched. Returns True when the
|
|
file was written.
|
|
"""
|
|
if not env:
|
|
return False
|
|
home = os.path.abspath(codex_home_dir)
|
|
config_path = os.path.join(home, "config.toml")
|
|
try:
|
|
existing = ""
|
|
if os.path.exists(config_path):
|
|
with open(config_path, encoding="utf-8") as fh:
|
|
existing = fh.read()
|
|
except OSError:
|
|
return False
|
|
start = existing.find("[mcp_servers.playwright.env]")
|
|
if start >= 0:
|
|
end = existing.find("\n[", start + 1)
|
|
if end < 0:
|
|
end = len(existing)
|
|
existing = existing[:start] + existing[end:]
|
|
lines = ["\n[mcp_servers.playwright.env]\n"]
|
|
for key in sorted(env):
|
|
lines.append("%s = %s\n" % (key, json.dumps(str(env[key]))))
|
|
try:
|
|
with open(config_path, "w", encoding="utf-8") as fh:
|
|
fh.write(existing)
|
|
if existing and not existing.endswith("\n"):
|
|
fh.write("\n")
|
|
fh.writelines(lines)
|
|
except OSError:
|
|
return False
|
|
return True
|