使用 Scrapeless 抓取服务器发送事件 (SSE) 流

一个实时博客的评论计数器，一个支持小部件的“代理正在输入”指示器，以及一个逐字填充的AI聊天回复共享一个特征：浏览器只打开了一次连接，从那时起，服务器一直在通过同一个开放响应推送每个更新。评论计数器没有第二个请求，输入指示器没有轮询循环——一个HTTP GET请求，保持开放状态，带有Content-Type: text/event-stream，并且每当发生变化时，服务器在其上写入data: {...}\n\n。获取页面一次并继续的工具永远看不到这些，因为数据在初始响应头之后到达，连接不会关闭。

将Playwright连接到Scrapeless Scraping Browser，地址为wss://browser.scrapeless.com/api/v2/browser，底层的CDP会话为您提供两种不同的方式来读取到达的推送帧：Playwright自己的流式响应读取器，以及来自Chrome开发者工具协议的原始Network.eventSourceMessageReceived事件。本指南连接到该云浏览器，打开一个真实的服务器推送事件（SSE）流，并以两种方式捕获帧，每种代码路径都针对一个实时公共源执行。

为什么SSE需要不同的捕获路径

page.goto()后跟一个DOM读取，只显示页面在那一刻包含的标记内容。一个SSE供给的小部件从不重新渲染整个页面——每当新的data:行到达时，它会追加或替换一个片段，因此单个DOM快照只能捕获您查看时刚好发生的更新。如果页面没有渲染它们，这些更新本身根本不会接触DOM；读取它们的唯一可靠地方是流本身。

SSE也是比浏览器可以打开的另外两种实时传输方式更为狭窄的情况。一个隐藏的JSON端点用一个响应回答一个请求——使用page.expect_response()读取它，您就完成了。WebSocket需要在任一方发送帧之前进行Upgrade: websocket握手，并且一旦打开就是全双工——任一方可以随时写入。SSE则不需要：WHATWG服务器推送事件规范将其定义为一种普通HTTP响应，其主体永远不会结束，这是对普通GET的响应，读取时只需要常规的流体读取器。服务器向其写入；客户端只读取。

Chrome开发者工具协议直接暴露了该流，就像它暴露WebSocket帧和拦截的HTTP响应一样。其网络域为每条页面的EventSource连接收到的消息触发一个专用的eventSourceMessageReceived事件，与普通获取触发的通用响应体事件分开。Scrapeless Scraping Browser是一个仅通过CDP可达的云Chromium会话——没有WebDriver/Selenium端点来驱动它——因此任何支持CDP的客户端，包括这里的Playwright，都可以读取任一层：浏览器自己的流式主体，或其下方的协议事件。

前提条件

您需要Python 3.9或更高版本——playwright 1.59.0在PyPI上声明Requires-Python >=3.9——需要playwright包，以及来自app.scrapeless.com免费计划的Scrapeless API密钥。无需本地Chrome二进制文件：connect_over_cdp可以达到已存在于Scrapeless云中的浏览器。

以下示例连接到维基媒体基金会的公共recentchange流，该流在维基媒体自己的EventStreams服务页面上进行了文档记录。它不需要API密钥和账户——所有维基媒体项目中的每个编辑都是公开的，由此设计，流的存在专门是为了让工具可以消费它。两个示例在五个真实帧后自动停止，因此两个运行都没有持久保持连接，时间限制在证明捕获有效的时间内。

安装

pip install playwright

export SCRAPELESS_API_KEY="your_scrapeless_api_key"

通过CDP连接

重用任何Playwright到Scraping-Browser脚本使用的相同URL构建模式：一个WSS端点上的三个查询参数。

import os
from urllib.parse import urlencode

API_KEY = os.environ["SCRAPELESS_API_KEY"]

def scraping_browser_url(proxy_country="US", session_ttl=60):
    params = urlencode({"token": API_KEY, "sessionTTL": session_ttl, "proxyCountry": proxy_country})
    return f"wss://browser.scrapeless.com/api/v2/browser?{params}"

与地理围栏市场数据插座不同，维基媒体的公共流接受来自任何地区的连接：proxyCountry="US" 和 proxyCountry="DE" 都能够完成握手并开始在会话的实时运行中传递帧，双方都没有关闭代码或连接失败。proxyCountry 对许多实际目标仍然很重要——针对特定市场的流在本系列中是一个真正的失败模式——但在这个特定的公共源上并不是约束条件。请根据您自己的目标进行确认，而不是假设任何结果。

使用 Playwright 的响应流捕获帧

