Real Time Speech Recognition API: Azure Speech SDK, FastAPI & WebSockets
Running continuous recognition with FastAPI and Websockets using webm audio streaming format
Image generated with AI
Real Time Speech Recognition API: Azure Speech SDK, FastAPI & WebSockets
Running continuous recognition with FastAPI and Websockets using webm audio streaming format
I’ve been trying to figure out a way to achieve it and after a week of several testings I’ve finally got to the solution (it’s not perfect but might help the community).
Demo vídeo
Demonstration of a continuous audio streaming with Azure Speech Service and FastAPI
My experience
After looking for the documentation and doing several tests, I realized you do not need to convert the audio format from the source for the generated chunks. You can use the same widely supported webm audio format using the audio/webm;codecs:opusfor the client side.
The approach here would be to take the audio chunks using the well-known MediaRecorder from the WebAudio API, configure it to take the samples using the ’audio/webm;codecs:opus’ format, and then, send each chunk to an API via websocket, for example, a FastAPI with WebSockets enabled. Please see the docs from Triangolo to understand how it works: WebSockets with FastAPI
In any case, it can be good to have an exported file (a final one) just after the websocket client is closed, so I’m considering a records/ folder to save the audio files in .webm format. Your project structure in Python should look like this:
.
├── records/
├── .env
├── Dockerfile
├── requirements.txt
└── websocket.pyLet’s get started
1. Install the required dependencies
First, make sure you have all the necessary packages in Python, you can first create a virtual environment with venv using the following command:
# Create a virtual environment and activate it
python -m venv base
source base/bin/activatePlease, see that I’ve used python instead of python3 since I have installed the symblink python-is-python3, but you can change it to python3 if the command above does not work for you.
After that, please install the following libraries and generate a requirements.txt file for reproducibility:
# Install the python packages
pip install fastapi websockets azure-cognitiveservices-speech python-dotenv asyncio
# Export the packages used in the project for further use
pip freeze > requirements.txtThen, install the GStreamer encoder so Azure Speech Recognition client can use it to transform the webm chunks from the browser to the required PCM audio format.
# Installing GStreamer for audio format conversion
sudo apt install libgstreamer1.0–0 \
gstreamer1.0-plugins-base \
gstreamer1.0-plugins-good \
gstreamer1.0-plugins-bad \
gstreamer1.0-plugins-uglyThese are the explicit words from the Azure documentation:
The Speech SDK and Speech CLI use GStreamer to support different kinds of input audio formats. GStreamer decompresses the audio before it’s sent over the wire to the Speech service as raw PCM.
The default audio streaming format is WAV (16 kHz or 8 kHz, 16-bit, and mono PCM). Outside WAV and PCM, the following compressed input formats are also supported through GStreamer:
- MP3
- OPUS/OGG
- FLAC
- ALAW in WAV container
- MULAW in WAV container
- ANY for MP4 container or unknown media format
Please note that we will need to useANYfor thewebm format.
You can read more about this in their article: How to use compressed input audio
Frontend: Simple HTML interface with Javascript
For simplicity, we can create a simple web app aplication using Javascript and plain HTML.
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Real Time Speech Recognition</title>
</head>
<body>
<!-- Your custom UI here -->
<button onclick="start_microphone()">Start Recording</button>
<button onclick="stop_microphone()">Stop Recording</button>
<!-- The Javascript code that handles the streaming process -->
<script>
let mediaRecorder;
let ws;
/* Handle the websocket audio chunk streaming */
const start_microphone = async () => {
ws = new WebSocket("ws://localhost:8000/ws");
const stream = await navigator.mediaDevices.getUserMedia({ audio: true });
mediaRecorder = new MediaRecorder(stream, { mimeType: 'audio/webm;codecs:opus' });
try {
// If websocket is opened
ws.onopen = () => {
/* Sends a message in the client console when the websocket is opened */
console.log("WebSocket connection opened.");
};
// If message is received from server
ws.onmessage = (e) => {
/* Sends the received message from the API when it's available*/
console.log("Received message:", e.data);
};
// If media recorder data stream is available
mediaRecorder.ondataavailable = (e) => {
/* Sends the audio chunk whenever is available and the websocket is opened */
if (e.data.size > 0 && ws.readyState === WebSocket.OPEN) {
ws.send(e.data);
/* Shows the chunk info that has been sent */
console.log(e.data);
}
};
// Waits 100ms before sending a new data chunk
mediaRecorder.start(100);
} catch (error) { // In case there is an error
console.error("Error accessing microphone:", error);
}
};
const stop_microphone = () => {
if (mediaRecorder && mediaRecorder.state === "recording") {
mediaRecorder.stop();
mediaRecorder.stream.getTracks().forEach(track => track.stop());
}
if (ws) {
ws.close();
ws.onclose = () => {
console.log("WebSocket connection closed.");
};
}
};
</script>
</body>
</html>Explain in detail
In the previous code, we have created two buttons that allows a simple interface in HTML to trigger two events, the start button will stablish a connection with the server on localhost:8000/ws and after that, it will start listening audio from the microphone and sending chunks of data every 100 milliseconds.
The stop button will stop the microphone and additionally, it will stop the websocket connection.
The code is very simple so you can twick it as you want.
Backend: Real Time Recognition with 🎙️Azure Speech SDK using ⚡FastAPI and Websocket
Let’s start by creating a new file inside your project folder called websocket.py.
Use the following code to test it out:
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import azure.cognitiveservices.speech as speechsdk
from dotenv import load_dotenv
from datetime import datetime
import os
import asyncio
load_dotenv() # Load the environmental variables from .env file
class bcolors: # Only to apply colors to the prints
HEADER = '\033[95m'
OKBLUE = '\033[94m'
OKCYAN = '\033[96m'
OKGREEN = '\033[92m'
WARNING = '\033[93m'
FAIL = '\033[91m'
ENDC = '\033[0m'
BOLD = '\033[1m'
UNDERLINE = '\033[4m'
app = FastAPI() # Create a new FastAPI app
# Speech key and region from your Azure Speech Recognition service
speech_key = os.getenv("SPEECH_KEY")
speech_region = os.getenv("SPEECH_REGION")
def create_speech_recognizer(loop, queue):
"""
Allows to create a client for the speech recognizer and a stream (buffer)
"""
# Create the configuration of the recognizer from your account of Azure
speech_config = speechsdk.SpeechConfig(
subscription=speech_key,
region=speech_region
)
# Configuration for the input audio format. The documentation specifies that you can use GStreamer (It also needs to be installed locally) to encode other formats to PCM (Pulse Code Modulation).
# Here I'm using speechsdk.AudioStreamContainerFormat.ANY since i'm sending streaming data using the audio/webm;codecs:opus format directl, that is supported for most of the modern browsers
format = speechsdk.audio.AudioStreamFormat(compressed_stream_format=speechsdk.AudioStreamContainerFormat.ANY) # To receive audio data in any format and process them with GStreamer
stream = speechsdk.audio.PushAudioInputStream(format) # Creates an audio stream to send data to the speech service
audio_config = speechsdk.audio.AudioConfig(stream=stream) # Adjust the audio config using the recently created stream
# Creates a speech recognizer client for the speech recognizer
speech_recognizer = speechsdk.SpeechRecognizer(
speech_config=speech_config,
audio_config=audio_config,
language="en-US" # Change to your desired language if supported. If not specified, 'en-US' will be used by default.
)
# Callbacks for the speech recognizer. They are automatically triggered based on event type
def recognizing_cb(evt: speechsdk.SpeechRecognitionEventArgs):
"""
Triggered everytime the recognizer has processed a set of audio chunks and recognized part of the speech
"""
print(f"{bcolors.OKGREEN}Azure Speech Recognition -> Recognizing: {evt.result.text}{bcolors.ENDC}")
def recognized_cb(evt: speechsdk.SpeechRecognitionEventArgs):
"""
Triggered when the speech recognition has processed an audio fragment and recognized the text in its entirety
"""
print(f"{bcolors.OKGREEN}Azure Speech Recognition -> Recognized: {evt.result.text}{bcolors.ENDC}")
asyncio.run_coroutine_threadsafe(queue.put(evt.result.text), loop)
def stop_cb(evt: speechsdk.SessionEventArgs):
"""
Triggered when speech recognition session is stopped
"""
print(f"{bcolors.WARNING}Azure Speech Recognition -> Session stopped due websocket close: {evt}{bcolors.ENDC}")
def canceled_cb(evt: speechsdk.SessionEventArgs):
"""
Triggered when the speech recognition session is cancelled due to an error
"""
print(f"{bcolors.FAIL}Azure Speech Recognition -> Session canceled due an error: {evt}{bcolors.ENDC}")
# Connect callbacks to the speech recognizer to be triggered when an event occurs.
speech_recognizer.recognizing.connect(recognizing_cb)
speech_recognizer.recognized.connect(recognized_cb)
speech_recognizer.session_stopped.connect(stop_cb)
speech_recognizer.canceled.connect(canceled_cb)
return speech_recognizer, stream
@app.websocket("/ws") # Change to your desired websocket endpoint
async def audio_streaming(websocket: WebSocket):
await websocket.accept() # Accept client connection
loop = asyncio.get_event_loop() # Get the asyncio event loop
message_queue = asyncio.Queue() # Create a message queue to store the results of speech recognition
speech_recognizer, stream = create_speech_recognizer(loop, message_queue) # Create the speech recognizer and the audio stream
async def receive_audio(websocket, stream):
audio_data = b"" # Store the audio data in bytes
print(f"{bcolors.OKGREEN}WebSocket -> Receiving audio from client and saving into stream...{bcolors.ENDC}")
while True: # As long as the customer is connected
try: # Attempt to receive audio data from the client
data = await websocket.receive_bytes() # Receive audio data from the client
audio_data += data # Store audio all data chunks in a variable
stream.write(data) # Write audio data to the stream buffer
print(f"{bcolors.OKCYAN}WebSocket -> Stream data en bytes: {len(data)}{bcolors.ENDC}", end="\n") # Data that are being sent from the client
except WebSocketDisconnect: # If the client is disconnected
print(f"{bcolors.FAIL}Azure Speech Recognition -> Stream closed{bcolors.ENDC}")
stream.close() # Close the stream
print(f"{bcolors.OKBLUE}API -> Websocket client disconnected!{bcolors.ENDC}")
print(f"{bcolors.OKBLUE}API -> Stopping continuous recognition...{bcolors.ENDC}")
speech_recognizer.stop_continuous_recognition() # Stop speech recognition
print(f"{bcolors.OKBLUE}API -> Continuous recognition stopped!{bcolors.ENDC}")
print(f"{bcolors.OKBLUE}API -> Exporting audio data to a file...{bcolors.ENDC}")
# Save received audio data to a file
with open(f"records/received_audio_{datetime.now()}.webm", "wb") as f: # Create an audio file in webm format
f.write(audio_data) # Write the whole audio data to the file
print(f"{bcolors.OKBLUE}API -> Audio data exported!{bcolors.ENDC}")
break
except Exception as e: # If an error occurs
print(f"Error: {e}")
break # Exiting the loop
async def send_messages():
"""
Allows messages recognized by the Azure service to be sent to the client via the websocket to the client
"""
while True: # As long as the customer is connected
message = await message_queue.get() # Get the recognized text from the queue
await websocket.send_text(message) # Send the text to the websocket client
try:
speech_recognizer.start_continuous_recognition() # Start continuous speech recognition
print("API -> Continuous recognition running, say something to process data...")
await asyncio.gather(receive_audio(websocket, stream), send_messages()) # Execute the functions of receiving audio and sending messages back to the client.
except Exception as e:
print(f"Error: {e}")Code explanation
The backend code begins by establishing a connection via WebSockets. Following this, it initializes an event queue using Python’s built-in asyncio.Queue. However, other queue systems like Azure Queue Storage or Kafka can also be used.
This queue serves to decouple the transcription intents generated when an event in the Azure Speech SDK is triggered. Since these callbacks run in a different thread and are blocking operations, we use asyncio.run_coroutine_threadsafe() to asynchronously place the recognized speech text into the queue. This method requires the current event loop, which is why we use asyncio.get_event_loop().
This setup allows us to have two concurrent functions: one to handle incoming data and another to manage the sending of transcriptions.
Additionally, the code configures the recognizer to use GStreamer, which automatically encodes the WebM audio to PCM format using the collected chunks over time.
format = speechsdk.audio.AudioStreamFormat(compressed_stream_format=speechsdk.AudioStreamContainerFormat.ANY)
stream = speechsdk.audio.PushAudioInputStream(format)
audio_config = speechsdk.audio.AudioConfig(stream=stream)After copying the code inside the websocket.py file, create an .env file with the environmental variables for testing.
Note: Don’t forget to exclude this file in a .gitignore file if you have initiated the repository.
SPEECH_KEY="<your-speech-key>"
SPEECH_REGION="<your-speech-region>"After these steps, you will be able to test the recongnizer. Execute the following command:
fastapi run websocket.pyYou should see something like this:
INFO Using path websocket.py
INFO Resolved absolute path /home/username/azure-speech-recognition-streaming-websocket/websocket.py
INFO Searching for package file structure from directories with __init__.py files
INFO Importing from /home/username/azure-speech-recognition-streaming-websocket
╭─ Python module file ─╮
│ │
│ 🐍 websocket.py │
│ │
╰──────────────────────╯
INFO Importing module websocket
INFO Found importable FastAPI app
╭── Importable FastAPI app ───╮
│ │
│ from websocket import app │
│ │
╰─────────────────────────────╯
INFO Using import string websocket:app
╭─────────── FastAPI CLI - Production mode ───────────╮
│ │
│ Serving at: http://0.0.0.0:8000 │
│ │
│ API docs: http://0.0.0.0:8000/docs │
│ │
│ Running in production mode, for development use: │
│ │
│ fastapi dev │
│ │
╰─────────────────────────────────────────────────────╯
INFO: Started server process [24953]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)Deploy the ⚡FastAPI app into a 📦Docker Container.
Finally, if you want to run the code inside a container, first you need to create the Dockerfile in the project folder as follows:
# Gets a lightweight version for Python
FROM python:3.12-slim
# Creates a directory and moves to it
WORKDIR /app
# Copy the source code from your current working directory to the newly created directory inside the container
ADD . /app
# Install all the dependencies
RUN pip install - no-cache-dir -r requirements.txt
# Install the GStreamer to convert from webm to PCM (important to install, otherwise won't work)
RUN apt-get update && apt-get install -y libgstreamer1.0–0 gstreamer1.0-plugins-base gstreamer1.0-plugins-good gstreamer1.0-plugins-bad gstreamer1.0-plugins-ugly
# Expose the port 8000, it's commonly used by FastAPI
EXPOSE 8000
# Runs the FastAPI application for production
CMD ["fastapi", "run", "websocket.py"]and build it using
docker build -t mycontainer:latest .and finally, you can run it using
docker run -d -p 8000:8000 mycontainer:latestI hope this information can be very helpul and send your valuable feedback if something can be improved.
메타데이터
- post_id
- 566f4e5e62ff
- slug
- real-time-speech-recognition-api-azure-speech-sdk-fastapi-websockets-566f4e5e62ff
- url
- https://medium.com/@kennethdiazgonzalez/real-time-speech-recognition-api-azure-speech-sdk-fastapi-websockets-566f4e5e62ff
- canonical_url
- https://medium.com/@kennethdiazgonzalez/real-time-speech-recognition-api-azure-speech-sdk-fastapi-websockets-566f4e5e62ff
- author_url
- https://medium.com/@kennethdiazgonzalez
- status
- ok
- fetched_at
- 2026-06-27 10:07:59
