Skyvern 远程浏览器流式传输通道（Channels）架构解析：VNC / Message / CDP 三类 WebSocket 通道的设计与实现

Skyvern 远程浏览器流式传输通道（Channels）架构解析：VNC / Message / CDP 三类 WebSocket 通道的设计与实现

【免费下载链接】skyvernAutomate browser based workflows with AI项目地址: https://gitcode.com/GitHub_Trending/sk/skyvern

本文基于 channels/README.md 整理，并对照 skyvern/forge/sdk/routes/streaming 目录下的实际源码进行印证与扩充。文章介绍 Skyvern 如何用一组"命名 WebSocket 通道"支撑远程浏览器的实时画面渲染、前后端命令交互与 Record Browser 用户行为录制，读完你可以掌握三类通道的职责边界、跨通道协作机制、生命周期管理，以及它们与持久化浏览器会话（Persistent Browser Session）之间的绑定关系。

Skyvern 的远程浏览器能力（远程实时查看、人类接管操作、录制回放）本质上依赖一组 WebSocket 连接。为了给不同的数据形态（像素流、JSON 控制消息、浏览器协议指令）提供清晰的职责边界，Skyvern 把这些 WebSocket 统一抽象为"通道（Channel）"：VNC 通道承载 RFB 协议字节流、Message 通道承载前端与 API 之间的 JSON 消息、CDP 通道承载面向浏览器的 DevTools 协议数据。本文结合源码逐层拆解这套机制。

一、什么是"通道"：命名 WebSocket 的抽象

channels/README.md 给出的定义非常直白：通道就是服务于某种特定用途的 WebSocket。它们被归类为"命名通道"只是为了便于理解，底层协议并无魔法——所有通道都是普通的 WebSocket 连接，区别只在于上面跑的数据是什么。

当前共有四类通道：

从源码结构看，前三类分别落在 channels/vnc.py、channels/message.py 与 channels/execution.py，而 Exfiltration 通道在 channels/exfiltration.py。每类通道文件头部都以 ASCII 图描述了自身的链路形态，例如 vnc.py 的注释是：

[Skyvern App] <--> [API Server] <--> [websockified noVNC] <--> [Browser]

而 message.py 是：

[Skyvern App] <--> [API Server]

三类通道在业务上的分工

README 用一段话总结了通道在真实业务中的配合方式：

• First-party browser sessions（自持浏览器会话）：通过 VNC 通道渲染实时画面；
• Record Browser：通过 Exfiltration 通道捕获用户事件，通过 Message 通道承载录制命令与解释（interpretation）更新；
• Vendor-held sessions（第三方托管会话）：当没有可中继的 RFB 端点时，改由 CDP screencast（cdp流式传输模式）渲染画面，而不是走 VNC。

这与 streaming/registries.py 头部的注释相互印证：传统的 VNC 实时查看会打开两条配对的 WebSocket（VNC 通道 + Message 通道）；而 CDP Record Browser 路径在 API 侧只打开 Message 通道，画面帧与用户事件捕获都走 CDP（screencast + ExfiltrationChannel），因此不存在跨 socket 配对被打断的问题。

二、总体架构与注册表设计

README 中给出了整幅高层组件图，这里将其核心结构重绘并标注源码位置：

┌─────────────────┐ │ Frontend App │ │ (Skyvern) │ └────────┬────────┘ │ 两条 WebSocket 连接（通过 client_id 配对） ┌────┴────┬──────────────────────────────────────────┐ │ │ │ │ VNC Channel Message Channel │ (RFB Protocol) (JSON Messages) │ │ │ ┌───▼─────────▼──────────────────────────────────────────▼────┐ │ API Server │ │ 注册表（进程内内存状态） │ │ - vnc_channels: dict[client_id -> VncChannel] │ │ - message_channels: dict[client_id -> MessageChannel] │ │ │ │ VNC 通道逻辑: Message 通道逻辑: │ │ - RFB 透传 - 复制/粘贴协调 │ │ - 键盘/鼠标过滤 - 控制权交接（agent/user） │ │ - Interactor 控制 - 剪贴板管理 │ │ - 复制/粘贴检测 - 通道协调 │ │ │ │ CDP 通道（按需创建）: │ │ - ExecutionChannel: JS 求值（粘贴、读取选中文本） │ │ - ExfiltrationChannel: Record Browser 用户事件捕获 │ └────┬─────────────────────────────────────────────────────┬──┘ │ WebSocket (RFB) Playwright (CDP) │ ┌────▼─────────────────────────────────────────────────────▼──┐ │ Persistent Browser Session │ │ noVNC Server (websockify) + Chrome/Chromium │ └─────────────────────────────────────────────────────────────┘

