Audio (Python)

Audio lives on RealtimeSession. The media transport and OS-device capture/playback both ship with the base install:

pip install cosmo-ai-sdk

On Linux, OS mic/speaker use needs the system PortAudio library (apt install libportaudio2); sessions that never touch OS audio don't load it.

async with agent.start() as session:
    await session.set_microphone_enabled(True)
    await session.set_speaker_enabled(True)
    async for event in session:
        ...

Microphone

await session.set_microphone_enabled(enabled: bool)

True captures and publishes the default OS microphone; False stops and unpublishes it, then gates the server side with set_muted(True). For non-mic audio input (synthetic generator, WAV replay) use publish_audio_source instead.

await session.set_muted(muted: bool)

Toggle the server-side mic gate only — no track is published or unpublished, so it is the right primitive for push-to-talk. The SDK re-asserts the last mute state automatically after a reconnect.

await session.publish_audio_source(source, *, track_name: str = "mic")

Publish a caller-supplied rtc.AudioSource as the mic track (requires the livekit extra). You own the source and keep it fed via source.capture_frame(...) — livekit-rtc for Python has no OS-mic capture of its own, so any audio input is captured caller-side and pushed in. Returns the transport's track publication handle. Publishing audio binds the agent's input to this client and unmutes it.

Speaker

await session.set_speaker_enabled(enabled: bool)

Play the agent's voice on the default OS output device, or stop. Idempotent. For custom playback consume agent_audio() directly.

session.set_agent_playback_volume(volume: float)

Software gain for OS playback: 0 mutes, 1 is unity; values outside 0…1 are clamped, NaN raises ValueError. Affects only set_speaker_enabled output, never agent_audio() frames. May be called before the speaker is enabled; the value persists.

Agent audio frames

async for frame in session.agent_audio():  # AgentAudioFrame
    ...

The agent's decoded voice — record it, pipe it elsewhere, or feed a custom player (requires the livekit extra). Frames flow once the agent publishes audio; the iterator finishes when the session ends. Multiple concurrent iterators each receive every frame; a stalled consumer drops its oldest frames.

Agent audio is decoded at a fixed geometry — 48 kHz, mono, 16-bit (cosmo_ai.audio.AGENT_AUDIO_SAMPLE_RATE == 48000):

from cosmo_ai.audio import AgentAudioFrame

@dataclass(frozen=True)
class AgentAudioFrame:
    data: bytes              # 16-bit little-endian PCM, channels interleaved
    sample_rate: int
    num_channels: int
    samples_per_channel: int

Level metering

async for levels in session.audio_levels():  # AudioLevels
    ...

Mic and agent RMS levels sampled at a fixed ~20 Hz cadence, latest-value — a slow consumer skips samples, it never lags (requires the livekit extra). mic is live while the microphone is publishing; agent is live once the agent's track exists and decays to 0.0 between utterances. Finishes when the session ends.

from cosmo_ai.audio import AudioLevels

@dataclass(frozen=True)
class AudioLevels:
    mic: float    # RMS 0…1, 0.0 while inactive
    agent: float  # RMS 0…1, 0.0 while inactive

See Audio for the conceptual model and gating directives.