流式传输

wss wss://ws.xsignalfeed.com/v1/stream

连接

实时流接收密钥的三种方式与 REST 完全相同。在服务器上请用请求头；只有在无法设置请求头的地方才用 URL 参数。如果密钥没有送达或不被接受，连接会先建立、随即以 4001 关闭，关闭原因会写明是哪种情况。

密钥的传法适用场景
Authorization: Bearer xsf_…服务器端推荐使用；与 REST 用的是同一个请求头
X-API-Key: xsf_…与上一种等价
?api_key=xsf_…浏览器，以及无法设置握手请求头的客户端。密钥会成为 URL 的一部分，注意不要让这类 URL 进入日志
python — 用请求头传密钥
import websockets

async with websockets.connect("wss://ws.xsignalfeed.com/v1/stream",
                              additional_headers={"Authorization": "Bearer xsf_your_key"}) as ws:
    print(await ws.recv())    # {"t": "system.hello", ...}
查询参数含义
api_key无法设置头部时用它传密钥
last_id从此 sid 之后续传。除非密钥有 replay scope，否则静默忽略
import asyncio, json, websockets

async def main():
    url = "wss://ws.xsignalfeed.com/v1/stream?api_key=xsf_your_key"
    async with websockets.connect(url) as ws:
        async for raw in ws:
            ev = json.loads(raw)
            if ev["t"].startswith("system."): continue
            print(ev["t"], ev["target"]["screen_name"], ev["d"].get("text", "")[:60])

asyncio.run(main())

帧

有三种帧是我们的，而不是 X 的：

帧何时
system.hello第一帧。targets 是此刻这个 Key 监控的账号数（之后新增的账号无需重连即会加入）；replay_from 回显你重连时带的 last_id，未带时为 null，此时不会重放任何事件
system.ping静默期间作为保活发送，以免空闲的代理关闭连接
system.error连接出了问题；随后会关闭

关闭码

握手总会成功，紧接着才关闭，这样你的客户端库看到的是关闭码和原因，而不是一个裸的 HTTP 错误。

关闭码含义
1012服务重启，发生在维护期间。请按退避策略重连，并把你处理过的最后一个 sid 作为 ?last_id= 传入：断开期间的事件会被重放，不会丢失
4001密钥未被接受，关闭原因会写明是哪种：missing_api_key（我们没收到任何密钥：检查请求头名称，或者你的代理、客户端库是否把它转发了），invalid_api_key（没有这个密钥：拼写错误，或者密钥已被轮换替换），key_revoked（密钥已被吊销）
4002此密钥的并发连接过多；关闭前会先发一帧 system.error
4003密钥缺少 stream scope

正确地重连

把这个一次写好，别写那个天真的版本。两件要紧事：指数退避，以及把 sid 持久化到能挺过进程重启的地方。

python——生产环境的写法
import asyncio, json, pathlib, random, websockets

STATE = pathlib.Path("xsf.cursor")          # survives restarts; a DB row is better still

async def run(key, handle):
    backoff = 1
    while True:
        last = STATE.read_text().strip() if STATE.exists() else None
        url = f"wss://ws.xsignalfeed.com/v1/stream?api_key={key}" + (f"&last_id={last}" if last else "")
        try:
            async with websockets.connect(url, ping_interval=20, max_size=2**22) as ws:
                backoff = 1
                async for raw in ws:
                    ev = json.loads(raw)
                    if ev["t"].startswith("system."): continue
                    handle(ev)                      # must be idempotent: dedupe on ev["id"]
                    if ev.get("sid"): STATE.write_text(ev["sid"])   # only after handling
        except Exception as e:
            print("disconnected:", e)
        await asyncio.sleep(backoff + random.random())   # jitter, or every client returns at once
        backoff = min(60, backoff * 2)