进程内注册表：client_id是唯一纽带

所有活跃通道都登记在 streaming/registries.py 的进程内内存字典中，以client_id为键：

• vnc_channels: dict[str, VncChannel]（第 45 行）
• message_channels: dict[str, MessageChannel]（第 71 行）

对应的增删查函数为add_vnc_channel/get_vnc_channel/del_vnc_channel与add_message_channel/get_message_channel/del_message_channel。其中几个细节值得注意：

• get_message_channel（第 78-92 行）在返回前会检查通道is_open，若已关闭会先将其从注册表清除再返回None，避免把失效通道交还给调用方；
• 删除函数都带expected参数做身份校验——当同一个client_id在路由切换或重连时被新通道复用，旧通道的关闭流程不能误删新通道（源码注释明确说明了这一点）；
• del_vnc_channel同样是"只在候选对象与 expected 一致时才删除"的防御式清理。

此外，该模块还维护着 CDP 输入通道注册表cdp_input_channels，以及一套流引用计数（try_stream_ref_inc/stream_ref_dec/finalize_stream_teardown），用于在活跃 CDP 流持有 workflow run 的浏览器时，推迟浏览器状态的清理。

VncChannel 与 MessageChannel 的数据结构

VncChannel（channels/vnc.py 第 145-257 行）的核心字段：

• client_id：唯一标识一个前端实例，是跨通道配对的纽带；
• organization_id、vnc_port、x_api_key：组织归属、VNC 端口与鉴权密钥；
• websocket：与前端建立的 FastAPI WebSocket；
• initial_interactor/_interactor：当前交互方，"agent" 或 "user"（二值状态）；
• browser_session/task/workflow_run：通道所绑定的后台实体（三者至多一个生效，identity属性据此拼接日志上下文）；
• key_state：跟踪 Ctrl/Alt/Cmd 按键状态的机内状态。

MessageChannel（channels/message.py 第 414-495 行）则带有一个out_queue异步队列：所有出站帧先入队，再由唯一写者任务backend_to_frontend串行写 WebSocket，从而避免不同协程并发写同一 socket 造成帧交错（源码注释中多次强调"single-writer"这一设计约束）。

三、通道配对与粘性会话：必须遵守的部署约束

README 用显著篇幅强调了一个关键设计约束：

同一个前端实例的 VNC 通道与 Message 通道必须连接到同一个 API Server 实例，因为它们通过以client_id为键的进程内注册表协调。

原因很直观：注册表是进程内存状态。如果两条 WebSocket 被负载均衡到不同后端实例，get_message_channel(client_id)在另一个进程里就查不到配对通道。README 中给出示例：

Frontend Instance (client_id="abc123") ├─→ VNC Channel ─────→ API Server Instance #2 │ vnc_channels["abc123"] = VncChannel └─→ Message Channel ──→ API Server Instance #2 message_channels["abc123"] = MessageChannel

streaming/registries.py 头部注释也印证了这一点：在 AWS 上他们不得不开启等价于粘性会话的机制，保证单个前端实例始终连接同一个后端 API 实例。

部署要求（Deployment Requirement）：负载均衡器必须启用粘性会话（如基于 Cookie 或基于 IP 的亲和性），确保来自同一client_id的两条 WebSocket 到达同一后端实例。这是多副本部署下实时查看功能可用的前提。

四、通道生命周期：验证、双循环与 fail-fast

创建时的实体校验

每个通道在创建时都要先验证其绑定的后台实体。README 的生命周期图指出创建流程为：verify_browser_session()/verify_task()/verify_workflow_run()→ 返回entity + browser_session→ 创建通道并加入注册表。

校验逻辑集中在 streaming/verify.py：

