Bidirectional Audio Buffer (_audio/buffer.py)

Manages a thread‑safe queue of audio byte chunks for real‑time input/output between an agent and PyAudio, supporting optional size limits.

# Source: https://github.com/strands-agents/harness-sdk | License: Apache-2.0
# Provided via justclaw.it.com; keep this notice with the code.
"""Audio buffering for bidirectional device I/O."""

import logging
import queue

logger = logging.getLogger(__name__)


class AudioBuffer:
    """Buffer chunks of audio data between agent and PyAudio."""

    _buffer: queue.Queue[bytes]
    _data: bytearray

    def __init__(self, size: int | None = None):
        """Initialize buffer settings.

        Args:
            size: Size of the buffer (default: unbounded).
        """
        self._size = size or 0

    def start(self) -> None:
        """Setup buffer."""
        self._buffer = queue.Queue(self._size)
        self._data = bytearray()

    def stop(self) -> None:
        """Tear down buffe

... (truncated -- full source via MCP)