Using Speech Revolutions with FastAPI

Integrate Speech Revolutions into a FastAPI service: submit a job from an endpoint, write progress from the async client's callbacks into a per-job store, expose a /progress/{job_id} endpoint for your frontend to poll, and receive signed completion webhooks. Your API key stays in the server environment.

Keep the key server-side

AsyncSpeechRevolutions() reads SPEECHREVOLUTIONS_API_KEY from the environment. The key stays on your FastAPI host; the browser only sees job IDs and progress values.

Install

bash

pip install fastapi uvicorn python-multipart speechrevolutions

export SPEECHREVOLUTIONS_API_KEY=stt_...

Submit in a background task and expose a progress endpoint

Start the transcription in a FastAPI BackgroundTasks job so the request returns immediately. The async client's on_upload_progress and on_progress callbacks write to a per-job store, and the /progress/{job_id} endpoint returns the latest snapshot. This follows the live_progress_webapp.py pattern from the cookbook.

main.py

import threading
import uuid
from dataclasses import dataclass, field

from fastapi import FastAPI, BackgroundTasks, UploadFile, HTTPException
from speechrevolutions import AsyncSpeechRevolutions, ProgressEvent

app = FastAPI()

# Weight upload + transcription into one 0-100 bar (see the live-progress guide).
UPLOAD_WEIGHT = 0.15
TRANSCRIBE_WEIGHT = 0.85


@dataclass
class JobProgress:
    phase: str = "starting"          # "upload" | "transcribe" | "done" | "failed"
    percent: float = 0.0             # overall 0-100
    text: str | None = None
    lock: threading.Lock = field(default_factory=threading.Lock, repr=False)

    def _set(self, phase: str, overall: float) -> None:
        with self.lock:
            self.phase = phase
            self.percent = max(self.percent, round(overall, 1))  # never go backwards

    def on_upload(self, e: ProgressEvent) -> None:
        self._set("upload", (e.percent or 0.0) * UPLOAD_WEIGHT)

    def on_transcribe(self, e: ProgressEvent) -> None:
        self._set("transcribe", UPLOAD_WEIGHT * 100 + (e.percent or 0.0) * TRANSCRIBE_WEIGHT)

    def snapshot(self) -> dict:
        with self.lock:
            return {"phase": self.phase, "percent": self.percent, "text": self.text}


JOBS: dict[str, JobProgress] = {}


async def run_job(job_id: str, audio: bytes) -> None:
    store = JOBS[job_id]
    try:
        async with AsyncSpeechRevolutions() as client:
            result = await client.transcribe(
                audio,
                on_upload_progress=store.on_upload,
                on_progress=store.on_transcribe,
                speaker_labels=True,
            )
        with store.lock:
            store.phase, store.percent, store.text = "done", 100.0, result.text
    except Exception:
        store._set("failed", store.percent)


@app.post("/transcribe")
async def transcribe(file: UploadFile, background: BackgroundTasks):
    job_id = uuid.uuid4().hex
    JOBS[job_id] = JobProgress()
    audio = await file.read()
    background.add_task(run_job, job_id, audio)  # returns immediately
    return {"job_id": job_id}


@app.get("/progress/{job_id}")
def progress(job_id: str):
    store = JOBS.get(job_id)
    if store is None:
        raise HTTPException(status_code=404, detail="unknown job")
    return store.snapshot()  # {"phase": "transcribe", "percent": 63.5, "text": null}

Multiple workers need a shared store

The in-memory JOBS dict works only when one worker handles both the submit and the poll. With multiple Uvicorn or Gunicorn workers, store the snapshot in Redis or your database so any worker can answer the poll. See Live progress for web apps for the weighting details.

Fire-and-forget with submit()

If you don't need live progress, submit() returns a job ID without holding the connection open. Poll get_job_status() (or set a callback_url) and fetch the transcript when the job finishes.

python

async with AsyncSpeechRevolutions() as client:
    job_id = await client.submit("meeting.mp3", speaker_labels=True)
    # ...later, from a poller or a webhook:
    status = await client.get_job_status(job_id)   # .status: processing|completed|failed
    if status.is_completed:
        result = await client.get_transcript(job_id)
        print(result.text)

Signed webhook receiver

For long jobs, pass a callback_url and Speech Revolutions POSTs to it when the job finishes. The raw body is signed with HMAC-SHA256 and the signature is sent in the X-SR-Signature: sha256=<hex> header. Verify it against the exact bytes you received, not a re-serialized dict, with a constant-time comparison. This follows webhook_receiver_fastapi.py from the cookbook. Copy your signing secret from the console (API Keys → Webhook signing secret) into SPEECHREVOLUTIONS_WEBHOOK_SECRET. The webhooks guide covers the payload, retries, and secret rotation.

webhooks.py

import hashlib
import hmac
import json
import os

from fastapi import Request, HTTPException

SIGNING_SECRET = os.environ["SPEECHREVOLUTIONS_WEBHOOK_SECRET"]


def verify_signature(raw_body: bytes, signature_header: str) -> bool:
    """Return True if X-SR-Signature matches the raw request body."""
    expected = "sha256=" + hmac.new(
        SIGNING_SECRET.encode(), raw_body, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature_header or "")


@app.post("/webhooks/stt")
async def receive(request: Request):
    raw = await request.body()  # verify against the exact bytes received
    if not verify_signature(raw, request.headers.get("X-SR-Signature", "")):
        raise HTTPException(status_code=401, detail="bad signature")

    event = json.loads(raw)
    # {job_id, status: "completed" | "failed", download_url?, step?, reason?}
    if event["status"] == "completed":
        ...  # mark done; fetch event["download_url"] or client.get_transcript(...)
    else:
        ...  # event["step"], event["reason"]
    return {"ok": True}  # a 2xx acks delivery; 5xx is retried

How it works

The async client runs the upload flow and waits on the SSE job stream, converting its completed/total counts into percent. See the Python SDK for the full API.