• verify_browser_session（第 42-135 行）：检查会话存在、非终态（is_final_status）、未超时（结合timeout_minutes、last_activity_at与MAX_LIFETIME_SECONDS判定），并解析出可寻址的浏览器地址browser_address；
• verify_task（第 138-188 行）：任务状态必须是created / queued / running（终态直接拒绝），并通过PERSISTENT_SESSIONS_MANAGER.get_session_by_runnable_id找到关联浏览器会话；
• verify_workflow_run（第 191-275 行）：workflow run 状态必须是created / queued / running / paused，同样解析其浏览器会话。

双循环并发执行

通道建立后，会并行跑两个循环（README 称为 "Loops"，源码中Loops = list[asyncio.Task]，注释戏称其为 "queue-less actors"）：

轮询间隔常量为POLL_INTERVAL_FOR_VERIFICATION_SECONDS = 5（verify.py 第 39 行）。验证循环会把校验结果写回通道字段（例如vnc_channel.task = task、vnc_channel.browser_session = browser_session），一旦实体进入终态，字段被置空，is_open属性随之返回False，数据流循环自然退出。

collect()：fail-fast 循环管理

两个循环通过 skyvern/forge/sdk/utils/aio.py 中的collect()聚合。README 说明了它的行为：

1. 等待任意一个循环失败或完成；
2. 取消所有其余循环；
3. 传播第一个异常。

以loop_stream_vnc为例（vnc.py 第 512-527 行），两个子任务frontend_to_browser与browser_to_frontend被collect(loops)管理，任何一侧异常都会触发另一侧取消，并在finally中执行vnc_channel.close(reason="loop-stream-vnc-closed")。

Message 通道略有不同（message.py 第 1032-1077 行）：loop_stream_messages用asyncio.wait(loops, return_when=FIRST_COMPLETED)替代collect，因为一侧的干净返回也必须拆掉另一侧（否则receive_json()/out_queue.get()会永远阻塞）；随后在finally中取消所有循环、取消遗留的剪贴板任务、停止 Exfiltration 通道并关闭 Message 通道。

清理：close() 的三件事

README 指出channel.close()总是执行三件事：

1. 将browser_session/task/workflow_run置空；
2. 关闭 WebSocket；
3. 从注册表移除自身。

对应 vnc.py 第 221-235 行与 message.py 第 450-463 行的实现。注意删除注册表条目时都传入了expected=self，防止误删复用同一client_id的新通道。

五、VNC 通道数据流：RFB 透传与输入拦截

VNC 通道是纯透传（pass-through）通道，但正因为透传经过 API Server，Skyvern 得以在中间监控并拦截 RFB 协议消息。README 绘制的数据流如下：

User 键盘/鼠标输入 ▼ Frontend (noVNC client) —— 将输入编码为 RFB 协议字节 ▼ WebSocket (bytes) API Server: VncChannel.loop_stream_vnc() frontend_to_browser() 协程: 1. 从前端接收 RFB 字节 2. 识别消息类型（keyboard=4, mouse=5） 3. 更新 key_state 跟踪 4. 检查特殊组合键： - Ctrl+C / Cmd+C → copy_text() 经 CDP - Ctrl+V / Cmd+V → ask_for_clipboard() 经 Message - Ctrl+O → 拦截（禁止） 5. 检查 interactor 模式： - interactor=="agent" → 拦截用户输入 - interactor=="user" → 放行 6. 屏蔽鼠标右键（安全原因） 7. 转发给 noVNC server ▼ WebSocket (bytes) Persistent Browser: noVNC Server (websockify) —— RFB → VNC 协议转换 ▼ Browser Display Update 屏幕更新（反向）： Browser → noVNC → VncChannel.browser_to_frontend() → Frontend

源码级实现对照

在 channels/vnc.py 中可以逐条找到对应实现：

