Python SDK Deep Dive: Advanced TTS Integration

The Python SDK covers basic requests in 3 lines. Real production code needs more.

Installation

pip install speeko

Basic Usage

from speeko import Client

client = Client(api_key="sk-...")

audio = client.tts(text="Hello world", voice="af_heart")

with open("output.mp3", "wb") as f:
    f.write(audio)

Async Client

For high-throughput workloads, use async:

from speeko import AsyncClient
import asyncio

async def main():
    async with AsyncClient(api_key="sk-...") as client:
        tasks = [
            client.tts(text=t, voice="af_heart")
            for t in TEXTS
        ]
        results = await asyncio.gather(*tasks)

asyncio.run(main())

Batch Processing

Process 10K items without overwhelming the API:

from asyncio import Semaphore

async def batch_process(texts, concurrency=10):
    sem = Semaphore(concurrency)

    async def bounded(text):
        async with sem:
            return await client.tts(text=text, voice="af_heart")

    return await asyncio.gather(*[bounded(t) for t in texts])

Streaming

For real-time applications:

async def stream_tts(text):
    async for chunk in client.tts_stream(text=text, voice="af_heart"):
        play_audio_chunk(chunk)

Error Handling

from speeko.exceptions import (
    AuthError,
    RateLimitError,
    InsufficientCreditsError,
    ServerError
)

try:
    audio = client.tts(text="Hello", voice="af_heart")
except RateLimitError as e:
    await asyncio.sleep(e.retry_after)
except InsufficientCreditsError:
    notify_billing_team()
except AuthError:
    rotate_api_key()
except ServerError:
    pass

Retry Logic

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=60)
)
async def robust_tts(text):
    return await client.tts(text=text, voice="af_heart")

Context Manager Pattern

from contextlib import asynccontextmanager

@asynccontextmanager
async def speeko_session():
    async with AsyncClient(api_key=os.getenv("SPEEKO_KEY")) as client:
        yield client

Observability

import logging

logging.basicConfig(level=logging.INFO)
logging.getLogger("speeko").setLevel(logging.DEBUG)

Logs include request IDs — share these in support tickets.

Testing

Mock the client in unit tests:

from unittest.mock import AsyncMock

client = AsyncMock()
client.tts.return_value = b"fake audio bytes"

Install the Python SDK.