Playwright 的 响应事件 API 一旦响应的头部到达，就会触发 page.on("response")，而不必等待主体完成——这在这里很重要，因为 SSE 响应本身永远不会完成。将此与页面自己的 fetch() 和一个 ReadableStream 读取器配对，您可以在接收块时读取主体，而不是等待一个从未到来的完成事件：

import json
import os
from urllib.parse import urlencode

from playwright.sync_api import sync_playwright

API_KEY = os.environ["SCRAPELESS_API_KEY"]
FRAME_LIMIT = 5
STREAM_URL = "https://stream.wikimedia.org/v2/stream/recentchange"

def scraping_browser_url(proxy_country="US", session_ttl=60):
    params = urlencode({"token": API_KEY, "sessionTTL": session_ttl, "proxyCountry": proxy_country})
    return f"wss://browser.scrapeless.com/api/v2/browser?{params}"

responses_seen = []

def handle_response(response):
    if response.url == STREAM_URL:
        responses_seen.append((response.status, response.headers.get("content-type")))

with sync_playwright() as p:
    browser = p.chromium.connect_over_cdp(scraping_browser_url())
    page = browser.new_page()
    page.on("response", handle_response)

    page.goto("about:blank")
    frames = page.evaluate(
        f"""async () => {{
            const resp = await fetch("{STREAM_URL}", {{ headers: {{ "Accept": "text/event-stream" }} }});
            const reader = resp.body.getReader();
            const decoder = new TextDecoder();
            let buffer = "";
            const out = [];
            while (out.length < {FRAME_LIMIT}) {{
                const {{ done, value }} = await reader.read();
                if (done) break;
                buffer += decoder.decode(value, {{ stream: true }});
                let idx;
                while ((idx = buffer.indexOf("\\n\\n")) !== -1 && out.length < {FRAME_LIMIT}) {{
                    const rawEvent = buffer.slice(0, idx);
                    buffer = buffer.slice(idx + 2);
                    const dataLine = rawEvent.split("\\n").find(l => l.startsWith("data:"));
                    if (dataLine) out.push(dataLine.slice(5).trim());
                }}
            }}
            await reader.cancel();
            return out;
        }}"""
    )
    browser.close()

print(f"response seen via page.on('response'): status={responses_seen[0][0]}, content-type={responses_seen[0][1]}")
print(f"captured {len(frames)} frames via in-page fetch() stream reader")
print(json.dumps(json.loads(frames[0]), indent=2))

在实时流上运行它将打印一个真实的维基媒体编辑事件，以及 Playwright 自身观察到的响应：

response seen via page.on('response'): status=200, content-type=text/event-stream; charset=utf-8
captured 5 frames via in-page fetch() stream reader
{
  "$schema": "/mediawiki/recentchange/1.0.0",
  "meta": {
    "uri": "https://de.wikipedia.org/wiki/Liste_der_Kulturdenkmale_in_Oschatz",
    "domain": "de.wikipedia.org",
    "stream": "mediawiki.recentchange",
    "dt": "2026-07-28T14:16:13.757Z"
  },
  "id": 382817780,
  "type": "edit"
}

page.on("response") 确认 Playwright 自身的网络层在外部 HTTP 响应上看到了 200 和 SSE 内容类型——这证明这是一个连接，而不是五个单独的请求。页面内的循环然后逐块读取该连接的主体，在 事件流格式 用于分隔记录的空白行上进行拆分，并从每个记录中提取 data: 行。reader.cancel() 在获取到五帧的瞬间关闭底层连接，因此这里不会让维基媒体的流超过所需的证明。

从 CDP 网络域捕获原始帧

手动读取 fetch() 流是可行的，但这并不会使 Chrome 的网络堆栈将连接识别为 EventSource——这种识别是触发 CDP eventSourceMessageReceived 事件的原因，而它只在页面使用浏览器原生的 EventSource 对象打开流时触发，而不是通过简单的 fetch() 调用。当你希望浏览器自己的流会计信息而不是自定义解析器时，可以直接获取原始 CDP 事件：它将 eventName、eventId 和 data 作为单独字段返回，而不是你自己需要拆分的原始文本。

import json
import os
from urllib.parse import urlencode

from playwright.sync_api import sync_playwright

API_KEY = os.environ["SCRAPELESS_API_KEY"]
FRAME_LIMIT = 5
STREAM_URL = "https://stream.wikimedia.org/v2/stream/recentchange"

def scraping_browser_url(proxy_country="US", session_ttl=60):
    params = urlencode({"token": API_KEY, "sessionTTL": session_ttl, "proxyCountry": proxy_country})
    return f"wss://browser.scrapeless.com/api/v2/browser?{params}"

cdp_events = []

def on_sse_message(event):
    cdp_events.append(event)

