listentome

listentome is an async-first audio I/O library for Python, built on PortAudio.

The key features are:

• Async native: streams are async context managers and async iterators. No callbacks, no queues, no call_soon_threadsafe.
• Bytes in, bytes out: audio blocks are plain bytes. No NumPy requirement, no raw/array duality.
• No global state: no mutable module defaults. You pass the device, samplerate, and dtype explicitly.
• Explicit backpressure: you decide what happens when the consumer falls behind - drop or raise.
• Testable without hardware: every stream accepts a Backend protocol, so tests inject a fake and never touch a device.
• Sync facade: play() and record() for scripts, built on the same async core.

Requirements

The PortAudio system library:

brew install portaudio        # macOS
apt install libportaudio2     # Debian/Ubuntu

Installation

uv add listentome

Example

Record

Create a file main.py with:

import anyio

import listentome as ltm


async def main() -> None:
    async with ltm.InputStream(samplerate=48_000, channels=1) as stream:
        async for block in stream:
            print(f"got {len(block)} bytes")


anyio.run(main)

Run it:

uv run main.py

That's it. The stream opens the default microphone, and each iteration gives you one block of raw float32 samples as bytes. Pass dtype="int16" (or int32, int8, uint8) for a different sample format, and blocksize to control how many frames each block carries.

[!WARNING] Backpressure is explicit. If you iterate slower than audio arrives, the oldest blocks are dropped so latency stays bounded. Pass on_overflow="raise" to get an Overflow exception instead, and max_buffered_blocks to size the buffer. A dropped block in a live conversation is a glitch; unbounded latency is a broken conversation.

Play

write() returns once the device has consumed the audio:

import anyio

import listentome as ltm


async def main() -> None:
    tone = bytes(48_000 * 4)  # one second of float32 silence
    async with ltm.OutputStream(samplerate=48_000, channels=1) as stream:
        await stream.write(tone)


anyio.run(main)

Duplex

DuplexStream captures and plays through the same device pair - read() (or async for) for capture, write() for playback:

import anyio

import listentome as ltm


async def main() -> None:
    async with ltm.DuplexStream(samplerate=48_000, channels=1) as stream:
        async for block in stream:
            await stream.write(block)


anyio.run(main)

Use with Pydantic AI

This is what listentome is built for: realtime speech-to-speech agents without audio plumbing.

With callback-based libraries, wiring a microphone to a realtime session takes a PortAudio callback, a thread-safe queue, loop.call_soon_threadsafe, and a hand-rolled drop-oldest policy. With listentome, the microphone is an async iterator and the speaker is an awaitable - the plumbing disappears into the library:

import anyio

import listentome as ltm
from pydantic_ai import Agent, PartDeltaEvent, SpeechPartDelta

# OpenAI's realtime models speak and listen in 24 kHz mono PCM16 audio.
SAMPLE_RATE = 24_000
BLOCK_SIZE = 2_400  # 100 ms per audio block

agent = Agent(
    instructions="You are a friendly voice assistant. Keep your replies short and conversational."
)


@agent.tool_plain
def get_weather(city: str) -> str:
    """Look up the current weather in a city."""
    return f"It is currently 21 degrees and sunny in {city}."


async def main() -> None:
    mic = ltm.InputStream(samplerate=SAMPLE_RATE, channels=1, dtype="int16", blocksize=BLOCK_SIZE)
    speaker = ltm.OutputStream(samplerate=SAMPLE_RATE, channels=1, dtype="int16", blocksize=BLOCK_SIZE)

    async with (
        agent.realtime("openai:gpt-realtime").session() as session,
        mic,
        speaker,
        anyio.create_task_group() as tg,
    ):

        async def stream_mic() -> None:
            async for block in mic:
                await session.send_audio(block)

        tg.start_soon(stream_mic)
        print("Listening - start talking (Ctrl-C to quit).")

        async for event in session:
            match event:
                case PartDeltaEvent(delta=SpeechPartDelta(audio_chunk=chunk)) if chunk:
                    await speaker.write(chunk)

        tg.cancel_scope.cancel()


anyio.run(main)

The mic loop is async for block in mic - the drop-oldest queue is the stream's own overflow policy. The speaker is await speaker.write(chunk) - write() suspends until the device consumed the audio, so the model's audio never runs unboundedly ahead of what the user hears.

[!NOTE] Barge-in. A production assistant also handles interruption: on RealtimeInputSpeechStartEvent, stop feeding the speaker and call session.interrupt() with how much was actually played. See the Pydantic AI realtime docs for the full event set.

Devices

There is no global default state. Query devices explicitly and pass an index to the stream:

import listentome as ltm

for device in ltm.devices():
    print(device.index, device.name, device.max_input_channels, device.max_output_channels)

ltm.default_input() and ltm.default_output() return the system defaults, or None when no device exists. Device is a frozen dataclass - there is nothing to mutate.

Sync facade

For scripts that do not need an event loop:

import listentome as ltm

data = ltm.record(2.0, samplerate=48_000, channels=1)
ltm.play(data, samplerate=48_000, channels=1)

These run the async streams on an internal anyio portal. There is a single implementation underneath - the sync functions are conveniences, not a parallel API.

Testing without hardware

Every stream and query function accepts a backend argument satisfying the Backend protocol:

class Backend(Protocol):
    def devices(self) -> list[Device]: ...
    def default_input(self) -> Device | None: ...
    def default_output(self) -> Device | None: ...
    def open(self, ...) -> RawStream: ...

Inject a fake backend in tests to drive streams without any audio device - including simulated device failures your real hardware would never produce. See tests/fake_backend.py for a reference implementation; listentome's own suite runs at 100% coverage without touching a device.

License

MIT