Hermes
139 lines
4.1 KiB
Python
139 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Voice MCP Server — STT via Moonshine, TTS via KittenTTS.
|
|
|
|
HTTP transport for Hermes MCP client. Tools:
|
|
voice_stt(file_path, language) → transcription text
|
|
voice_tts(text, voice, speed) → path to generated .wav
|
|
voice_pipeline(audio_path) → full STT → TTS roundtrip
|
|
"""
|
|
|
|
import hashlib
|
|
import os
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
app = FastMCP("voice")
|
|
|
|
# ── Lazy model loading (RAM-friendly) ────────────────────
|
|
|
|
_stt = None
|
|
_tts = None
|
|
_model_dir = Path(os.environ.get("MODEL_DIR", "/models"))
|
|
|
|
|
|
def get_stt():
|
|
global _stt
|
|
if _stt is None:
|
|
from moonshine_voice import Transcriber, ModelArch
|
|
from moonshine_voice.download import get_model_for_language
|
|
model_path, model_arch = get_model_for_language(
|
|
"en", ModelArch.TINY, cache_root=str(_model_dir)
|
|
)
|
|
_stt = Transcriber(model_path=model_path, model_arch=model_arch)
|
|
return _stt
|
|
|
|
|
|
def get_tts():
|
|
global _tts
|
|
if _tts is None:
|
|
from kittentts import KittenTTS
|
|
_tts = KittenTTS(
|
|
"KittenML/kitten-tts-nano-0.8-int8",
|
|
cache_dir=str(_model_dir),
|
|
)
|
|
return _tts
|
|
|
|
|
|
# ── Tools ────────────────────────────────────────────────
|
|
|
|
@app.tool()
|
|
async def voice_stt(file_path: str, language: str = "auto") -> str:
|
|
"""Transcribe speech audio to text using Moonshine.
|
|
|
|
Args:
|
|
file_path: Absolute path to audio file (.wav)
|
|
language: Language code or 'auto' (default: auto)
|
|
Returns:
|
|
Transcribed text
|
|
"""
|
|
from moonshine_voice.utils import load_wav_file
|
|
stt = get_stt()
|
|
audio_data, sample_rate = load_wav_file(file_path)
|
|
result = stt.transcribe_without_streaming(audio_data, sample_rate)
|
|
return " ".join(l.text for l in result.lines)
|
|
|
|
|
|
@app.tool()
|
|
async def voice_tts(
|
|
text: str,
|
|
voice: str = "Jasper",
|
|
speed: float = 1.0,
|
|
) -> str:
|
|
"""Convert text to speech using KittenTTS.
|
|
|
|
Args:
|
|
text: Text to speak
|
|
voice: One of Bella, Jasper, Luna, Bruno, Rosie, Hugo, Kiki, Leo
|
|
speed: Speech speed (0.5-2.0, default 1.0)
|
|
Returns:
|
|
Absolute path to generated .wav file
|
|
"""
|
|
tts = get_tts()
|
|
|
|
# Stable filename based on content hash
|
|
h = hashlib.sha256(f"{text}_{voice}_{speed}".encode()).hexdigest()[:12]
|
|
out_path = os.path.join(tempfile.gettempdir(), f"tts_{h}.wav")
|
|
|
|
tts.generate_to_file(text, out_path, voice=voice, speed=speed)
|
|
return out_path
|
|
|
|
|
|
@app.tool()
|
|
async def voice_pipeline(audio_path: str, tts_voice: str = "Jasper") -> str:
|
|
"""Full voice pipeline: STT → echo back as TTS.
|
|
Useful for testing the roundtrip.
|
|
|
|
Args:
|
|
audio_path: Path to input audio file
|
|
tts_voice: Voice for TTS response
|
|
Returns:
|
|
JSON with transcribed text and path to response audio
|
|
"""
|
|
import json
|
|
|
|
# STT
|
|
from moonshine_voice.utils import load_wav_file
|
|
stt = get_stt()
|
|
audio_data, sample_rate = load_wav_file(audio_path)
|
|
result = stt.transcribe_without_streaming(audio_data, sample_rate)
|
|
text = " ".join(l.text for l in result.lines)
|
|
|
|
# TTS
|
|
tts = get_tts()
|
|
h = hashlib.sha256(text.encode()).hexdigest()[:12]
|
|
out_path = os.path.join(tempfile.gettempdir(), f"tts_{h}.wav")
|
|
tts.generate_to_file(text, out_path, voice=tts_voice)
|
|
|
|
return json.dumps({
|
|
"transcription": text,
|
|
"response_audio": out_path,
|
|
})
|
|
|
|
|
|
@app.tool()
|
|
async def voice_available_voices() -> str:
|
|
"""List available TTS voices."""
|
|
tts = get_tts()
|
|
return str(tts.available_voices)
|
|
|
|
|
|
# ── Entry Point ──────────────────────────────────────────
|
|
|
|
if __name__ == "__main__":
|
|
import sys, uvicorn
|
|
port = int(os.environ.get("PORT", "8000"))
|
|
print(f"Voice MCP starting on port {port}...", file=sys.stderr)
|
|
uvicorn.run(app.streamable_http_app(), host="0.0.0.0", port=port)
|