with sync_playwright() as p:
    browser = p.chromium.connect_over_cdp(scraping_browser_url())
    page = browser.new_page()

    cdp = page.context.new_cdp_session(page)
    cdp.send("Network.enable")
    cdp.on("Network.eventSourceMessageReceived", on_sse_message)

    page.goto("about:blank")
    page.evaluate(
        f"""() => {{
            window.__count = 0;
            const es = new EventSource("{STREAM_URL}");
            window.__es = es;
            es.onmessage = () => {{
                window.__count += 1;
                if (window.__count >= {FRAME_LIMIT}) {{ es.close(); }}
            }};
        }}"""
    )
    page.wait_for_function(f"window.__count >= {FRAME_LIMIT}", timeout=20000)
    page.wait_for_timeout(300)
    browser.close()

print(f"捕获到 {len(cdp_events)} 个原始 CDP eventSourceMessageReceived 事件")
event = cdp_events[0]
print(f"eventName={event['eventName']}")
print(f"eventId={event['eventId']}")
print(json.dumps(json.loads(event["data"]), indent=2))

原始协议事件包含相同的编辑有效负载以及 fetch 基于解析器必须跳过的字段：

捕获到 5 个原始 CDP eventSourceMessageReceived 事件
eventName=message
eventId=[{"topic":"eqiad.mediawiki.recentchange","partition":0,"timestamp":1785248186774},{"topic":"codfw.mediawiki.recentchange","partition":0,"offset":-1}]
{
  "$schema": "/mediawiki/recentchange/1.0.0",
  "meta": {
    "uri": "https://www.wikidata.org/wiki/Q100886493",
    "domain": "www.wikidata.org",
    "stream": "mediawiki.recentchange",
    "dt": "2026-07-28T14:16:26.773Z"
  },
  "id": 2602974671,
  "type": "edit"
}

page.context.new_cdp_session(page) 打开一个会话，其 CDPSession 接口 提供 send() 用于协议命令，以及 on() 用于协议事件；Network.enable 打开事件报告，每个后续的 eventSourceMessageReceived 事件都以 DevTools 网络面板读取 EventSource 行的确切 eventName/eventId/data 形状触发。这里的 eventId 不是一个简单的计数器——维基媒体将 Kafka 主题、分区和偏移量元数据编码到其中，因为该值同时充当恢复游标：重新连接时用它作为 Last-Event-ID 请求头，服务从该确切位置继续，而不是从头开始重放。

你得到的回报

这两种路径返回相同的底层编辑事件，因为它们都在读取相同打开连接的推送记录——一个通过基于 fetch 的手动解析器，一个直接来自协议自己的 SSE 会计。

维基媒体自己的文档建议过滤掉 meta.domain 等于 "canary" 的记录——该服务为自身监控注入的合成心跳事件，而非真实的编辑。任何使用此数据源的消费者都应在将记录视为用户活动之前丢弃这些记录，正如你会丢弃某些 SSE 服务器发送的保持活动的注释行（: keepalive\n\n）；该格式允许以 : 开头的 SSE 行为完全没有 data: 字段的注释，而上述两种捕获路径也已经自然跳过了它，因为它们都没有在此查找内容。

获取您的免费计划的 API 密钥：app.scrapeless.com

SSE 与 WebSocket 的实际比较

这两种协议解决了重叠的问题，但有不同的权衡，选择错误的协议会浪费宝贵的调试时间。WebSocket 在任何数据帧传输之前需要进行 101 Switching Protocols 握手，并保持全双工连接；根据本系列的WebSocket 捕获指南，没有任何方式会自动重新连接掉线的 WebSocket。SSE 对简单的 GET 请求返回 200 和 Content-Type: text/event-stream，只有服务器进行写入，而浏览器的本机 EventSource 对象会自行重新连接。根据 WHATWG 规范，如果连接关闭，用户代理会等待一个实现定义的延迟（通常为几秒，可以通过格式定义的专用重新连接延迟字段进行调整），然后重新打开请求，自动附加 Last-Event-ID，以便服务器可以恢复而非重播所有内容。

这种重新连接的行为就是本指南中 CDP 级别区分在实践中重要的原因：原始的 fetch() 读取器必须手动重新实现重新连接和 Last-Event-ID 跟踪，而真正的 EventSource 对象则从浏览器免费获得这一功能——代价是失去对新连接打开时机的直接控制。当您希望浏览器管理重新连接时，请使用真实的 EventSource 读取目标的流量；当您需要自己控制连接生命周期（例如在达到一定数量的记录后取消）时，请使用 fetch() 及流读取器，如同上述两个示例所示。

从一个已打开的真实页面读取 SSE 连接

