hiozjl 的技术笔记

Linux、OpenWrt、VPS、网络与自托管实践记录

2026年5月5日 hiozjl 7 mins read

让 ESP32 语音助手播放 NAS 音乐：Navidrome + MCP 桥接实践

Linux

最近在玩 xiaozhi-esp32-server（一个开源的 ESP32 语音助手方案），想着能不能直接对着小智说一句「放首歌」就播 NAS 里的音乐。NAS 上跑着 Navidrome 作为音乐流媒体服务端，于是中间搭了一座 MCP（Model Context Protocol） 桥接服务，打通了整个链路。

整体架构

整条链路很简单：

  • 用户对 ESP32 说「放歌」
  • ESP32 把语音送到 xiaozhi-esp32-server（ASR 转文字）
  • xiaozhi LLM 识别意图 → 调用 MCP 工具
  • MCP 桥接服务通过 Subsonic API 搜索 Navidrome 曲库
  • 流媒体通过 HTTP 代理转码后推给 ESP32 播放

全程走局域网 HTTP 直连，ESP32 不需要处理 HTTPS 认证。

MCP 工具（SSE 协议）

桥接服务基于 Python 的 fastmcp，注册了以下几个 MCP 工具供 LLM 调用：

  • navidrome_search(query) — 搜索歌曲、歌手、专辑
  • navidrome_get_stream_url(song_id) — 获取歌曲的局域网 HTTP 流地址和歌词地址
  • navidrome_random_song() — 随机来一首

xiaozhi 的 LLM 通过 SSE 连接注册这些工具，用户说「放一首周杰伦的歌」时，LLM 会自动调用 navidrome_search 找到歌曲。

音频流代理

Navidrome 默认返回 HTTPS 流 URL（含密码参数），ESP32 处理不了。桥接服务在 /api/stream/{song_id} 做了一个 HTTP 代理：

  • 接收 ESP32 的 HTTP 请求
  • 桥接服务内部用 Subsonic API 从 Navidrome 拉取音频流
  • 加 &format=mp3 参数请求转码
  • 以 audio/mpeg 流式返回给 ESP32

这样 ESP32 只需访问一个简单的 HTTP URL，不需要处理任何认证。

歌词显示支持

为了让 ESP32 播放音乐时能同步显示歌词，桥接服务新增了歌词代理接口。Navidrome 的 Subsonic API 提供了 getLyricsBySongId 端点返回结构化歌词，桥接服务将其转换为标准的 LRC 格式供设备解析。

LRC 格式示例：

[00:13.97]风吹过我的双脚
[00:19.30]怀念夏天的味道
[00:24.56]你的微笑我舍不得一口吃掉

每个 MCP 工具返回的歌曲信息中，新增了 lyric_url 字段，指向歌词代理接口：

http://桥接IP:8651/api/lyrics/{song_id}

xiaozhi 设备端收到播放指令时，调用 self.music.play_url(url, title, artist, lyric, lyric_url) 即可将歌词 URL 传递给播放器，设备会自动请求并解析 LRC 歌词，在屏幕上同步滚动显示。

歌词接口内部处理流程：

  1. 调用 Navidrome getLyricsBySongId 获取结构化歌词（含每行歌词和起始时间戳，单位毫秒）
  2. 解析 structuredLyrics 中的 line 数组
  3. 将每行时间戳从毫秒转为 [MM:SS.CC] 格式
  4. 返回纯文本 LRC 内容

如果歌曲没有歌词数据，接口返回空内容，不影响正常播放。

Subsonic API 集成

Navidrome 实现了 Subsonic API，这是成熟的音乐服务器协议。桥接服务通过以下端点与 Navidrome 交互：

  • search3 — 全文搜索（歌手/专辑/歌曲）
  • getRandomSongs — 随机歌曲
  • stream — 获取音频流
  • getLyricsBySongId — 获取歌词
  • ping — 健康检查

Subsonic API 使用 HTTP Basic Auth + 查询参数认证，简单可靠。

配置要点

xiaozhi 的配置关键修改：

  • LLM 配置接入 DeepSeek
  • MCP 注册桥接 SSE 端点
  • TTS 配置本地引擎

注意：Navidrome 流 URL 返回的是 HTTPS，ESP32 需要走本地 HTTP 代理中转。

效果

现在对着 ESP32 说「随便放首歌」或者「播放周杰伦的歌」，小智会通过 LLM 理解意图 → 搜索 Navidrome 曲库 → 返回歌曲信息 → 播放音乐 → 屏幕同步滚动显示歌词。整个过程从说话到出声大约几秒钟，体验还不错。

#!/usr/bin/env python3
"""
MCP 桥接服务 — Navidrome 音乐搜索 + 歌词
"""
import asyncio, json, os, urllib.parse
import httpx, uvicorn
from fastapi import FastAPI
from mcp.server.fastmcp import FastMCP
from starlette.responses import StreamingResponse

