5分で作るTuneプラグイン入門

Tune Serverはオープンであるように設計されています。バージョン0.7.90以降、Pythonプラグインシステムによって、ソースコードを変更することなくサーバーを拡張できます。新しいオーディオ出力の追加、サードパーティサービスとの連携、独自APIエンドポイントの作成、再生イベントへの反応などが可能です。この記事では、数分で動作する最初のプラグインを作ってみましょう。

前提条件

• マシンにPython 3.11以上がインストールされていること
• pip（Pythonに同梱）
• 動作するTuneサーバー（バージョン0.7.90以上）
• テキストエディタ（VS Code、PyCharm、vimなど）

Pythonのバージョンを確認します:

python3 --version
# Python 3.11.x 以上

コンセプト

Tuneのプラグインシステムは3つの要素で構成されています。

1. TunePlugin Protocol

各プラグインは、TunePlugin protocolを実装したPythonクラスです。このprotocolは2つの属性（nameとversion）と2つのライフサイクルメソッド（setupとteardown）を定義します。

from typing import Protocol, runtime_checkable

@runtime_checkable
class TunePlugin(Protocol):
    name: str       # 一意の識別子、例: "hello"
    version: str    # semverバージョン、例: "0.1.0"

    async def setup(self, ctx: PluginContext) -> None:
        """サーバー起動時に呼ばれる。"""
        ...

    async def teardown(self) -> None:
        """サーバー停止時に呼ばれる。"""
        ...

2. PluginContext

setupの際、プラグインはPluginContextオブジェクトを受け取ります。これはTuneの内部サービスへ安全にアクセスするための入口です。

• ctx.event_bus — 再生イベント（トラック開始、ゾーン作成など）を購読する
• ctx.zone_manager — 新しいオーディオ出力タイプを登録する
• ctx.api_app — 独自のFastAPIルートをマウントする
• ctx.settings — サーバー設定の読み取り（読み取り専用）
• ctx.db — データベースへのアクセス

3. ライフサイクル

起動時、TuneはPythonのentry pointsを通じてインストール済みプラグインを自動的に発見し、次の順序で初期化します。

1. 発見 — Tuneがentry_points(group="tune_server.plugins")をスキャンする
2. setup(ctx) — プラグインがルート、フック、出力タイプを登録する
3. 通常動作 — サーバーが稼働し、プラグインがアクティブになる
4. teardown() — 停止時、プラグインがリソースを解放する

ホットリロードはありません。変更後は必ずサーバーの再起動が必要です。これは安定性を保証するための意図的な選択です（再生中に状態が壊れることを避けるため）。

ステップバイステップのクイックスタート

/api/v1/hello というAPIエンドポイントを追加し、トラックが変わるたびにメッセージをログ出力する「Hello」プラグインを作ってみましょう。

ステップ1 — プロジェクト構造を作る

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

最終的な構造は次のようになります:

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

ステップ2 — pyproject.tomlを書く

pyproject.tomlファイルは、あなたのプラグインをPythonとTuneに対して宣言します。鍵となるのは[project.entry-points]セクションで、これによりTuneが起動時にプラグインを発見できます。

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

[project]
name = "tune-hello-plugin"
version = "0.1.0"
description = "私の最初のTuneプラグイン"
requires-python = ">=3.11"
dependencies = []

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

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

hello = "tune_hello:HelloPlugin"という行の意味は、「tune_helloパッケージ内のHelloPluginクラスを、'hello'という名前のプラグインとして登録する」ということです。

ステップ3 — プラグインを書く

src/tune_hello/__init__.pyを作成します:

from tune_hello.plugin import HelloPlugin

__all__ = ["HelloPlugin"]

次にsrc/tune_hello/plugin.py:

import logging
from fastapi import APIRouter

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


class HelloPlugin:
    """エンドポイントを追加しトラックをログ出力する最小プラグイン。"""

    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": "Helloプラグインからこんにちは！"}

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

    async def setup(self, ctx) -> None:
        """Tune起動時に呼ばれる。"""
        # APIルートを登録する
        ctx.register_router(self.router, prefix="/api/v1")

        # 再生イベントを購読する
        ctx.event_bus.on("track.started", self._on_track_started)

        logger.info("Helloプラグイン起動 (v%s)", self.version)

    async def teardown(self) -> None:
        """Tune停止時に呼ばれる。"""
        logger.info("Helloプラグイン停止")

    async def _on_track_started(self, event) -> None:
        """新しいトラックごとに呼ばれる。"""
        track = event.track
        logger.info(
            "新しいトラック: %s - %s",
            track.artist,
            track.title,
        )

ステップ4 — プラグインをインストールする

プラグインのフォルダから、Tuneと同じPython環境に開発モードでインストールします:

# 必要ならTuneのvirtualenvを有効化する
source /chemin/vers/tune-server/venv/bin/activate

# editableモードでプラグインをインストール
pip install -e .

-e（editable）モードなら、毎回再インストールせずにコードを変更できます。

ステップ5 — Tuneを再起動する

Tuneサーバーを再起動します。ログに次のように表示されるはずです:

INFO  plugin_loaded  name=hello  version=0.1.0
INFO  Helloプラグイン起動 (v0.1.0)

systemdを使っている場合:

sudo systemctl restart tune-server