上述两种捕获方法都从一个空白页面打开连接，因为这可以保持目标简洁且公开。实时仪表盘、聊天用户界面或通知源以相同的方式打开自己的 EventSource 或流式 fetch()，正如相关组件一旦装载 — 此时 page.on("response") 和 CDP Network 域的事件一样。请在对真实目标（而不是 about:blank）调用 page.goto() 之前附加相同的处理程序，页面自己的脚本接收它们时，帧随之到达；捕获逻辑没有因为连接属于页面而非 page.evaluate() 调用而改变。改变的是发现 — 只需打开目标自己的网络面板一次，按 EventSource 或 Fetch/XHR 进行过滤，在围绕其编写处理程序之前确认端点及其 Content-Type，而不是提前猜测流 URL。

结论

SSE 连接是浏览器可以打开的三种实时传输中最简单的 — 一个 GET、一个开放的响应，没有握手 — 而这种简单正是使得只读取 DOM 或等待响应结束的工具从未见过它的原因。Playwright 的 page.on("response") 与内置流读取器以及原始 CDP Network.eventSourceMessageReceived 事件都读取相同的推送数据，一个通过手动解析器，另一个直接通过协议，两者在 Scrapeless Scraping Browser 的 CDP 连接上针对真实的公共维基百科编辑流表现一致。在关闭连接之前限制您捕获的记录数量，如果重新连接，请尊重流的 id: 字段携带的恢复游标，其余的脚本就是本指南已经介绍的一小部分 Playwright 调用。请查看 Scraping Browser 产品页面 上的当前会话和流出限制，并查看 定价页面 上的计划限制。关于本指南所基于的连接机制， Chrome DevTools 协议解释 详细介绍了 CDP 在网络域之外所暴露的内容。

加入我们的社区以申请免费计划并与其他开发者交流构建浏览器自动化的经验：Discord · Telegram。

常见问题

问：我需要 Selenium 或 WebDriver 才能以这种方式捕获 SSE 帧吗？
不。无损刮削浏览器只能通过Chrome开发工具协议（CDP）访问，因此任何支持CDP的客户端——在这里是Playwright或Puppeteer——都可以连接并读取流。没有WebDriver端点，因此Selenium无法驱动此连接。

问：为什么普通的 fetch() 读取器不会触发CDP的 eventSourceMessageReceived 事件？

Chrome的网络栈只将通过原生 EventSource 对象打开页面的连接分类为EventSource，并通过该专用事件报告。fetch()调用在传输级别以相同的方式传输字节，但Chrome不会将其解析或标记为SSE，因此读取它需要自己解析 text/event-stream 格式，如本指南中的响应流示例所示。

问：如果SSE连接中断，会自动重新连接吗？

只有在使用原生 EventSource 对象打开时。根据WHATWG规范，浏览器会等待一个实现定义的延迟，然后使用设置为最后一个 id: 值的 Last-Event-ID 头重新打开请求，因此一个良好行为的服务器可以恢复，而不是从头重播。基于fetch()的读取器不会自动获得这一点——重新连接和 Last-Event-ID 跟踪必须手动编写。

问：这与本系列中的网络请求拦截技术有什么不同？

拦截读取离散的请求/响应对——每当页面需要新数据时，它就会发起新的HTTP请求，您可以捕获每一个。SSE连接是一个单一请求，其响应体永远不会结束；没有东西可以重复拦截，因为服务器不断向它已经发送的一个响应写入。

问：Wikimedia的流中的 : comment 行或 canary 事件会发生什么？

在SSE格式中，以下划线 : 开头的行是注释，没有 data: 字段，由一些服务器发送以保持连接处于空闲状态；本指南中的捕获路径只查找 data: 行，因此会自动跳过注释行。Wikimedia还为自己的监控注入合成的 meta.domain: "canary" 记录——在将记录视为真实编辑之前，过滤掉这些。

问：在我找到的任何SSE端点上运行这个安全么？

只能针对您被允许读取的公共、未经身份验证的端点，并且只能在端点自身文档允许的数量范围内。这里的示例针对Wikimedia的文档公开编辑流，不需要密钥，并在五条记录后关闭连接，而不是无限期保持打开状态。

问：SSE连接能保持多长时间？

sessionTTL 查询参数以秒为单位限制浏览器会话。一个较短的值足以支持本指南中的限制性捕获；一个较长的值可以让会话——以及任何开放的流连接——保持更长时间的活动。

问：我可以在完全没有浏览器的情况下读取SSE流吗？

可以，对于像这样的公共未经身份验证的端点——一个支持分块读取的普通HTTP客户端可以直接解析相同的 text/event-stream 格式。当目标需要真实的Chromium指纹来建立连接，或者当页面作为其自身JavaScript的副作用打开流而不暴露文档中的独立端点时，本指南中的浏览器会话是值得的。