### Quickstart Installation and Setup
Source: https://www.assemblyai.com/docs/integrations/recall
This code snippet outlines the initial steps to clone the repository, install dependencies, and run the bot, including setting up ngrok and configuring environment variables.
```bash
# 1. Clone and install
git clone https://github.com/AssemblyAI/assemblyai-recallai-zoom-bot.git
cd assemblyai-recallai-zoom-bot
npm install
# 2. Run ngrok and copy the ngrok URL
# ngrok http 8000
# 3. Configure your .env file and edit it with your API keys and ngrok URL
cp .env.example .env
# 4a. Open a new terminal and run
# node webhook.js
# 4b. Open another terminal and run
# node zoomBot.js
```
--------------------------------
### Install and start the buffer-clearing filter
Source: https://www.assemblyai.com/docs/voice-agents/livekit-u3-rt-pro
Example of installing the buffer-clearing filter on the session and then starting the session with your agent.
```python
from filters.short_utterance_buffer import install_short_utterance_filter
install_short_utterance_filter(session)
await session.start(room=ctx.room, agent=MyAgent())
```
--------------------------------
### Quickstart
Source: https://www.assemblyai.com/docs/pre-recorded-audio/guides/audio-duration-fix
A quickstart example demonstrating how to use the audio duration fix functionality.
```python
audio_file="./audio.mp4"
if __name__ == "__main__":
file_path = f"{audio_file}"
main(file_path)
```
--------------------------------
### Clone the example repo and install dependencies
Source: https://www.assemblyai.com/docs/integrations/recall
Commands to clone the project repository and install the necessary Node.js dependencies.
```bash
git clone https://github.com/AssemblyAI/assemblyai-recallai-zoom-bot.git
cd assemblyai-recallai-zoom-bot
npm install
```
--------------------------------
### Browser Quickstart Example
Source: https://www.assemblyai.com/docs/voice-agents/voice-agent-api/browser-integration
A basic HTML and JavaScript example demonstrating how to establish a WebSocket connection to the Voice Agent API and handle messages.
```html
Voice Agent Browser Integration
Voice Agent Example
```
--------------------------------
### Python Example
Source: https://www.assemblyai.com/docs/guides/task-endpoint-ai-coach
Example of how to process the response from the AI Coach Task Endpoint in Python.
```python
result = response.json()
if "error" in result:
print(f"\nError from LLM Gateway: {result['error']}")
else:
response_text = result['choices'][0]['message']['content']
print(f"\nResponse ID: {result['request_id']}\n")
print(response_text)
```
--------------------------------
### Front-load your most important rule
Source: https://www.assemblyai.com/docs/voice-agents/voice-agent-api/prompting-guide
Example of front-loading the most important rule.
```text
BE SHORT. This is the most important rule. Keep every response under two sentences.
You are a customer support agent for Acme Corp...
```
--------------------------------
### JavaScript Example
Source: https://www.assemblyai.com/docs/guides/task-endpoint-ai-coach
Example of how to process the response from the AI Coach Task Endpoint in JavaScript.
```javascript
if (response.error) {
console.log(`\nError from LLM Gateway: ${response.error}`);
} else {
const responseText = response.choices[0].message.content;
console.log(`\nResponse ID: ${response.request_id}\n`);
console.log(responseText);
}
```
--------------------------------
### Keyterms Prompt Example
Source: https://www.assemblyai.com/docs/streaming/universal-3-pro/prompting
Example of how to use the `keyterms_prompt` parameter to boost recognition of specific terms.
```python
keyterms_prompt=["Keanu Reeves", "AssemblyAI", "Universal-2"]
```
--------------------------------
### Quickstart
Source: https://www.assemblyai.com/docs/guides/task-endpoint-ai-coach
This Python code snippet demonstrates how to set up an AI coach using AssemblyAI's LLM Gateway. It shows how to authenticate, specify an audio source (either a public URL or an uploaded file), and initiate a request.
```python
import requests
import time
base_url = "https://api.assemblyai.com"
headers = {"authorization": ""}
# Use a publicly-accessible URL:
audio_url = "https://storage.googleapis.com/aai-web-samples/meeting.mp4"
# with open("/your_audio_file.mp3", "rb") as f:
# response = requests.post(base_url + "/v2/upload", headers=headers, data=f)
# if response.status_code != 200:
# print(f"Error: {response.status_code}, Response: {response.text}")
# response.raise_for_status()
# upload_json = response.json()
```
--------------------------------
### OpenAI Authentication and SDK Setup
Source: https://www.assemblyai.com/docs/pre-recorded-audio/migration-guides/oai_to_aai
Example of setting up OpenAI API key and client.
```python
from openai import OpenAI
api_key = "YOUR_OPENAI_API_KEY"
client = OpenAI(api_key)
```
--------------------------------
### Quickstart Example
Source: https://www.assemblyai.com/docs/streaming/universal-3-pro
This code snippet demonstrates how to set up and use the Universal 3-Pro model for real-time audio transcription.
```javascript
const { AssemblyAI } = require("@assemblyai/sdk");
const { Readable } = require("stream");
const client = new AssemblyAI({ apiKey: "YOUR_API_KEY" });
const run = async () => {
try {
const recording = await client.realtime.createRecording({
sampleRate: 16_000,
audioType: "wav",
});
Readable.toWeb(recording.stream()).pipeTo(transcriber.stream());
process.on("SIGINT", async function () {
console.log();
console.log("Stopping recording");
recording.stop();
console.log("Closing streaming transcript connection");
await transcriber.close();
process.exit();
});
} catch (error) {
console.error(error);
}
};
run();
```
--------------------------------
### Start Live Session with Gladia API
Source: https://www.assemblyai.com/docs/streaming/migration-guides/gladia-to-aai-streaming
This snippet shows how to initiate a live session with the Gladia API to get the session URL.
```python
response = requests.post(
f"{GLADIA_API_URL}/v2/live",
headers={\"X-Gladia-Key\": GLADIA_API_KEY},
json=config,
timeout=3,
)
if not response.ok:
print(f"{response.status_code}: {response.text or response.reason}")
exit(response.status_code)
session_data = response.json()
```
--------------------------------
### Python Quickstart Example
Source: https://www.assemblyai.com/docs/guides/task-endpoint-structured-QA
This Python script demonstrates the end-to-end process of transcribing audio, defining questions with context and answer formats, constructing an LLM prompt, and querying the LLM Gateway for structured Q&A.
```python
import requests
import time
import xml.etree.ElementTree as ET
API_KEY = "YOUR_API_KEY"
audio_url = "https://storage.googleapis.com/aai-web-samples/meeting.mp4"
# -------------------------------
# Step 1: Transcribe the audio
# -------------------------------
transcript_request = requests.post(
"https://api.assemblyai.com/v2/transcript",
headers={"authorization": API_KEY, "content-type": "application/json"},
json={"audio_url": audio_url, "speech_models": ["universal-3-pro"]},
)
transcript_id = transcript_request.json()["id"]
# Poll for completion
while True:
polling_response = requests.get(
f"https://api.assemblyai.com/v2/transcript/{transcript_id}",
headers={"authorization": API_KEY},
)
status = polling_response.json()["status"]
if status == "completed":
break
elif status == "error":
raise RuntimeError(f"Transcription failed: {polling_response.json()['error']}")
else:
print(f"Transcription status: {status}")
time.sleep(3)
# -------------------------------
# Step 2: Build question helper functions
# -------------------------------
def construct_question(question):
question_str = f"Question: {question['question']}"
if question.get("context"):
question_str += f"\nContext: {question['context']}"
# Default answer_format
if not question.get("answer_format"):
question["answer_format"] = "short sentence"
question_str += f"\nAnswer Format: {question['answer_format']}"
if question.get("answer_options"):
options_str = ", ".join(question["answer_options"])
question_str += f"\nOptions: {options_str}"
return question_str + "\n"
def escape_xml_characters(xml_string):
return xml_string.replace("&", "&")
# -------------------------------
# Step 3: Define questions
# -------------------------------
questions = [
{
"question": "What are the top level KPIs for engineering?",
"context": "KPI stands for key performance indicator",
"answer_format": "short sentence",
},
{
"question": "How many days has it been since the data team has gotten updated metrics?",
"answer_options": ["1", "2", "3", "4", "5", "6", "7", "more than 7"],
},
{"question": "What are the future plans for the project?"},
]
question_str = "\n".join(construct_question(q) for q in questions)
# -------------------------------
# Step 4: Build the LLM prompt
# -------------------------------
prompt = f"""You are an expert at giving accurate answers to questions about texts.
No preamble.
Given the series of questions, answer the questions.
Each question may follow up with answer format, answer options, and context for each question.
It is critical that you follow the answer format and answer options for each question.
When context is provided with a question, refer to it when answering the question.
You are useful, true and concise, and write in perfect English.
Only the question is allowed between the tag. Do not include the answer format, options, or question context in your response.
Only text is allowed between the and tags.
XML tags are not allowed between the and tags.
End your response with a closing tag.
For each question-answer pair, format your response according to the template provided below:
Template for response:
The question
Your answer
...
...
These are the questions:
{question_str}
Transcript:
{{{{ transcript }}}}
"""
# -------------------------------
# Step 5: Query LLM Gateway
# -------------------------------
headers = {"authorization": API_KEY}
response = requests.post(
"https://llm-gateway.assemblyai.com/v1/chat/completions",
headers=headers,
json={
"model": "claude-sonnet-4-5-20250929",
"messages": [{"role": "user", "content": prompt}],
"transcript_id": transcript_id,
"max_tokens": 2000,
},
)
response_json = response.json()
llm_output = response_json["choices"][0]["message"]["content"]
# -------------------------------
# Step 6: Parse and print XML response
```
--------------------------------
### AssemblyAI Event Handlers (Python)
Source: https://www.assemblyai.com/docs/streaming/migration-guides/speechmatics_to_aai_streaming
Example Python code for handling WebSocket events (on_open, on_error, on_close) for AssemblyAI. This includes starting a separate thread for audio streaming upon connection and handling errors and disconnections.
```python
import threading
def on_open(ws):
"""Called when the WebSocket connection is established."""
print("WebSocket connection opened.")
print(f"Connected to: {API_ENDPOINT}")
# Start sending audio data in a separate thread
def stream_audio():
global stream
print("Starting audio streaming...")
while not stop_event.is_set():
try:
audio_data = stream.read(FRAMES_PER_BUFFER, exception_on_overflow=False)
# Send audio data as binary message
ws.send(audio_data, websocket.ABNF.OPCODE_BINARY)
except Exception as e:
print(f"Error streaming audio: {e}")
# If stream read fails, likely means it's closed, stop the loop
break
print("Audio streaming stopped.")
global audio_thread
audio_thread = threading.Thread(target=stream_audio)
audio_thread.daemon = (
True # Allow main thread to exit even if this thread is running
)
audio_thread.start()
def on_error(ws, error):
"""Called when a WebSocket error occurs."""
print(f"\nWebSocket Error: {error}") # Attempt to signal stop on error
stop_event.set()
def on_close(ws, close_status_code, close_msg):
"""Called when the WebSocket connection is closed."""
print(f"\nWebSocket Disconnected: Status={close_status_code}, Msg={close_msg}")
# Ensure audio resources are released
global stream, audio
stop_event.set() # Signal audio thread just in case it's still running
if stream:
if stream.is_active():
stream.stop_stream()
stream.close()
stream = None
if audio:
audio.terminate()
audio = None
# Try to join the audio thread to ensure clean exit
if audio_thread and audio_thread.is_alive():
audio_thread.join(timeout=1.0)
```
--------------------------------
### Python Quickstart
Source: https://www.assemblyai.com/docs/streaming/guides/stream_prerecorded_file_realtime
This Python code snippet demonstrates the initial setup for streaming a pre-recorded file in real time. It imports necessary libraries and sets up basic variables for WebSocket communication and audio file handling.
```python
import websocket
import json
import threading
import time
import wave
import os
from urllib.parse import urlencode
```
--------------------------------
### Speechmatics Event Handlers (Python)
Source: https://www.assemblyai.com/docs/streaming/migration-guides/speechmatics_to_aai_streaming
Example Python code for handling WebSocket events (on_open, on_error, on_close) when migrating from Speechmatics. This includes sending the initial recognition start message and handling connection errors and closures.
```python
import json
def on_open(ws):
"""Called when the WebSocket connection is established."""
print("WebSocket connection opened.")
print(f"Connected to: {API_ENDPOINT}")
# Send StartRecognition message
start_message = {
"message": "StartRecognition",
"audio_format": {
"type": "raw",
"encoding": "pcm_f32le",
"sample_rate": SAMPLE_RATE
},
"transcription_config": {
"language": CONNECTION_PARAMS["language"],
"enable_partials": CONNECTION_PARAMS["enable_partials"],
"max_delay": CONNECTION_PARAMS["max_delay"]
}
}
ws.send(json.dumps(start_message))
def on_error(ws, error):
"""Called when a WebSocket error occurs."""
print(f"\nWebSocket Error: {error}")
# Attempt to signal stop on error
stop_event.set()
def on_close(ws, close_status_code, close_msg):
"""Called when the WebSocket connection is closed."""
print(f"\nWebSocket Disconnected: Status={close_status_code}, Msg={close_msg}")
# Ensure audio resources are released
global stream, audio
stop_event.set() # Signal audio thread just in case it's still running
if stream:
if stream.is_active():
stream.stop_stream()
stream.close()
stream = None
if audio:
audio.terminate()
audio = None
# Try to join the audio thread to ensure clean exit
if audio_thread and audio_thread.is_alive():
audio_thread.join(timeout=1.0)
```
--------------------------------
### Configure environment variables
Source: https://www.assemblyai.com/docs/integrations/recall
Instructions to copy the example environment file and set up API keys and region.
```bash
cp .env.example .env
RECALL_API_KEY=your_recall_api_key
RECALL_REGION=us-west-2
```
--------------------------------
### Pair bad examples with good ones
Source: https://www.assemblyai.com/docs/voice-agents/voice-agent-api/prompting-guide
Example of pairing bad examples with good ones to teach a rule.
```text
When the user describes their project, don't give a feature tour:
Bad: "You could build A, B, or C. What problem are you trying to solve?"
Good: "Yeah, like a receptionist."
```
--------------------------------
### AssemblyAI SDK setup
Source: https://www.assemblyai.com/docs/pre-recorded-audio/migration-guides/aws_to_aai
AssemblyAI SDK setup for transcription.
```python
import assemblyai as aai
aai.settings.api_key = "YOUR-API-KEY"
transcriber = aai.Transcriber()
```
--------------------------------
### AWS SDK setup
Source: https://www.assemblyai.com/docs/pre-recorded-audio/migration-guides/aws_to_aai
AWS SDK setup for transcription.
```python
import boto3
import time
transcribe_client = boto3.client("transcribe")
```
--------------------------------
### AssemblyAI installation snippet
Source: https://www.assemblyai.com/docs/pre-recorded-audio/migration-guides/google_to_aai
Installation snippet for AssemblyAI SDK.
```python
import assemblyai as aai
aai.settings.api_key = "YOUR-API-KEY"
transcriber = aai.Transcriber()
```
--------------------------------
### Quickstart pattern (Python sketch)
Source: https://www.assemblyai.com/docs/coding-agent-prompts
A Python sketch demonstrating the basic pattern for using the Voice Agent API with websockets, sounddevice, and numpy for real-time audio processing and communication.
```python
# pip install websockets sounddevice numpy
import asyncio, base64, json, os
import sounddevice as sd
import websockets
URL = "wss://agents.assemblyai.com/v1/ws"
SAMPLE_RATE = 24_000
async def main():
headers = {"Authorization": f"Bearer {os.environ['ASSEMBLYAI_API_KEY']}"}
async with websockets.connect(URL, additional_headers=headers) as ws:
await ws.send(json.dumps({
"type": "session.update",
"session": {
"system_prompt": "You are a helpful assistant.",
"greeting": "Hi! How can I help?",
"output": {"voice": "ivy"},
},
}))
ready = asyncio.Event()
loop = asyncio.get_running_loop()
mic_q: asyncio.Queue = asyncio.Queue()
def on_mic(indata, *_):
if ready.is_set():
loop.call_soon_threadsafe(mic_q.put_nowait, bytes(indata))
async def pump_mic():
while True:
chunk = await mic_q.get()
await ws.send(json.dumps({
"type": "input.audio",
"audio": base64.b64encode(chunk).decode(),
}))
with sd.InputStream(samplerate=SAMPLE_RATE, channels=1,
dtype="int16", callback=on_mic), \
sd.OutputStream(samplerate=SAMPLE_RATE, channels=1,
dtype="int16") as speaker:
asyncio.create_task(pump_mic())
async for raw in ws:
ev = json.loads(raw)
if ev["type"] == "session.ready":
ready.set()
elif ev["type"] == "reply.audio":
import numpy as np
speaker.write(np.frombuffer(base64.b64decode(ev["data"]), dtype=np.int16))
elif ev["type"] == "reply.done" and ev.get("status") == "interrupted":
speaker.abort(); speaker.start()
asyncio.run(main())
```
--------------------------------
### API Explorer - Get subtitles for transcript
Source: https://www.assemblyai.com/docs/api-reference/transcripts/get-subtitles?explorer=true
This is an example of the cURL command to get subtitles for a transcript, as shown in the API Explorer.
```bash
$ curl https://api.assemblyai.com/v2/transcript/transcript_id/srt \
-H "Authorization: "
```
--------------------------------
### Install ws for Node.js
Source: https://www.assemblyai.com/docs/streaming/guides/stream_prerecorded_file_realtime
Install the ws library for Node.js to enable WebSocket communication.
```bash
npm install ws
```
--------------------------------
### Python Quickstart
Source: https://www.assemblyai.com/docs/streaming/whisper-streaming
A Python example demonstrating how to use the StreamingClient to transcribe audio in real-time, including handling partial and full turn utterances, language detection, and session termination.
```python
if event.utterance:
print(f"[PARTIAL TURN UTTERANCE]: {event.utterance}")
# Display language detection info if available
if event.language_code:
print(f"[UTTERANCE LANGUAGE DETECTION]: {event.language_code} - {event.language_confidence:.2%}")
if event.end_of_turn:
print(f"[FULL TURN TRANSCRIPT]: {event.transcript}")
# Display language detection info if available
if event.language_code:
print(f"[END OF TURN LANGUAGE DETECTION]: {event.language_code} - {event.language_confidence:.2%}")
def on_terminated(self: Type[StreamingClient], event: TerminationEvent):
print(
f"Session terminated: {event.audio_duration_seconds} seconds of audio processed"
)
def on_error(self: Type[StreamingClient], error: StreamingError):
print(f"Error occurred: {error}")
def main():
client = StreamingClient(
StreamingClientOptions(
api_key=api_key,
api_host="streaming.assemblyai.com",
)
)
client.on(StreamingEvents.Begin, on_begin)
client.on(StreamingEvents.Turn, on_turn)
client.on(StreamingEvents.Termination, on_terminated)
client.on(StreamingEvents.Error, on_error)
client.connect(
StreamingParameters(
sample_rate=48000,
speech_model="whisper-rt",
language_detection=True,
)
)
try:
client.stream(
aai.extras.MicrophoneStream(sample_rate=48000)
)
finally:
client.disconnect(terminate=True)
if __name__ == "__main__":
main()
```
--------------------------------
### Installation Command
Source: https://www.assemblyai.com/docs/streaming/label-speakers-and-separate-channels
Command to install the necessary Python packages for running the example.
```bash
pip install assemblyai numpy pyaudio
```
--------------------------------
### Tool Description Example
Source: https://www.assemblyai.com/docs/voice-agents/voice-agent-api/tool-calling
An example of a tool description that guides the model on when to invoke the tool.
```json
{
"description": "Get current weather for any city. Use this whenever the user asks about weather, temperature, conditions, what to wear, or anything weather-dependent. Prefer calling this over guessing."
}
```
--------------------------------
### Quickstart
Source: https://www.assemblyai.com/docs/streaming/guides/turn_detection_improvement_using_async
This code snippet sets up the necessary imports and configuration for audio processing and real-time transcription.
```python
import requests
import time
import json
import pyaudio
import websocket
import threading
from urllib.parse import urlencode
from datetime import datetime
import os
from pathlib import Path
YOUR_API_KEY = "" # Replace with your API key
AUDIO_FOLDER_PATH = "" # Folder containing audio files
# Audio Configuration
SAMPLE_RATE = 16000
CHANNELS = 1
FORMAT = pyaudio.paInt16
FRAMES_PER_BUFFER = 800 # 50ms of audio (0.05s * 16000Hz)
# Global variables for audio stream and websocket
audio = None
stream = None
ws_app = None
audio_thread = None
stop_event = threading.Event()
recorded_frames = []
recording_lock = threading.Lock()
```
--------------------------------
### Python Quickstart Example
Source: https://www.assemblyai.com/docs/pre-recorded-audio/guides/speaker-diarization-with-async-chunking
This Python script demonstrates the setup and core logic for performing speaker diarization using AssemblyAI and Nvidia's NeMo framework. It includes functions for transcription polling, downloading audio files, and identifying the longest monologues for each speaker.
```python
import assemblyai as aai
import requests
import json
import time
import requests
import copy
from pydub import AudioSegment
import os
import nemo.collections.asr as nemo_asr
from pydub import AudioSegment
speaker_model = nemo_asr.models.EncDecSpeakerLabelModel.from_pretrained("nvidia/speakerverification_en_titanet_large")
assemblyai_key = "YOUR_API_KEY"
headers = {
"authorization": assemblyai_key
}
def get_transcript(transcript_id):
polling_endpoint = f"https://api.assemblyai.com/v2/transcript/{transcript_id}"
while True:
transcription_result = requests.get(polling_endpoint, headers=headers).json()
if transcription_result['status'] == 'completed':
# print("Transcript ID:", transcript_id)
return(transcription_result)
break
elif transcription_result['status'] == 'error':
raise RuntimeError(f"Transcription failed: {transcription_result['error']}")
else:
time.sleep(3)
def download_wav(presigned_url, output_filename):
# Download the WAV file from the presigned URL
response = requests.get(presigned_url)
if response.status_code == 200:
print("downloading...")
with open(output_filename, 'wb') as f:
f.write(response.content)
print("successfully downloaded file:", output_filename)
else:
raise Exception("Failed to download file, status code: {}".format(response.status_code))
# Function to identify the longest monologue of each speaker from each clip
# you pass in the utterances and it returns the longest monologue from each speaker on that file
def find_longest_monologues(utterances):
longest_monologues = {}
current_monologue = {}
last_speaker = None # Track the last speaker to identify interruptions
for utterance in utterances:
speaker = utterance['speaker']
start_time = utterance['start']
end_time = utterance['end']
if speaker not in current_monologue:
current_monologue[speaker] = {"start": start_time, "end": end_time}
longest_monologues[speaker] = []
else:
# Extend monologue only if it's the same speaker speaking continuously
if current_monologue[speaker]["end"] == start_time and last_speaker == speaker:
current_monologue[speaker]["end"] = end_time
else:
monologue_length = current_monologue[speaker]["end"] - current_monologue[speaker]["start"]
new_entry = (monologue_length, copy.deepcopy(current_monologue[speaker]))
if len(longest_monologues[speaker]) < 1 or monologue_length > min(longest_monologues[speaker], key=lambda x: x[0])[0]:
if len(longest_monologues[speaker]) == 1:
longest_monologues[speaker].remove(min(longest_monologues[speaker], key=lambda x: x[0]))
longest_monologues[speaker].append(new_entry)
current_monologue[speaker] = {"start": start_time, "end": end_time}
last_speaker = speaker # Update the last speaker
# Check the last monologue for each speaker
for speaker, monologue in current_monologue.items():
monologue_length = monologue["end"] - monologue["start"]
new_entry = (monologue_length, monologue)
if len(longest_monologues[speaker]) < 1 or monologue_length > min(longest_monologues[speaker], key=lambda x: x[0])[0]:
if len(longest_monologues[speaker]) == 1:
longest_monologues[speaker].remove(min(longest_monologues[speaker], key=lambda x: x[0]))
longest_monologues[speaker].append(new_entry)
return longest_monologues
# Create clips of each long monologue and embed the clip
# you pass in the file path and the longest monologue objects returned by the find_longest_monologues function.
```
--------------------------------
### Python Quickstart
Source: https://www.assemblyai.com/docs/streaming/universal-3-pro
This Python code snippet demonstrates how to use the AssemblyAI streaming client with the Universal 3 Pro model.
```python
def on_error(self: Type[StreamingClient], error: StreamingError):
print(f"Error occurred: {error}")
def main():
client = StreamingClient(
StreamingClientOptions(
api_key=api_key,
api_host="streaming.assemblyai.com",
)
)
client.on(StreamingEvents.Begin, on_begin)
client.on(StreamingEvents.Turn, on_turn)
client.on(StreamingEvents.Termination, on_terminated)
client.on(StreamingEvents.Error, on_error)
client.connect(
StreamingParameters(
sample_rate=16000,
speech_model="u3-rt-pro",
)
)
try:
client.stream(
aai.extras.MicrophoneStream(sample_rate=16000)
)
finally:
client.disconnect(terminate=True)
if __name__ == "__main__":
main()
```
--------------------------------
### Node.js Quickstart
Source: https://www.assemblyai.com/docs/streaming/whisper-streaming
A Node.js example demonstrating how to use WebSockets to transcribe audio in real-time, including configuration, audio recording, and handling WebSocket messages.
```javascript
const WebSocket = require("ws");
const mic = require("mic");
const querystring = require("querystring");
const fs = require("fs");
// --- Configuration ---
const YOUR_API_KEY = "YOUR-API-KEY"; // Replace with your actual API key
const CONNECTION_PARAMS = {
sample_rate: 48000,
speech_model: "whisper-rt",
language_detection: true,
};
const API_ENDPOINT_BASE_URL = "wss://streaming.assemblyai.com/v3/ws";
const API_ENDPOINT = `${API_ENDPOINT_BASE_URL}?${querystring.stringify(CONNECTION_PARAMS)}`;
// Audio Configuration
const SAMPLE_RATE = CONNECTION_PARAMS.sample_rate;
const CHANNELS = 1;
// Global variables
let micInstance = null;
let micInputStream = null;
let ws = null;
let stopRequested = false;
// WAV recording variables
let recordedFrames = []; // Store audio frames for WAV file
// --- Helper functions ---
function clearLine() {
process.stdout.write("\r" + " ".repeat(80) + "\r");
}
function formatTimestamp(timestamp) {
return new Date(timestamp * 1000).toISOString();
}
function createWavHeader(sampleRate, channels, dataLength) {
const buffer = Buffer.alloc(44);
// RIFF header
buffer.write("RIFF", 0);
buffer.writeUInt32LE(36 + dataLength, 4);
buffer.write("WAVE", 8);
// fmt chunk
buffer.write("fmt ", 12);
buffer.writeUInt32LE(16, 16); // fmt chunk size
buffer.writeUInt16LE(1, 20); // PCM format
buffer.writeUInt16LE(channels, 22);
buffer.writeUInt32LE(sampleRate, 24);
buffer.writeUInt32LE(sampleRate * channels * 2, 28); // byte rate
buffer.writeUInt16LE(channels * 2, 32); // block align
buffer.writeUInt16LE(16, 34); // bits per sample
// data chunk
buffer.write("data", 36);
buffer.writeUInt32LE(dataLength, 40);
return buffer;
}
function saveWavFile() {
if (recordedFrames.length === 0) {
console.log("No audio data recorded.");
return;
}
// Generate filename with timestamp
const timestamp = new Date().toISOString().replace(/[:.]/g, "-").slice(0, 19);
const filename = `recorded_audio_${timestamp}.wav`;
try {
// Combine all recorded frames
const audioData = Buffer.concat(recordedFrames);
const dataLength = audioData.length;
// Create WAV header
const wavHeader = createWavHeader(SAMPLE_RATE, CHANNELS, dataLength);
// Write WAV file
const wavFile = Buffer.concat([wavHeader, audioData]);
fs.writeFileSync(filename, wavFile);
console.log(`Audio saved to: ${filename}`);
console.log(
`Duration: ${(dataLength / (SAMPLE_RATE * CHANNELS * 2)).toFixed(2)} seconds`
);
} catch (error) {
console.error(`Error saving WAV file: ${error}`);
}
}
// --- Main function ---
async function run() {
console.log("Starting AssemblyAI real-time transcription...");
console.log("Audio will be saved to a WAV file when the session ends.");
console.log(`Connecting websocket to url ${API_ENDPOINT}`);
// Initialize WebSocket connection
ws = new WebSocket(API_ENDPOINT, {
headers: {
Authorization: YOUR_API_KEY,
},
});
// Setup WebSocket event handlers
ws.on("open", () => {
console.log("WebSocket connection opened.");
console.log("Receiving SessionBegins ...");
// Start the microphone
startMicrophone();
});
ws.on("message", (message) => {
try {
const data = JSON.parse(message);
const msgType = data.type;
if (msgType === "Begin") {
```
--------------------------------
### Google Speech-to-Text installation snippet
Source: https://www.assemblyai.com/docs/pre-recorded-audio/migration-guides/google_to_aai
Installation snippet for Google Speech-to-Text client.
```python
from google.cloud import speech
client = speech.SpeechClient()
```
--------------------------------
### cURL quickstart — Step 1: Transcribe with key phrases
Source: https://www.assemblyai.com/docs/getting-started/end-to-end-examples/content-repurposing
This cURL command initiates a transcription process with key phrases enabled, suitable for extracting important terms from audio content.
```bash
curl https://api.assemblyai.com/v2/transcript \
--header "Authorization: YOUR_API_KEY" \
--header "Content-Type: application/json" \
--data '{ \
"audio_url": "https://assembly.ai/wildfires.mp3", \
"speech_models": ["universal-3-pro", "universal-2"], \
"language_detection": true, \
"auto_highlights": true \
}'
```
--------------------------------
### API Error Response Example
Source: https://www.assemblyai.com/docs/faq/i-am-getting-an-error-what-should-i-do
Example of an API response containing an 'error' key with details about the issue.
```json
{
"status": "error",
"error": "Download error, unable to access file at https://example.com/audio.mp3",
...
}
```
--------------------------------
### Python SDK Quick Start
Source: https://www.assemblyai.com/docs/agent-instructions.md
Demonstrates how to transcribe an audio file using the AssemblyAI Python SDK, including configuration for speech models and speaker labels.
```python
# pip install assemblyai
import assemblyai as aai
import os
aai.settings.api_key = os.environ["ASSEMBLYAI_API_KEY"]
config = aai.TranscriptionConfig(
speech_models=["universal-3-pro", "universal-2"], # fallback handled by SDK
speaker_labels=True,
)
transcript = aai.Transcriber(config=config).transcribe("https://assembly.ai/wildfires.mp3")
# Or a local path: .transcribe("./recording.wav")
if transcript.status == aai.TranscriptStatus.error:
raise RuntimeError(transcript.error)
print(transcript.text)
```
--------------------------------
### Fetch Transcript Details
Source: https://www.assemblyai.com/docs/faq/i-am-getting-an-error-what-should-i-do
Example of how to make a GET request to fetch transcript details using curl.
```bash
curl https://api.assemblyai.com/v2/transcript/ \
-H "Authorization: "
```
--------------------------------
### Quickstart
Source: https://www.assemblyai.com/docs/pre-recorded-audio/guides/batch_transcription
This code snippet shows how to set up the AssemblyAI SDK, define folders for audio and transcripts, configure transcription settings, and then transcribe multiple audio files concurrently using threads.
```python
import assemblyai as aai
import threading
import os
aai.settings.api_key = "YOUR_API_KEY"
batch_folder = "audio"
transcription_result_folder = "transcripts"
config = aai.TranscriptionConfig(speech_models=["universal-3-pro", "universal-2"])
transcriber = aai.Transcriber()
def transcribe_audio(audio_file):
transcriber = aai.Transcriber()
transcript = transcriber.transcribe(os.path.join(batch_folder, audio_file), config)
if transcript.status == "completed":
with open(f"{transcription_result_folder}/{audio_file}.txt", "w") as f:
f.write(transcript.text)
elif transcript.status == "error":
print("Error: ", transcript.error)
threads = []
for filename in os.listdir(batch_folder):
thread = threading.Thread(target=transcribe_audio, args=(filename,)))
threads.append(thread)
thread.start()
for thread in threads:
thread.join()
print("All transcriptions are complete.")
```
--------------------------------
### Python Quickstart Example
Source: https://www.assemblyai.com/docs/guides/transcript-citations
This Python script demonstrates how to upload an audio file, transcribe it, retrieve sentences, and use the LLM Gateway for question answering. It also includes functions for semantic search using sentence embeddings.
```python
import datetime
import numpy as np
import requests
import time
from sklearn.neighbors import NearestNeighbors
from sentence_transformers import SentenceTransformer
# Configuration
api_key = ""
base_url = "https://api.assemblyai.com"
headers = {"authorization": api_key}
def upload_file(file_path):
"""Upload a local audio file to AssemblyAI"""
with open(file_path, "rb") as f:
response = requests.post(f"{base_url}/v2/upload", headers=headers, data=f)
if response.status_code != 200:
print(f"Error uploading: {response.status_code}, {response.text}")
response.raise_for_status()
return response.json()["upload_url"]
def transcribe_audio(audio_url):
"""Submit audio for transcription with sentences enabled and poll until complete"""
data = {
"audio_url": audio_url,
"speech_models": ["universal-3-pro"],
"auto_highlights": False,
"sentiment_analysis": False,
"entity_detection": False
}
response = requests.post(f"{base_url}/v2/transcript", headers=headers, json=data)
if response.status_code != 200:
print(f"Error submitting transcription: {response.status_code}, {response.text}")
response.raise_for_status()
transcript_id = response.json()["id"]
polling_endpoint = f"{base_url}/v2/transcript/{transcript_id}"
print("Transcribing...")
while True:
transcript = requests.get(polling_endpoint, headers=headers).json()
if transcript["status"] == "completed":
print("Transcription completed!")
return transcript
elif transcript["status"] == "error":
raise RuntimeError(f"Transcription failed: {transcript['error']}")
else:
time.sleep(3)
def get_sentences(transcript_id):
"""Get sentences from a completed transcript"""
sentences_endpoint = f"{base_url}/v2/transcript/{transcript_id}/sentences"
response = requests.get(sentences_endpoint, headers=headers)
if response.status_code != 200:
print(f"Error getting sentences: {response.status_code}, {response.text}")
response.raise_for_status()
return response.json()["sentences"]
def process_with_llm_gateway(transcript_text, question, context=""):
"""Send transcript to LLM Gateway for question answering"""
prompt = f"""Based on the following transcript, please answer this question:
Question: {question}
Context: {context}
Transcript: {transcript_text}
Please provide a clear and specific answer."""
llm_gateway_data = {
"model": "claude-sonnet-4-5-20250929",
"messages": [
{
"role": "user",
"content": prompt
}
],
"max_tokens": 2000
}
response = requests.post(
"https://llm-gateway.assemblyai.com/v1/chat/completions",
headers=headers,
json=llm_gateway_data
)
result = response.json()
if "error" in result:
raise RuntimeError(f"LLM Gateway error: {result['error']}")
return result['choices'][0]['message']['content']
def sliding_window(elements, distance, stride):
"""Create sliding windows of elements"""
idx = 0
results = []
while idx + distance < len(elements):
results.append(elements[idx:idx + distance])
idx += (distance - stride)
return results
# Main execution
# If using a local file:
audio_url = upload_file("")
# If using a public URL:
# audio_url = ""
# Transcribe audio
transcript = transcribe_audio(audio_url)
transcript_text = transcript["text"]
transcript_id = transcript["id"]
# Get sentences
print("Getting sentences...")
sentences = get_sentences(transcript_id)
# Initialize embedder
embedder = SentenceTransformer("multi-qa-mpnet-base-dot-v1")
embeddings = {}
```
--------------------------------
### Quick Upgrade Example
Source: https://www.assemblyai.com/docs/streaming/migration-guides/universal-to-u3-pro-streaming
Compares the connection parameters for Universal Streaming and Universal-3 Pro Streaming.
```python
# Before (Universal Streaming)
CONNECTION_PARAMS = {
"sample_rate": 16000,
"format_turns": True,
}
# After (Universal-3 Pro Streaming)
CONNECTION_PARAMS = {
"sample_rate": 16000,
"speech_model": "u3-rt-pro",
}
```
--------------------------------
### Python Quickstart
Source: https://www.assemblyai.com/docs/streaming/universal-3-pro/supported-languages
A Python example demonstrating how to set up a WebSocket connection for real-time transcription using the Universal 3 Pro model.
```python
ws_thread = threading.Thread(target=ws_app.run_forever)
ws_thread.daemon = True
ws_thread.start()
try:
while ws_thread.is_alive():
time.sleep(0.1)
except KeyboardInterrupt:
print("\nStopping...")
stop_event.set()
if ws_app and ws_app.sock and ws_app.sock.connected:
ws_app.send(json.dumps({"type": "Terminate"}))
time.sleep(2)
if ws_app:
ws_app.close()
ws_thread.join(timeout=2.0)
if __name__ == "__main__":
run()
```
--------------------------------
### Accessing Named Entities (Gladia Example)
Source: https://www.assemblyai.com/docs/pre-recorded-audio/migration-guides/gladia_to_aai
Example of how to access named entities from a Gladia transcript.
```python
for entity in transcript['result']['named_entity_recognition']['results']:
print(entity['text'])
print(entity['entity_type'])
print(f"Timestamp: {entity['start']} - {entity['end']}\n")
```
--------------------------------
### Install websocket-client for Python
Source: https://www.assemblyai.com/docs/streaming/guides/stream_prerecorded_file_realtime
Install the websocket-client library for Python to enable WebSocket communication.
```bash
pip install websocket-client
```
--------------------------------
### Get subtitles for transcript
Source: https://www.assemblyai.com/docs/api-reference/transcripts/get-subtitles?explorer=true
This example shows how to retrieve subtitles for a transcript in SRT format using cURL.
```bash
$ curl https://api.assemblyai.com/v2/transcript/transcript_id/srt \
-H "Authorization: "
```