NAVIDROME_BASE = os.getenv("NAVIDROME_BASE", "https://navidrome.example.com")
NAVIDROME_USER = os.getenv("NAVIDROME_USER", "user")
NAVIDROME_PASS = os.getenv("NAVIDROME_PASS", "pass")

def _subsonic_url(endpoint, **params):
    p = {"u": NAVIDROME_USER, "p": NAVIDROME_PASS, "v": "1.16.0", "c": "mcp-bridge"}
    p.update(params)
    return f"{NAVIDROME_BASE}/rest/{endpoint}?{urllib.parse.urlencode(p)}"

async def _subsonic_get(endpoint, **params):
    import xml.etree.ElementTree as ET
    async with httpx.AsyncClient(timeout=15) as c:
        resp = await c.get(_subsonic_url(endpoint, **params))
        resp.raise_for_status()
        root = ET.fromstring(resp.text)
        songs = []
        for song in root.iter("{http://subsonic.org/restapi}song"):
            songs.append({"id": song.get("id"), "title": song.get("title"),
                          "artist": song.get("artist"), "album": song.get("album"),
                          "duration": int(song.get("duration", 0))})
        return {"songs": songs}

def _stream_url(song_id):
    return _subsonic_url("stream", id=song_id)

# MCP 服务
mcp = FastMCP("mcp-bridge", host="0.0.0.0", port=8650)

@mcp.tool()
async def navidrome_search(query, limit=5):
    """搜索 Navidrome 曲库"""
    result = await _subsonic_get("search3", query=query, songCount=limit)
    return json.dumps(result.get("songs", []), ensure_ascii=False) or "未找到"

@mcp.tool()
async def navidrome_get_stream_url(song_id):
    """获取歌曲的播放信息，包括音频流URL和歌词URL。
    返回: {"url": "...", "lyric_url": "..."}
    xiaozhi收到后应调用 self.music.play_url(url, title, artist, lyric, lyric_url) 播放。
    """
    stream_url = f"http://桥接IP:8651/api/stream/{song_id}"
    lyric_url = f"http://桥接IP:8651/api/lyrics/{song_id}"
    return json.dumps({"url": stream_url, "lyric_url": lyric_url})

@mcp.tool()
async def navidrome_random_song():
    """随机获取一首歌曲的播放信息，含歌词URL"""
    result = await _subsonic_get("getRandomSongs", size=1)
    songs = result.get("songs", [])
    if songs:
        s = songs[0]
        s["stream_url"] = f"http://桥接IP:8651/api/stream/{s['id']}"
        s["lyric_url"] = f"http://桥接IP:8651/api/lyrics/{s['id']}"
        return json.dumps(s, ensure_ascii=False)
    return "曲库为空"

# FastAPI — 音频流 + 歌词代理
api_app = FastAPI()

@api_app.get("/api/stream/{song_id}")
async def stream_proxy(song_id):
    """代理 Navidrome 音频流"""
    url = _subsonic_url("stream", id=song_id) + "&format=mp3"
    async def audio_stream():
        async with httpx.AsyncClient(timeout=300) as c:
            async with c.stream("GET", url) as resp:
                async for chunk in resp.aiter_bytes():
                    yield chunk
    return StreamingResponse(audio_stream(), media_type="audio/mpeg")

@api_app.get("/api/lyrics/{song_id}")
async def lyrics_proxy(song_id):
    """代理 Navidrome 歌词，返回 LRC 格式文本"""
    url = _subsonic_url("getLyricsBySongId", id=song_id) + "&f=json"
    async with httpx.AsyncClient(timeout=15) as c:
        resp = await c.get(url)
        resp.raise_for_status()
        data = resp.json()

    lyrics_list = (
        data.get("subsonic-response", {})
            .get("lyricsList", {})
            .get("structuredLyrics", [])
    )
    if isinstance(lyrics_list, dict):
        lyrics_list = [lyrics_list]

    from starlette.responses import PlainTextResponse
    for item in lyrics_list:
        lines = item.get("line", [])
        if isinstance(lines, dict):
            lines = [lines]
        lrc_lines = []
        for line in lines:
            start = line.get("start")
            text = line.get("value") or line.get("#text") or ""
            if start is None or not text:
                continue
            ms = int(start)
            lrc_lines.append(
                f"[{ms//60000:02d}:{ms%60000//1000:02d}.{ms%1000//10:02d}]{text}"
            )
        if lrc_lines:
            return PlainTextResponse("\n".join(lrc_lines))
    return PlainTextResponse("")

def run():
    import threading
    t = threading.Thread(
        target=lambda: uvicorn.run(api_app, host="0.0.0.0", port=8651),
        daemon=True
    )
    t.start()
    asyncio.run(mcp.run_sse_async())

if __name__ == "__main__":
    run()

核心依赖：fastmcp、FastAPI、uvicorn、httpx。

MCP 作为 LLM 的「工具箱」，让语音助手具备了调用外部服务的能力。桥接模式很好地解决了 HTTPS 认证和格式转码的问题，给 ESP32 语音助手加了一个实用的音乐播放技能。加上歌词代理后，播放体验更完整了——不止能听歌，还能看歌词。