veretube-bot 0.1.6

Python bot library for veretube/sync v4.0 channels

veretube-bot

Python library for writing bots on sync 4.0 aka veretube channels.

Bots connect over socket.io for real-time events (chat, user list, playlist changes) and use a REST API for commands (queue media, kick/ban, manage emotes, read/write settings). Both surfaces are covered by this library.

Installation

pip install veretube-bot

Getting a token

Tokens are issued in the channel settings modal on the Bots tab. You need at least moderator rank to see it. Fill in a name and rank, click Issue Token, and copy the token immediately — it starts with cbt_ and is only shown once.

Quick start

from veretube_bot import Bot

bot = Bot(
    token="cbt_...",
    channel="mychannel",
    socket_url="http://your-server:1337",   # socket.io port (default 1337)
    api_url="http://your-server:8080/api/v1",  # HTTP port (default 8080)
)

@bot.on("chatMsg")
def on_chat(data):
    if data["msg"] == "!hello":
        bot.send_message("Hello!", to=data["username"])

    if data["msg"] == "!np":
        if bot.now_playing:
            bot.send_message(f"Now playing: {bot.now_playing['title']}")

@bot.on("changeMedia")
def on_media(data):
    print(f"Now playing: {data['title']}")

bot.run()  # connect and block until disconnected

Async quick start

import asyncio
from veretube_bot import AsyncBot

async def main():
    bot = AsyncBot(
        token="cbt_...",
        channel="mychannel",
        socket_url="http://your-server:1337",
        api_url="http://your-server:8080/api/v1",
        transports=["websocket"],  # optional
    )

    @bot.on("login")
    async def on_login(data):
        if data.get("success"):
            await bot.send_message("hello from async bot")

    await bot.connect(timeout=10)
    await asyncio.sleep(1)
    await bot.disconnect()

asyncio.run(main())

Connection

Bot(
    token,       # str  — bot token starting with 'cbt_'
    channel,     # str  — channel name (lowercase, no '#')
    socket_url,  # str  — socket.io server URL (uses io.port, not the HTTP port)
    api_url,     # str  — REST API base URL, e.g. http://host:8080/api/v1
    transports=None,         # e.g. ["websocket"] to bypass polling/upgrade issues
    reconnection=False,       # set True to reconnect automatically on drop
    reconnection_delay=3,     # seconds between reconnect attempts
)

Method	Description
bot.run()	Connect and block until disconnected
bot.connect()	Open the connection and wait until / namespace is ready
bot.wait()	Block until disconnected (call after connect())
bot.disconnect()	Close the connection

By default reconnection=False — the expectation is that an external process supervisor (systemd, supervisord, etc.) handles restarts. Pass reconnection=True for bots that should self-recover.

Event handling

Register handlers with @bot.on(event_name). Multiple handlers for the same event are all called. Exceptions in a handler are logged and skipped so one bad handler doesn't affect the others.

@bot.on("chatMsg")
def on_chat(data):
    print(data["username"], data["msg"])

@bot.on("userLeave")
def on_leave(data):
    print(f"{data['name']} left")

Socket events

Event	Payload	Description
connect	None	Socket connected
disconnect	reason string or None	Socket disconnected
connect_error	error object	Connection failed
login	{ success, name, guest }	Server accepted the token
chatMsg	{ username, msg, meta, time }	Chat message sent
pm	{ username, msg, meta }	Private message received
userlist	[{ name, rank, meta }, ...]	Full user list on join
addUser	{ name, rank, meta }	User joined
userLeave	{ name }	User left
setUserMeta	{ name, meta }	User AFK/muted state changed
setUserRank	{ name, rank }	User rank changed
changeMedia	{ id, type, title, seconds, ... }	New video started
playlist	[{ uid, media, queueby, temp }, ...]	Full playlist on join
queue	{ item, after }	Item added to playlist
delete	{ uid }	Playlist item removed
channelOpts	settings dict	Channel options updated
clearchat	None	Chat cleared by a moderator
errorMsg	{ msg }	Error from the server
kick	{ reason }	Bot was kicked
announcement	{ title, text }	Server-wide announcement
updateEmote	{ name, image }	Emote added or changed
removeEmote	{ name }	Emote removed

meta.is_bot is True in chatMsg and userlist payloads for bots.

In-memory state

These are updated automatically from socket events before your handlers are called:

bot.users        # list of { name, rank, meta } — current user list
bot.now_playing  # { id, type, title, seconds, ... } or None
bot.playlist     # list of playlist items
bot.channel_opts # channel options dict

Look up a specific user:

user = bot.get_user("Alice")  # returns dict or None

Chat

bot.send_message("Hello!")
bot.send_message("Hey!", to="Alice")  # sends "Alice: Hey!"
bot.send_action("waves")              # sends "/me waves"
bot.send_pm("Alice", "Hello privately")

Playlist

bot.queue("dQw4w9WgXcQ", "yt")          # add to end (rank >= MOD)
bot.queue("dQw4w9WgXcQ", "yt", "next")  # add as next up
bot.skip()                               # skip to next item (rank >= MOD)
bot.skip_to(uid)                         # jump to specific uid (rank >= MOD)
bot.delete_item(uid)                     # remove by uid (rank >= ADMIN)
bot.shuffle_playlist()                   # shuffle (rank >= ADMIN)
bot.clear_playlist()                     # clear all (rank >= ADMIN)

