or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

audio-filters.mdevents-exceptions.mdindex.mdnode-management.mdplayer-control.mdqueue-system.mdtrack-search.md
tile.json

tessl/pypi-wavelink

A robust and powerful, fully asynchronous Lavalink wrapper built for discord.py in Python.

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/wavelink@3.4.x

To install, run

npx @tessl/cli install tessl/pypi-wavelink@3.4.0

index.mddocs/

Wavelink

A robust and powerful, fully asynchronous Lavalink wrapper built for discord.py in Python. Wavelink enables Discord bot developers to integrate advanced audio playback capabilities through a comprehensive, object-oriented API that supports Lavalink v4+ servers, advanced AutoPlay functionality, full type annotation compliance, and seamless integration with Discord.py's voice system.

Package Information

  • Package Name: wavelink
  • Language: Python
  • Installation: pip install wavelink
  • Version: 3.4.1
  • License: MIT
  • Documentation: https://wavelink.dev/en/latest

Core Imports

import wavelink

Common imports for typical use cases:

from wavelink import Pool, Node, Player, Playable, Playlist, Queue
from wavelink import Filters, AutoPlayMode, QueueMode, TrackSource

Basic Usage

import discord
from discord.ext import commands
import wavelink

class Bot(commands.Bot):
    def __init__(self):
        super().__init__(command_prefix='!', intents=discord.Intents.all())

    async def on_ready(self):
        print(f'{self.user} is ready!')
        
        # Connect to Lavalink node
        await wavelink.Pool.connect(
            uri="http://localhost:2333",
            password="youshallnotpass",
            client=self
        )

    async def on_wavelink_track_end(self, payload):
        # Handle track end events
        player = payload.player
        if not player.queue.is_empty:
            await player.play(player.queue.get())

bot = Bot()

@bot.command()
async def play(ctx, *, query: str):
    """Play a song from YouTube, SoundCloud, etc."""
    if not ctx.voice_client:
        player = await ctx.author.voice.channel.connect(cls=wavelink.Player)
    else:
        player = ctx.voice_client

    # Search for tracks
    tracks = await wavelink.Pool.fetch_tracks(query)
    if not tracks:
        return await ctx.send("No tracks found!")

    if isinstance(tracks, wavelink.Playlist):
        # Add entire playlist
        player.queue.put(tracks)
        await ctx.send(f"Added playlist: {tracks.name}")
    else:
        # Add single track
        track = tracks[0]
        player.queue.put(track)
        await ctx.send(f"Added: {track.title}")

    # Start playing if nothing is currently playing
    if not player.playing:
        await player.play(player.queue.get())

@bot.command()
async def skip(ctx):
    """Skip the current track."""
    player = ctx.voice_client
    if player:
        await player.skip()
        await ctx.send("Skipped!")

bot.run('YOUR_BOT_TOKEN')

Architecture

Wavelink follows a hierarchical architecture that provides comprehensive control over audio playback:

  • Pool: Global node manager that handles connections to multiple Lavalink servers, load balancing, and track searching
  • Node: Individual Lavalink server connection with REST API access, WebSocket communication, and player management
  • Player: Discord voice client extension providing playback control, queue management, filters, and event handling
  • Queue: Thread-safe track queue with multiple modes (normal, loop, loop_all), history tracking, and AutoPlay support
  • Tracks: Rich track and playlist objects with metadata, source information, and search capabilities

This design enables seamless integration with discord.py while providing advanced features like AutoPlay recommendations, comprehensive audio filtering, and robust event handling for sophisticated music bot development.

Capabilities

Node Management

Connection and management of Lavalink servers including pool management, node discovery, health monitoring, and load balancing across multiple nodes.

class Pool:
    @classmethod
    async def connect(
        cls,
        uri: str,
        password: str,
        *,
        identifier: str | None = None,
        client: discord.Client,
        **kwargs
    ) -> Node: ...
    
    @classmethod
    def get_node(cls, identifier: str | None = None) -> Node: ...
    
    @classmethod
    async def fetch_tracks(
        cls,
        query: str,
        *,
        node: Node | None = None
    ) -> list[Playable] | Playlist: ...

class Node:
    @property
    def status(self) -> NodeStatus: ...
    
    @property
    def players(self) -> dict[int, Player]: ...
    
    async def fetch_stats(self) -> StatsResponsePayload: ...

Node Management

Player Control

Main audio player interface providing playback control, volume management, seeking, and voice channel operations with full discord.py VoiceProtocol integration.

class Player(discord.VoiceProtocol):
    queue: Queue
    auto_queue: Queue
    
    @property
    def current(self) -> Playable | None: ...
    
    @property
    def playing(self) -> bool: ...
    
    @property
    def paused(self) -> bool: ...
    
    async def play(
        self,
        track: Playable,
        *,
        replace: bool = True,
        start: int = 0,
        end: int | None = None,
        volume: int | None = None,
        paused: bool | None = None,
        add_history: bool = True
    ) -> None: ...
    
    async def pause(self, value: bool) -> None: ...
    
    async def seek(self, position: int = 0) -> None: ...
    
    async def skip(self, *, force: bool = True) -> Playable | None: ...

