llms.txt
Compare with other sites' llms.txt files →# FastRTC
### The Real-Time Communication Library for Python.
Turn any python function into a real-time audio and video stream over WebRTC or WebSockets.
[](https://github.com/user-attachments/assets/a297aa1e-ff42-448c-a58c-389b0a575d4d)
## Installation
```bash
pip install fastrtc
```
to use built-in pause detection (see [ReplyOnPause](userguide/audio/#reply-on-pause)), speech-to-text (see [Speech To Text](userguide/audio/#speech-to-text)), and text to speech (see [Text To Speech](userguide/audio/#text-to-speech)), install the `vad`, `stt`, and `tts` extras:
```bash
pip install "fastrtc[vad, stt, tts]"
```
## Quickstart
Import the [Stream](userguide/streams) class and pass in a [handler](userguide/streams/#handlers). The `Stream` has three main methods:
- `.ui.launch()`: Launch a built-in UI for easily testing and sharing your stream. Built with [Gradio](https://www.gradio.app/).
- `.fastphone()`: Get a free temporary phone number to call into your stream. Hugging Face token required.
- `.mount(app)`: Mount the stream on a [FastAPI](https://fastapi.tiangolo.com/) app. Perfect for integrating with your already existing production system.
```python
from fastrtc import Stream, ReplyOnPause
import numpy as np
def echo(audio: tuple[int, np.ndarray]):
# The function will be passed the audio until the user pauses
# Implement any iterator that yields audio
# See "LLM Voice Chat" for a more complete example
yield audio
stream = Stream(
handler=ReplyOnPause(echo),
modality="audio",
mode="send-receive",
)
```
```py
import os
from fastrtc import (ReplyOnPause, Stream, get_stt_model, get_tts_model)
from openai import OpenAI
sambanova_client = OpenAI(
api_key=os.getenv("SAMBANOVA_API_KEY"), base_url="https://api.sambanova.ai/v1"
)
stt_model = get_stt_model()
tts_model = get_tts_model()
def echo(audio):
prompt = stt_model.stt(audio)
response = sambanova_client.chat.completions.create(
model="Meta-Llama-3.2-3B-Instruct",
messages=[{"role": "user", "content": prompt}],
max_tokens=200,
)
prompt = response.choices[0].message.content
for audio_chunk in tts_model.stream_tts_sync(prompt):
yield audio_chunk
stream = Stream(ReplyOnPause(echo), modality="audio", mode="send-receive")
```
```python
from fastrtc import Stream
import numpy as np
def flip_vertically(image):
return np.flip(image, axis=0)
stream = Stream(
handler=flip_vertically,
modality="video",
mode="send-receive",
)
```
```python
from fastrtc import Stream
import gradio as gr
import cv2
from huggingface_hub import hf_hub_download
from .inference import YOLOv10
model_file = hf_hub_download(
repo_id="onnx-community/yolov10n", filename="onnx/model.onnx"
)
# git clone https://huggingface.co/spaces/fastrtc/object-detection
# for YOLOv10 implementation
model = YOLOv10(model_file)
def detection(image, conf_threshold=0.3):
image = cv2.resize(image, (model.input_width, model.input_height))
new_image = model.detect_objects(image, conf_threshold)
return cv2.resize(new_image, (500, 500))
stream = Stream(
handler=detection,
modality="video",
mode="send-receive",
additional_inputs=[
gr.Slider(minimum=0, maximum=1, step=0.01, value=0.3)
]
)
```
Run:
```py
stream.ui.launch()
```
```py
stream.fastphone()
```
```py
app = FastAPI()
stream.mount(app)
# Optional: Add routes
@app.get("/")
async def _():
return HTMLResponse(content=open("index.html").read())
# uvicorn app:app --host 0.0.0.0 --port 8000
```
Learn more about the [Stream](userguide/streams) in the user guide.
## Key Features
Automatic Voice Detection and Turn Taking built-in, only worry about the logic for responding to the user.
Automatic UI - Use the `.ui.launch()` method to launch the webRTC-enabled built-in Gradio UI.
Automatic WebRTC Support - Use the `.mount(app)` method to mount the stream on a FastAPI app and get a webRTC endpoint for your own frontend!
Websocket Support - Use the `.mount(app)` method to mount the stream on a FastAPI app and get a websocket endpoint for your own frontend!
Automatic Telephone Support - Use the `fastphone()` method of the stream to launch the application and get a free temporary phone number!
Completely customizable backend - A `Stream` can easily be mounted on a FastAPI app so you can easily extend it to fit your production application. See the [Talk To Claude](https://huggingface.co/spaces/fastrtc/talk-to-claude) demo for an example on how to serve a custom JS frontend.
## Examples
See the [cookbook](/cookbook).
Follow and join our [organization](https://huggingface.co/fastrtc) on Hugging Face!
# Connecting via API
Before continuing, select the `modality`, `mode` of your `Stream` and whether you're using `WebRTC` or `WebSocket`s.
Connection
WebRTC WebSocket
Modality
Audio Video Audio-Video
Mode
Send-Receive Receive Send
### Sample Code
### Message Format
Over both WebRTC and WebSocket, the server can send messages of the following format:
```json
{
"type": `send_input` | `fetch_output` | `stopword` | `error` | `warning` | `log`,
"data": string | object
}
```
- `send_input`: Send any input data for the handler to the server. See [`Additional Inputs`](#additional-inputs) for more details.
- `fetch_output`: An instance of [`AdditionalOutputs`](#additional-outputs) is sent to the server.
- `stopword`: The stopword has been detected. See [`ReplyOnStopWords`](../audio/#reply-on-stopwords) for more details.
- `error`: An error occurred. The `data` will be a string containing the error message.
- `warning`: A warning occurred. The `data` will be a string containing the warning message.
- `log`: A log message. The `data` will be a string containing the log message.
The `ReplyOnPause` handler can also send the following `log` messages.
```json
{
"type": "log",
"data": "pause_detected" | "response_starting" | "started_talking"
}
```
Tip
When using WebRTC, the messages will be encoded as strings, so parse as JSON before using.
### Additional Inputs
When the `send_input` message is received, update the inputs of your handler using the `set_input` method. The first argument is the `webrtc_id`, followed by your custom arguments.
```python
from pydantic import BaseModel, Field
class InputData(BaseModel):
webrtc_id: str
conf_threshold: float = Field(ge=0, le=1)
@app.post("/input_hook")
async def _(data: InputData):
stream.set_input(data.webrtc_id, data.conf_threshold)
```
**Accessing inputs in handler startup:**
In your handler's `start_up()` method, use `wait_for_args()` to wait for inputs, then access them via `latest_args[1:]` (the first element is metadata):
```python
from fastrtc import AsyncStreamHandler
class CustomHandler(AsyncStreamHandler):
async def start_up(self) -> None:
await self.wait_for_args()
conf_threshold = self.latest_args[1]
# Use conf_threshold for initialization
```
The updated data will be passed to the handler on the **next** call.
### Additional Outputs
The `fetch_output` message is sent to the client whenever an instance of [`AdditionalOutputs`](../streams/#additional-outputs) is available. You can access the latest output data by calling the `fetch_latest_output` method of the `Stream` object.
However, rather than fetching each output manually, a common pattern is to fetch the entire stream of output data by calling the `output_stream` method.
Here is an example:
```python
from fastapi.responses import StreamingResponse
@app.get("/updates")
async def stream_updates(webrtc_id: str):
async def output_stream():
async for output in stream.output_stream(webrtc_id):
# Output is the AdditionalOutputs instance
# Be sure to serialize it however you would like
yield f"data: {output.args[0]}\n\n"
return StreamingResponse(
output_stream(),
media_type="text/event-stream"
)
```
### Handling Errors
When connecting via `WebRTC`, the server will respond to the `/webrtc/offer` route with a JSON response. If there are too many connections, the server will respond with a 200 error.
```json
{
"status": "failed",
"meta": {
"error": "concurrency_limit_reached",
"limit": 10
}
```
Over `WebSocket`, the server will send the same message before closing the connection.
Tip
The server will sends a 200 status code because otherwise the gradio client will not be able to process the json response and display the error.
# Audio-Video Streaming
You can simultaneously stream audio and video using `AudioVideoStreamHandler` or `AsyncAudioVideoStreamHandler`. They are identical to the audio `StreamHandlers` with the addition of `video_receive` and `video_emit` methods which take and return a `numpy` array, respectively.
Here is an example of the video handling functions for connecting with the Gemini multimodal API. In this case, we simply reflect the webcam feed back to the user but every second we'll send the latest webcam frame (and an additional image component) to the Gemini server.
Please see the "Gemini Audio Video Chat" example in the [cookbook](../../cookbook) for the complete code.
Async Gemini Video Handling
```python
async def video_receive(self, frame: np.ndarray):
"""Send video frames to the server"""
if self.session:
# send image every 1 second
# otherwise we flood the API
if time.time() - self.last_frame_time > 1:
self.last_frame_time = time.time()
await self.session.send(encode_image(frame))
if self.latest_args[2] is not None:
await self.session.send(encode_image(self.latest_args[2]))
self.video_queue.put_nowait(frame)
async def video_emit(self) -> VideoEmitType:
"""Return video frames to the client"""
return await self.video_queue.get()
```
## Reply On Pause
Typically, you want to run a python function whenever a user has stopped speaking. This can be done by wrapping a python generator with the `ReplyOnPause` class and passing it to the `handler` argument of the `Stream` object. The `ReplyOnPause` class will handle the voice detection and turn taking logic automatically!
```python
from fastrtc import ReplyOnPause, Stream
def response(audio: tuple[int, np.ndarray]): # (1)
sample_rate, audio_array = audio
# Generate response
for audio_chunk in generate_response(sample_rate, audio_array):
yield (sample_rate, audio_chunk) # (2)
stream = Stream(
handler=ReplyOnPause(response),
modality="audio",
mode="send-receive"
)
```
1. The python generator will receive the **entire** audio up until the user stopped. It will be a tuple of the form (sampling_rate, numpy array of audio). The array will have a shape of (1, num_samples). You can also pass in additional input components.
1. The generator must yield audio chunks as a tuple of (sampling_rate, numpy audio array). Each numpy audio array must have a shape of (1, num_samples).
1. The python generator will receive the **entire** audio up until the user stopped. It will be a tuple of the form (sampling_rate, numpy array of audio). The array will have a shape of (1, num_samples). You can also pass in additional input components.
1. The generator must yield audio chunks as a tuple of (sampling_rate, numpy audio array). Each numpy audio array must have a shape of (1, num_samples).
Asynchronous
You can also use an async generator with `ReplyOnPause`.
Parameters
You can customize the voice detection parameters by passing in `algo_options` and `model_options` to the `ReplyOnPause` class.
```python
from fastrtc import AlgoOptions, SileroVadOptions
stream = Stream(
handler=ReplyOnPause(
response,
algo_options=AlgoOptions(
audio_chunk_duration=0.6,
started_talking_threshold=0.2,
speech_threshold=0.1
),
model_options=SileroVadOptions(
threshold=0.5,
min_speech_duration_ms=250,
min_silence_duration_ms=100
)
)
)
```
### Interruptions
By default, the `ReplyOnPause` handler will allow you to interrupt the response at any time by speaking again. If you do not want to allow interruption, you can set the `can_interrupt` parameter to `False`.
```python
from fastrtc import Stream, ReplyOnPause
stream = Stream(
handler=ReplyOnPause(
response,
can_interrupt=True,
)
)
```
[](https://github.com/user-attachments/assets/dba68dd7-7444-439b-b948-59171067e850)
Muting Response Audio
You can directly talk over the output audio and the interruption will still work. However, in these cases, the audio transcription may be incorrect. To prevent this, it's best practice to mute the output audio before talking over it.
### Startup Function
You can pass in a `startup_fn` to the `ReplyOnPause` class. This function will be called when the connection is first established. It is helpful for generating initial responses.
```python
from fastrtc import get_tts_model, Stream, ReplyOnPause
tts_client = get_tts_model()
def echo(audio: tuple[int, np.ndarray]):
# Implement any iterator that yields audio
# See "LLM Voice Chat" for a more complete example
yield audio
def startup():
for chunk in tts_client.stream_tts_sync("Welcome to the echo audio demo!"):
yield chunk
stream = Stream(
handler=ReplyOnPause(echo, startup_fn=startup),
modality="audio",
mode="send-receive",
ui_args={"title": "Echo Audio"},
)
```
[](https://github.com/user-attachments/assets/c6b1cb51-5790-4522-80c3-e24e58ef9f11)
## Reply On Stopwords
You can configure your AI model to run whenever a set of "stop words" are detected, like "Hey Siri" or "computer", with the `ReplyOnStopWords` class.
The API is similar to `ReplyOnPause` with the addition of a `stop_words` parameter.
```py
from fastrtc import Stream, ReplyOnStopWords
def response(audio: tuple[int, np.ndarray]):
"""This function must yield audio frames"""
...
for numpy_array in generated_audio:
yield (sampling_rate, numpy_array, "mono")
stream = Stream(
handler=ReplyOnStopWords(generate,
input_sample_rate=16000,
stop_words=["computer"]), # (1)
modality="audio",
mode="send-receive"
)
```
1. The `stop_words` can be single words or pairs of words. Be sure to include common misspellings of your word for more robust detection, e.g. "llama", "lamma". In my experience, it's best to use two very distinct words like "ok computer" or "hello iris".
1. The `stop_words` can be single words or pairs of words. Be sure to include common misspellings of your word for more robust detection, e.g. "llama", "lamma". In my experience, it's best to use two very distinct words like "ok computer" or "hello iris".
Extra Dependencies
The `ReplyOnStopWords` class requires the `stopword` extra. Run `pip install fastrtc[stopword]` to install it.
English Only
The `ReplyOnStopWords` class is currently only supported for English.
## Stream Handler
`ReplyOnPause` and `ReplyOnStopWords` are implementations of a `StreamHandler`. The `StreamHandler` is a low-level abstraction that gives you arbitrary control over how the input audio stream and output audio stream are created. The following example echos back the user audio.
```py
import gradio as gr
from fastrtc import StreamHandler
from queue import Queue
class EchoHandler(StreamHandler):
def __init__(self) -> None:
super().__init__()
self.queue = Queue()
def receive(self, frame: tuple[int, np.ndarray]) -> None: # (1)
self.queue.put(frame)
def emit(self) -> None: # (2)
return self.queue.get()
def copy(self) -> StreamHandler:
return EchoHandler()
def shutdown(self) -> None: # (3)
pass
def start_up(self) -> None: # (4)
pass
stream = Stream(
handler=EchoHandler(),
modality="audio",
mode="send-receive"
)
```
1. The `StreamHandler` class implements three methods: `receive`, `emit` and `copy`. The `receive` method is called when a new frame is received from the client, and the `emit` method returns the next frame to send to the client. The `copy` method is called at the beginning of the stream to ensure each user has a unique stream handler.
1. The `emit` method SHOULD NOT block. If a frame is not ready to be sent, the method should return `None`. If you need to wait for a frame, use [`wait_for_item`](../../utils#wait_for_item) from the `utils` module.
1. The `shutdown` method is called when the stream is closed. It should be used to clean up any resources.
1. The `start_up` method is called when the stream is first created. It should be used to initialize any resources. See [Talk To OpenAI](https://huggingface.co/spaces/fastrtc/talk-to-openai-gradio) or [Talk To Gemini](https://huggingface.co/spaces/fastrtc/talk-to-gemini-gradio) for an example of a `StreamHandler` that uses the `start_up` method to connect to an API.
1. The `StreamHandler` class implements three methods: `receive`, `emit` and `copy`. The `receive` method is called when a new frame is received from the client, and the `emit` method returns the next frame to send to the client. The `copy` method is called at the beginning of the stream to ensure each user has a unique stream handler.
1. The `emit` method SHOULD NOT block. If a frame is not ready to be sent, the method should return `None`. If you need to wait for a frame, use [`wait_for_item`](../../utils#wait_for_item) from the `utils` module.
1. The `shutdown` method is called when the stream is closed. It should be used to clean up any resources.
1. The `start_up` method is called when the stream is first created. It should be used to initialize any resources. See [Talk To OpenAI](https://huggingface.co/spaces/fastrtc/talk-to-openai-gradio) or [Talk To Gemini](https://huggingface.co/spaces/fastrtc/talk-to-gemini-gradio) for an example of a `StreamHandler` that uses the `start_up` method to connect to an API.
Tip
See this [Talk To Gemini](https://huggingface.co/spaces/fastrtc/talk-to-gemini-gradio) for a complete example of a more complex stream handler.
Warning
The `emit` method should not block. If you need to wait for a frame, use [`wait_for_item`](../../utils#wait_for_item) from the `utils` module.
## Async Stream Handlers
It is also possible to create asynchronous stream handlers. This is very convenient for accessing async APIs from major LLM developers, like Google and OpenAI. The main difference is that `receive`, `emit`, and `start_up` are now defined with `async def`.
Here is a simple example of using `AsyncStreamHandler`:
```py
from fastrtc import AsyncStreamHandler, wait_for_item, Stream
import asyncio
import numpy as np
class AsyncEchoHandler(AsyncStreamHandler):
"""Simple Async Echo Handler"""
def __init__(self) -> None:
super().__init__(input_sample_rate=24000)
self.queue = asyncio.Queue()
async def receive(self, frame: tuple[int, np.ndarray]) -> None:
await self.queue.put(frame)
async def emit(self) -> None:
return await wait_for_item(self.queue)
def copy(self):
return AsyncEchoHandler()
async def shutdown(self):
pass
async def start_up(self) -> None:
pass
```
Tip
See [Talk To Gemini](https://huggingface.co/spaces/fastrtc/talk-to-gemini), [Talk To Openai](https://huggingface.co/spaces/fastrtc/talk-to-openai) for complete examples of `AsyncStreamHandler`s.
## Text To Speech
You can use an on-device text to speech model if you have the `tts` extra installed. Import the `get_tts_model` function and call it with the model name you want to use. At the moment, the only model supported is `kokoro`.
The `get_tts_model` function returns an object with three methods:
- `tts`: Synchronous text to speech.
- `stream_tts
