host
c9a483282d
- playwright_browser.Browser：单页 Chromium，navigate / snapshot（正文 8000 字截断标记）/ click / fill / evaluate / screenshot（工作区内）/ close；启动参数 --no-sandbox + --disable-dev-shm-usage（runsc + 1003 用户实测必需）， 代理显式传给 launch（Chromium 在 Linux 下不稳定地读 env 代理），浏览器路径 PLAYWRIGHT_BROWSERS_PATH=/opt/ms-playwright。 - playwright_mcp：stdio MCP，7 个 browser_* 工具，同一进程 单 Browser 实例；测试用 fake browser / fake factory。 - worker：--playwright-mcp flag，_tenant_dirs 播种两个脚本 + [mcp_servers.playwright]，_prepare_tenant 写 env（代理 + 浏览器路径，MCP 子进程不继承 Codex 进程 env，镜像 ENV 在此必须重写，缺了会报 Executable doesn't exist）。 - 技能 browser-automation：JS 页、翻页、表单、截图时用， 静态页优先 web_fetch / web_search；截图须 PIL 读回验证。 - 镜像（层构建，非本 PR）：FROM latest + fonts-noto-cjk + playwright + chromium（npmmirror）+ install-deps，3.77GB； runsc + 1003 + docker-codex 条件下 JS 执行、example.com 标题、截图均已实测通过。
293 lines
11 KiB
Python
293 lines
11 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 = ""
|
|
|
|
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 = ""
|
|
return {"url": self._current_url, "title": title}
|
|
|
|
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
|
|
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 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 = []
|
|
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.
|
|
|
|
Mirrors ``ensure_codex_home_imagegen``: copies
|
|
``playwright_browser.py`` and ``playwright_mcp.py`` only when
|
|
missing (a tenant-modified script is never silently
|
|
overwritten, it is reported in ``script_differs``), and
|
|
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 = []
|
|
for name in ("playwright_browser.py", "playwright_mcp.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:
|
|
script_differs.append(name)
|
|
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}
|
|
|
|
|
|
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
|