Cree su primer plugin de Tune en 5 minutos

Tune Server está diseñado para ser abierto. Desde la versión 0.7.90, un sistema de plugins en Python permite extender el servidor sin modificar su código fuente: añadir nuevas salidas de audio, integrar servicios de terceros, crear endpoints de API personalizados o reaccionar a los eventos de reproducción. En este artículo vamos a crear un primer plugin funcional en unos minutos.

Requisitos previos

• Python 3.11+ instalado en su máquina
• pip (incluido con Python)
• Un servidor Tune funcional (versión 0.7.90 o superior)
• Un editor de texto (VS Code, PyCharm, vim...)

Compruebe su versión de Python:

python3 --version
# Python 3.11.x o superior

El concepto

El sistema de plugins de Tune se basa en tres elementos:

1. El Protocol TunePlugin

Cada plugin es una clase Python que implementa el protocol TunePlugin. Este protocol define dos atributos (name y version) y dos métodos de ciclo de vida (setup y teardown).

from typing import Protocol, runtime_checkable

@runtime_checkable
class TunePlugin(Protocol):
    name: str       # identificador único, ej.: "hello"
    version: str    # versión semver, ej.: "0.1.0"

    async def setup(self, ctx: PluginContext) -> None:
        """Llamado al arrancar el servidor."""
        ...

    async def teardown(self) -> None:
        """Llamado al detener el servidor."""
        ...

2. El PluginContext

Durante el setup, su plugin recibe un objeto PluginContext que da acceso a los servicios internos de Tune de forma segura:

• ctx.event_bus — suscribirse a los eventos de reproducción (pista iniciada, zona creada, etc.)
• ctx.zone_manager — registrar nuevos tipos de salida de audio
• ctx.api_app — montar rutas API FastAPI personalizadas
• ctx.settings — leer la configuración del servidor (solo lectura)
• ctx.db — acceso a la base de datos

3. El ciclo de vida

Al arrancar, Tune descubre automáticamente los plugins instalados a través de los entry points de Python, y luego los inicializa en orden:

1. Descubrimiento — Tune escanea los entry_points(group="tune_server.plugins")
2. setup(ctx) — el plugin registra sus rutas, hooks y tipos de salida
3. Funcionamiento normal — el servidor funciona, el plugin está activo
4. teardown() — al detenerse, el plugin libera sus recursos

Sin recarga en caliente: tras cualquier modificación es necesario reiniciar el servidor. Es una decisión deliberada para garantizar la estabilidad (sin corrupción de estado durante la reproducción).

Quick start paso a paso

Creemos un plugin "Hello" que añada un endpoint API /api/v1/hello y registre un mensaje en cada cambio de pista.

Paso 1 — Crear la estructura del proyecto

mkdir tune-hello-plugin
cd tune-hello-plugin
mkdir -p src/tune_hello

La estructura final será:

tune-hello-plugin/
├── pyproject.toml
└── src/
    └── tune_hello/
        ├── __init__.py
        └── plugin.py

Paso 2 — Escribir el pyproject.toml

El archivo pyproject.toml declara su plugin ante Python y ante Tune. La sección [project.entry-points] es la clave: permite que Tune descubra su plugin al arrancar.

[build-system]
requires = ["setuptools>=68.0"]
build-backend = "setuptools.build_meta"

[project]
name = "tune-hello-plugin"
version = "0.1.0"
description = "Mi primer plugin de Tune"
requires-python = ">=3.11"
dependencies = []

[project.entry-points."tune_server.plugins"]
hello = "tune_hello:HelloPlugin"

[tool.setuptools.packages.find]
where = ["src"]

La línea hello = "tune_hello:HelloPlugin" significa: "registra el plugin llamado 'hello', cuya clase es HelloPlugin en el paquete tune_hello".

Paso 3 — Escribir el plugin

Cree src/tune_hello/__init__.py:

from tune_hello.plugin import HelloPlugin

__all__ = ["HelloPlugin"]

Luego src/tune_hello/plugin.py:

import logging
from fastapi import APIRouter

logger = logging.getLogger("tune.plugin.hello")


class HelloPlugin:
    """Un plugin mínimo que añade un endpoint y registra las pistas."""

    name = "hello"
    version = "0.1.0"

    def __init__(self):
        self.router = APIRouter(tags=["hello"])
        self._setup_routes()

    def _setup_routes(self):
        @self.router.get("/hello")
        async def hello():
            return {"message": "¡Hola desde el plugin Hello!"}

        @self.router.get("/hello/status")
        async def status():
            return {
                "plugin": self.name,
                "version": self.version,
                "status": "running",
            }

    async def setup(self, ctx) -> None:
        """Llamado al arrancar Tune."""
        # Registrar nuestras rutas API
        ctx.register_router(self.router, prefix="/api/v1")

        # Suscribirse a los eventos de reproducción
        ctx.event_bus.on("track.started", self._on_track_started)

        logger.info("Plugin Hello iniciado (v%s)", self.version)

    async def teardown(self) -> None:
        """Llamado al detener Tune."""
        logger.info("Plugin Hello detenido")

    async def _on_track_started(self, event) -> None:
        """Llamado en cada nueva pista."""
        track = event.track
        logger.info(
            "Nueva pista: %s - %s",
            track.artist,
            track.title,
        )

