Audio stream in

core · v0.1.0

Stream microphone audio in over the socket and get transcripts back, from any speech-to-text provider. Defines the transcriber contract; provider kits plug in a stream. Bounded streams, per-user limits, gap detection and watcher fan-out.

Installed for you

With deepgram, elevenlabs, fake-voice, openai, pipecat-bridge, voice-agent. Open it to write your own provider, or as the reference for transcriber@1.

Install	copit add @chanx-kit/audio-stream-in
Import from	audio_stream_in
Requires kits	media-stream-in
Contract	transcriber@1
UI kits	transcriber
Providers	deepgram, elevenlabs, fake-voice, openai, pipecat-bridge
Python packages	chanx>=2.10.2,<3
Keywords	audio, speech-to-text, transcription, streaming
Source	kits/audio_stream_in

Stream microphone audio in over a chanx websocket and get transcripts back, from any speech-to-text provider. This kit defines the transcriber contract (transcriber@1) and does everything around the provider: format negotiation, ordering, limits, keepalive, fan-out. A provider kit only opens a stream:

Provider kit
fake-voice	scripted, no key: demos, UI work, CI
deepgram, elevenlabs, openai	their stt part
pipecat-bridge	any Pipecat speech-to-text service

copit add @chanx-kit/fake-voice --only stt   # pulls in audio-stream-in and media-stream-in

Two topics

Topic	Pattern	Who	Direction
TranscriberTopic	transcribe:{session}	the session's owner	audio in, transcripts and errors out
TranscriptTopic	transcript:{session}	anyone authorize admits	transcripts out only

Sending audio and reading captions need different permissions, so they are separate topics. Override authorize on both: by default anyone may subscribe.

class VoiceConsumer(AsyncJsonWebsocketConsumer):
    topics = [MyTranscriberTopic, TranscriptTopic]

The conversation

1. On subscribe the server sends audio_config: PCM16 mono at 24 kHz and 100 ms chunks by default, the one format Deepgram, ElevenLabs and OpenAI all take.
2. The client sends audio_start with the format it captured (a mismatch is bad_format), then audio_chunks numbered from 0, then audio_end.
3. The server sends speech_started, transcript_partial (the segment being spoken, replacing the previous partial), transcript_final (a segment that will not change) and utterance_end. An utterance is its finals in order, under one utterance_id.

Chunks carry bytes as base64 (see media-stream-in); handlers and providers see bytes. Chunks sent right behind audio_start are kept while the provider connects, and chunks that arrive out of order are put back in order; ones that never arrive are reported as audio_gap rather than fed to the provider as silence it never heard.

Limits

Providers bill per second of audio, whoever is logged in, so every stream is bounded:

Attribute	Default
max_session_seconds	300	one stream's length
idle_timeout_seconds	15	no audio for this long closes the stream
max_sessions_per_user	2	per current_user_id(), or per connection without one
keepalive_seconds	4	the provider is kept open while the user pauses

Set any to None to turn it off. A limit that trips sends transcriber_error with limit_reached. Streams live in the process holding the socket, since the audio arrives there; only transcripts cross processes, through the channel layer.

Doing more with transcripts

Every transcript goes through a hook, then to this connection and to transcript:<session> watchers. Override a hook to store text, start an agent run, or filter:

class MyTranscriberTopic(DeepgramTranscriberTopic):
    async def on_final(self, transcript: TranscriptFinal) -> None:
        await super().on_final(transcript)
        await save_line(self.session_id, transcript.text)

on_speech_started, on_partial, on_final, on_utterance_end and on_gap are all overridable. If you subclass TranscriptTopic, point transcript_topic at your subclass: group names carry the class name.

Writing a provider

Subclass TranscriberTopic and return a TranscriberStream from open_stream: it takes audio (send), keeps alive, finishes or closes, and yields frames (SpeechStartedFrame, PartialFrame, FinalFrame, UtteranceEndFrame). Then run the shared suite against it, the way every provider here does:

from ..audio_stream_in.tests.contract import TranscriberContract

class TestMyProvider(TranscriberContract):
    @pytest.fixture
    def transcriber_topic(self):
        return MyProviderTopic  # set up to hear "hello world" in the suite's audio

Message reference

Generated from the kit's message classes.

audio_chunk

AudioChunkMessage

Field	Type	Required
index	integer	yes
data	string	yes

audio_config

AudioConfigMessage

Field	Type	Required
encoding	pcm16	no
sample_rate	integer	no
channels	integer	no
chunk_ms	integer	no

audio_config_request

AudioConfigRequestMessage

Payload: null

audio_end

AudioEndMessage

Payload: null

audio_start

AudioStartMessage

Field	Type	Required
encoding	pcm16	no
sample_rate	integer	no
channels	integer	no
language	string | null	no

speech_started

SpeechStartedMessage

Field	Type	Required
at_ms	integer	yes

transcriber_error

TranscriberErrorMessage

Field	Type	Required
code	TranscriberErrorCode	yes
message	string	yes

transcript_final

TranscriptFinalMessage

Field	Type	Required
utterance_id	string	yes
text	string	yes
start_ms	integer	yes
end_ms	integer	yes
confidence	number | null	no

transcript_partial

TranscriptPartialMessage

Field	Type	Required
utterance_id	string	yes
text	string	yes
start_ms	integer	yes
end_ms	integer	yes

utterance_end

UtteranceEndMessage

Field	Type	Required
utterance_id	string	yes