feat(voice): local Whisper transcription with pluggable backends

README.md

@@ -30,6 +30,7 @@ Message sent successfully:
 - [Quick Start](#quick-start)
 - [MCP Client Configuration](#mcp-client-configuration)
 - [Multi-Account Setup](#multi-account-setup)
+- [Voice Transcription](#voice-transcription)
 - [Proxy Support](#proxy-support)
 - [File Path Security](#file-path-security)
 - [Docker](#docker)
@@ -49,6 +50,7 @@ The server currently includes 80+ MCP tools grouped into these areas:
 - **Media:** send files, download media, upload files, send voice notes, stickers, GIFs, and inspect message media.
 - **Profile and privacy:** get your own account info, update profile fields, set or delete profile photos, inspect privacy settings, get user info/photos/status, and manage bot commands.
 - **Folders and drafts:** list, create, update, reorder, and delete Telegram folders; save, list, and clear drafts.
+- **Voice transcription (optional):** download a voice/audio message and transcribe it locally with Whisper (Intel CPU/Iris/Arc, NVIDIA CUDA, Apple Silicon, or CPU fallback). See [Voice Transcription](#voice-transcription).
 
 All tool results that include Telegram user-controlled content are sanitized and, where practical, returned as structured JSON.
 
@@ -167,6 +169,81 @@ Example prompts:
 - "Show unread messages from all accounts"
 - "Send this from my work account to @example"
 
+## Voice Transcription
+
+Voice and audio messages can be transcribed locally with Whisper. Audio
+never leaves your host — the model runs against whichever accelerator
+your machine has. Two MCP tools are exposed:
+
+- `transcribe_voice(chat_id, message_id, language=None)` — downloads the
+  voice/audio attachment and returns transcribed text.
+- `voice_transcription_info()` — reports the active backend, device,
+  model, and config.
+
+### Install one of the backend extras
+
+The transcription dependencies are **optional** — pick the one that
+matches your hardware. Without any of them installed the tools return a
+clear `transcription_unavailable` error and the rest of the server keeps
+working unchanged.
+
+| Hardware | Extras | Backend |
+|---|---|---|
+| Universal CPU on any OS, NVIDIA CUDA | `pip install -e ".[voice]"` | `faster-whisper` (CTranslate2) |
+| Intel CPU / Iris Xe iGPU / Arc dGPU | `pip install -e ".[voice-openvino]"` | OpenVINO GenAI |
+| Apple Silicon (M1/M2/M3/M4) | `pip install -e ".[voice-mlx]"` | `mlx-whisper` |
+
+You can install several extras at once — `auto`-detection picks the best
+available accelerator at startup. If only CPU is available the server
+emits a one-time warning ("будет долго" / `whisper_cpu_only`) — disable
+with `WHISPER_WARN_CPU=false`.
+
+### Configuration
+
+All settings are env vars. Defaults are sensible for most setups.
+
+| Var | Default | Meaning |
+|---|---|---|
+| `WHISPER_ENABLED` | `true` | Set to `false` to disable transcription tools entirely. |
+| `WHISPER_BACKEND` | `auto` | `auto` / `faster_whisper` / `openvino` / `mlx`. |
+| `WHISPER_DEVICE` | `auto` | `auto` / `cpu` / `cuda` / `gpu` / `gpu.0` / `gpu.1` … |
+| `WHISPER_MODEL` | `base` | `tiny` / `base` / `small` / `medium` / `large-v3` / `large-v3-turbo`. |
+| `WHISPER_LANGUAGE` | _(unset = auto-detect)_ | ISO 639-1 code (`ru`, `en`, …). |
+| `WHISPER_WARN_CPU` | `true` | Log a one-time warning when running CPU-only. |
+| `WHISPER_CACHE_DIR` | `~/.cache/telegram-mcp/whisper` | Where weights are downloaded. |
+
+### Auto-detect priority
+
+1. Apple Silicon → `mlx-whisper` (MPS).
+2. NVIDIA CUDA → `faster-whisper` (`cuda`, fp16).
+3. Intel discrete GPU (Arc) or iGPU (Iris/UHD) → OpenVINO (`GPU.N`).
+4. CPU fallback via `faster-whisper` (or OpenVINO if it's the only one
+   installed).
+
+Performance reference (Whisper-base, 11 s English clip):
+
+| Device | Inference | × realtime |
+|---|---|---|
+| Intel Arc A550M (OpenVINO) | ~160 ms | 67× |
+| Intel Iris Xe (OpenVINO) | ~330 ms | 33× |
+| i7-12700H, 20 threads (OpenVINO CPU) | ~600 ms | 18× |
+
+Larger models (`large-v3-turbo` etc.) are correspondingly slower; pick
+the smallest model that gives you acceptable accuracy for your language
+and noise level.
+
+### Example
+
+```text
+> Find the latest voice message in my Saved Messages and transcribe it
+  in Russian.
+
+  list_messages(chat_id="me", limit=20, with_media=true)
+  → finds a voice message with id=12345
+  transcribe_voice(chat_id="me", message_id=12345, language="ru")
+  → "Привет, проверка транскрипции через локальный Whisper."
+```
+
 ## Proxy Support
 
 Route Telegram traffic through a proxy by setting the `TELEGRAM_PROXY_*`

pyproject.toml

@@ -39,13 +39,38 @@ proxy = [
     "python-socks>=2.4.3",
 ]
 
+# Voice transcription (universal): faster-whisper covers CPU on every OS and
+# NVIDIA CUDA. See README "Voice transcription" for backend selection.
+voice = [
+    "faster-whisper>=1.1.0",
+]
+
+# Voice transcription on Intel CPU / Iris Xe iGPU / Arc dGPU via OpenVINO.
+voice-openvino = [
+    "openvino>=2025.0.0",
+    "openvino-genai>=2025.0.0.0",
+    "librosa>=0.10",
+    "huggingface-hub>=0.26",
+    "numpy>=1.24",
+]
+
+# Voice transcription on Apple Silicon (M1/M2/M3/M4) via Apple's MLX.
+voice-mlx = [
+    "mlx-whisper>=0.4",
+]
+
 [project.urls]
 "Homepage" = "https://github.com/chigwell/telegram-mcp"
 "Bug Tracker" = "https://github.com/chigwell/telegram-mcp/issues"
 
 [tool.setuptools]
 py-modules = ["main", "sanitize", "session_string_generator"]
-packages = ["telegram_mcp", "telegram_mcp.tools"]
+packages = [
+    "telegram_mcp",
+    "telegram_mcp.tools",
+    "telegram_mcp.voice",
+    "telegram_mcp.voice.backends",
+]
 
 [project.scripts]
 telegram-mcp = "main:main"

telegram_mcp/tools/__init__.py

@@ -8,5 +8,6 @@
 from telegram_mcp.tools.media import *
 from telegram_mcp.tools.profile import *
 from telegram_mcp.tools.folders import *
+from telegram_mcp.tools.voice import *
 
 __all__ = [name for name in globals() if not name.startswith("_")]

telegram_mcp/tools/voice.py

@@ -0,0 +1,100 @@
+"""Voice transcription MCP tools (local Whisper)."""
+
+from __future__ import annotations
+
+import json
+import tempfile
+from pathlib import Path
+
+from telegram_mcp import voice
+from telegram_mcp.runtime import *  # noqa: F401,F403
+
+
+@mcp.tool(
+    annotations=ToolAnnotations(
+        title="Transcribe Voice",
+        readOnlyHint=True,
+        openWorldHint=False,
+    )
+)
+@with_account(readonly=True)
+@validate_id("chat_id")
+async def transcribe_voice(
+    chat_id: Union[int, str],
+    message_id: int,
+    language: Optional[str] = None,
+    account: str = None,
+) -> str:
+    """
+    Download a voice message and transcribe it locally via Whisper.
+
+    The active backend is auto-detected at startup (Apple MLX → NVIDIA CUDA →
+    Intel OpenVINO → CPU) and configurable via ``WHISPER_BACKEND`` /
+    ``WHISPER_DEVICE`` / ``WHISPER_MODEL`` env vars. No audio is uploaded
+    anywhere — transcription runs entirely on the host.
+
+    Args:
+        chat_id: The chat ID or username containing the voice message.
+        message_id: The message ID with the voice (or audio) attachment.
+        language: ISO 639-1 code (e.g. ``"ru"``, ``"en"``). ``None`` =
+            auto-detect from audio (slightly slower).
+
+    Returns:
+        Transcribed text, or a JSON error blob if transcription is
+        unavailable, the message has no voice attachment, or download fails.
+    """
+    try:
+        cl = get_client(account)
+        await ensure_connected(cl)
+        entity = await resolve_entity(chat_id, cl)
+        msg = await cl.get_messages(entity, ids=message_id)
+        if msg is None:
+            return f"Message {message_id} in chat {chat_id} not found."
+        if not (msg.voice or msg.audio):
+            return (
+                f"Message {message_id} in chat {chat_id} has no voice/audio attachment "
+                "(only voice notes and audio messages can be transcribed)."
+            )
+
+        with tempfile.TemporaryDirectory(prefix="telegram-mcp-voice-") as td:
+            target = Path(td) / "voice"  # Telethon will append correct extension.
+            downloaded = await cl.download_media(msg, file=str(target))
+            if not downloaded:
+                return f"Failed to download voice from message {message_id}."
+            try:
+                text = voice.transcribe(Path(downloaded), language=language)
+            except voice.VoiceTranscriptionUnavailable as exc:
+                return json.dumps(
+                    {
+                        "error": "transcription_unavailable",
+                        "message": str(exc),
+                        "config": voice.get_backend_info(),
+                    },
+                    ensure_ascii=False,
+                )
+        return text
+    except Exception as e:
+        return log_and_format_error(
+            "transcribe_voice",
+            e,
+            chat_id=chat_id,
+            message_id=message_id,
+            language=language,
+        )
+
+
+@mcp.tool(
+    annotations=ToolAnnotations(
+        title="Voice Transcription Info",
+        readOnlyHint=True,
+        openWorldHint=False,
+    )
+)
+async def voice_transcription_info() -> str:
+    """
+    Report current voice transcription configuration and active backend.
+
+    Useful to verify which Whisper backend/device the server picked up
+    (Intel Arc, NVIDIA CUDA, Apple MPS, or CPU fallback).
+    """
+    return json.dumps(voice.get_backend_info(), ensure_ascii=False, indent=2)

telegram_mcp/voice/__init__.py

@@ -0,0 +1,116 @@
+"""Local Whisper voice transcription with pluggable backends.
+
+Public API:
+    transcribe(audio_path, language=None) -> str
+    get_backend_info() -> dict   (for introspection)
+    VoiceTranscriptionUnavailable
+
+Backend selection is driven by env vars (see config.py) and runtime hardware
+detection (see detect.py). Backends are loaded lazily so users only pay for
+the deps they have installed.
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+from pathlib import Path
+from typing import Any, Dict, Optional
+
+from .backends.base import VoiceTranscriptionUnavailable, WhisperBackend
+from .config import WhisperConfig, get_config
+from .detect import resolve_backend
+
+logger = logging.getLogger(__name__)
+
+_backend: Optional[WhisperBackend] = None
+_backend_lock = threading.Lock()
+_warned_cpu = False
+
+
+def _build_backend(name: str, model: str, device: str, cache_dir: Path) -> WhisperBackend:
+    if name == "faster_whisper":
+        from .backends.faster_whisper import FasterWhisperBackend
+
+        return FasterWhisperBackend(model=model, device=device, cache_dir=cache_dir)
+    if name == "openvino":
+        from .backends.openvino import OpenVINOBackend
+
+        return OpenVINOBackend(model=model, device=device, cache_dir=cache_dir)
+    if name == "mlx":
+        from .backends.mlx import MLXBackend
+
+        return MLXBackend(model=model, device=device, cache_dir=cache_dir)
+    raise VoiceTranscriptionUnavailable(f"Unknown whisper backend: {name}")
+
+
+def _maybe_warn_cpu(cfg: WhisperConfig, device: str) -> None:
+    global _warned_cpu
+    if _warned_cpu:
+        return
+    if device.lower() != "cpu":
+        return
+    if not cfg.warn_cpu:
+        return
+    logger.warning(
+        "whisper_cpu_only: транскрипция работает на CPU без GPU-ускорения — будет долго. "
+        "Поставь optional extras для своего железа: "
+        "[voice-openvino] (Intel CPU/iGPU/Arc), [voice] с CUDA (NVIDIA), [voice-mlx] (Apple Silicon). "
+        "Отключить варнинг: WHISPER_WARN_CPU=false. Отключить транскрипцию: WHISPER_ENABLED=false."
+    )
+    _warned_cpu = True
+
+
+def get_backend() -> WhisperBackend:
+    """Return the active backend, building it on first access."""
+    global _backend
+    if _backend is not None:
+        return _backend
+    with _backend_lock:
+        if _backend is not None:
+            return _backend
+        cfg = get_config()
+        if not cfg.enabled:
+            raise VoiceTranscriptionUnavailable(
+                "Voice transcription is disabled (WHISPER_ENABLED=false)."
+            )
+        backend_name, device = resolve_backend(cfg)
+        _maybe_warn_cpu(cfg, device)
+        logger.info(
+            "whisper_backend_selected", extra={"backend": backend_name, "device": device, "model": cfg.model}
+        )
+        _backend = _build_backend(backend_name, cfg.model, device, cfg.cache_dir)
+        return _backend
+
+
+def transcribe(audio_path: Path, language: Optional[str] = None) -> str:
+    """Transcribe an audio file. Raises VoiceTranscriptionUnavailable if disabled/unsupported."""
+    cfg = get_config()
+    lang = language if language is not None else cfg.language
+    return get_backend().transcribe(Path(audio_path), language=lang)
+
+
+def get_backend_info() -> Dict[str, Any]:
+    """Return info about the active or to-be-built backend, without forcing instantiation if not yet built."""
+    cfg = get_config()
+    info: Dict[str, Any] = {
+        "enabled": cfg.enabled,
+        "backend": cfg.backend,
+        "device": cfg.device,
+        "model": cfg.model,
+        "language": cfg.language,
+        "warn_cpu": cfg.warn_cpu,
+        "cache_dir": str(cfg.cache_dir),
+    }
+    if _backend is not None:
+        info["active_backend"] = _backend.name
+        info["active_device"] = _backend.device
+    return info
+
+
+__all__ = [
+    "transcribe",
+    "get_backend",
+    "get_backend_info",
+    "VoiceTranscriptionUnavailable",
+]

telegram_mcp/voice/backends/__init__.py

@@ -0,0 +1,9 @@
+"""Whisper backend implementations.
+
+Each backend module is imported lazily via the facade in
+`telegram_mcp.voice` so users only pay for the deps they have installed.
+"""
+
+from .base import VoiceTranscriptionUnavailable, WhisperBackend
+
+__all__ = ["VoiceTranscriptionUnavailable", "WhisperBackend"]