Python SDK
Official Python client for the Speech Revolutions STT API. Ships a synchronous SpeechRevolutions client and an asyncio AsyncSpeechRevolutions client with the same one-line transcribe() API.
Source on GitHub · report an issue · PyPI
Install
pip install speechrevolutions
# optional console progress bars (tqdm)
pip install "speechrevolutions[progress]"Quickstart
Point transcribe() at a local path, a URL, raw bytes, or a file object. With the default output_type="json" the SDK parses the response into a transcript-first result object.
from speechrevolutions import SpeechRevolutions
client = SpeechRevolutions() # reads SPEECHREVOLUTIONS_API_KEY
result = client.transcribe("meeting.mp3", speaker_labels=True)
print(result.text)
for u in result.utterances:
print(f"Speaker {u.speaker}: {u.text}")From a URL or file
transcribe() auto-detects http(s) URLs; the transcribe_url() / transcribe_file() aliases make intent explicit.
# auto-detected
result = client.transcribe("https://example.com/audio.mp3")
# explicit aliases
result = client.transcribe_url("https://example.com/audio.mp3")
result = client.transcribe_file("./local.wav")Options
Pass options as keyword arguments (ElevenLabs / Deepgram style) or as a TranscribeOptions config object (AssemblyAI style):
from speechrevolutions import TranscribeOptions
# kwargs — `diarize` is a Deepgram-compatible alias for `speaker_labels`
result = client.transcribe("a.mp3", diarize=True, output_type="srt")
# config object
result = client.transcribe(
"a.mp3",
options=TranscribeOptions(speaker_labels=True),
)| Kwarg | Type | Default | Notes |
|---|---|---|---|
output_type | str | "json" | txt | json | srt | vtt | docx | pdf |
word_timestamps | bool | True | Per-word start/end times |
speaker_labels | bool | True | Label who spoke each segment |
diarize | bool | — | Alias for speaker_labels |
nltk | bool | True | Restore punctuation & capitalization |
tier | str | "standard" | standard — the only tier currently available |
custom_vocabulary | list[str] | None | None | Domain terms to bias toward |
on_progress | Callable | None | Transcription-progress callback (see below) |
on_upload_progress | Callable | None | Upload byte-progress callback |
progress | bool | False | Render live console bars |
Live progress
Unlike AssemblyAI and Deepgram — which expose no percentage for pre-recorded audio — you get real-time progress for both the file upload and the transcription, as a console bar, a callback, or both. They compose: the bars render and your callbacks still fire for every event.
# 1. Console bars — an "Uploading" byte bar, then a "Transcribing" bar.
# Uses tqdm if installed: pip install "speechrevolutions[progress]"
result = client.transcribe("meeting.mp3", progress=True)
# 2. Programmatic — read event.percent (0-100, or None until total is known)
def on_progress(event): # transcription
print(event.percent, event.step) # e.g. 42.0 "transcribe"
def on_upload(event): # upload (event.step == "upload")
print("upload", event.percent)
result = client.transcribe(
"meeting.mp3",
on_progress=on_progress,
on_upload_progress=on_upload,
progress=True, # bars AND callbacks together
)Each ProgressEvent carries .completed, .total, .step, .elapsed_seconds, and a computed .percent (0–100, or None when the total is not yet known).
Building a UI? Live progress for web apps shows how to fold both callbacks into a single 0–100 bar you can serve to your frontend.
Async
The asyncio client mirrors the sync API. Use async with so the underlying httpx client is closed on exit.
import asyncio
from speechrevolutions import AsyncSpeechRevolutions
async def main():
async with AsyncSpeechRevolutions() as client:
result = await client.transcribe(
"meeting.mp3",
speaker_labels=True,
progress=True,
)
print(result.text)
asyncio.run(main())Result shape
With output_type="json" the SDK returns a transcript-first object. The client never writes files unless you call save().
| Member | Description |
|---|---|
result.text | Full transcript (AssemblyAI / ElevenLabs style) |
result.transcript | Deepgram-style alias for the same text |
result.words | Word + start / end / speaker |
result.utterances | AssemblyAI-style speaker turns |
result.to_deepgram() | Deepgram-shaped dict for migrations |
result.to_dict() | Normalized JSON dict |
result.content / result.save(path) | Raw bytes / write to disk |
dg = result.to_deepgram()
print(dg["results"]["channels"][0]["alternatives"][0]["transcript"])
# save() appends the output type when the path has no extension
out = result.save("output") # -> "output.json"
print("saved to", out)Webhooks
Pass callback_url to be notified when a job finishes instead of holding the call open — the right pattern for server and background workloads. On completion or permanent failure the platform POSTs a signed JSON body to your URL:
client.transcribe("meeting.mp3", callback_url="https://you.example.com/hook")The POST body is {job_id, status: "completed"|"failed", download_url?, step?, reason?}, signed with HMAC-SHA256 over the raw body in the X-SR-Signature: sha256=<hex> header (plus X-SR-Event with the status and a unique X-SR-Delivery id). Always verify the signature against the raw bytes you received — not a re-serialized dict — using a constant-time comparison.
import hashlib
import hmac
def verify_signature(raw_body: bytes, signature_header: str, signing_secret: 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 "")
# FastAPI receiver
# @app.post("/webhooks/speechrevolutions")
# async def receive(request: Request):
# raw = await request.body()
# if not verify_signature(raw, request.headers.get("X-SR-Signature", ""), SECRET):
# raise HTTPException(status_code=401, detail="bad signature")
# event = json.loads(raw)
# if event["status"] == "completed":
# ... # mark done; fetch event["download_url"]
# else:
# ... # event["step"], event["reason"]
# return {"ok": True} # a 2xx acks delivery (we retry on 5xx)Retrieve results later
Fetch a job by id anytime — useful after a webhook, or when rebuilding state after a restart. The download URL is regenerated on demand, so results are fetchable long after the original upload.
# Get a job's current status (processing | completed | failed)
status = client.get_job_status(job_id)
print(status.status)
if status.is_completed:
result = client.get_transcript(job_id) # downloads + parses into a Transcript
print(result.text)
elif status.is_failed:
print(status.failed_stage, status.reason)
# List your most-recent jobs (newest first), cursor-paginated
page = client.list_jobs(limit=50)
print(page["jobs"], page["next_before"])
for job in page["jobs"]:
print(job["job_id"], job["created_at"])Robustness
Configure retries, backoff, and an outbound proxy on the client. Transient 429/5xx/network errors are retried automatically (honoring the Retry-After header).
client = SpeechRevolutions(
max_retries=3,
retry_backoff=0.5, # seconds, exponential
proxies={"https": "http://proxy.internal:8080"},
)Errors are typed and carry a .status_code and the server .request_id for correlating with support.
from speechrevolutions.exceptions import RateLimitError, AuthenticationError
try:
result = client.transcribe("meeting.mp3")
except RateLimitError as e:
print(e.status_code, e.request_id) # e.g. 429 "req_..."
except AuthenticationError as e:
print(e.status_code, e.request_id)Auth
The SDK reads the key from either environment variable:
export SPEECHREVOLUTIONS_API_KEY=stt_...Or pass it explicitly:
client = SpeechRevolutions(api_key="stt_...")Under the hood
The SDK drives the /api/v1/upload flow (create job → presigned PUT → complete) and waits on the SSE job stream, converting the server's completed/total counts into percent for you.