Deepgram Python Speech-to-Text

import os
from dotenv import load_dotenv
load_dotenv()

from deepgram import DeepgramClient

client = DeepgramClient()  # reads DEEPGRAM_API_KEY from env
# or: DeepgramClient(api_key=os.environ["DEEPGRAM_API_KEY"])

response = client.listen.v1.media.transcribe_url(
    url="https://dpgr.am/spacewalk.wav",
    model="nova-3",
    smart_format=True,
    punctuate=True,
)
transcript = response.results.channels[0].alternatives[0].transcript

with open("audio.wav", "rb") as f:
    audio_bytes = f.read()

response = client.listen.v1.media.transcribe_file(
    request=audio_bytes,
    model="nova-3",
)

import threading
from deepgram.core.events import EventType
from deepgram.listen.v1.types import (
    ListenV1Results, ListenV1Metadata,
    ListenV1SpeechStarted, ListenV1UtteranceEnd,
)

with client.listen.v1.connect(
    model="nova-3",
    interim_results=True,    # ← emit partial results while user is still speaking
    utterance_end_ms=1000,   # silence (ms) before server emits UtteranceEnd
    vad_events=True,         # SpeechStarted events
    smart_format=True,
) as conn:
    # Mutable container so the on_message closure can update state without `global`
    state = {"last_interim_len": 0}

    def on_message(m):
        if isinstance(m, ListenV1Results) and m.channel and m.channel.alternatives:
            transcript = m.channel.alternatives[0].transcript
            if not transcript:
                return
            if m.is_final:
                # Final segment: overwrite the running interim line, newline if utterance ended
                pad = " " * max(0, state["last_interim_len"] - len(transcript))
                end = "\n" if m.speech_final else ""
                print(f"\r{transcript}{pad}", end=end, flush=True)
                state["last_interim_len"] = 0
            else:
                # Interim: keep overwriting the same console line as the user speaks
                print(f"\r{transcript}", end="", flush=True)
                state["last_interim_len"] = len(transcript)
        elif isinstance(m, ListenV1UtteranceEnd):
            print()  # newline; UtteranceEnd fires after final results when audio goes silent
        elif isinstance(m, ListenV1SpeechStarted):
            pass  # optional: reset UI when a new utterance begins

    conn.on(EventType.OPEN,    lambda _: print("connected"))
    conn.on(EventType.MESSAGE, on_message)
    conn.on(EventType.CLOSE,   lambda _: print("\nclosed"))
    conn.on(EventType.ERROR,   lambda e: print(f"\nerr: {e}"))

    # Start receive loop in background so we can send concurrently
    threading.Thread(target=conn.start_listening, daemon=True).start()

    for chunk in audio_chunks:         # raw PCM bytes at declared encoding/sample_rate
        conn.send_media(chunk)

    conn.send_finalize()               # flush final partial before closing

from deepgram import AsyncDeepgramClient
client = AsyncDeepgramClient()

response = await client.listen.v1.media.transcribe_url(url=..., model="nova-3")

async with client.listen.v1.connect(model="nova-3") as conn:
    # same .on(...) handlers, then:
    await conn.start_listening()

import asyncio
from deepgram import AsyncDeepgramClient

client = AsyncDeepgramClient()

async def transcribe(url: str) -> str:
    response = await client.listen.v1.media.transcribe_url(
        url=url,
        model="nova-3",
        smart_format=True,
    )
    # `response` is the FULL transcription — no polling, no callback, just await.
    return response.results.channels[0].alternatives[0].transcript

text = asyncio.run(transcribe("https://dpgr.am/spacewalk.wav"))

response = client.listen.v1.media.transcribe_url(
    url="https://dpgr.am/spacewalk.wav",
    callback="https://your.app/deepgram-webhook",
    callback_method="POST",         # or "PUT"
    model="nova-3",
    smart_format=True,
)
print(f"Accepted; tracking id: {response.request_id}")
# response is a "listen accepted" — NOT the transcript. Wait for your webhook.