Media types: yt YouTube, sc SoundCloud, tw Twitch stream, tc Twitch clip, vm Vimeo, dm Dailymotion, fi direct file URL, cu custom embed.

Shows

Show endpoints manage scheduled playlist runs. They support bot Bearer auth and session auth.

shows = bot.list_shows()                 # rank >= MOD
public_shows = bot.list_public_shows()   # read-only public schedule
resolved = bot.resolve_show_media([      # rank >= MOD, max 50 entries
    {"type": "yt", "id": "dQw4w9WgXcQ"}
])
show = bot.get_show(show_id)             # rank >= MOD

payload = {
    "name": "Friday Prime",
    "notes": "Opening block",
    "color": "#337AB7",
    "scheduled_for": "2026-05-22T19:00:00.000Z",
    "estimated_end_at": "2026-05-22T21:00:00.000Z",
    "timezone": "America/New_York",
    "recurrence": "weekly",
    "fill_mode": "replace",
    "conflict_mode": "force",
    "start_playback": True,
    "playlist": [
        {"type": "yt", "id": "dQw4w9WgXcQ", "pos": "end"}
    ],
    "status": "scheduled",
}

created = bot.create_show(payload)       # rank >= MOD
updated = bot.update_show(show_id, payload)  # rank >= MOD
bot.delete_show(show_id)                 # rank >= ADMIN

bot.show_action(show_id, "pause")        # rank >= MOD
bot.show_action(show_id, "resume")       # rank >= MOD
bot.show_action(show_id, "schedule")     # rank >= MOD
bot.show_action(show_id, "run")          # rank >= ADMIN
bot.show_action(show_id, "cancel")       # rank >= ADMIN

Create/update payload constraints:

• name: required, 1-100 chars
• scheduled_for: required date string or unix timestamp (ms)
• estimated_end_at: optional date string/timestamp; must be later than scheduled_for when present
• timezone: required non-empty IANA time zone region string from the IANA Time Zone Database (for example Europe/Berlin, America/New_York)
• recurrence: none | daily | weekly
• fill_mode: append | replace
• conflict_mode: force | skip
• playlist: required non-empty array of media entries with type, id, and optional pos (next | end)
• status: draft | scheduled | paused | running | completed | failed | canceled (running is accepted and normalized to scheduled on write)
• notes: optional string | null, trimmed and capped to 20000 chars
• color: optional string | null, must match ^#[0-9A-Fa-f]{6}$ when provided

Action payload schema:

{ "action": "run" }

Moderation

bot.kick("BadUser")                    # rank >= MOD
bot.kick("BadUser", reason="Spamming")
bot.ban("BadUser")                     # rank >= ADMIN
bot.ban("BadUser", reason="Evading")
bot.unban("BadUser")                   # rank >= ADMIN
bot.set_rank("Alice", Rank.MOD)        # rank >= OWNER

Emotes

Emote endpoints read/write the database directly and work even when the channel is offline.

emotes = bot.get_emotes()             # list of { name, image, source }
bot.add_emote("KEKW", "https://...")  # rank >= OWNER
bot.update_emote("KEKW", image="https://...")
bot.update_emote("KEKW", new_name="KEKWait")
bot.delete_emote("KEKW")             # rank >= OWNER

Settings

settings = bot.get_settings()           # rank >= OWNER, channel must be active
bot.update_settings(pagetitle="Now Playing: Chill Beats")
bot.update_settings(allow_voteskip=False, afk_timeout=300)

Available setting keys: allow_voteskip, allow_dupes, voteskip_ratio, maxlength, playlist_max_duration_per_user, afk_timeout, enable_link_regex, chat_antiflood, chat_antiflood_burst, chat_antiflood_sustained, new_user_chat_delay, new_user_chat_link_delay, pagetitle, password, externalcss, externaljs, show_public, torbanned, block_anonymous_users, allow_ascii_control, playlist_max_per_user.

Direct REST access

Every REST endpoint is also accessible on bot.api if you need something not covered by the shortcuts:

playlist = bot.api.get_playlist()   # { items, currentIndex, locked }
bot.api.skip_to(uid)
bot.api.set_user_rank("Alice", 2)
bot.api.list_shows()
bot.api.show_action(show_id, "run")

Error handling

REST calls raise BotAPIError on failure:

from veretube_bot import Bot, BotAPIError

@bot.on("chatMsg")
def on_chat(data):
    if data["msg"].startswith("!add "):
        _, type, id = data["msg"].split(None, 2)
        try:
            bot.queue(id, type)
        except BotAPIError as e:
            bot.send_message(f"Error: {e}")  # e.status_code has the HTTP code

Rank constants

from veretube_bot import Rank

Rank.MOD      # 2
Rank.ADMIN    # 3
Rank.OWNER    # 4
Rank.CREATOR  # 5

A bot's effective rank is capped at the rank of the user who issued its token.