Custom speech providers

To use a speech provider that is not built in, write your own engine. Custom engines are in beta. You write a Python class that subclasses ASREngine or TTSEngine, and set name to its module path (path.to.module.ClassName). Every other key under asr: or tts: is passed to your class’s from_config_dict as a dictionary.

integrations.yml

channels:
  inspector:
    enabled: true
    asr:
      name: addons.custom_asr.MyASR
      endpoint: "wss://api.example.com/v1/speech"
      language_map:
        en:
          language: en-US
    tts:
      name: addons.custom_tts.MyTTS
      endpoint: "https://api.example.com/v1/speak"
      timeout: 20
      language_map:
        en:
          language: en-US
          voice: en-US-JennyNeural

Put the code in a package next to agent.yml, here addons/ with an __init__.py, and start rasa from the project root so the module can be imported.

Custom speech recognition

Subclass ASREngine from rasa.core.channels.voice_stream.asr.asr_engine. The engine receives the caller’s audio as RasaAudioBytes, which may need converting to the format your service expects, and turns the service’s messages into ASREvents.

addons/custom_asr.py

import json
import os
from typing import Any, Dict, List, Optional

from websockets.asyncio.client import ClientConnection, connect

from rasa.core.channels.voice_stream.asr.asr_engine import (
    ASREngine,
    ASREngineConfig,
    ASRLanguageMapEntry,
)
from rasa.core.channels.voice_stream.asr.asr_event import (
    ASREvent,
    NewTranscript,
    UserIsSpeaking,
)
from rasa.core.channels.voice_stream.audio_bytes import AudioFormat, RasaAudioBytes


class MyASRConfig(ASREngineConfig):
    endpoint: Optional[str] = None


class MyASR(ASREngine[MyASRConfig]):
    required_env_vars = ("MY_ASR_API_KEY",)

    @classmethod
    def name(cls) -> str:
        return "my_asr"

    @staticmethod
    def get_default_config(rasa_language: str) -> MyASRConfig:
        return MyASRConfig(
            endpoint="wss://api.example.com/v1/speech",
            language_map={rasa_language: ASRLanguageMapEntry(language="en-US")},
        )

    @classmethod
    def from_config_dict(
        cls,
        config: Dict,
        format: AudioFormat,
        rasa_language: str,
        additional_languages: Optional[List[str]] = None,
    ) -> "MyASR":
        return cls(
            rasa_language=rasa_language,
            format=format,
            config=MyASRConfig.model_validate(config),
            additional_languages=additional_languages,
        )

    async def open_websocket_connection(self) -> ClientConnection:
        language = self.current_language_config.engine_language_key
        return await connect(
            f"{self.config.endpoint}?language={language}",
            additional_headers={
                "Authorization": f"Bearer {os.environ['MY_ASR_API_KEY']}"
            },
        )

    async def signal_audio_done(self) -> None:
        if self.asr_socket:
            await self.asr_socket.send(json.dumps({"type": "end_of_audio"}))

    def rasa_audio_bytes_to_engine_bytes(self, chunk: RasaAudioBytes) -> bytes:
        return chunk.data  # convert here if your service needs another format

    def engine_event_to_asr_event(self, e: Any) -> Optional[ASREvent]:
        message = json.loads(e)
        if message.get("is_final"):
            return NewTranscript(message["text"])
        if message.get("text"):
            return UserIsSpeaking(message["text"])
        return None

Method	Required	Purpose
`name()`	yes	Class method returning a short engine name, used in logs and traces
`get_default_config(rasa_language)`	yes	Static method returning the default config, including a `language_map` entry for the agent’s language
`from_config_dict(config, format, rasa_language, additional_languages)`	yes	Class method that builds the engine from the dictionary in `integrations.yml`
`open_websocket_connection()`	yes	Open the connection to your service
`rasa_audio_bytes_to_engine_bytes(chunk)`	yes	Convert a `RasaAudioBytes` chunk to what your service expects
`signal_audio_done()`	yes	Tell your service that no more audio is coming
`engine_event_to_asr_event(event)`	yes	Convert a message from your service to an `ASREvent`, or return `None` to ignore it
`send_keep_alive()`	no	Keep the connection open. The default does nothing.
`close_connection()`	no	Extra cleanup when the call ends

engine_event_to_asr_event returns UserIsSpeaking(text) for interim transcripts and NewTranscript(text) for final ones, both from rasa.core.channels.voice_stream.asr.asr_event.

Emit a complete caller turn

Some services mark each recognized word or segment as final before the caller has finished. Do not map each of those segments to NewTranscript: the channel can process each event as a separate user turn. Accumulate the segments and emit a transcript at the provider’s end-of-utterance boundary. The is_final field in the example above assumes the example service finalizes a complete turn. Also handle short turns whose final message contains text but had no interim update. Test that sequence, successive turns, and empty provider messages offline. A missing interim transcript must not discard the words in a final message. See Test the speech path separately.

