feat: 浏览器自动化 MCP（Playwright）+ browser-automation 技能
- 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 标题、截图均已实测通过。
This commit is contained in:
host committed
1 parent
872b23a296
commit
c9a483282d
5 files changed
+821
-2
No files matched your search
@@ -0,0 +1,292 @@
|
||||
"""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
|
||||
Reference in new issue
Block a user