Skip to content

Python SDK

Coming Soon

The official Python SDK is under development. In the meantime, you can use the REST API directly with requests or any HTTP client.

Using the API with requests

Here's a complete example using the requests library:

Installation

pip install requests

Full Example

import time
import requests

API_KEY = "your-api-key"
BASE_URL = "https://api.icana.ai"
HEADERS = {"X-API-Key": API_KEY}


def upload_file(file_path: str) -> dict:
    """Upload an audio file and return the response."""
    with open(file_path, "rb") as f:
        response = requests.post(
            f"{BASE_URL}/upload",
            headers=HEADERS,
            files={"file": f},
        )
    response.raise_for_status()
    return response.json()


def submit_transcription(s3_uri: str, language: str = "en", prompt: str = None) -> dict:
    """Submit a transcription job and return the response."""
    payload = {"s3_uri": s3_uri, "language": language}
    if prompt:
        payload["prompt"] = prompt

    response = requests.post(
        f"{BASE_URL}/transcribe",
        headers={**HEADERS, "Content-Type": "application/json"},
        json=payload,
    )
    response.raise_for_status()
    return response.json()


def poll_status(job_id: str, interval: int = 10) -> dict:
    """Poll until the job completes or fails."""
    while True:
        response = requests.get(
            f"{BASE_URL}/status/{job_id}",
            headers=HEADERS,
        )
        response.raise_for_status()
        result = response.json()

        status = result["status"]
        print(f"Job {job_id}: {status}")

        if status in ("COMPLETE", "FAILED", "DELETED", "ERROR"):
            return result

        time.sleep(interval)


def delete_files(job_id: str, s3_uri: str) -> dict:
    """Delete input and output files for a job."""
    response = requests.delete(
        f"{BASE_URL}/delete/{job_id}",
        headers={**HEADERS, "Content-Type": "application/json"},
        json={"s3_uri": s3_uri},
    )
    response.raise_for_status()
    return response.json()


# --- Usage ---

# 1. Upload
upload = upload_file("meeting.mp3")
print(f"Uploaded: {upload['s3_uri']}")

# 2. Transcribe
job = submit_transcription(upload["s3_uri"], language="en")
print(f"Job submitted: {job['aws_batch_job_id']}")

# 3. Wait for result
result = poll_status(job["aws_batch_job_id"])

if result["status"] == "COMPLETE":
    print(f"\nTranscription:\n{result['transcription']}")
    print(f"\nDiarisation:\n{result['diarization']}")
else:
    print(f"Job failed with status: {result['status']}")

# 4. Clean up
delete_files(job["aws_batch_job_id"], upload["s3_uri"])

Queue Status

def get_queue_status() -> dict:
    """Get the current queue status."""
    response = requests.get(f"{BASE_URL}/queue/status", headers=HEADERS)
    response.raise_for_status()
    return response.json()

status = get_queue_status()
print(f"Total jobs in queue: {status['total_jobs']}")
print(f"Breakdown: {status['status_counts']}")