Paso 4 — Instalar el plugin

Desde la carpeta de su plugin, instálelo en modo desarrollo en el mismo entorno Python que Tune:

# Active el virtualenv de Tune si es necesario
source /ruta/hacia/tune-server/venv/bin/activate

# Instale el plugin en modo editable
pip install -e .

El modo -e (editable) permite modificar el código sin reinstalar cada vez.

Paso 5 — Reiniciar Tune

Reinicie el servidor Tune. En los logs debería ver:

INFO  plugin_loaded  name=hello  version=0.1.0
INFO  Plugin Hello iniciado (v0.1.0)

Si utiliza systemd:

sudo systemctl restart tune-server

O a través de la API (si el servidor ya está en ejecución):

curl -X POST http://localhost:8888/api/v1/system/restart

Paso 6 — Verificar

Confirme que el plugin se ha cargado correctamente:

# Lista de plugins cargados
curl http://localhost:8888/api/v1/system/plugins

Respuesta esperada:

{
  "plugins": [
    {
      "name": "hello",
      "version": "0.1.0",
      "status": "running"
    }
  ]
}

Pruebe su endpoint:

curl http://localhost:8888/api/v1/hello

{"message": "¡Hola desde el plugin Hello!"}

Y lance una pista en Tune — verá aparecer el log Nueva pista: Artista - Título en la salida del servidor.

Ir más lejos

El plugin Hello es deliberadamente simple. Estas son las principales capacidades del sistema de plugins.

Añadir un tipo de salida de audio personalizado

Puede crear nuevos tipos de salida para transmitir hacia dispositivos o servicios no soportados de forma nativa. Por ejemplo, un plugin para emitir hacia un servidor Icecast:

from tune_server.plugins.api import OutputTypeSpec
from tune_server.audio.formats import AudioCapabilities


class IcecastPlugin:
    name = "icecast"
    version = "0.1.0"

    async def setup(self, ctx) -> None:
        ctx.register_output_type(OutputTypeSpec(
            name="icecast",
            factory=self._create_output,
            capabilities=AudioCapabilities(
                max_sample_rate=48000,
                max_bit_depth=16,
                formats=["flac", "mp3"],
            ),
            description="Emisión hacia un servidor Icecast",
        ))

    def _create_output(self, zone_id: str | None):
        return IcecastOutput(
            host=os.environ.get("TUNE_ICECAST_HOST", "localhost"),
            port=int(os.environ.get("TUNE_ICECAST_PORT", "8000")),
            mount=os.environ.get("TUNE_ICECAST_MOUNT", "/stream"),
            password=os.environ.get("TUNE_ICECAST_PASSWORD", "hackme"),
        )

    async def teardown(self) -> None:
        pass

Una vez instalado el plugin, el tipo de salida icecast aparece en la lista de salidas disponibles para cada zona.

Enganchar los eventos de reproducción

El EventBus de Tune emite eventos en cada acción. Un plugin puede suscribirse para implementar scrobbling (Last.fm, ListenBrainz), estadísticas de escucha o notificaciones.

class ScrobblerPlugin:
    name = "lastfm-scrobbler"
    version = "0.1.0"

    async def setup(self, ctx) -> None:
        ctx.event_bus.on("track.started", self._on_track_started)
        ctx.event_bus.on("track.finished", self._on_track_finished)

    async def _on_track_started(self, event) -> None:
        # Señalar "now playing" a Last.fm
        await self.lastfm_client.update_now_playing(
            artist=event.track.artist,
            title=event.track.title,
            album=event.track.album,
        )

    async def _on_track_finished(self, event) -> None:
        # Scrobble si la pista se ha escuchado al menos al 50%
        if event.progress >= 0.5:
            await self.lastfm_client.scrobble(
                artist=event.track.artist,
                title=event.track.title,
                album=event.track.album,
                timestamp=event.started_at,
            )

    async def teardown(self) -> None:
        await self.lastfm_client.close()

Añadir rutas API

Cada plugin puede aportar un APIRouter FastAPI completo, con tantas rutas como sea necesario. Las rutas se documentan automáticamente en el Swagger UI de Tune (/docs).

router = APIRouter(tags=["stats"])

@router.get("/stats/top-artists")
async def top_artists(limit: int = 10):
    """Devuelve los artistas más escuchados."""
    return await get_top_artists(limit=limit)

