Build a Discord Music Bot with Python

What you will build

A Discord music bot connects a text command to a voice player. In this tutorial, Python receives commands through discord.py, yt-dlp resolves a YouTube audio source, and FFmpeg converts that source into the PCM audio consumed by Discord. Each server has its own queue and playback controls.

You will build a small, trusted-server bot with !play, pause, resume, skip, queue management and buttons. You will also handle the less visible work: stopping a request while extraction is still running, moving between voice channels, rejecting commands in other text channels and recovering from a failed track.

This edition comes from Nicode-Music, a Nicode Studio project whose owner confirmed audio playback in the original installation. The tutorial has its own clean entry point and configuration. It does not include the original server IDs, private photographs, token or machine-specific launcher.

The documented setup uses Windows and PowerShell. You should be comfortable creating files and running terminal commands; previous Discord bot experience is not required. Keep the article and download together: every application file below is included verbatim in the package.

Download the complete project and this guide. You can also build it by copying the named file blocks below. The tests exercise the player without logging in to Discord. A real server check is a separate final step.

1. Prepare Python, FFmpeg and Deno

Create a folder named discord-music-bot. Use Python 3.14 on 64-bit Windows for the installation path validated here. Install Python from python.org, then open PowerShell in your folder.

python --version
python -m venv .venv

The first command should identify Python 3.14. The second creates a private package environment inside .venv; it does not start a bot. If Windows opens the Store or cannot find Python, correct your Python installation or use its full executable path before continuing.

Install FFmpeg using a Windows build linked from FFmpeg's download page. Extract it and locate bin\ffmpeg.exe. Installing a similarly named Python package does not provide this executable.

Install Deno from its official installation instructions. yt-dlp's current EJS setup guide requires an external JavaScript runtime for YouTube challenges. This project uses Deno, with a minimum supported version of 2.3.0, and installs the matching EJS scripts through yt-dlp[default].

Either put both executables on PATH or set their full paths for the current terminal. Replace these two example paths with the files you actually extracted:

$env:FFMPEG_PATH = "C:\Tools\ffmpeg\bin\ffmpeg.exe"
$env:DENO_PATH = "C:\Tools\deno\deno.exe"
& $env:FFMPEG_PATH -version
& $env:DENO_PATH --version

Each command must print a version. These environment assignments last only for this terminal session. You need to set them again in a new terminal unless you deliberately configure PATH. No Windows task or background service is required.

Create the two dependency files below. requirements.txt records the direct dependencies; requirements-lock.txt records the resolved versions used for this Windows test. Install the lock file for the matching starting point.

Create requirements.txt in the project folder with this complete content:

discord.py[voice]==2.7.1
yt-dlp[default]==2026.8.19

Create requirements-lock.txt in the project folder with this complete content:

aiohappyeyeballs==2.7.1
aiohttp==3.14.5
aiosignal==1.4.0
attrs==26.1.0
audioop-lts==0.2.2
brotli==1.2.0
certifi==2026.10.19
cffi==2.1.1
charset-normalizer==3.5.2
davey==0.1.7
discord.py==2.7.1
frozenlist==1.8.0
idna==3.20
multidict==7.1.0
mutagen==1.48.1
propcache==0.5.4
pycparser==3.11
pycryptodomex==3.24.1
PyNaCl==1.5.0
requests==2.34.2
urllib3==2.8.0
websockets==17.3
yarl==1.25.1
yt-dlp==2026.8.19
yt-dlp-ejs==0.8.0

.venv\Scripts\python.exe -m pip install -r requirements-lock.txt
.venv\Scripts\python.exe -m pip check

The second command should report no broken requirements. The package list includes voice support and the EJS companion package. FFmpeg and Deno remain separate executables. Versions are a reproducible starting point, not a claim that these packages will remain current: YouTube changes can require an extractor update and fresh verification.

2. Create your Discord application and channels

In the Discord Developer Portal, create an application and use its bot settings to obtain a bot token. Keep that token private. The supplied program asks for it in a hidden terminal prompt and does not write it to disk.

Enable Message Content Intent for the bot. This tutorial uses prefix commands such as !play, so the application needs that intent enabled both in the portal and in Python. See the discord.py intents guide.

Use a guild installation link for your application with the bot scope and invite it to a server you manage. Discord's application setup guide explains its installation settings. This bot uses Gateway events rather than an HTTP interactions endpoint; you do not need a public web server.

Create a text channel called music-commands and a normal voice channel called Music. Give the bot View Channel, Send Messages and Read Message History in the text channel, and View Channel, Connect and Speak in the voice channel. Check category overrides as well. Administrator is unnecessary.

Restrict the text channel to people you trust to control music. This edition does not implement DJ roles or voting: anyone able to use its configured text channel can operate its commands and buttons. Voice moves additionally check that the requesting member can access the destination.

The bot connects to the configured voice channel on startup. For the first run, use a test server or a quiet channel where you have permission to run this check.

3. Configure your own server

Create this example file and copy it to voice_channels.json. Edit the copy, leaving the example reusable:

Create voice_channels.example.json in the project folder with this complete content:

{
  "channels": [
    {
      "guild_name": "Your server",
      "guild_id": "",
      "channel_name": "Music",
      "channel_id": "",
      "command_channel_name": "music-commands"
    }
  ]
}

Copy-Item voice_channels.example.json voice_channels.json

Replace Your server with your exact server name. Keep Music and music-commands only if those are the channel names you created. Names must identify a unique match. Duplicate text-channel names are rejected.

