YouTube WebSocket API quick start for new uploads

# pip install "websockets>=13"   (Python 3.9+)
import asyncio, json, os
from websockets.asyncio.client import connect

# ws_url and ws_key from GET /v1/dashboard (or the API access page)
WS_URL = os.environ["YT_WS_URL"] + "?key=" + os.environ["YT_WS_KEY"]

def handle(msg):
    kind = msg.get("type")
    if kind == "upload":
        video = msg.get("video") or {}
        channel = msg.get("channel") or {}
        print(msg.get("subtype"), channel.get("name"), video.get("title"), video.get("url"))
    elif kind == "upgrade":
        upgrade = msg.get("upgrade") or {}
        print("better thumbnail for", upgrade.get("video_id"), upgrade.get("url"))
    elif kind == "deletion":
        print("deleted", (msg.get("video") or {}).get("id"))

async def main():
    backoff = 1
    while True:
        try:
            async with connect(WS_URL) as ws:
                backoff = 1
                async for raw in ws:
                    try:
                        event = json.loads(raw)
                    except ValueError:
                        continue
                    if isinstance(event, dict):
                        handle(event)
        except Exception as exc:
            print("disconnected:", exc)
        await asyncio.sleep(backoff)
        backoff = min(backoff * 2, 30)

asyncio.run(main())

// npm install ws, save as consumer.mjs, run: node consumer.mjs   (Node 18+)
import WebSocket from "ws"

// ws_url and ws_key from GET /v1/dashboard (or the API access page)
const BASE_URL = process.env.YT_WS_URL
const WS_KEY = process.env.YT_WS_KEY
if (!BASE_URL || !WS_KEY) throw new Error("Set YT_WS_URL and YT_WS_KEY")
const WS_URL = BASE_URL + "?key=" + encodeURIComponent(WS_KEY)

function handle(msg) {
  if (msg.type === "upload") {
    console.log(msg.subtype, msg.channel?.name, msg.video?.title, msg.video?.url)
  } else if (msg.type === "upgrade") {
    console.log("better thumbnail for", msg.upgrade?.video_id, msg.upgrade?.url)
  } else if (msg.type === "deletion") {
    console.log("deleted", msg.video?.id)
  }
}

let backoff = 1000

function connect() {
  const ws = new WebSocket(WS_URL)
  ws.on("open", () => { backoff = 1000 })
  ws.on("message", (raw) => {
    let event
    try { event = JSON.parse(raw.toString()) } catch { return }
    if (event && typeof event === "object" && !Array.isArray(event)) handle(event)
  })
  ws.on("error", (err) => console.error("socket error:", err.message))
  ws.on("close", () => {
    setTimeout(connect, backoff)
    backoff = Math.min(backoff * 2, 30000)
  })
}

connect()

{
  "type": "upload",
  "subtype": "video",
  "detected_at": "2026-10-15T14:03:22Z",
  "channel": {
    "id": "UCexampleChannel0000001a",
    "name": "Example Studio",
    "url": "https://www.youtube.com/channel/UCexampleChannel0000001a",
    "avatar": null
  },
  "video": {
    "id": "aBcDeF12345",
    "url": "https://www.youtube.com/watch?v=aBcDeF12345",
    "title": "Studio tour, part 2",
    "description": null,
    "published_at": "2026-10-15T14:03:00Z",
    "updated_at": null
  },
  "images": {
    "seed": null,
    "chosen": "https://example.com/thumbs/aBcDeF12345.jpg",
    "thumbnail_small": null
  }
}

curl "https://youtube.1322.io/v1/dashboard" \
  -H "X-Api-Key: $YT_API_KEY"

curl "https://youtube.1322.io/v1/list" \
  -H "X-Api-Key: $YT_API_KEY"

curl -X POST "https://youtube.1322.io/v1/track" \
  -H "Content-Type: application/json" \
  -H "X-Api-Key: $YT_API_KEY" \
  -d '{ "channel": "@examplestudio" }'

curl -X POST "https://youtube.1322.io/v1/untrack" \
  -H "Content-Type: application/json" \
  -H "X-Api-Key: $YT_API_KEY" \
  -d '{ "channel": "UCexampleChannel0000001a" }'

curl "https://youtube.1322.io/v1/limits" \
  -H "X-Api-Key: $YT_API_KEY"