Player Control

Track Search & Management

Comprehensive track searching across multiple platforms (YouTube, SoundCloud, YouTube Music) with rich metadata, playlist support, and advanced search capabilities.

class Playable:
    title: str
    author: str
    length: int
    uri: str
    source: TrackSource
    
    @classmethod
    async def search(
        cls,
        query: str,
        *,
        source: TrackSource | None = None,
        node: Node | None = None
    ) -> list[Playable]: ...

class Playlist:
    name: str
    tracks: list[Playable]
    
    def pop(self, index: int = -1) -> Playable: ...

# Search result type
Search = list[Playable] | Playlist

Track Search & Management

Queue System

Advanced queue management with multiple modes, history tracking, AutoPlay functionality, and thread-safe operations for robust track management.

class Queue:
    mode: QueueMode
    history: Queue | None
    
    @property
    def is_empty(self) -> bool: ...
    
    @property
    def count(self) -> int: ...
    
    def put(
        self,
        item: list[Playable] | Playable | Playlist,
        *,
        atomic: bool = True
    ) -> int: ...
    
    def get() -> Playable: ...
    
    async def get_wait() -> Playable: ...
    
    def shuffle(self) -> None: ...
    
    def clear(self) -> None: ...

# Queue modes
class QueueMode(enum.Enum):
    normal = 0
    loop = 1
    loop_all = 2

# AutoPlay modes  
class AutoPlayMode(enum.Enum):
    enabled = 0
    partial = 1
    disabled = 2

Queue System

Audio Filters

Comprehensive audio filter system including equalizer, distortion, karaoke, timescale, and plugin filters for advanced audio processing and effects.

class Filters:
    volume: float | None
    equalizer: Equalizer
    karaoke: Karaoke
    timescale: Timescale
    tremolo: Tremolo
    vibrato: Vibrato
    rotation: Rotation
    distortion: Distortion
    channel_mix: ChannelMix
    low_pass: LowPass
    plugin_filters: PluginFilters
    
    def set_filters(self, **filters) -> None: ...
    
    def reset(self) -> None: ...
    
    @classmethod
    def from_filters(cls, **filters) -> Self: ...

class Equalizer:
    def set(self, *, bands: list[dict] | None = None) -> Self: ...
    
    def reset(self) -> Self: ...
    
    @property
    def payload(self) -> dict[int, dict]: ...

Audio Filters

Events & Exceptions

Rich event system for track lifecycle, player state changes, and WebSocket events, plus comprehensive exception hierarchy for robust error handling.

# Exception hierarchy
class WavelinkException(Exception): ...

class NodeException(WavelinkException):
    status: int | None

class LavalinkException(WavelinkException):
    timestamp: int
    status: int
    error: str
    trace: str | None
    path: str

class QueueEmpty(WavelinkException): ...

# Event payloads (data structures for event handlers)
class TrackStartEventPayload: ...
class TrackEndEventPayload: ...
class PlayerUpdateEventPayload: ...

Events & Exceptions

Types

Core type definitions used across the wavelink API.

# Enums
class NodeStatus(enum.Enum):
    DISCONNECTED = 0
    CONNECTING = 1
    CONNECTED = 2

class TrackSource(enum.Enum):
    YouTube = 0
    YouTubeMusic = 1
    SoundCloud = 2

class DiscordVoiceCloseType(enum.Enum):
    """Discord Voice WebSocket close codes."""
    CLOSE_NORMAL = 1000
    UNKNOWN_OPCODE = 4001
    FAILED_DECODE_PAYLOAD = 4002
    NOT_AUTHENTICATED = 4003
    AUTHENTICATION_FAILED = 4004
    ALREADY_AUTHENTICATED = 4005
    SESSION_INVALID = 4006
    SESSION_TIMEOUT = 4009
    SERVER_NOT_FOUND = 4011
    UNKNOWN_PROTOCOL = 4012
    DISCONNECTED = 4014
    VOICE_SERVER_CRASHED = 4015
    UNKNOWN_ENCRYPTION_MODE = 4016

# Namespace for extras and metadata
class ExtrasNamespace:
    """
    A namespace for additional track metadata and custom attributes.
    
    Supports construction with dict of str keys and Any values, 
    keyword pairs, or a mix of both. Can access dict version 
    by calling dict() on instance.
    """
    def __init__(self, __dict: dict[str, Any] = {}, /, **kwargs: Any) -> None: ...

# Cache implementation for internal use
class LFUCache:
    """
    Least Frequently Used cache implementation used internally 
    by Pool for track caching.
    """
    def __init__(self, *, capacity: int) -> None: ...
    
    @property
    def capacity(self) -> int: ...
    
    def get(self, key: Any, default: Any = ...) -> Any: ...
    
    def put(self, key: Any, value: Any) -> None: ...