pip install fish-audio-sdk

# With audio playback utilities
pip install fish-audio-sdk[utils]

export FISH_API_KEY=your_api_key_here

from fishaudio import FishAudio

client = FishAudio(api_key="your_api_key")

from fishaudio import FishAudio
from fishaudio.utils import play, save

client = FishAudio()

# Generate audio
audio = client.tts.convert(text="Hello, world!")

# Play or save
play(audio)
save(audio, "output.mp3")

import asyncio
from fishaudio import AsyncFishAudio
from fishaudio.utils import play, save

async def main():
    client = AsyncFishAudio()
    audio = await client.tts.convert(text="Hello, world!")
    play(audio)
    save(audio, "output.mp3")

asyncio.run(main())

# Use a specific voice by ID
audio = client.tts.convert(
    text="Custom voice",
    reference_id="802e3bc2b27e49c2995d23ef70e6ac89"
)

audio = client.tts.convert(
    text="Speaking faster!",
    speed=1.5  # 1.5x speed
)

from fishaudio.types import TTSConfig, Prosody

config = TTSConfig(
    prosody=Prosody(speed=1.2, volume=-5),
    reference_id="933563129e564b19a115bedd57b7406a",
    format="wav",
    latency="balanced"
)

# Reuse across generations
audio1 = client.tts.convert(text="First message", config=config)
audio2 = client.tts.convert(text="Second message", config=config)

# Stream and process chunks as they arrive
for chunk in client.tts.stream(text="Long content..."):
    send_to_websocket(chunk)

# Or collect all chunks
audio = client.tts.stream(text="Hello!").collect()

# Transcribe audio
with open("audio.wav", "rb") as f:
    result = client.asr.transcribe(audio=f.read(), language="en")

print(result.text)

# Access timestamped segments
for segment in result.segments:
    print(f"[{segment.start:.2f}s - {segment.end:.2f}s] {segment.text}")

def text_chunks():
    yield "Hello, "
    yield "this is "
    yield "streaming!"

audio_stream = client.tts.stream_websocket(text_chunks(), latency="balanced")
play(audio_stream)

import asyncio
from fishaudio import AsyncFishAudio

async def text_chunks():
    yield "Hello, "
    yield "this is "
    yield "streaming!"

async def main():
    async with AsyncFishAudio() as client:
        # stream_websocket is an async generator — iterate it, don't await the call
        audio_stream = client.tts.stream_websocket(text_chunks(), latency="balanced")
        with open("out.mp3", "wb") as f:
            async for chunk in audio_stream:
                f.write(chunk)

asyncio.run(main())

from fishaudio.types import ReferenceAudio

# Clone voice on-the-fly
with open("reference.wav", "rb") as f:
    audio = client.tts.convert(
        text="Cloned voice speaking",
        references=[ReferenceAudio(
            audio=f.read(),
            text="Text spoken in reference"
        )]
    )

# Create voice model for reuse
with open("voice_sample.wav", "rb") as f:
    voice = client.voices.create(
        title="My Voice",
        voices=[f.read()],
        description="Custom voice clone"
    )

# Use the created model
audio = client.tts.convert(
    text="Using my saved voice",
    reference_id=voice.id
)

from fishaudio.exceptions import (
    AuthenticationError,
    RateLimitError,
    NotFoundError,
    APIError,
    FishAudioError,
)

try:
    audio = client.tts.convert(text="Hello!")
except AuthenticationError:
    print("Invalid API key")
except RateLimitError:
    print("Rate limit exceeded")
except NotFoundError:
    print("Voice model not found")
except APIError as e:
    print(f"API error {e.status}: {e.message}")  # any other HTTP error, including 422 validation
except FishAudioError as e:
    print(f"SDK error: {e}")