• 消息类型识别：MessageType枚举（第 74-77 行）定义了Keyboard = 4、Mouse = 5、ClientCutText = 6；frontend_to_browser循环里用data[0]判别（第 388 行）；
• 按键状态机：KeyState（第 114-142 行）跟踪 Ctrl/Alt/Cmd 的按下与抬起，并提供is_copy()（Ctrl/Cmd+C）、is_paste()（Ctrl/Cmd+V）、is_forbidden()/is_ctrl_o()（Ctrl+O 禁止，防止在远程浏览器中打开本地文件）；
• 复制/粘贴钩子：命中 Ctrl+C 时调用copy_text(vnc_channel)（第 260-280 行），通过 ExecutionChannel 执行get_selected_text()读取选中文本，再经 Message 通道send_copied_text()回传前端；命中 Ctrl+V 且远端剪贴板未在 2 秒宽限期内同步过（REMOTE_CLIPBOARD_SYNC_PASTE_GRACE_SECONDS = 2.0，第 80 行）时，调用ask_for_clipboard(vnc_channel)（第 282-298 行），经 Message 通道向前端请求剪贴板内容；
• Interactor 拦截：第 413-418 行，当interactor != "user"时，键盘与鼠标消息一律continue丢弃——即自动化（agent）控制期间用户无法插手；
• 右键屏蔽：is_rmb(data)判断data[0:2] == b"\x05\x04"（鼠标按下消息 + 右按钮位），命中则丢弃（第 408-411 行）；
• VNC 地址构建：_build_vnc_url_from_browser_address（第 304-344 行）根据浏览器会话的browser_address拼接出带凭证的 VNC 路由地址——回环地址（本地开发/单容器自托管）直接用ws://{host}:{vnc_port}，其余地址必须走wss://路由器（路径形如/vnc/{session_id}?token=...），且wss://连接会附加x-api-key请求头（第 370-375 行）。对无法构建路由地址的会话，直接抛出MissingRoutedVncAddressError，避免把私网地址的裸流发给客户。

VNC 通道的三种创建入口

vnc.py 的 WebSocket 路由暴露了三个端点，分别对应三种后台实体，均在通过auth()鉴权与require_client_id()检查后创建通道：

对应工厂函数为 channels/vnc.py 中的get_vnc_channel_for_browser_session/get_vnc_channel_for_task/get_vnc_channel_for_workflow_run。三者都返回(VncChannel, Loops)二元组，其中Loops恒为[验证循环, 流式循环]。值得注意的是initial_interactor一律初始化为"agent"，即通道建立之初用户输入是被拦截的。

六、Message 通道：JSON 消息协议与命令处理

Message 通道承载前端与 API Server 之间的 JSON 消息。README 强调其语义是fire-and-forget（发后即忘），请求-响应模式建立在消息类型之上。通道形态为[Skyvern App] <--> [API Server]。

消息类型全表（MessageKind）

channels/message.py 第 54-79 行定义了完整的MessageKind枚举，双向消息共约 27 种：

入站消息由reify_channel_message(data)（第 325-379 行）按kind字段反序列化为对应的 dataclass（未知 kind 抛ValueError）；出站消息经message_to_dict（第 382-400 行）序列化，其中枚举转为值、NaN/Infinity 替换为None（_replace_non_finite，防止JSON.parse整帧拒绝）。

消息处理主循环

loop_stream_messages（第 519-1077 行）内部有两组协程：

• frontend_to_backend：websocket.receive_json()→handle_data(data)分发处理，遇断开抛异常退出；
• backend_to_frontend：唯一的 WebSocket 写者，从out_queue取消息并send_json。

handle_data（第 555-977 行）是一个巨大的match message.kind分发器，每条分支要么直接执行，要么通过execution_for_message_channel打开 CDP 执行通道完成浏览器操作。几个值得展开的分支：

• TAKE_CONTROL / CEDE_CONTROL（第 951-962、738-748 行）：通过注册表get_vnc_channel(client_id)找到配对 VNC 通道，把vnc_channel.interactor置为"user"/"agent"——这就是"接管控制"的实现本质；
• CLIPBOARD_PASTE / ASK_FOR_CLIPBOARD_RESPONSE：先做剪贴板大小限制校验（MAX_CLIPBOARD_PASTE_BYTES = 1MB，见 payload_limits.py），超限直接回error；同时用信号量式的计数（_MAX_IN_FLIGHT_CLIPBOARD_TASKS = 2）限制并发剪贴板任务，忙时提示 "Clipboard is busy; try again."；
• NAVIGATE：URL 先经_normalize_url补全 scheme、再经validate_fetch_url校验；被拒地址（BlockedHost等）只记录服务器日志，不回显到 toast，避免调用方借报错文本枚举内网地址段；
• BEGIN_EXFILTRATION：在 VNC 通道存在时复用其上下文，否则退化为MessageChannelContext，创建并启动ExfiltrationChannel，同时可启用 live interpretation 会话；
• TAKE_SCREENSHOT：截图字节超MAX_SCREENSHOT_BYTES = 5MB拒绝，成功则返回 base64 编码的 PNG（MessageOutScreenshot）。

