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 统一抽象为通道ChannelVNC 通道承载 RFB 协议字节流、Message 通道承载前端与 API 之间的 JSON 消息、CDP 通道承载面向浏览器的 DevTools 协议数据。本文结合源码逐层拆解这套机制。一、什么是通道命名 WebSocket 的抽象channels/README.md 给出的定义非常直白通道就是服务于某种特定用途的 WebSocket。它们被归类为命名通道只是为了便于理解底层协议并无魔法——所有通道都是普通的 WebSocket 连接区别只在于上面跑的数据是什么。当前共有四类通道通道数据形态链路主要用途VNC 通道noVNC 的 RFB 协议原始字节流前端 ⇄ API Server ⇄ websockify noVNC Server ⇄ 浏览器实时渲染浏览器画面、透传键盘/鼠标输入Message 通道JSON 消息前端 ⇄ API Server复制/粘贴协调、控制权交接agent/user、录制命令、解释更新等CDP Execution 通道CDP 协议数据经 PlaywrightAPI Server ⇄ 浏览器一次性执行JS 求值、粘贴文本、读取选中文本、导航等CDP Exfiltration 通道原始 JavaScript 事件JSONAPI Server ⇄ 浏览器Record Browser 的用户行为捕获与导出从源码结构看前三类分别落在 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 screencastcdp流式传输模式渲染画面而不是走 VNC。这与 streaming/registries.py 头部的注释相互印证传统的 VNC 实时查看会打开两条配对的 WebSocketVNC 通道 Message 通道而 CDP Record Browser 路径在 API 侧只打开 Message 通道画面帧与用户事件捕获都走 CDPscreencast 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 的数据结构VncChannelchannels/vnc.py 第 145-257 行的核心字段client_id唯一标识一个前端实例是跨通道配对的纽带organization_id、vnc_port、x_api_key组织归属、VNC 端口与鉴权密钥websocket与前端建立的 FastAPI WebSocketinitial_interactor/_interactor当前交互方agent 或 user二值状态browser_session/task/workflow_run通道所绑定的后台实体三者至多一个生效identity属性据此拼接日志上下文key_state跟踪 Ctrl/Alt/Cmd 按键状态的机内状态。MessageChannelchannels/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_idabc123) ├─→ VNC Channel ─────→ API Server Instance #2 │ vnc_channels[abc123] VncChannel └─→ Message Channel ──→ API Server Instance #2 message_channels[abc123] MessageChannelstreaming/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.pyverify_browser_session第 42-135 行检查会话存在、非终态is_final_status、未超时结合timeout_minutes、last_activity_at与MAX_LIFETIME_SECONDS判定并解析出可寻址的浏览器地址browser_addressverify_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循环职责退出条件验证循环每 5 秒轮询数据库刷新通道持有的实体状态实体失效置空或 WebSocket 关闭数据流循环VNC 双向 RFB 透传 / Message 双向 JSON 收发断开连接轮询间隔常量为POLL_INTERVAL_FOR_VERIFICATION_SECONDS 5verify.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 说明了它的行为等待任意一个循环失败或完成取消所有其余循环传播第一个异常。以loop_stream_vnc为例vnc.py 第 512-527 行两个子任务frontend_to_browser与browser_to_frontend被collect(loops)管理任何一侧异常都会触发另一侧取消并在finally中执行vnc_channel.close(reasonloop-stream-vnc-closed)。Message 通道略有不同message.py 第 1032-1077 行loop_stream_messages用asyncio.wait(loops, return_whenFIRST_COMPLETED)替代collect因为一侧的干净返回也必须拆掉另一侧否则receive_json()/out_queue.get()会永远阻塞随后在finally中取消所有循环、取消遗留的剪贴板任务、停止 Exfiltration 通道并关闭 Message 通道。清理close() 的三件事README 指出channel.close()总是执行三件事将browser_session/task/workflow_run置空关闭 WebSocket从注册表移除自身。对应 vnc.py 第 221-235 行与 message.py 第 450-463 行的实现。注意删除注册表条目时都传入了expectedself防止误删复用同一client_id的新通道。五、VNC 通道数据流RFB 透传与输入拦截VNC 通道是纯透传pass-through通道但正因为透传经过 API ServerSkyvern 得以在中间监控并拦截 RFB 协议消息。README 绘制的数据流如下User 键盘/鼠标输入 ▼ Frontend (noVNC client) —— 将输入编码为 RFB 协议字节 ▼ WebSocket (bytes) API Server: VncChannel.loop_stream_vnc() frontend_to_browser() 协程: 1. 从前端接收 RFB 字节 2. 识别消息类型keyboard4, mouse5 3. 更新 key_state 跟踪 4. 检查特殊组合键 - CtrlC / CmdC → copy_text() 经 CDP - CtrlV / CmdV → ask_for_clipboard() 经 Message - CtrlO → 拦截禁止 5. 检查 interactor 模式 - interactoragent → 拦截用户输入 - interactoruser → 放行 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 6frontend_to_browser循环里用data[0]判别第 388 行按键状态机KeyState第 114-142 行跟踪 Ctrl/Alt/Cmd 的按下与抬起并提供is_copy()Ctrl/CmdC、is_paste()Ctrl/CmdV、is_forbidden()/is_ctrl_o()CtrlO 禁止防止在远程浏览器中打开本地文件复制/粘贴钩子命中 CtrlC 时调用copy_text(vnc_channel)第 260-280 行通过 ExecutionChannel 执行get_selected_text()读取选中文本再经 Message 通道send_copied_text()回传前端命中 CtrlV 且远端剪贴板未在 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()检查后创建通道端点后台实体/stream/vnc/browser_session/{browser_session_id}base_router持久化浏览器会话/stream/vnc/task/{task_id}legacy_base_routerTask/stream/vnc/workflow_run/{workflow_run_id}legacy_base_routerWorkflow Run对应工厂函数为 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]。消息类型全表MessageKindchannels/message.py 第 54-79 行定义了完整的MessageKind枚举双向消息共约 27 种方向kind含义入站ask-for-clipboard-response前端返回剪贴板内容入站begin-exfiltration/end-exfiltration开始/结束录制捕获入站take-control/cede-control用户接管 / 交还控制权入站clipboard-copy/clipboard-paste复制选中文本 / 粘贴文本入站get-browser-url/browser-url出站查询/返回当前 URL入站navigate/reload/go-back/go-forward浏览器导航控制入站take-screenshot→ 出站screenshot截图请求与响应入站clear-cookies/clear-history/clear-all-data清理浏览器数据入站recording-capture-pause/resume/rearm录制捕获暂停/恢复/重新武装出站error后端处理失败前端弹 toast出站exfiltrated-event用户事件录制数据回传出站recording-interpretation-update录制解释增量/快照更新入站消息由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_backendwebsocket.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.NAVIGATEURL 先经_normalize_url补全 scheme、再经validate_fetch_url校验被拒地址BlockedHost等只记录服务器日志不回显到 toast避免调用方借报错文本枚举内网地址段BEGIN_EXFILTRATION在 VNC 通道存在时复用其上下文否则退化为MessageChannelContext创建并启动ExfiltrationChannel同时可启用 live interpretation 会话TAKE_SCREENSHOT截图字节超MAX_SCREENSHOT_BYTES 5MB拒绝成功则返回 base64 编码的 PNGMessageOutScreenshot。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(contextvnc_channel) try: await channel.connect() yield channel finally: await channel.stop() # 注意用 stop() 而非 close()execution.py 第 334-357 行注释解释了为何用stop()裸close()会触发浏览器 disconnected 事件而重连回调会复活一个永远没人释放的驱动。ExecutionChannel提供的能力与 Message 命令一一对应方法实现要点get_selected_text()执行 JSwindow.getSelection().toString()返回选中文本get_current_url()读取page.urlpaste_text(text)page.keyboard.insert_text(text)模拟插入navigate(url)规范化 URL →validate_fetch_url校验 →page.goto(wait_untildomcontentloaded)reload(hard)hardTrue时先经 CDPNetwork.clearBrowserCache再 reloadgo_back()/go_forward()Playwright 页面历史导航take_screenshot()PNG 截图非全页超 5MB 拒绝返回 base64clear_cookies()/clear_storage()/clear_history()经 CDPStorage.clearDataForOrigin/Network.clearBrowserCache等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 reconnectstop()与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去重与排序事件带exfilDocIdexfilSeq组合去重键_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 归纳的六条关键设计模式已逐条与源码核对Channel Pairing两条 WebSocket 通过进程内注册表按client_id协调registries.pyFail-Fast Loopscollect()保证任一循环失败即关闭整条通道utils/aio.pyInteractor Modeagent/user二值状态控制用户输入是否放行vnc.pyinteractor属性 数据流循环拦截逻辑On-Demand CDPExecutionChannel 每次操作临时建连、用完即关保持无状态execution.pyexecution_channel上下文管理器Polling Verification每 5 秒验证后台实体有效性verify.pyPOLL_INTERVAL_FOR_VERIFICATION_SECONDSPass-Through ProxyAPI Server 拦截但不转换 RFB 数据只在透传路径上做监控与过滤vnc.pyloop_stream_vnc。十、阅读源码的路线图如果你想深入这套机制推荐按以下顺序阅读仓库源码channels/README.md —— 本篇文章的原始依据架构总览streaming/registries.py —— 进程内注册表与流引用计数streaming/verify.py —— 实体校验与 5 秒验证循环channels/vnc.py —— RFB 透传、输入拦截、剪贴板协调channels/message.py —— JSON 消息协议与全量命令分发channels/cdp.py → channels/execution.py → channels/exfiltration.py —— CDP 连接基建、一次性执行与录制捕获vnc.py 与 messages.py —— WebSocket 路由端点与鉴权入口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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考