Add Quickdial plugin for STT and TTS

.github/next-release/changeset-quickdial-plugin.md

@@ -0,0 +1,5 @@
+---
+"livekit-plugins-quickdial": minor
+---
+
+Add Quickdial plugin for STT and TTS. Provides speech-to-text and text-to-speech via the Quickdial API (CPU-optimized, real-time, priced per character).

examples/voice_agents/quickdial_agent.py

@@ -0,0 +1,42 @@
+import logging
+
+from dotenv import load_dotenv
+
+from livekit.agents import Agent, AgentSession, JobContext, WorkerOptions, cli
+from livekit.plugins import openai, quickdial, silero
+
+# Quickdial provides cheap, fast, real-time STT + TTS on CPU (priced per character).
+# Set QUICKDIAL_API_KEY in your environment (get a key at https://web.quickdial.ai).
+
+logger = logging.getLogger("quickdial-agent")
+
+load_dotenv()
+
+
+class MyAgent(Agent):
+    def __init__(self) -> None:
+        super().__init__(
+            instructions=(
+                "You are a friendly voice assistant powered by Quickdial. "
+                "Keep replies short, natural, and conversational. "
+                "Do not use emojis, asterisks, or markdown."
+            )
+        )
+
+    async def on_enter(self) -> None:
+        self.session.generate_reply(instructions="Warmly greet the user and ask how you can help.")
+
+
+async def entrypoint(ctx: JobContext):
+    session = AgentSession(
+        stt=quickdial.STT(language="en"),
+        llm=openai.LLM(model="gpt-4o-mini"),
+        tts=quickdial.TTS(voice="alba"),
+        vad=silero.VAD.load(),
+    )
+
+    await session.start(agent=MyAgent(), room=ctx.room)
+
+
+if __name__ == "__main__":
+    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))

livekit-agents/pyproject.toml

@@ -107,6 +107,7 @@ nvidia = ["livekit-plugins-nvidia>=1.6.6"]
 openai = ["livekit-plugins-openai>=1.6.6"]
 perplexity = ["livekit-plugins-perplexity>=1.6.6"]
 protoface = ["livekit-plugins-protoface>=1.6.6"]
+quickdial = ["livekit-plugins-quickdial>=1.6.6"]
 resemble = ["livekit-plugins-resemble>=1.6.6"]
 respeecher = ["livekit-plugins-respeecher>=1.6.6"]
 rime = ["livekit-plugins-rime>=1.6.6"]

livekit-plugins/livekit-plugins-quickdial/README.md