Message 通道的路由端点在 messages.py：/stream/messages/browser_session/{browser_session_id}（base_router）与/stream/messages/workflow_run/{workflow_run_id}（legacy_base_router）。

七、控制权交接：Agent ↔ User 交互模型

README 特别说明：当前系统里并不存在真正的 "AI agent"——任何非用户发起的浏览器控制都"有点像 agent"，未来若引入自动操作的 agent 可复用这套机制。当前实现中，interactor 是一个二值状态机：

初始状态: interactor = "agent" - 用户键盘/鼠标输入被拦截 - （未来的）agent 可通过 CDP 控制浏览器 │ 用户在前端点 "Take Control" ▼ Frontend → MessageChannel: {"kind": "take-control"} ▼ MessageChannel.handle_data() → vnc_channel.interactor = "user" ▼ 新状态: interactor = "user" - 用户键盘/鼠标输入放行 - agent 应暂停自动化 │ 用户在前端点 "Cede Control" ▼ Frontend → MessageChannel: {"kind": "cede-control"} ▼ MessageChannel.handle_data() → vnc_channel.interactor = "agent" ▼ 回到初始状态

实现上，interactor是VncChannel的一个带日志的属性（vnc.py 第 198-206 行），切换时打印日志；拦截逻辑则位于 VNC 数据流循环的第 413-418 行——这也解释了为什么"接管控制"的消息走 Message 通道、而拦截发生在 VNC 通道：两条通道通过client_id配对，Message 通道修改 VNC 通道的状态，VNC 通道据此放行或拦截输入。

八、CDP 通道：按需创建的一次性执行

ExecutionChannel（一次性执行）

channels/execution.py 中的ExecutionChannel继承自 channels/cdp.py 的CdpChannel。README 指出它是"one-off executions"——每次操作都新建连接、用完即关，保持无状态：

@asynccontextmanager async def execution_channel(vnc_channel: VncChannel): channel = ExecutionChannel(context=vnc_channel) try: await channel.connect() yield channel finally: await channel.stop() # 注意用 stop() 而非 close()

（execution.py 第 334-357 行，注释解释了为何用stop()：裸close()会触发浏览器 "disconnected" 事件，而重连回调会复活一个永远没人释放的驱动。）

ExecutionChannel提供的能力与 Message 命令一一对应：

CdpChannel：连接基建

channels/cdp.py 的CdpChannel是抽象基类（直接实例化会抛TypeError），封装了：

• connect()（幂等）：通过 Playwright 的async_playwright()启动驱动，connect_over_cdp_with_diagnostics建立浏览器连接，优先复用已有 browser context / page；
• 鉴权头：x-api-key（以及本地 PBS 地址的X-Session-Id）；
• 下载行为：apply_download_behavior通过 CDPBrowser.setDownloadBehavior将下载落到/app/downloads/{org_id}/{browser_session_id}；
• 断线自愈：注册浏览器disconnected回调，非终止状态下自动close()后重连（源码注释标注了TODO: avoid blind reconnect）；
• stop()与close()的语义区分：close()可被connect()复用来回收掉线连接，stop()是终态关闭，置_closing = True防止重连回调复活驱动（对应 SKY-12524 的修复）；
• JS 资产加载：_load_js_asset以模块级lru_cache缓存 channels/js 目录下的脚本（adorn.js、decorate.js、exfiltrate.js、undecorate.js）。

ExfiltrationChannel（用户事件捕获）

channels/exfiltration.py 实现了 Record Browser 的事件捕获通道，链路为[Skyvern App] <-- [API Server] <--> [Browser (CDP)]，数据形态是"原始 JavaScript 事件（JSON）经 WebSocket 传输"。

核心机制：

