autourgos-micinput

Standalone, provider-agnostic microphone capture for the Autourgos framework. Extracted out of autourgos-live so any package -- autourgos-live, autourgos-openaichat, autourgos-responses, autourgos-agent, or plain application code -- can capture raw PCM audio from the system microphone without depending on autourgos-live for it. This package knows nothing about any specific LLM API; it just yields bytes.

import asyncio
from autourgos_micinput import MicrophoneStream

async def main():
    async with MicrophoneStream(sample_rate=16000) as mic:
        async for chunk in mic:
            print(len(chunk), "bytes captured, mime_type =", mic.mime_type)
            break  # just capture one chunk for this example

asyncio.run(main())

Install

pip install "autourgos-micinput[mic]"

sounddevice is required to actually open a MicrophoneStream and is gated behind the mic extra -- import autourgos_micinput alone never requires it. Requires Python 3.10+.

Usage

With autourgos-live (voice input to Gemini Live)

autourgos-live does not depend on this package or re-export MicrophoneStream -- that was tried and reverted (see autourgos-live's CHANGELOG, 0.3.0/0.4.0) because it made this package a mandatory install even for callers who never touch local audio (telephony, browser WebRTC, text-only). Install both separately and forward chunks yourself -- three lines:

from autourgos_live import GeminiLiveSession
from autourgos_micinput import MicrophoneStream

async def pipe_microphone(session, **kwargs):
    async with MicrophoneStream(**kwargs) as mic:
        async for chunk in mic:
            await session.send_audio_chunk(chunk, mime_type=mic.mime_type)

With autourgos-openaichat / autourgos-responses (voice input to a request/response call)

These packages don't integrate with autourgos-micinput directly -- capture audio yourself, run it through your own speech-to-text step, then pass the resulting text to invoke()/create():

import asyncio
from autourgos_micinput import MicrophoneStream

async def record_utterance(seconds: float = 3.0) -> bytes:
    chunks = []
    frames_needed = int(seconds * 1000 / 100)  # default chunk_ms=100
    async with MicrophoneStream(sample_rate=16000) as mic:
        async for chunk in mic:
            chunks.append(chunk)
            if len(chunks) >= frames_needed:
                break
    return b"".join(chunks)

# pcm = asyncio.run(record_utterance())
# text = my_own_speech_to_text(pcm)
# response = my_openaichat_model.invoke(text)

With autourgos-agent (as a tool)

Wrap capture in an @tool-decorated function the same way any other tool is defined:

from autourgos_agent import tool
from autourgos_micinput import MicrophoneStream
import asyncio

@tool
def record_audio(seconds: float = 3.0) -> bytes:
    """Record `seconds` of audio from the microphone and return raw 16kHz PCM bytes."""
    async def _record():
        chunks = []
        async with MicrophoneStream(sample_rate=16000) as mic:
            async for chunk in mic:
                chunks.append(chunk)
                if len(chunks) * 0.1 >= seconds:
                    break
        return b"".join(chunks)
    return asyncio.run(_record())

API Reference

MicrophoneStream

Async context manager; async-iterates raw 16-bit PCM chunks from the input device.

Raises ImportError (with an install hint) on __aenter__ if the mic extra isn't installed -- never on import autourgos_micinput.

License

Apache License 2.0, Copyright (c) 2026 Jitin Kumar Sengar