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, speaker playback 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, *, capture: MicrophoneCapture | None = None
)

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 start_audio_stream instead.

Capture runs inside WebRTC's audio device module, which applies echo cancellation, noise suppression, and automatic gain control before the audio is encoded. capture selects which of the three run. On the websocket transport, echo cancellation runs when selected (the default); noise suppression and gain control are unavailable there, and asking for them is logged when capture starts:

from cosmo_ai import MicrophoneCapture

await session.set_microphone_enabled(
    True,
    capture=MicrophoneCapture(
        echo_cancellation=True,
        noise_suppression=False,
        auto_gain_control=False,
    ),
)

All three default to True. Keep echo cancellation on for any session played through speakers — without it the agent's own voice re-enters the microphone and it interrupts itself. Noise suppression and gain control attenuate the speaker's level, which occasionally works against you: on laptop speakers they can duck someone who talks while the agent is talking, far enough that the agent stops registering the interruption.

This is client-side processing applied to the raw capture. It is separate from AudioConfig(noise_cancellation=...), which asks the server to denoise the received stream.

A host with no usable input device — headless CI, a denied permission, a device another process holds exclusively — raises AudioUnavailableError before anything is published.

await session.set_muted(muted: bool)

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

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

Publish a caller-owned PcmAudioSource as the session's voice. You own the source and keep it fed through source.capture_frame(...). Use it for audio the SDK cannot capture itself: a synthetic generator, WAV replay, a load generator, or any pipeline running where there is no input device. For the OS microphone use set_microphone_enabled. Publishing binds the agent's input to this client and clears the server-side mute gate; stop_audio_stream unpublishes and closes it again.

from cosmo_ai import PcmAudioSource

source = PcmAudioSource(48_000, 1)
await session.start_audio_stream(source)
await source.capture_frame(frame)  # 16-bit PCM, feed at the pace it plays

A frame is anything carrying data, sample_rate and num_channels, so livekit.rtc.AudioFrame works as-is. Feed frames at the rate they would play: turn detection reads a continuous stream, and a burst arriving faster than real time is heard as one long utterance. PcmAudioSource publishes on either transport; an rtc.AudioSource publishes on the room transport only.

A session carries one voice, so the microphone and an audio stream are mutually exclusive and only one stream runs at a time. Starting a stream while either holds the voice raises SessionStateError with code audio_publish_already_active, as does enabling the microphone while a stream is running — stop the active one first. stop_audio_stream is idempotent.

The equivalents elsewhere take each platform's own audio type: TypeScript's startAudioStream(stream) takes a MediaStream, and Swift's startAudioStream() is fed by pushAudioBuffer(_:).

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. 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. 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.