Browser Harness 解剖课

def _read_task(args):
    ...
    code = sys.stdin.read()      # 把你管道喂进来的 Python 全文读进来

你每次 @'...'@ | browser-use,竖线左边的文字就是从这里进门的——整段代码被当成一个字符串读进变量 code。

    ensure_daemon()              # 确保保活者在岗(不在就拉起)
    _install_helper_trace()      # 给每个 helper 套一层记录器(遥测用)
    exec(code, globals())        # 执行你的代码——厨房大门在此

这就是全部灵魂。exec 是 Python 的"照单全做"指令:你给什么代码它跑什么。注意第二个参数 globals()——意思是"在我这个房间里跑",而这个房间里已经预先摆好了所有 helper(文件顶部那句 from .helpers import *)。所以你的脚本里 new_tab、js 张口就来,不用 import。开放厨房 = 一句 exec + 一屋子现成厨具。

# Windows default stdout/stderr encoding is cp1252
# which can't encode the 🐴 marker helpers prepend to tab titles...
for _stream in (sys.stdout, sys.stderr):
    _stream.reconfigure(encoding="utf-8", errors="replace")

开头第一件事是修一个 Windows 专属 bug:默认编码打印不了 🐴 这个 emoji 会当场崩溃。等等,🐴?——对,你之前在标签页标题里看到的马头,是这个工具的官方记号,谜底在第二节揭晓。

# An explicit BU_CDP_URL or BU_CDP_WS also blocks the spawn ...
# silently replacing the user's explicit endpoint *and* billing
# them for a cloud browser they never asked for.

这段注释值得单独看:作者解释为什么要加一道防线——不加的话,某些条件下会悄悄给用户开一个收费的云浏览器。好代码的注释讲"为什么",不讲"是什么";这和你笔记里"记做了什么+为什么"是同一个家法。

def cdp(method, session_id=None, **params):
    """Raw CDP. cdp('Page.navigate', url='...')"""
    return _send({"method": method, "params": params,
                  "session_id": session_id}).get("result", {})

整个"翻译官"就这三行:把你的指令装进信封,交给水管(_send),拆开回信。没有任何智能、任何判断——CDP 的原话直接过。博客说"别包装工具",这就是不包装的样子。

def click_at_xy(x, y, button="left", clicks=1):
    cdp("Input.dispatchMouseEvent", type="mousePressed", x=x, y=y, ...)
    cdp("Input.dispatchMouseEvent", type="mouseReleased", x=x, y=y, ...)

点击 = 两句话:按下、抬起。就是精确复述"人手指做的两个动作"。Azure 那种嵌套八层 iframe 的页面为什么拦不住它?因为这是在合成器层面戳坐标,压根不经过网页的"户口本"。

def _mark_tab():
    """Prepend horse emoji to tab title so the user can see
    which tab the agent controls."""
    ... document.title = '🐴 ' + document.title ...

🐴 谜底揭晓:agent 每接管一个标签页,就在标题前贴一个马头,让你一眼看出"这个标签现在归 AI 管";切走时再摘掉(马头在 JS 字符串里占 2 个码元,加空格共 3 个,所以摘的时候精确切 3 位——连这种细节都有注释)。为什么是马?harness 本义是马具。这是整个项目最俏皮的一行代码:护栏没了,但透明度留着。

def upload_file(selector, path):
    doc = cdp("DOM.getDocument", depth=-1)
    nid = cdp("DOM.querySelector", ..., selector=selector)["nodeId"]
    cdp("DOM.setFileInputFiles", files=[path], nodeId=nid)

还记得博客"神奇时刻"里那个官方忘了写、agent 任务中途自己现写的上传函数吗?就是它——后来被收编进了正式版。你现在看到的这四行,初稿作者是某次任务里的一个 AI。

def goto_url(url):
    r = cdp("Page.navigate", url=url)
    if os.environ.get("BH_DOMAIN_SKILLS") != "1":
        return r
    d = (AGENT_WORKSPACE / "domain-skills" / 域名)
    return {**r, "domain_skills": 该域名下的笔记清单}

这就是被你冻结的第 5 课的开关本尊:每次导航,若 BH_DOMAIN_SKILLS=1,顺手把"这个网站的历史笔记清单"附在返回值里递给 agent。机制全貌就这几行——不神秘,就是"进店先翻小本本"。

def _load_agent_helpers():
    p = AGENT_WORKSPACE / "agent_helpers.py"
    if not p.exists(): return
    ...把里面的函数全部注入到厨房...

_load_agent_helpers()   # 文件末尾:每次启动都执行

自愈循环的接口:官方 helpers 之外,还有一个专属于 agent 的自留地文件,每次启动自动加载。agent 哪次任务里发明了新工具,写进这个文件,以后每次开工厨房里就多一件趁手家伙。"工具会生长"落到实处,就是文件末尾这一次调用。

"""CDP WS holder + IPC relay (Unix socket on POSIX,
TCP loopback on Windows). One daemon per BU_NAME."""

第一行自我介绍:我抱着那条通往 Chrome 的 websocket 不撒手,并且当中继站。为什么需要它?你每次管道喂代码都是一个短命进程,跑完就死;和 Chrome 握手授权很贵,不能每次重来——所以要一个长命的家伙抱着连接,短命进程来借。

_WINDOWS_PROFILES = (
    "Google/Chrome/User Data", "Google/Chrome SxS/User Data",
    "Chromium/User Data", "Microsoft/Edge/User Data",
    "BraveSoftware/Brave-Browser/User Data", ...)

它找浏览器的方式是穷举所有已知浏览器的安置点,挨家挨户敲门看谁开着调试端口——Chrome 正式版、金丝雀版、Edge 四个版本、Brave,全在名单上。还记得那次连接失败,报错列出一长串路径吗?就是这张名单在喊话。

if e.code == 403:
    raise RuntimeError("permission-blocked: Chrome is reachable,
      but the per-session Allow remote debugging popup
      has not been accepted")

今天你亲手点过的那个 Allow,在源码里有专属报错:HTTP 403 = "Chrome 活着,但这一届会话的授权弹窗还没人点"。我们白天撞的墙,晚上在源码里找到了墙的图纸——这就是读源码的乐趣。