For stable server and voice-channel selection, enable Discord Developer Mode and copy their IDs into the corresponding string fields. IDs are optional on first use: the auto-connect module saves the resolved server and voice-channel IDs into your local configuration. Text-channel selection still uses its exact name, so update the configuration if you rename it.

Add another object to channels to use the same bot in another authorised server. Each server gets a separate player. Do not add the same server twice: ambiguous configuration deliberately prevents command access.

Create .gitignore before using Git. It keeps local configuration and runtime state out of the repository; it cannot remove a secret that was already committed.

Create .gitignore in the project folder with this complete content:

.venv/
__pycache__/
voice_channels.json
music_welcome.json
*.tmp
.env
*.log
tools/

The finished project has the Python files below, the configuration example, both dependency files and four test files. Keep them in one folder. The program resolves local paths relative to those files, rather than assuming a particular Windows username or drive.

4. Connect commands to audio

Start with bot.py. Reading it from top to bottom shows the boundary of each responsibility: validate configuration, find dependencies, validate requests, construct an audio source, then register Discord events and commands.

normalize_query() accepts HTTPS YouTube URLs or converts search words into a single-result YouTube search. It rejects arbitrary URL hosts; the bot is not intended as a general URL-fetching service. This is an input restriction, not a process sandbox.

get_audio_source() performs blocking network extraction and starts FFmpeg. The queue module calls it in a worker thread with asyncio.to_thread(), keeping the event loop available for commands. The extracted media URL is temporary: the queue stores the user's request and resolves it just before playback.

FFmpegPCMAudio supplies audio frames to the voice client. The example uses the interface documented in the discord.py API reference. It does not download a library of tracks or keep a copy of the audio.

The final if __name__ == '__main__' guard matters: importing the module in a test registers objects but does not prompt for a token or connect to Discord.

Create bot.py in the project folder with this complete content:

"""Nicode Studio music tutorial. Run with Python; never put a token in this file."""
import asyncio
import getpass
import json
import os
from pathlib import Path
import shutil
from urllib.parse import urlsplit

import discord
from discord.ext import commands
import yt_dlp
import voice_autoconnect
from voice_move import change
from music import install

ROOT = Path(__file__).resolve().parent

def load_settings():
    path = voice_autoconnect.CONFIG
    if not path.is_file():
        raise ValueError('Copy voice_channels.example.json to voice_channels.json and edit it.')
    data = json.loads(path.read_text(encoding='utf-8-sig'))
    channels = data.get('channels')
    if not isinstance(channels, list) or not channels:
        raise ValueError('channels must be a nonempty list.')
    for entry in channels:
        for key in ('guild_name', 'channel_name', 'command_channel_name'):
            if not isinstance(entry.get(key), str) or not entry[key].strip():
                raise ValueError('Missing configuration field: ' + key)
        for key in ('guild_id', 'channel_id'):
            if entry.get(key) and not str(entry[key]).isdigit():
                raise ValueError(key + ' must be an ID copied from Discord, or empty.')
    return data

def configured_target(guild):
    if guild is None:
        return None
    matches = [
        t for t in load_settings()['channels']
        if (str(guild.id) == str(t['guild_id']) if t.get('guild_id')
            else guild.name == t['guild_name'])
    ]
    return matches[0] if len(matches) == 1 else None

def binary(name):
    configured = os.environ.get(name.upper() + '_PATH')
    found = configured or shutil.which(name)
    if not found or not Path(found).is_file():
        raise ValueError(name + ' not found. Add it to PATH or set ' + name.upper() + '_PATH.')
    return str(Path(found).resolve())

def normalize_query(query):
    query = query.strip()
    if not query or len(query) > 500:
        raise ValueError('Use a YouTube URL or a search of 1–500 characters.')
    if '://' in query or query.startswith('//'):
        parsed = urlsplit(query)
        if (parsed.scheme != 'https' or parsed.username or parsed.password
                or parsed.port not in (None, 443)
                or parsed.hostname not in ('youtube.com', 'www.youtube.com', 'm.youtube.com', 'youtu.be')):
            raise ValueError('Only HTTPS YouTube links are accepted.')
        return query
    return 'ytsearch1:' + query

class QuietLogger:
    # Extractor diagnostics can contain signed media URLs. Keep the public log minimal.
    def debug(self, message): pass
    def warning(self, message): pass
    def error(self, message): pass

def get_audio_source(query):
    ffmpeg, deno = binary('ffmpeg'), binary('deno')
    options = {
        'format': 'bestaudio/best',
        'noplaylist': True,
        'quiet': True,
        'logger': QuietLogger(),
        'socket_timeout': 20,
        'retries': 2,
        'extractor_retries': 2,
        'ffmpeg_location': ffmpeg,
        'js_runtimes': {'deno': {'path': deno}},
    }
    with yt_dlp.YoutubeDL(options) as extractor:
        info = extractor.extract_info(normalize_query(query), download=False)
    if info and 'entries' in info:
        info = next((entry for entry in info['entries'] if entry), None)
    if not info or not info.get('url'):
        raise ValueError('No playable audio found.')
    return discord.FFmpegPCMAudio(
        info['url'], executable=ffmpeg,
        before_options='-nostdin -reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5',
        options='-vn',
    )

intents = discord.Intents.default()
intents.message_content = True
bot = commands.Bot(command_prefix='!', intents=intents,
                   allowed_mentions=discord.AllowedMentions.none())

