Python SDK

SDK Download

Download koldan-sdk-python-9.1.3.zip

Setup

Download the Python SDK archive from SDK Overview, extract it, then install it into a virtual environment.

python -m venv .venv
.venv\Scripts\activate
pip install <PATH_TO_EXTRACTED_PYTHON_SDK>

Configure the shared environment variables from Getting Started, and point KOLDAN_FILE_PATH to the audio file you want to transcribe.

Upload And Transcribe With `upload_and_transcribe`

The example below keeps the request intentionally minimal: upload one file, start transcription with the general model alias, poll until the job finishes, and print either the transcript or the error details.

import os
import time

import koldan_sdk
from koldan_sdk.api.speech_transcriptions_api import SpeechTranscriptionsApi
from koldan_sdk.models.speech_service_diarization_options import SpeechServiceDiarizationOptions
from koldan_sdk.models.speech_service_transcription_language_options import SpeechServiceTranscriptionLanguageOptions
from koldan_sdk.models.speech_service_transcription_options import SpeechServiceTranscriptionOptions
from koldan_sdk.models.transcription_job_status import TranscriptionJobStatus

config = koldan_sdk.Configuration(host=os.environ["KOLDAN_BASE_URL"])
config.api_key["apiKey"] = os.environ["KOLDAN_API_KEY"]

with koldan_sdk.ApiClient(config) as client:
    api = SpeechTranscriptionsApi(client)

    with open(os.environ["KOLDAN_FILE_PATH"], "rb") as source:
        transcription_options = SpeechServiceTranscriptionOptions(
            model="general",
            language=SpeechServiceTranscriptionLanguageOptions(),
            punctuation=False,
            capitalization=False
        )
        diarization_options = SpeechServiceDiarizationOptions(
            enabled=False
        )
        response = api.upload_and_transcribe(
            name="meeting-recording",
            transcription=transcription_options.to_json(),
            file=(os.path.basename(source.name), source.read()),
            diarization=diarization_options.to_json(),
        )

    job = response.job
    while True:
        job = api.get_job(job.id)
        print(f"Job status: {job.status}")

        if job.status in {TranscriptionJobStatus.COMPLETED, TranscriptionJobStatus.FAILED, TranscriptionJobStatus.CANCELLED}:
            break

        time.sleep(2)

    if job.status == TranscriptionJobStatus.COMPLETED:
        print(job.result.text)
    else:
        print(job.errors)

Run

python upload_file.py

Expected Result

A successful run prints Job status: ... updates and then the transcription text from your uploaded audio.

If the job fails, the example prints the error array returned by the API.

Next Step: Webhooks and Automatic Cleanup

For long-running or production workloads, use a webhook to receive completion events without polling. The example below verifies signed requests, persists the result, handles retries, and can optionally purge the uploaded file and its derived resources.

Persist before purge

Purging is irreversible. Save every result your application needs outside Koldan before enabling automatic purge. After the source file is purged, its media, listening audio, transcription results, and summary results cannot be retrieved from Koldan.

How the Flow Works

sequenceDiagram
    participant U as Upload client
    participant K as Koldan
    participant W as Webhook server
    participant S as Application storage

    U->>K: Upload and transcribe with webhook
    K-->>U: File ID and job ID
    K->>W: transcription.completed
    W->>W: Verify signature and event ID
    W->>S: Persist the result
    opt Automatic purge enabled
        W->>K: DELETE file?purge=true
        K-->>W: 204 No Content
    end
    W-->>K: 204 No Content

Koldan also sends transcription.failed when processing fails. The receiver below stores failed events but only purges automatically after transcription.completed. You can apply a different cleanup policy if failed uploads must also be removed immediately.

Prerequisites

An API key with speech:files:write, speech:transcriptions:write, and speech:transcriptions:read
The speech:files:delete scope when automatic purge is enabled
A webhook URL that Koldan can reach

Webhook URLs must use HTTPS by default. For local development, expose the webhook server through an HTTPS tunnel or use an appropriately configured development deployment. A webhook URL such as http://localhost:8000 usually refers to the Koldan server itself and is also blocked by the default webhook security settings.

Install the webhook server dependencies in the same environment as the generated SDK:

pip install fastapi uvicorn