• 页面内注入：通过page.expose_binding("__skyvern_exfiltrate_event", ...)暴露绑定 +add_init_script注入 js/exfiltrate.js，页面事件以[EXFIL]前缀 JSON 写入window.__skyvern_exfil_queue；
• 双通道采集：既监听 Playwright 的ConsoleMessage，也经 CDPRuntime.consoleAPICalled抓取（_attach_page_cdp_console_capture）；
• 队列排空：_drain_queue_loop每 0.25 秒执行一次_DRAIN_QUEUE_JS，按"所有权令牌"（_drain_token）校验后最多取 1000 条；页内队列上限与QUEUE_DRAIN_MAX_ITEMS镜像（注释指明对应 exfiltrate.js 中的EXFIL_QUEUE_LIMIT）；
• 去重与排序：事件带exfilDocId+exfilSeq组合去重键（_event_dedup_key），并带单调递增的capture_seq供解释器恢复时间顺序；
• 刷新循环：每 1 秒对每个页面重新注入捕获脚本与装饰（鼠标跟随器等，decorate.js/undecorate.js/adorn.js），覆盖导航后的新文档；
• 捕获控制：pause_capture()/resume_capture()配合 Message 通道的recording-capture-pause/resume命令；
• 事件出口：通过on_event回调转成MessageOutExfiltratedEvent经 Message 通道发给前端，并同步喂给interpretation_registry做 live interpretation。

九、实体关系与关键设计模式总结

README 给出了通道与后台实体的关联关系：

Organization ├─→ BrowserSession ────────┐ ├─→ Task ──────────────────┤ （可选绑定 browser_session） └─→ WorkflowRun ───────────┤ （可选绑定 browser_session） ▼ VncChannel + MessageChannel （进程内内存，按 client_id 配对）

通道本身不持久化，是"挂在实体上的瞬时连接对象"；实体一旦终态，验证循环在 5 秒内发现并拆除通道。

最后，README 归纳的六条关键设计模式（已逐条与源码核对）：

1. Channel Pairing：两条 WebSocket 通过进程内注册表按client_id协调（registries.py）；
2. Fail-Fast Loops：collect()保证任一循环失败即关闭整条通道（utils/aio.py）；
3. Interactor Mode："agent"/"user"二值状态控制用户输入是否放行（vnc.pyinteractor属性 + 数据流循环拦截逻辑）；
4. On-Demand CDP：ExecutionChannel 每次操作临时建连、用完即关，保持无状态（execution.pyexecution_channel上下文管理器）；
5. Polling Verification：每 5 秒验证后台实体有效性（verify.pyPOLL_INTERVAL_FOR_VERIFICATION_SECONDS）；
6. Pass-Through Proxy：API Server 拦截但不转换 RFB 数据，只在透传路径上做监控与过滤（vnc.pyloop_stream_vnc）。

十、阅读源码的路线图

如果你想深入这套机制，推荐按以下顺序阅读仓库源码：

1. channels/README.md —— 本篇文章的原始依据，架构总览；
2. streaming/registries.py —— 进程内注册表与流引用计数；
3. streaming/verify.py —— 实体校验与 5 秒验证循环；
4. channels/vnc.py —— RFB 透传、输入拦截、剪贴板协调；
5. channels/message.py —— JSON 消息协议与全量命令分发；
6. channels/cdp.py → channels/execution.py → channels/exfiltration.py —— CDP 连接基建、一次性执行与录制捕获；
7. vnc.py 与 messages.py —— WebSocket 路由端点与鉴权入口；
8. channels/js —— 注入到页面侧的捕获与装饰脚本（exfiltrate.js/adorn.js/decorate.js/undecorate.js）。

理解这套"命名通道"架构，是理解 Skyvern 远程浏览器实时查看、人工接管与录制回放三大能力的前提：VNC 通道解决"看得见"，Message 通道解决"控得住"，CDP 通道解决"执行与记录"。三者通过client_id在 API Server 进程内配对协作，配合粘性会话与 fail-fast 生命周期管理，构成一套完整、可运维的浏览器流式传输方案。

【免费下载链接】skyvernAutomate browser based workflows with AI项目地址: https://gitcode.com/GitHub_Trending/sk/skyvern