@bot.check
async def command_channel_only(ctx):
    target = configured_target(ctx.guild)
    if not target:
        return False
    matches = [c for c in ctx.guild.text_channels
               if c.name == target['command_channel_name']]
    return len(matches) == 1 and matches[0].id == ctx.channel.id

@bot.command(name='join', aliases=['canal', 'move'])
async def join(ctx, *, destination: str = None):
    await change(ctx, destination, configured_target(ctx.guild))

@bot.event
async def on_ready():
    print('Connected. Use !help in the configured text channel.', flush=True)
    await voice_autoconnect.auto_connect(bot)

@bot.event
async def on_command_error(ctx, error):
    if isinstance(error, (commands.CommandNotFound, commands.CheckFailure)):
        return
    if isinstance(error, commands.MissingRequiredArgument):
        await ctx.send('Use !play followed by a YouTube link or search words.')
        return
    await ctx.send('Command failed: ' + type(getattr(error, 'original', error)).__name__)

music = install(bot, configured_target, get_audio_source, join, ROOT)

def main():
    load_settings()
    binary('ffmpeg')
    binary('deno')
    token = os.environ.get('DISCORD_TOKEN') or getpass.getpass('Discord bot token (hidden): ')
    if not token.strip():
        raise ValueError('A bot token is required.')
    bot.run(token.strip())

if __name__ == '__main__':
    main()

5. Join and move between voice channels

voice_autoconnect.py handles startup. Its lock prevents overlapping ready events from creating competing connection attempts. Successfully handled servers are remembered for the lifetime of the process. On a normal reconnect, discord.py manages the existing voice connection; after a permanent loss, use !join and !resume.

The configuration update uses a temporary file and replacement so it does not write a partially serialized JSON document in the ordinary successful path. This is local persistence of channel selection, not persistence of music queues.

Create voice_autoconnect.py in the project folder with this complete content:

import asyncio
import json
from pathlib import Path

CONFIG = Path(__file__).with_name('voice_channels.json')
_lock = asyncio.Lock()
_completed = set()

async def auto_connect(bot):
    async with _lock:
        settings = json.loads(CONFIG.read_text(encoding='utf-8-sig'))
        for target in settings['channels']:
            guilds = [g for g in bot.guilds if (str(g.id) == str(target['guild_id']) if target.get('guild_id') else g.name == target['guild_name'])]
            if len(guilds) != 1:
                print('Auto-connect: missing or ambiguous server: ' + target['guild_name'], flush=True)
                continue
            guild = guilds[0]
            if guild.id in _completed:
                continue
            channels = [c for c in guild.voice_channels if (str(c.id) == str(target['channel_id']) if target.get('channel_id') else c.name == target['channel_name'])]
            if len(channels) != 1:
                print('Auto-connect: missing or ambiguous voice channel: ' + target['channel_name'], flush=True)
                continue
            channel = channels[0]
            permissions = channel.permissions_for(guild.me)
            if not (permissions.view_channel and permissions.connect and permissions.speak):
                print('Auto-connect: missing View Channel, Connect or Speak in ' + channel.name, flush=True)
                continue
            # Persist stable IDs before connecting; future name changes remain safe.
            target.update(guild_id=str(guild.id), channel_id=str(channel.id))
            temporary = CONFIG.with_suffix('.tmp')
            temporary.write_text(json.dumps(settings, ensure_ascii=False, indent=2), encoding='utf-8')
            temporary.replace(CONFIG)
            if guild.voice_client:
                _completed.add(guild.id)
                continue
            try:
                await channel.connect(timeout=30.0, reconnect=True)
            except Exception as exc:
                print('Auto-connect failed in ' + guild.name + ': ' + type(exc).__name__, flush=True)
            else:
                _completed.add(guild.id)
                print('Voice connected: ' + guild.name + ' / ' + channel.name, flush=True)

voice_move.py resolves a name, ID or channel mention only within the requesting server. It checks permissions for the member and the bot before joining or moving. A lock serializes simultaneous move requests in that server.

If Discord times out while confirming a move, the operation may still be underway. The response tells you to inspect the actual location instead of assuming that retrying is harmless. Moving does not construct a new player, so it preserves the existing queue and paused state.

Create voice_move.py in the project folder with this complete content:

"""Move within the requesting guild without replacing its player or queue."""
import asyncio
import re
import discord

_locks = {}

def resolve(guild, destination, target):
    if destination:
        value = destination.strip().strip('"')
        match = re.fullmatch(r'<#(\d+)>|(\d+)', value)
        if match:
            channel_id = int(match.group(1) or match.group(2))
            matches = [c for c in guild.voice_channels if c.id == channel_id]
        else:
            matches = [c for c in guild.voice_channels if c.name.casefold() == value.casefold()]
    else:
        matches = [c for c in guild.voice_channels if
                   (str(c.id) == str(target['channel_id']) if target.get('channel_id')
                    else c.name == target['channel_name'])]
    if len(matches) != 1:
        raise ValueError('No unique voice channel found in this server. Use !move with its exact name, mention or ID.')
    return matches[0]