Configure the Webhook

Keep the three shared environment variables from Getting Started, then add:

WindowsmacOS / Linux

$env:KOLDAN_WEBHOOK_URL = "https://app.example.com/webhooks/koldan/transcriptions"
$env:KOLDAN_WEBHOOK_SECRET = "replace-with-a-random-secret"
$env:KOLDAN_PURGE_AFTER_RECEIPT = "false"
$env:KOLDAN_RESULTS_DIR = ".\koldan-results"

export KOLDAN_WEBHOOK_URL="https://app.example.com/webhooks/koldan/transcriptions"
export KOLDAN_WEBHOOK_SECRET="replace-with-a-random-secret"
export KOLDAN_PURGE_AFTER_RECEIPT="false"
export KOLDAN_RESULTS_DIR="./koldan-results"

Start with KOLDAN_PURGE_AFTER_RECEIPT=false. Change it to true only after verifying that your application persists everything it needs.

Create the Webhook Server

Save this as webhook_server.py:

import asyncio
import hashlib
import hmac
import json
import os
from pathlib import Path
from uuid import UUID

import koldan_sdk
from fastapi import FastAPI, HTTPException, Request, Response
from koldan_sdk.api.speech_files_api import SpeechFilesApi
from koldan_sdk.api.speech_transcriptions_api import SpeechTranscriptionsApi
from koldan_sdk.exceptions import ApiException

app = FastAPI()
event_lock = asyncio.Lock()

WEBHOOK_SECRET = os.environ["KOLDAN_WEBHOOK_SECRET"]
PURGE_AFTER_RECEIPT = os.getenv(
    "KOLDAN_PURGE_AFTER_RECEIPT", "false"
).lower() == "true"
RESULTS_DIR = Path(os.getenv("KOLDAN_RESULTS_DIR", "./koldan-results"))
RESULTS_DIR.mkdir(parents=True, exist_ok=True)


def verify_signature(raw_body: bytes, signature_header: str | None) -> None:
    expected = "sha256=" + hmac.new(
        WEBHOOK_SECRET.encode("utf-8"),
        raw_body,
        hashlib.sha256,
    ).hexdigest()

    if not signature_header or not hmac.compare_digest(
        expected, signature_header
    ):
        raise HTTPException(status_code=401, detail="Invalid webhook signature")


