> ## Documentation Index
> Fetch the complete documentation index at: https://rasa.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Custom speech providers

> Write your own speech recognition or speech synthesis provider integration and use it on a voice channel (beta).

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.

```yaml integrations.yml theme={null}
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
`ASREvent`s.

```python addons/custom_asr.py theme={null}
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](https://rasa.com/docs/mantle/voice/testing.md#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`.

```python addons/custom_tts.py theme={null}
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:

```python theme={null}
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](https://github.com/RasaHQ/rasa-community-resources/tree/4aa0c4419dc193fef7a969c12d59edcf720f2606/patterns/voice-vendor-router)
contains examples of custom engines, including Speechmatics recognition and
audio-format conversion. Its
[Speechmatics adapter source](https://github.com/RasaHQ/rasa-community-resources/blob/4aa0c4419dc193fef7a969c12d59edcf720f2606/patterns/voice-vendor-router/voicerouter/providers/speechmatics.py)
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.

## See also

* [Speech providers](https://rasa.com/docs/mantle/reference/speech-providers.md): the built-in providers, their defaults, and their settings
* [`integrations.yml` → Voice channels](https://rasa.com/docs/mantle/reference/integrations-yml#voice-channels): `language_map`, `interruptions`, and `silence_timeout`


## Related topics

- [Speech providers](https://rasa.com/docs/mantle/reference/speech-providers.md)
- [Changelog](https://rasa.com/docs/mantle/changelog.md)
- [Voice agents](https://rasa.com/docs/mantle/voice/overview.md)
- [Speech Integrations](https://rasa.com/docs/reference/integrations/speech-integrations.md)
- [Studio Environment Variables](https://rasa.com/docs/reference/config/studio-environment-variables.md)


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.