@@ -0,0 +1,57 @@
+# LiveKit Plugins Quickdial
+
+[Quickdial](https://quickdial.ai/) plugin for LiveKit Agents. Quickdial is a
+Real-time voice API — lifelike text-to-speech and accurate
+speech-to-text over REST + WebSocket, priced per character.
+
+See [https://docs.livekit.io/agents/integrations/](https://docs.livekit.io/agents/integrations/) for more information.
+
+## Installation
+
+```bash
+pip install livekit-plugins-quickdial
+```
+
+## Pre-requisites
+
+You'll need an API key from Quickdial. Sign up at
+[web.quickdial.ai](https://web.quickdial.ai) (1000 free credits, no card
+required) and set it as an environment variable: `QUICKDIAL_API_KEY`.
+
+## Usage
+
+```python
+from livekit.agents import AgentSession
+from livekit.plugins import quickdial, silero
+
+session = AgentSession(
+    stt=quickdial.STT(language="en"),   # POST /v1/stt (whisper.cpp)
+    tts=quickdial.TTS(voice="alba"),    # POST /v1/tts, 24 kHz
+    vad=silero.VAD.load(),
+    llm=...,
+)
+```
+
+## Parameters
+
+### `quickdial.TTS`
+
+| Parameter     | Default                    | Description                                             |
+| ------------- | -------------------------- | ------------------------------------------------------- |
+| `voice`       | `"alba"`                   | Voice name (see `GET /v1/voices`).                      |
+| `api_key`     | `QUICKDIAL_API_KEY`        | Quickdial API key.                                      |
+| `base_url`    | `https://api.quickdial.ai` | API base URL.                                           |
+| `sample_rate` | `24000`                    | Output sample rate (Hz).                                |
+
+### `quickdial.STT`
+
+| Parameter     | Default                    | Description               |
+| ------------- | -------------------------- | ------------------------- |
+| `language`    | `"en"`                     | Transcription language.   |
+| `api_key`     | `QUICKDIAL_API_KEY`        | Quickdial API key.        |
+| `base_url`    | `https://api.quickdial.ai` | API base URL.             |
+
+## Additional resources
+
+- [Quickdial API docs](https://quickdial.ai/docs)
+- [Source repository](https://github.com/samay-ai/livekit-plugins-quickdial)

livekit-plugins/livekit-plugins-quickdial/livekit/plugins/quickdial/__init__.py

@@ -0,0 +1,43 @@
+# Copyright 2026 Samay AI (Quickdial)
+# Licensed under the Apache License, Version 2.0
+"""Quickdial plugin for LiveKit Agents — cheap, fast, real-time TTS & STT.
+
+from livekit.plugins import quickdial
+
+session = AgentSession(
+    tts=quickdial.TTS(voice="alba"),
+    stt=quickdial.STT(language="en"),
+    vad=silero.VAD.load(),
+    llm=...,
+)
+"""
+
+from livekit.agents import Plugin
+
+from .log import logger
+from .stt import STT
+from .tts import TTS, ChunkedStream
+from .version import __version__
+
+__all__ = [
+    "TTS",
+    "STT",
+    "ChunkedStream",
+    "__version__",
+]
+
+
+class QuickdialPlugin(Plugin):
+    def __init__(self) -> None:
+        super().__init__(__name__, __version__, __package__, logger)
+
+
+Plugin.register_plugin(QuickdialPlugin())
+
+# hide internal submodules from `dir()` per LiveKit plugin convention
+_module = dir()
+NOT_IN_ALL = [m for m in _module if m not in __all__]
+
+__pdoc__ = {}
+for _n in NOT_IN_ALL:
+    __pdoc__[_n] = False

livekit-plugins/livekit-plugins-quickdial/livekit/plugins/quickdial/log.py

@@ -0,0 +1,3 @@
+import logging
+
+logger = logging.getLogger("livekit.plugins.quickdial")

livekit-plugins/livekit-plugins-quickdial/livekit/plugins/quickdial/models.py

@@ -0,0 +1,21 @@
+from typing import Literal
+
+# Output containers Quickdial can return over REST.
+TTSEncoding = Literal["pcm", "wav", "opus"]
+
+# A few well-known voices (see GET /v1/voices for the full, live list).
+TTSVoices = Literal[
+    "alba",
+    "jane",
+    "charles",
+    "anna",
+    "george",
+    "vera",
+    "estelle",
+    "giovanni",
+    "juergen",
+    "lola",
+    "rafael",
+]
+
+STTLanguages = Literal["en", "fr", "de", "es", "it", "pt"]

livekit-plugins/livekit-plugins-quickdial/livekit/plugins/quickdial/stt.py

@@ -0,0 +1,122 @@
+# Copyright 2026 Samay AI (Quickdial)
+# Licensed under the Apache License, Version 2.0
+"""Quickdial STT for LiveKit Agents.
+
+Non-streaming transcription over ``POST /v1/stt`` (multipart WAV) via
+``_recognize_impl``. The STT declares ``streaming=False``; pair it with a VAD
+(e.g. ``silero.VAD``) so the AgentSession segments speech and calls
+``_recognize_impl`` per utterance, emitting a FINAL_TRANSCRIPT (no interim
+results yet). This mirrors the OpenAI Whisper STT plugin.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+from dataclasses import dataclass
+
+import aiohttp
+
+from livekit import rtc
+from livekit.agents import (
+    APIConnectionError,
+    APIConnectOptions,
+    APIStatusError,
+    APITimeoutError,
+    stt,
+    utils,
+)
+from livekit.agents.types import DEFAULT_API_CONNECT_OPTIONS, NOT_GIVEN, NotGivenOr
+from livekit.agents.utils import AudioBuffer, is_given
+
+from .models import STTLanguages
+
+DEFAULT_BASE_URL = "https://api.quickdial.ai"
+
+
+@dataclass
+class _STTOptions:
+    language: str
+    base_url: str
+    api_key: str
+    params: dict | None
+
+
+class STT(stt.STT):
+    def __init__(
+        self,
+        *,
+        language: STTLanguages | str = "en",
+        api_key: NotGivenOr[str] = NOT_GIVEN,
+        base_url: str = DEFAULT_BASE_URL,
+        params: NotGivenOr[dict] = NOT_GIVEN,
+        http_session: aiohttp.ClientSession | None = None,
+    ) -> None:
+        # Quickdial returns a transcript per utterance (no interim results), so we run
+        # as a NON-streaming STT: the AgentSession's VAD segments speech and calls
+        # _recognize_impl (POST /v1/stt) per utterance.
+        super().__init__(capabilities=stt.STTCapabilities(streaming=False, interim_results=False))
+        key = api_key if is_given(api_key) else os.environ.get("QUICKDIAL_API_KEY", "")
+        if not key:
+            raise ValueError("Quickdial API key required — pass api_key= or set QUICKDIAL_API_KEY")
+        self._opts = _STTOptions(
+            language=language,
+            base_url=base_url.rstrip("/"),
+            api_key=key,
+            params=params if is_given(params) else None,
+        )
+        self._session = http_session
+
+    @property
+    def provider(self) -> str:
+        return "Quickdial"
+
+    def _ensure_session(self) -> aiohttp.ClientSession:
+        if not self._session:
+            self._session = utils.http_context.http_session()
+        return self._session
+
+    async def _recognize_impl(
+        self,
+        buffer: AudioBuffer,
+        *,
+        language: NotGivenOr[str] = NOT_GIVEN,
+        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
+    ) -> stt.SpeechEvent:
+        wav = rtc.combine_audio_frames(buffer).to_wav_bytes()
+        form = aiohttp.FormData()
+        form.add_field("audio", wav, filename="audio.wav", content_type="audio/wav")
+        cfg = dict(self._opts.params or {})
+        cfg["language"] = language if is_given(language) else self._opts.language
+        form.add_field("params", json.dumps(cfg))
+        try:
+            async with self._ensure_session().post(
+                f"{self._opts.base_url}/v1/stt",
+                headers={"Authorization": f"Bearer {self._opts.api_key}"},
+                data=form,
+                timeout=aiohttp.ClientTimeout(total=30, sock_connect=conn_options.timeout),
+            ) as resp:
+                resp.raise_for_status()
+                data = await resp.json()
+                # tag the transcript with the *effective* language (per-request override,
+                # falling back to the configured default) when the server omits it
+                return _to_speech_event(data, cfg["language"])
+        except asyncio.TimeoutError as e:
+            raise APITimeoutError() from e
+        except aiohttp.ClientResponseError as e:
+            raise APIStatusError(message=e.message, status_code=e.status) from e
+        except Exception as e:
+            raise APIConnectionError() from e
+
+
+def _to_speech_event(data: dict, language: str) -> stt.SpeechEvent:
+    return stt.SpeechEvent(
+        type=stt.SpeechEventType.FINAL_TRANSCRIPT,
+        alternatives=[
+            stt.SpeechData(
+                language=data.get("language", language),
+                text=data.get("text", ""),
+            )
+        ],
+    )