@router.get("/stats/listening-time")
async def listening_time(period: str = "week"):
    """Tiempo de escucha total en el periodo."""
    return await get_listening_time(period=period)

Configuración mediante variables de entorno

Los plugins pueden declarar un esquema de configuración basado en pydantic-settings. Las variables de entorno siguen la convención TUNE_<PLUGIN>_<CLAVE>:

from pydantic_settings import BaseSettings


class IcecastSettings(BaseSettings):
    host: str = "localhost"
    port: int = 8000
    mount: str = "/stream"
    password: str = "hackme"

    class Config:
        env_prefix = "TUNE_ICECAST_"


class IcecastPlugin:
    name = "icecast"
    version = "0.1.0"

    def config_schema(self) -> type[BaseSettings]:
        return IcecastSettings

    async def setup(self, ctx) -> None:
        self.config = IcecastSettings()
        # self.config.host vale $TUNE_ICECAST_HOST o "localhost"
        ...

Las variables se leen desde el entorno o desde un archivo .env en la raíz del servidor.

Enganchar los eventos del Player

Para casos de uso que requieren un control más fino que el EventBus, los plugins pueden registrar hooks directamente en el Player de cada zona:

async def setup(self, ctx) -> None:
    ctx.register_player_hook("before_track", self._before_track)
    ctx.register_player_hook("after_track", self._after_track)
    ctx.register_player_hook("play", self._on_play)
    ctx.register_player_hook("pause", self._on_pause)

La diferencia con el EventBus: los hooks del Player se llaman de forma síncrona dentro del pipeline de reproducción, lo que permite intervenir antes de que se produzca una acción (por ejemplo, preparar un búfer antes del inicio de una pista).

Publicar en PyPI

¿Su plugin está listo? Compártalo con la comunidad publicándolo en PyPI.

Convención de nomenclatura

Los plugins de Tune siguen la convención tune-plugin-<nombre> para el nombre del paquete PyPI:

• tune-plugin-icecast
• tune-plugin-lastfm
• tune-plugin-prometheus

El nombre interno del plugin (atributo name) sigue siendo corto: icecast, lastfm, prometheus.

Packaging

Asegúrese de que su pyproject.toml esté completo:

[project]
name = "tune-plugin-icecast"
version = "0.1.0"
description = "Plugin de Tune para emitir hacia Icecast"
readme = "README.md"
license = {text = "MIT"}
requires-python = ">=3.11"
dependencies = [
    "tune-server>=0.7.90",
]

[project.urls]
Homepage = "https://github.com/su-nombre/tune-plugin-icecast"

[project.entry-points."tune_server.plugins"]
icecast = "tune_icecast:IcecastPlugin"

Publique:

pip install build twine
python -m build
twine upload dist/*

Los usuarios podrán después instalar su plugin con un solo comando:

pip install tune-plugin-icecast

Basta con reiniciar el servidor para que el plugin sea descubierto y activado automáticamente.

Compatibilidad de versiones

Declare siempre el rango de versiones de Tune compatibles en sus dependencias. El servidor expone un PROTOCOL_VERSION que su plugin puede verificar en el setup:

from tune_server.plugins.api import PROTOCOL_VERSION

async def setup(self, ctx) -> None:
    if PROTOCOL_VERSION != "1":
        raise RuntimeError(
            f"El plugin {self.name} requiere PROTOCOL_VERSION 1, "
            f"encontrado {PROTOCOL_VERSION}"
        )

Buenas prácticas

• Gestión de errores — Las excepciones de un plugin nunca deben hacer caer el servidor. Tune captura los errores en la carga y en los hooks, pero gestione sus propios try/except en la lógica de negocio.
• Logging — Utilice logging.getLogger("tune.plugin.<nombre>") para que sus logs sean coherentes con los del servidor.
• Async — Los métodos setup, teardown y los callbacks de eventos deben ser async. Tune es totalmente asíncrono (asyncio).
• Tests — Pruebe su plugin de forma independiente con pytest y pytest-asyncio. Simule (mock) el PluginContext para las pruebas unitarias.
• Prefijado — Si su plugin crea tablas en la base de datos, prefíjelas con el nombre del plugin (ej.: icecast_sessions) para evitar colisiones.
• Recursos — Libere siempre sus recursos (conexiones de red, archivos abiertos, tareas asyncio) en teardown.

Conclusión

Con unas pocas líneas de Python y un pyproject.toml, puede extender Tune Server para responder a sus necesidades específicas: integraciones domóticas, scrobbling, salidas de audio exóticas, paneles personalizados... Las posibilidades están abiertas.

Para ir más lejos:

• Consulte la documentación completa del SDK de plugins
• Explore los ejemplos en GitHub
• Comparta sus creaciones y plantee sus preguntas en el foro de Mozaik Labs

Estamos deseando ver lo que va a construir.