The Diffio Python SDK helps you call the Diffio API from Python. This version covers project creation, upload, generation, progress checks, and download URLs. Requires Python 3.8 or later.
pip install diffioFor local development:
cd diffio-python
pip install -e .Set the API key with DIFFIO_API_KEY. If you need to set the base URL explicitly, use the production endpoint with DIFFIO_API_BASE_URL.
export DIFFIO_API_KEY="diffio_live_..."
export DIFFIO_API_BASE_URL="https://api.diffio.ai/v1"Use request options to override headers, timeouts, retries, or the API key per request.
You can also pass timeoutInSeconds as an alias for timeout.
Generation creation is retried only when you supply a non-empty idempotencyKey,
even when maxRetries is configured globally or per request. Without a key, a
timeout or lost response can hide an accepted generation, so the SDK returns the
error after the first attempt. With a key, retries send the same key and payload.
Reuse that key when manually retrying the same operation; use a new key for a new
generation. The audio isolation and restore helpers do not supply a key and do
not automatically retry their generation-creation step.
from diffio import DiffioClient, RequestOptions
client = DiffioClient(apiKey="diffio_live_...")
projects = client.list_projects(
requestOptions=RequestOptions(
headers={"X-Debug": "1"},
timeout=30.0,
maxRetries=2,
retryBackoff=0.5,
)
)create_project uploads the file and returns the project metadata.
from diffio import DiffioClient
client = DiffioClient(apiKey="diffio_live_...")
file_path = "sample.wav"
project = client.create_project(
filePath=file_path,
)
generation = client.create_generation(
apiProjectId=project.apiProjectId,
model="diffio-4.0-flash",
sampling={"steps": 12, "guidance": 1.5},
idempotencyKey="restore-job-2026-001",
requestOptions={"maxRetries": 2},
)
print(generation.generationId)
print(generation.idempotentReplay)Use one stable idempotencyKey for every retry of the same logical generation request.
The response's optional idempotentReplay value is True when the API returns the
result of an earlier request with that key. Use a new key for a different generation.
from diffio import DiffioClient
client = DiffioClient(apiKey="diffio_live_...")
result = client.audio_isolation.isolate(
filePath="sample.wav",
model="diffio-4.0-flash",
sampling={"steps": 12, "guidance": 1.5},
)
print(result.generation.generationId)This helper runs the full flow and returns the downloaded bytes plus a metadata dict.
from diffio import DiffioClient
client = DiffioClient(apiKey="diffio_live_...")
audio_bytes, info = client.restore_audio(
filePath="sample.wav",
model="diffio-4.0-flash",
sampling={"steps": 12, "guidance": 1.5},
onProgress=lambda progress: print(progress.status),
)
if info["error"]:
print(info["error"])
else:
with open("restored.mp3", "wb") as handle:
handle.write(audio_bytes)
print(info["apiProjectId"], info["generationId"])wait_for_generation and generations.wait_for_complete wait for the overall
status to become complete. Individual stages can reach 100% while video
restoration or final settlement is still pending; stage progress alone does not
indicate overall completion.
For Diffio 2.0, complete means restored media is ready. Transcription can still
be pending, become available later, or finish as unavailable. Read
progress.transcription.status independently; progress.transcription is None
for older responses that do not report availability. A completed generation
remains successful if transcription is unavailable. Completion webhooks expose
the same optional event.transcription object; a later transcript does not emit
another generation.completed event.
from diffio import DiffioClient
client = DiffioClient(apiKey="diffio_live_...")
progress = client.generations.get_progress(
generationId="gen_123",
apiProjectId="proj_123",
)
print(progress.status)
if progress.transcription is not None:
print(progress.transcription.status)from diffio import DiffioClient
client = DiffioClient(apiKey="diffio_live_...")
download = client.generations.download(
generationId="gen_123",
apiProjectId="proj_123",
downloadType="mp3",
downloadFilePath="restored.mp3",
)
print(download.downloadUrl)If you only need the URL, use client.generations.get_download.
Set downloadType="transcript" to download the transcript JSON artifact when the generation has one.
Pending transcripts return DiffioApiError with statusCode == 409 and
responseBody["code"] == "TRANSCRIPT_PENDING". Unavailable transcripts return
statusCode == 404 and responseBody["code"] == "TRANSCRIPT_UNAVAILABLE". These
responses include responseBody["transcription"]["status"]. Check the error code
to distinguish them from other 409 or 404 errors.
from diffio import DiffioApiError
try:
transcript = client.generations.download(
generationId="gen_123",
apiProjectId="proj_123",
downloadType="transcript",
downloadFilePath="word_timestamps.json",
)
except DiffioApiError as exc:
body = exc.responseBody if isinstance(exc.responseBody, dict) else {}
if exc.statusCode == 409 and body.get("code") == "TRANSCRIPT_PENDING":
print("Transcript is pending; check progress and retry later.")
elif exc.statusCode == 404 and body.get("code") == "TRANSCRIPT_UNAVAILABLE":
print("Transcript is unavailable; restored media remains available.")
else:
raiserestore_audio(downloadType="transcript") also makes one download request after
media completion. It does not wait for a pending transcript. With its default
raiseOnError=False, it returns (None, info) and preserves the API error in
info["statusCode"] and info["responseBody"]; info["status"] can still be
complete because media restoration succeeded. With raiseOnError=True, it raises
the same DiffioApiError and attaches the metadata as exc.restoreInfo. Callers
can poll progress and retry the transcript download explicitly. Audio and video
downloads proceed independently of transcription availability.
Agent keys can manage account settings, scoped keys, usage, and webhook endpoints.
settings = client.account.get_settings()
key = client.api_keys.create(
label="Backend worker",
scopes=["projects:read", "projects:write", "generations:read", "generations:write", "artifacts:read"],
)
usage = client.usage.summary(apiKeyId=key.keyId)
webhook = client.webhooks.configure(
mode="live",
url="https://example.com/webhooks/diffio",
eventTypes=["generation.completed", "generation.failed"],
apiKeyId=key.keyId,
)from diffio import DiffioClient
client = DiffioClient(apiKey="diffio_live_...")
projects = client.projects.list()
for project in projects.projects:
print(project.apiProjectId, project.status)from diffio import DiffioClient
client = DiffioClient(apiKey="diffio_live_...")
generations = client.projects.list_generations(apiProjectId="proj_123")
for generation in generations.generations:
print(generation.generationId, generation.status)from diffio import DiffioClient
client = DiffioClient(apiKey="diffio_live_...")
event = client.webhooks.send_test_event(
eventType="generation.completed",
mode="live",
samplePayload={"apiProjectId": "proj_123"},
)
print(event.svixMessageId)Use the raw request body (bytes) plus the svix-* headers and your webhook signing secret.
from fastapi import FastAPI, Request, HTTPException
from diffio import DiffioClient
import os
app = FastAPI()
client = DiffioClient(apiKey=os.environ["DIFFIO_API_KEY"])
@app.post("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/webhooks/diffio")
async def diffio_webhook(request: Request):
payload = await request.body()
headers = request.headers
try:
event = client.webhooks.verify_signature(
payload=payload,
headers=headers,
secret=os.environ["DIFFIO_WEBHOOK_SECRET"],
)
except Exception:
raise HTTPException(status_code=400, detail="Invalid signature")
print("Webhook received", event.eventType)
return {"ok": True}- Audio restoration CLI tutorial:
tutorials/audio-restoration-cli/README.md
Use Python 3.8 or later.
cd diffio-python
python -m pip install -r requirements-dev.txt
python -m pytest