Custom speech synthesis

Subclass TTSEngine from rasa.core.channels.voice_stream.tts.tts_engine. The engine must produce RasaAudioBytes, in the audio format the channel passes to from_config_dict.

addons/custom_tts.py

from typing import AsyncIterator, Dict, List, Optional

from rasa.core.channels.voice_stream.audio_bytes import AudioFormat, RasaAudioBytes
from rasa.core.channels.voice_stream.tts.tts_engine import (
    TTSEngine,
    TTSEngineConfig,
    TTSLanguageMapEntry,
)


class MyTTSConfig(TTSEngineConfig):
    endpoint: Optional[str] = None


class MyTTS(TTSEngine[MyTTSConfig]):
    required_env_vars = ("MY_TTS_API_KEY",)

    @classmethod
    def name(cls) -> str:
        return "my_tts"

    @staticmethod
    def get_default_config(rasa_language: str) -> MyTTSConfig:
        return MyTTSConfig(
            endpoint="https://api.example.com/v1/speak",
            language_map={
                rasa_language: TTSLanguageMapEntry(
                    language="en-US", voice="en-US-JennyNeural"
                )
            },
        )

    @classmethod
    def from_config_dict(
        cls,
        config: Dict,
        format: AudioFormat,
        rasa_language: str,
        additional_languages: Optional[List[str]] = None,
    ) -> "MyTTS":
        return cls(
            rasa_language=rasa_language,
            format=format,
            config=MyTTSConfig.model_validate(config),
            additional_languages=additional_languages,
        )

    async def synthesize(
        self, text: str, config: Optional[MyTTSConfig] = None
    ) -> AsyncIterator[RasaAudioBytes]:
        voice = self.current_language_config.voice
        # Call your TTS service with `text` and `voice`, then yield audio chunks.
        yield self.engine_bytes_to_rasa_audio_bytes(b"\x00" * 160)

    def engine_bytes_to_rasa_audio_bytes(self, chunk: bytes) -> RasaAudioBytes:
        return RasaAudioBytes(chunk, format=self.audio_format)  # convert if needed

Method	Required	Purpose
`name()`	yes	Class method returning a short engine name, used in logs and traces
`get_default_config(rasa_language)`	yes	Static method returning the default config, including a `language_map` entry for the agent’s language
`from_config_dict(config, format, rasa_language, additional_languages)`	yes	Class method that builds the engine from the dictionary in `integrations.yml`
`synthesize(text, config)`	yes	Async generator that yields audio chunks for the full text
`engine_bytes_to_rasa_audio_bytes(chunk)`	yes	Convert your service’s audio bytes to `RasaAudioBytes`

synthesize handles whole utterances, such as the greeting. To also start speaking while the model is still writing the reply, set streaming_input = True and implement:

Method	Purpose
`connect(config)` and `close_connection()`	Open and close the connection to your service
`send_text_chunk(text)`	Send a piece of text to your service
`signal_text_done()`	Tell your service that the text is complete
`signal_interrupt()`	Optional. Drop buffered text and audio when the caller interrupts, if your provider can cancel a request. The default does nothing.
`stream_audio()`	Async generator that yields audio chunks as your service produces them

Config and environment variables

Define your own settings as fields on a config class that extends ASREngineConfig or TTSEngineConfig, as MyASRConfig and MyTTSConfig do above. Settings from integrations.yml are validated against it and are available as self.config. The base classes already provide language_map, and timeout for speech synthesis. Set required_env_vars to the environment variables your service needs, and required_packages to the Python packages it imports. The engine fails to start with the missing name if one is absent:

required_env_vars = ("MY_TTS_API_KEY",)
required_packages = ("my_tts_sdk",)

Keys in language_map must be the agent’s language codes, as for the built-in engines. Your engine reads the matching entry for the current language from self.current_language_config, which has engine_language_key (the entry’s language), voice, and model.

Learn from a community adapter

The community voice vendor router contains examples of custom engines, including Speechmatics recognition and audio-format conversion. Its Speechmatics adapter source shows how to assemble final segments when end-of-utterance detection is configured. Read this implementation alongside the Mantle interface above when writing your own adapter. This is community example code, not a built-in Mantle provider or a compatibility guarantee. Read its dependency pins, setup instructions, and per-adapter test status before adapting it. Install only the optional dependencies needed for your chosen provider, then verify the engine against your Rasa version and channel audio format. The examples on this page define the Mantle extension interface; the community package does not change that interface.

Reference

Custom speech recognition

Emit a complete caller turn

Custom speech synthesis

Config and environment variables

Learn from a community adapter

See also

​Custom speech recognition

​Emit a complete caller turn

​Custom speech synthesis

​Config and environment variables

​Learn from a community adapter

​See also

Custom speech recognition

Emit a complete caller turn

Custom speech synthesis

Config and environment variables

Learn from a community adapter

See also