ストリーミング

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

接続する

ストリームは REST と同じ 3 通りでキーを受け付けます。サーバーからはヘッダーを使い、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 スコープがなければ黙って無視されます
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())

フレーム

3 つのフレームは X ではなく当社が送るものです:

フレームタイミング
system.hello最初のフレーム。targets はこの時点でこのキーが監視しているアカウント数です（後から追加したアカウントは再接続なしで加わります）。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 スコープがありません

正しく再接続する

素朴な版ではなく、これを一度書いておいてください。重要なのは指数バックオフと、プロセスの再起動をまたいで残る場所に sid を永続化することの 2 点です。

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)