async def change(ctx, destination, target):
    if not target:
        return
    async with _locks.setdefault(ctx.guild.id, asyncio.Lock()):
        try:
            channel = resolve(ctx.guild, destination, target)
        except ValueError as exc:
            await ctx.send(str(exc))
            return
        member_permissions = channel.permissions_for(ctx.author)
        if not (member_permissions.view_channel and member_permissions.connect):
            await ctx.send('You cannot access that voice channel.')
            return
        permissions = channel.permissions_for(ctx.guild.me)
        if not (permissions.view_channel and permissions.connect and permissions.speak):
            await ctx.send('The bot needs View Channel, Connect and Speak permissions there.')
            return
        try:
            vc = ctx.voice_client
            if vc:
                if not vc.is_connected():
                    await ctx.send('Voice is reconnecting. Wait a few seconds and try again.')
                    return
                if vc.channel.id != channel.id:
                    await vc.move_to(channel, timeout=30.0)
            else:
                await channel.connect(timeout=30.0, reconnect=True)
        except asyncio.TimeoutError:
            await ctx.send('Discord did not confirm the move in time; it may still be in progress. Check the bot location before retrying.')
            return
        except Exception as exc:
            await ctx.send('Could not change voice channel (' + type(exc).__name__ + ').')
            return
        await ctx.send('Voice channel: ' + channel.name + '. The queue is retained.', allowed_mentions=discord.AllowedMentions.none())

6. Add a queue and persistent buttons

The core is Player. Each instance owns one deque of pending requests, one current request and one asynchronous playback task. The Music manager stores players by guild ID. A busy server therefore does not consume another server's queue, although both still share the same machine and process.

There are three different ways to stop work. !clear removes pending items but lets the current track finish. !stop stops the current track and retains pending items; resume starts the next queued item rather than seeking back into the stopped track. !reset stops and clears everything.

Stopping during extraction is more subtle. A background thread cannot simply be cancelled by changing the player's task state. The player increments a generation counter; when extraction returns, it checks the counter and cleans up a stale source instead of playing it. This prevents an old request from starting after a reset. The extraction thread can still take time to finish, so this is not a hard execution deadline.

Discord invokes its playback completion callback from an audio thread. The callback uses loop.call_soon_threadsafe() to notify the event loop, which then advances the queue. A failed track produces a short message and allows the next item to run.

Buttons have stable custom IDs and are registered again after startup. Their callbacks operate on the current player's state. A button on an old message does not refer to the track that was originally shown there. Every click checks the configured text channel again.

The welcome message ID is saved locally so startup can edit the existing message. If you delete music_welcome.json, the bot no longer knows which message to reuse and can send a new welcome. This state file contains no token, but it belongs to your installation.

Create music.py in the project folder with this complete content:

"""Queue and Discord controls. No credentials, one player per guild."""
import asyncio
import json
from collections import deque
from pathlib import Path
import discord

HELP = ('Welcome. Use **!play YouTube link or search words** to add music.\n'
        '**!skip** next · **!pause** pause · **!resume** resume\n'
        '**!stop** stop and keep queued tracks · **!queue** show queue\n'
        '**!clear** clear queue, keep current track · **!reset** stop and clear queue\n'
        '**!canal voice channel name** move without clearing the queue.\n'
        '**!join** return to the configured channel; restarting uses that channel.\n'
        '**!help** show this help. Buttons control the current player.\n'
        'The queue lives only while the bot is running.')

class Player:
    def __init__(self, guild, channel, source_factory):
        self.guild, self.channel, self.source_factory = guild, channel, source_factory
        self.queue = deque()
        self.current = None
        self.task = None
        self.generation = 0
        self.halted = False
        self.done = None

    def kick(self):
        if not self.halted and (self.task is None or self.task.done()):
            self.task = asyncio.create_task(self.run())

    async def say(self, text, **kwargs):
        try:
            await self.channel.send(text, allowed_mentions=discord.AllowedMentions.none(), **kwargs)
        except discord.HTTPException as exc:
            print('Could not send controls: ' + type(exc).__name__, flush=True)

    def interrupt(self):
        self.generation += 1
        vc = self.guild.voice_client
        if vc:
            vc.stop()
        if self.done:
            self.done.set()

    async def run(self):
        while self.queue and not self.halted:
            vc = self.guild.voice_client
            if not vc or not vc.is_connected():
                await self.say('No voice connection. Use !join and !resume; queued tracks are retained.')
                self.halted = True
                return
            query = self.queue.popleft()
            self.current = query
            generation = self.generation
            source = None
            try:
                source = await asyncio.to_thread(self.source_factory, query)
                if generation != self.generation or self.halted:
                    source.cleanup()
                    continue
                if not vc.is_connected():
                    source.cleanup()
                    self.queue.appendleft(query)
                    self.halted = True
                    await self.say('Voice connection lost. Use !join and !resume.')
                    return
                done = self.done = asyncio.Event()
                loop = asyncio.get_running_loop()
                errors = []
                def after(error, event=done):
                    if error:
                        errors.append(type(error).__name__)
                    loop.call_soon_threadsafe(event.set)
                vc.play(source, after=after)
                source = None  # Discord now owns and cleans it.
                await self.say('▶ Playing: ' + discord.utils.escape_markdown(query[:400])
                               + '\n!skip · !pause · !resume · !stop · !queue · !clear · !reset',
                               view=Controls(self.manager))
                await done.wait()
                if errors:
                    await self.say('Audio interrupted (' + errors[0] + '); continuing with the next track.')
            except Exception as exc:
                if source is not None:
                    source.cleanup()
                await self.say('Could not play this track (' + type(exc).__name__ + '). Continuing with the next track.')
            finally:
                self.current = None
                self.done = None

    def action(self, action):
        vc = self.guild.voice_client
        if action == 'queue':
            lines = ['Current: ' + (self.current or 'none')]
            lines += [str(i+1) + '. ' + q[:100] for i, q in enumerate(list(self.queue)[:10])]
            lines.append('Queued: ' + str(len(self.queue)))
            if self.halted:
                lines.append('Stopped. Use !resume to continue.')
            return discord.utils.escape_markdown('\n'.join(lines)[:1800])
        if action == 'clear':
            self.queue.clear()
            return 'Queue cleared. Current track continues.'
        if action in ('stop', 'reset'):
            self.halted = True
            if action == 'reset':
                self.queue.clear()
            self.interrupt()
            return 'Stopped and queue cleared.' if action == 'reset' else 'Stopped. Queued tracks retained; use !resume.'
        if action == 'skip':
            if not self.current:
                return 'No current track. Use !resume to play queued tracks.'
            self.interrupt()
            return 'Skipping to the next queued track.'
        if action == 'pause':
            if vc and vc.is_playing():
                vc.pause()
                return 'Paused. Use Resume or !resume.'
            return 'No playing audio to pause.'
        if action == 'resume':
            if vc and vc.is_paused():
                vc.resume()
                return 'Playback resumed.'
            self.halted = False
            self.kick()
            return 'Continuing the queue.' if self.queue or self.current else 'Queue empty. Add a track with !play.'

