WebSocket stream: connection rules and close codes

wss://ws.twitterapi.io/twitter/tweet/websocket
Header: x-api-key: YOUR_API_KEY

{
  "event_type": "error",
  "error_code": 4029,
  "message": "Reconnecting too fast. Wait retry_after_seconds before connecting again.",
  "retry_after_seconds": 5
}

import asyncio, json
import websockets  # pip install websockets

URL = "wss://ws.twitterapi.io/twitter/tweet/websocket"
API_KEY = "YOUR_API_KEY"
STOP_CODES = {4001, 4003, 4009}

async def run():
    delay = 5
    while True:
        try:
            async with websockets.connect(URL, additional_headers={"x-api-key": API_KEY}) as ws:
                delay = 5
                async for raw in ws:
                    msg = json.loads(raw)
                    if msg.get("event_type") in ("connected", "ping"):
                        continue
                    if msg.get("event_type") == "replaced":
                        print("Replaced by a newer connection; stopping.")
                        continue
                    print(msg)  # handle tweet events here
        except websockets.exceptions.InvalidStatus as e:  # refused before opening
            try:
                body = json.loads(e.response.body or b"{}")
            except ValueError:  # not JSON, e.g. a proxy error page
                body = {}
            if body.get("error_code") in STOP_CODES:
                print("Stopping:", body.get("message"))
                return
            delay = int(e.response.headers.get("Retry-After", delay))
        except websockets.exceptions.ConnectionClosed as e:
            if e.rcvd and e.rcvd.code in STOP_CODES:
                print("Stopping: closed with", e.rcvd.code)
                return
        except OSError:
            pass  # network error
        await asyncio.sleep(delay)
        delay = min(delay * 2, 60)

asyncio.run(run())