またはAPI経由（サーバーが既に稼働中の場合）:

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

ステップ6 — 確認する

プラグインが正しくロードされたか確認します:

# ロード済みプラグインの一覧
curl http://localhost:8888/api/v1/system/plugins

期待されるレスポンス:

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

エンドポイントをテストします:

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

{"message": "Helloプラグインからこんにちは！"}

そしてTuneでトラックを再生してみてください — サーバーの出力に新しいトラック: アーティスト - タイトルというログが現れます。

さらに先へ

Helloプラグインは意図的にシンプルにしてあります。ここからはプラグインシステムの主な機能を紹介します。

独自のオーディオ出力タイプを追加する

ネイティブにサポートされていない機器やサービスへストリーミングするために、新しい出力タイプを作成できます。例として、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="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

プラグインをインストールすると、各ゾーンで利用可能な出力の一覧にicecast出力タイプが表示されます。

再生イベントにフックする

TuneのEventBusは、あらゆるアクションごとにイベントを発行します。プラグインはこれを購読して、scrobbling（Last.fm、ListenBrainz）、再生統計、通知などを実装できます。

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:
        # Last.fmに"now playing"を通知する
        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:
        # トラックを50%以上聴いた場合にscrobbleする
        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()

APIルートを追加する

各プラグインは、必要なだけルートを持つ完全なFastAPIのAPIRouterを提供できます。ルートはTuneのSwagger UI（/docs）に自動的にドキュメント化されます。

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

@router.get("/stats/top-artists")
async def top_artists(limit: int = 10):
    """最も再生されたアーティストを返す。"""
    return await get_top_artists(limit=limit)

@router.get("/stats/listening-time")
async def listening_time(period: str = "week"):
    """期間内の合計再生時間。"""
    return await get_listening_time(period=period)

環境変数による設定

プラグインはpydantic-settingsベースの設定スキーマを宣言できます。環境変数はTUNE_<PLUGIN>_<KEY>という規約に従います:

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 は $TUNE_ICECAST_HOST または "localhost"
        ...

変数は環境、またはサーバーのルートにある.envファイルから読み込まれます。

Playerのイベントにフックする

EventBusより細かい制御が必要なユースケースでは、プラグインは各ゾーンのPlayerに直接フックを登録できます:

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)

EventBusとの違い: Playerのフックは再生パイプライン内で同期的に呼ばれるため、アクションが発生する前に介入できます（たとえば、トラック開始前にバッファを準備する、など）。

PyPIで公開する

プラグインが完成したら、PyPIで公開してコミュニティと共有しましょう。

命名規約

Tuneプラグインは、PyPIパッケージ名としてtune-plugin-<名前>という規約に従います:

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

プラグインの内部名（name属性）は短く保ちます: icecast、lastfm、prometheus。

パッケージング

pyproject.tomlが完全であることを確認します:

[project]
name = "tune-plugin-icecast"
version = "0.1.0"
description = "Icecastへ配信するTuneプラグイン"
readme = "README.md"
license = {text = "MIT"}
requires-python = ">=3.11"
dependencies = [
    "tune-server>=0.7.90",
]

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

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

公開:

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

ユーザーはその後、1つのコマンドであなたのプラグインをインストールできます:

pip install tune-plugin-icecast

サーバーを再起動するだけで、プラグインは自動的に発見され有効化されます。

バージョン互換性

依存関係には必ず互換性のあるTuneのバージョン範囲を宣言してください。サーバーはPROTOCOL_VERSIONを公開しており、プラグインはsetup時にこれを検証できます:

from tune_server.plugins.api import PROTOCOL_VERSION

async def setup(self, ctx) -> None:
    if PROTOCOL_VERSION != "1":
        raise RuntimeError(
            f"プラグイン {self.name} は PROTOCOL_VERSION 1 を必要とします。"
            f"検出されたバージョン: {PROTOCOL_VERSION}"
        )

ベストプラクティス

• エラー処理 — プラグイン内の例外がサーバーをクラッシュさせてはなりません。Tuneはロード時とフック内のエラーを捕捉しますが、ビジネスロジックでは自分でtry/exceptを扱ってください。
• ロギング — ログをサーバーのものと一貫させるため、logging.getLogger("tune.plugin.<名前>")を使ってください。
• Async — setup、teardown、およびイベントコールバックはasyncである必要があります。Tuneは完全に非同期（asyncio）です。
• テスト — pytestとpytest-asyncioでプラグインを独立してテストしてください。ユニットテストではPluginContextをモックします。
• 接頭辞 — プラグインがデータベースにテーブルを作成する場合、衝突を避けるためプラグイン名を接頭辞に付けてください（例: icecast_sessions）。
• リソース — リソース（ネットワーク接続、開いたファイル、asyncioタスク）は必ずteardownで解放してください。

結論

数行のPythonと1つのpyproject.tomlがあれば、Tune Serverを自分の固有のニーズに合わせて拡張できます。ホームオートメーション連携、scrobbling、変わり種のオーディオ出力、独自ダッシュボード……可能性は開かれています。

さらに先へ進むために:

• プラグインSDKの完全なドキュメントを参照してください
• GitHubのサンプルを探してみてください
• 作品の共有や質問はMozaik Labsフォーラムで

あなたが何を作り上げるのか、楽しみにしています。