class Controls(discord.ui.View):
    def __init__(self, manager):
        super().__init__(timeout=None)
        self.manager = manager
        for index, (action, label) in enumerate([
            ('skip', 'Skip'), ('pause', 'Pause'), ('resume', 'Resume'),
            ('stop', 'Stop'), ('queue', 'Queue'), ('clear', 'Clear queue'),
            ('reset', 'Stop and clear'), ('help', 'Help')]):
            button = discord.ui.Button(label=label, custom_id='nicode:music:' + action,
                                       row=index // 4, style=discord.ButtonStyle.secondary)
            async def callback(interaction, action=action):
                if not manager.allowed(interaction.guild, interaction.channel):
                    await interaction.response.send_message('Use the configured music text channel.', ephemeral=True)
                    return
                await interaction.response.defer(ephemeral=True)
                result = HELP if action == 'help' else manager.player(interaction.guild, interaction.channel).action(action)
                await interaction.followup.send(result, ephemeral=True, allowed_mentions=discord.AllowedMentions.none())
            button.callback = callback
            self.add_item(button)

class Music:
    def __init__(self, bot, target, source_factory, root):
        self.bot, self.target, self.source_factory = bot, target, source_factory
        self.players = {}
        self.path = Path(root) / 'music_welcome.json'
        self.ready_lock = asyncio.Lock()
        self.registered = False
        self.welcomed = set()

    def text_channel(self, guild):
        target = self.target(guild)
        channels = [c for c in guild.text_channels if c.name == target.get('command_channel_name')] if target else []
        return channels[0] if len(channels) == 1 else None

    def allowed(self, guild, channel):
        configured = self.text_channel(guild) if guild else None
        return configured is not None and channel.id == configured.id

    def player(self, guild, channel):
        if guild.id not in self.players:
            player = self.players[guild.id] = Player(guild, channel, self.source_factory)
            player.manager = self
        return self.players[guild.id]

    async def ready(self):
        async with self.ready_lock:
            from presentation import apply_avatar, welcome_image
            try:
                await apply_avatar(self.bot, self.path.parent)
            except (discord.HTTPException, OSError, ValueError, RuntimeError) as exc:
                print('Avatar pending: ' + type(exc).__name__, flush=True)
            if not self.registered:
                self.bot.add_view(Controls(self))
                self.registered = True
            try:
                saved = json.loads(self.path.read_text(encoding='utf-8')) if self.path.exists() else {}
            except (OSError, ValueError):
                print('Unreadable welcome state; no duplicate welcome sent.', flush=True)
                return
            for guild in self.bot.guilds:
                channel = self.text_channel(guild)
                if channel is None or channel.id in self.welcomed:
                    continue
                try:
                    message = None
                    if str(channel.id) in saved:
                        try:
                            message = await channel.fetch_message(saved[str(channel.id)])
                        except discord.NotFound:
                            pass
                    content = HELP + '\nInitial voice channel: **' + self.target(guild)['channel_name'] + '**.'
                    embed, image = welcome_image(self.path.parent)
                    try:
                        extra = {'embed': embed} if embed else {}
                        if message:
                            if image:
                                extra['attachments'] = [image]
                            await message.edit(content=content, view=Controls(self), allowed_mentions=discord.AllowedMentions.none(), **extra)
                        else:
                            if image:
                                extra['file'] = image
                            message = await channel.send(content, view=Controls(self), allowed_mentions=discord.AllowedMentions.none(), **extra)
                    finally:
                        if image:
                            image.close()
                    self.welcomed.add(channel.id)
                    saved[str(channel.id)] = message.id
                    temporary = self.path.with_suffix('.tmp')
                    temporary.write_text(json.dumps(saved), encoding='utf-8')
                    temporary.replace(self.path)
                except (discord.HTTPException, OSError) as exc:
                    print('Welcome pending: ' + type(exc).__name__, flush=True)

    async def enqueue(self, ctx, query, join):
        player = self.player(ctx.guild, ctx.channel)
        if len(player.queue) >= 100:
            await ctx.send('Queue limit: 100 tracks. Wait or clear it before adding more.')
            return
        player.queue.append(query)
        await ctx.send('Queued: ' + discord.utils.escape_markdown(query[:300]), allowed_mentions=discord.AllowedMentions.none())
        if not ctx.voice_client or not ctx.voice_client.is_connected():
            await ctx.invoke(join)
        player.halted = False
        player.kick()

def install(bot, target, source_factory, join, root):
    music = Music(bot, target, source_factory, root)
    for name in ('play', 'stop', 'leave', 'help'):
        bot.remove_command(name)
    @bot.command(name='play')
    async def play(ctx, *, url: str):
        await music.enqueue(ctx, url, join)
    for name in ('skip', 'pause', 'resume', 'stop', 'queue', 'clear', 'reset'):
        def make_command(action):
            async def command(ctx):
                await ctx.send(music.player(ctx.guild, ctx.channel).action(action), allowed_mentions=discord.AllowedMentions.none())
            return command
        bot.command(name=name)(make_command(name))
    @bot.command(name='help')
    async def help_command(ctx):
        await ctx.send(HELP, view=Controls(music))
    @bot.command(name='leave')
    async def leave(ctx):
        music.player(ctx.guild, ctx.channel).action('reset')
        if ctx.voice_client:
            await ctx.voice_client.disconnect()
        await ctx.send('Disconnected and queue cleared.')
    bot.add_listener(music.ready, 'on_ready')
    return music

The original installation also updated the bot's avatar and attached a welcome illustration. This tutorial keeps those hooks as explicit no-ops so running it does not rename your application or copy someone else's identity. Replace these functions later only if you want your own presentation layer.

Create presentation.py in the project folder with this complete content:

"""The tutorial does not change your application's name or avatar."""
async def apply_avatar(bot, root):
    return

def welcome_image(root):
    return None, None

7. Check the environment and start the bot

Create the environment checker. It imports dependencies, prints versions, runs the two executable version commands and parses the local configuration. It does not log in to Discord or request YouTube media.

Create check_environment.py in the project folder with this complete content:

"""No Discord login, messages or YouTube requests."""
import importlib.metadata as metadata
import subprocess
import sys
from bot import binary, load_settings

def main():
    print('Python:', sys.version.split()[0])
    for package in ('discord.py', 'PyNaCl', 'davey', 'yt-dlp', 'yt-dlp-ejs'):
        print(package + ':', metadata.version(package))
    for name in ('ffmpeg', 'deno'):
        path = binary(name)
        result = subprocess.run([path, '-version' if name == 'ffmpeg' else '--version'],
                                capture_output=True, text=True, timeout=15, check=True)
        print(result.stdout.splitlines()[0])
    print('Configured servers:', len(load_settings()['channels']))
    print('Environment OK. This check does not prove voice playback.')

if __name__ == '__main__':
    main()

.venv\Scripts\python.exe -B check_environment.py
.venv\Scripts\python.exe -B bot.py

Do not continue to the second command until the checker reports success. At the hidden prompt, paste the bot token from your own application. The prompt does not echo characters. If you already configured DISCORD_TOKEN through your own secret management, that value takes precedence; the tutorial does not require storing it in an environment file.

Expected startup: the terminal reports a connection, the bot joins the configured voice channel, and the text channel shows its command help and buttons. Use the following sequence with audio you are permitted to play. Replace the placeholder with your YouTube URL; do not type the angle brackets.

!play <your authorised YouTube URL>
!play another search phrase
!queue
!pause
!resume
!skip
!stop
!resume
!reset
!leave

First verify audible sound while you are in the same voice channel. Confirm that the second request waits rather than replacing the first. Pause and resume should affect the current track; skip should advance. Stop should retain pending tracks, reset should clear them, and leave should disconnect and clear the player.

Then join again, queue two tracks and try !move Other voice channel. Use a destination where both you and the bot have access. Repeat with a restricted destination to confirm refusal. Commands and buttons outside the configured text channel should not control playback.

Keep the terminal open. Ctrl+C stops the process and all in-memory queues are lost. Run only one copy of this bot installation at a time; these locks do not coordinate multiple processes.

8. Test the queue without a Discord connection

The four files below can run without a token. They replace Discord voice and channel objects with small fakes and exercise transitions that are hard to time manually. They do not substitute for hearing audio in Discord.

The queue test verifies FIFO order, natural completion, continuation after a bad track, pause/resume, skip, stop/reset semantics, stale extraction cleanup, per-server isolation, persistent controls and welcome reuse. The voice test checks destination resolution, permission refusal, timeout handling and preservation of player state. The auto-connect test covers ambiguity, missing permissions, repeat-ready behavior and stable IDs. The input test rejects external URL hosts and malformed requests.

import asyncio
import threading
from types import SimpleNamespace
from pathlib import Path
import tempfile
import music as m

class Source:
    def __init__(self): self.cleaned = False
    def cleanup(self): self.cleaned = True
class Voice:
    def __init__(self): self.playing = False; self.paused = False; self.after = None; self.connected = True
    def is_connected(self): return self.connected
    def is_playing(self): return self.playing
    def is_paused(self): return self.paused
    def play(self, source, after):
        assert not self.playing and not self.paused
        self.playing = True; self.after = after; self.source = source
    def stop(self):
        callback, self.after = self.after, None
        self.playing = self.paused = False
        if callback: self.source.cleanup(); callback(None)
    def pause(self): self.playing = False; self.paused = True
    def resume(self): self.playing = True; self.paused = False
class Channel:
    id = 10
    name = 'playlist'
    def __init__(self): self.sent = []; self.edits = 0
    async def send(self, content, **kwargs): self.sent.append(content); return SimpleNamespace(id=123)
    async def fetch_message(self, mid):
        async def edit(**kw): self.edits += 1
        return SimpleNamespace(id=mid, edit=edit)
async def until(predicate):
    for _ in range(500):
        if predicate(): return
        await asyncio.sleep(.002)
    raise AssertionError('Timed out')

async def main():
    channel, voice = Channel(), Voice()
    guild = SimpleNamespace(id=1, voice_client=voice, text_channels=[channel])
    created = []
    def factory(query):
        if query == 'bad': raise ValueError('test')
        created.append(query)
        return Source()
    with tempfile.TemporaryDirectory() as tmp:
        bot = SimpleNamespace(guilds=[guild], add_view=lambda v: None)
        target = lambda g: {'command_channel_name':'playlist', 'channel_name':'Aterrizaje'}
        manager = m.Music(bot, target, factory, tmp)
        assert manager.allowed(guild, channel)
        assert not manager.allowed(guild, SimpleNamespace(id=999))
        assert not manager.allowed(None, channel)
        assert m.Controls(manager).is_persistent()
        await manager.ready(); await manager.ready()
        assert len(channel.sent) == 1
        other = m.Music(bot, target, factory, tmp)
        await other.ready()
        assert len(channel.sent) == 1 and channel.edits == 1
        p = manager.player(guild, channel)
        p.queue.extend(['one', 'two', 'bad', 'three']); p.kick()
        await until(voice.is_playing)
        p.action('pause'); assert voice.is_paused()
        p.action('resume'); assert voice.is_playing()
        p.action('skip')
        await until(lambda: p.current == 'two' and voice.is_playing())
        voice.stop()
        await until(lambda: p.current == 'three' and voice.is_playing())
        assert created == ['one','two','three']
        p.queue.extend(['four','five']); p.action('clear')
        assert not p.queue and voice.is_playing()
        p.queue.append('four'); p.action('stop')
        await until(lambda: p.task.done())
        assert list(p.queue) == ['four'] and not voice.is_playing()
        p.action('resume'); await until(voice.is_playing)
        p.queue.append('five'); p.action('reset')
        await until(lambda: p.task.done())
        assert not p.queue and not voice.is_playing()
        # Stop while extraction is running must clean source and never play it.
        entered, release = threading.Event(), threading.Event()
        stale = Source()
        def slow(query): entered.set(); release.wait(5); return stale
        p.source_factory = slow; p.halted = False; p.queue.append('slow'); p.kick()
        await until(entered.is_set)
        p.action('reset'); release.set()
        await until(lambda: p.task.done())
        assert stale.cleaned and not voice.is_playing()
        # Two servers do not share their queue.
        g2 = SimpleNamespace(id=2,voice_client=Voice())
        p2 = manager.player(g2,channel); p2.queue.append('isolated')
        assert not p.queue
    print('PASS: FIFO, natural finish, skip, failed-track continuation, pause/resume, stop/resume, clear/reset, extraction race, guild isolation, persistent controls, channel restriction, welcome deduplication')

asyncio.run(main())

import asyncio
from types import SimpleNamespace as NS
from unittest.mock import AsyncMock
from voice_move import change, resolve

async def main():
    perms=NS(view_channel=True,connect=True,speak=True)
    a=NS(id=10,name='Sala uno',permissions_for=lambda m:perms,connect=AsyncMock())
    b=NS(id=20,name='Sala dos',permissions_for=lambda m:perms,connect=AsyncMock())
    vc=NS(channel=a,is_connected=lambda:True,move_to=AsyncMock(),queue=['keep'],paused=True)
    guild=NS(id=1,voice_channels=[a,b],me=object())
    ctx=NS(guild=guild,author=object(),voice_client=vc,send=AsyncMock())
    target={'channel_id':'10'}
    for value in ['Sala dos','sala DOS','<#20>','20','"Sala dos"']:
        assert resolve(guild,value,target) is b
    assert resolve(guild,None,target) is a
    await change(ctx,'Sala dos',target)
    vc.move_to.assert_awaited_once_with(b,timeout=30.0)
    assert vc.queue==['keep'] and vc.paused
    vc.move_to.reset_mock()
    await change(ctx,'Sala uno',target)
    vc.move_to.assert_not_awaited()
    for value in ['999','missing']:
        await change(ctx,value,target)
    vc.move_to.assert_not_awaited()
    guild.voice_channels.append(NS(id=30,name='Sala dos'))
    try: resolve(guild,'Sala dos',target)
    except ValueError: pass
    else: raise AssertionError('Ambiguity accepted')
    guild.voice_channels.pop()
    b.permissions_for=lambda m:NS(view_channel=False,connect=True,speak=True)
    await change(ctx,'20',target)
    vc.move_to.assert_not_awaited()
    b.permissions_for=lambda m:perms if m is ctx.author else NS(view_channel=True,connect=True,speak=False)
    await change(ctx,'20',target)
    vc.move_to.assert_not_awaited()
    b.permissions_for=lambda m:perms
    vc.move_to.side_effect=asyncio.TimeoutError()
    await change(ctx,'20',target)
    assert 'in time' in ctx.send.call_args.args[0]
    ctx.voice_client=None
    await change(ctx,'20',target)
    b.connect.assert_awaited_once()
    other=NS(voice_channels=[a])
    try: resolve(other,'20',target)
    except ValueError: pass
    else: raise AssertionError('Cross-guild channel accepted')
    import bot
    assert bot.bot.get_command('canal') is bot.join
    assert bot.bot.get_command('move') is bot.join
    print('PASS voice move: names, IDs, mentions, ambiguity, guild isolation, permissions, timeout, connect, queue/pause preservation, command registration')
asyncio.run(main())

import asyncio, json, tempfile, unittest
from pathlib import Path
from types import SimpleNamespace as NS
import voice_autoconnect as module

class Channel:
    def __init__(self, ident, name, allowed=True):
        self.id, self.name, self.calls, self.allowed = ident, name, 0, allowed
    def permissions_for(self, member):
        return NS(view_channel=True, connect=self.allowed, speak=True)
    async def connect(self, **kwargs):
        self.calls += 1

def guild(ident, name, channels):
    return NS(id=ident, name=name, voice_channels=channels, me=None, voice_client=None)

class Tests(unittest.IsolatedAsyncioTestCase):
    async def asyncSetUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        module.CONFIG = Path(self.tmp.name) / 'voice_channels.json'
        module._lock = asyncio.Lock()
        module._completed.clear()
    async def asyncTearDown(self):
        self.tmp.cleanup()
    def config(self, targets):
        module.CONFIG.write_text(json.dumps({'channels': targets}), encoding='utf-8')
    async def test_separate_servers_and_repeated_ready(self):
        a,b = Channel(10,'music'),Channel(20,'other')
        self.config([{'guild_name':'A','channel_name':'music'}, {'guild_name':'B','channel_name':'other'}])
        bot=NS(guilds=[guild(1,'A',[a]),guild(2,'B',[b])])
        await module.auto_connect(bot)
        await module.auto_connect(bot)
        self.assertEqual((a.calls,b.calls),(1,1))
        saved=json.loads(module.CONFIG.read_text())
        self.assertEqual(saved['channels'][1]['channel_id'],'20')
        module._completed.clear()
        bot.guilds[0].name='renamed'
        a.name='renamed'
        await module.auto_connect(bot)
        self.assertEqual(a.calls,2)
    async def test_ambiguous_channel(self):
        a,b=Channel(10,'music'),Channel(11,'music')
        self.config([{'guild_name':'A','channel_name':'music'}])
        await module.auto_connect(NS(guilds=[guild(1,'A',[a,b])]))
        self.assertEqual(a.calls+b.calls,0)
    async def test_permissions(self):
        a=Channel(10,'music',False)
        self.config([{'guild_name':'A','channel_name':'music'}])
        await module.auto_connect(NS(guilds=[guild(1,'A',[a])]))
        self.assertEqual(a.calls,0)
    async def test_unconfigured_server(self):
        a=Channel(10,'music')
        self.config([{'guild_name':'A','channel_name':'music'}])
        await module.auto_connect(NS(guilds=[guild(2,'B',[a])]))
        self.assertEqual(a.calls,0)

if __name__=='__main__': unittest.main()

import unittest
from bot import normalize_query

class InputTests(unittest.TestCase):
    def test_search(self):
        self.assertEqual(normalize_query('  piano study  '), 'ytsearch1:piano study')
    def test_youtube(self):
        for url in ('https://youtu.be/example', 'https://www.youtube.com/watch?v=example'):
            self.assertEqual(normalize_query(url), url)
    def test_reject_other_urls(self):
        for url in ('http://youtube.com/watch?v=x', 'https://localhost/x',
                    'https://youtube.com.evil.test/x', 'file:///etc/passwd',
                    'https://user:pass@youtube.com/x', 'https://youtube.com:444/x',
                    '//youtube.com/x', '', 'x' * 501):
            with self.subTest(url=url), self.assertRaises(ValueError):
                normalize_query(url)

if __name__ == '__main__':
    unittest.main()

.venv\Scripts\python.exe -B test_queue.py
.venv\Scripts\python.exe -B test_voice_move.py
.venv\Scripts\python.exe -B test_autoconnect.py
.venv\Scripts\python.exe -B test_input.py

The first two scripts print a PASS summary. The unittest scripts report OK. A failure is a reason to inspect the change; do not replace an assertion with a weaker expectation just to make the output green.

9. Diagnose the failing boundary

The extractor logger deliberately avoids printing complete signed media URLs. When diagnosing a failing dependency, keep detailed output local and redact tokens, cookies and media URLs before sharing it. Do not turn a transient extraction failure into an endless retry loop.

If yt-dlp needs an update, create a separate test environment, install an updated yt-dlp[default], and repeat the offline and real playback checks before replacing your working installation. Keep the old lock file until the new combination is verified.

Where this design stops

This bot is suitable for learning and small trusted servers. It has a 100-item pending queue limit per server and caps request length, but it does not include DJ roles, user quotas, durable queues, a distributed player, metrics, guaranteed reconnect recovery or a maximum track duration. Multiple servers can extract at once and compete for CPU and network resources.

For an unattended service, the next design decisions would be explicit request limits, idle disconnect, structured diagnostics with redaction, lifecycle cleanup on server removal and a policy for long-running extraction. Those choices should come from the service you intend to operate; they are not hidden capabilities of this example.

Use media you have permission to play and follow the applicable platform terms. The fact that a URL can be extracted does not establish permission to use its content.

You now have the complete path from a command to audio, plus tests for the state transitions that can otherwise make a music bot feel unpredictable. Change one boundary at a time, rerun the tests and confirm the user-visible result in your own server.