def save_json_atomically(path: Path, value: dict) -> None:
    temporary_path = path.with_suffix(path.suffix + ".tmp")
    temporary_path.write_text(
        json.dumps(value, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    temporary_path.replace(path)


def purge_source_file(job_id: UUID) -> None:
    config = koldan_sdk.Configuration(host=os.environ["KOLDAN_BASE_URL"])
    config.api_key["apiKey"] = os.environ["KOLDAN_API_KEY"]

    with koldan_sdk.ApiClient(config) as client:
        transcriptions_api = SpeechTranscriptionsApi(client)
        files_api = SpeechFilesApi(client)

        try:
            job = transcriptions_api.get_job(job_id)
            files_api.delete_file(job.file.id, purge=True)
        except ApiException as exc:
            # A retry may arrive after a previous attempt already purged the file.
            if exc.status not in {404, 409}:
                raise


@app.post("/webhooks/koldan/transcriptions", status_code=204)
async def receive_transcription_webhook(request: Request) -> Response:
    raw_body = await request.body()
    verify_signature(
        raw_body,
        request.headers.get("X-Webhook-Signature"),
    )

    webhook_id = request.headers.get("X-Webhook-Id")
    if not webhook_id:
        raise HTTPException(status_code=400, detail="Missing X-Webhook-Id")

    payload = json.loads(raw_body)
    event = payload.get("event")
    job_id_value = payload.get("jobId")
    if event not in {"transcription.completed", "transcription.failed"}:
        raise HTTPException(status_code=400, detail="Unsupported event")
    if not job_id_value:
        raise HTTPException(status_code=400, detail="Missing jobId")

    try:
        job_id = UUID(job_id_value)
        webhook_uuid = UUID(webhook_id)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail="Invalid event ID") from exc

    result_path = RESULTS_DIR / f"{job_id}.json"
    completed_path = RESULTS_DIR / f"{webhook_uuid}.done"

    # The lock is sufficient for this single-process example. Production
    # deployments should enforce webhook-ID uniqueness in durable storage.
    async with event_lock:
        if completed_path.exists():
            return Response(status_code=204)

        # Persist first. Never place the purge call before this operation.
        save_json_atomically(result_path, payload)

        if PURGE_AFTER_RECEIPT and event == "transcription.completed":
            purge_source_file(job_id)

        completed_path.touch()

    return Response(status_code=204)

The signature is an HMAC-SHA256 digest of the exact raw HTTP request body. Verify it before parsing JSON. X-Webhook-Id remains the same when Koldan retries a delivery, so it can be used as an idempotency key.

Run the server:

uvicorn webhook_server:app --host 0.0.0.0 --port 8000

Submit a Transcription with a Webhook

Save this as submit_transcription.py:

import os

import koldan_sdk
from koldan_sdk.api.speech_transcriptions_api import SpeechTranscriptionsApi
from koldan_sdk.models.speech_service_webhook import SpeechServiceWebhook
from koldan_sdk.models.speech_service_diarization_options import SpeechServiceDiarizationOptions
from koldan_sdk.models.speech_service_transcription_language_options import SpeechServiceTranscriptionLanguageOptions
from koldan_sdk.models.speech_service_transcription_options import SpeechServiceTranscriptionOptions

config = koldan_sdk.Configuration(host=os.environ["KOLDAN_BASE_URL"])
config.api_key["apiKey"] = os.environ["KOLDAN_API_KEY"]

with koldan_sdk.ApiClient(config) as client:
    api = SpeechTranscriptionsApi(client)

    transcription_options = SpeechServiceTranscriptionOptions(
        model="general",
        language=SpeechServiceTranscriptionLanguageOptions(),
        punctuation=False,
        capitalization=False,
    )
    diarization_options = SpeechServiceDiarizationOptions(enabled=False)
    webhook = SpeechServiceWebhook(
        url=os.environ["KOLDAN_WEBHOOK_URL"],
        secret=os.environ["KOLDAN_WEBHOOK_SECRET"],
        include_segments=True,
    )

    with open(os.environ["KOLDAN_FILE_PATH"], "rb") as source:
        response = api.upload_and_transcribe(
            name="meeting-recording",
            transcription=transcription_options.to_json(),
            file=(os.path.basename(source.name), source.read()),
            diarization=diarization_options.to_json(),
            webhook=webhook.to_json(),
            generate_listening_audio=False,
        )

    print(f"Uploaded file: {response.file.id}")
    print(f"Created job: {response.job.id}")

Run the submission client after the webhook server is reachable:

python submit_transcription.py

The client exits after Koldan accepts the job. The webhook server later writes the terminal event to KOLDAN_RESULTS_DIR.

Choose a Cleanup Policy

Policy	API operation	Result
Retain	No cleanup request	Koldan's configured retention policy applies.
Discard source content	`POST /files/{id}/discard-content`	Removes stored media while keeping file metadata and transcription results available.
Purge the complete file tree	`DELETE /files/{id}?purge=true`	Permanently removes media, listening audio, transcription result/error data, and summary result/error data associated with the file.

The example implements the retain and purge the complete file tree policies. Set KOLDAN_PURGE_AFTER_RECEIPT=true to enable the latter.

Webhook payload with segments

When include_segments=True is set in the webhook configuration (as shown above), the completed webhook payload includes the full result with segments, words, and timing — saving a roundtrip to retrieve the full result. When include_segments is False (default), the payload contains only the transcript text and detected languages.

Production Considerations

Store results and webhook IDs in durable storage rather than local files.
Enforce a unique constraint on X-Webhook-Id so multiple workers cannot process the same delivery concurrently.
Return a 2xx response only after the event is durably recorded. For longer processing, durably enqueue the event and acknowledge it afterward.
Expect retries after timeouts or non-2xx responses. Cleanup and result persistence must be idempotent.
Monitor transcription.failed events and decide whether failed uploads should be retained, discarded, or purged.
Keep the webhook secret separate from the Koldan API key and rotate both according to your security policy.

See Transcription Webhook Events for the payload and header contract, and Files for complete purge semantics.