I/O Channels

I/O channels handle the flow of data between your application and the bidi-agent. They manage input sources (microphone, keyboard, WebSocket) and output destinations (speakers, console, UI) while the agent focuses on conversation logic and model communication.

flowchart LR
    A[Microphone]
    B[Keyboard]
    A --> C[Bidi-Agent]
    B --> C
    C --> D[Speakers]
    C --> E[Console]

I/O Interfaces

The bidi agent uses two protocol interfaces that define how data flows in and out of conversations:

BidiInput: A callable protocol for reading data from sources such as a microphone, keyboard, or WebSocket and returning BidiAgentInput.
BidiOutput: A callable protocol for receiving BidiOutputEvent objects from the agent and handling them appropriately.

Both protocols include optional lifecycle methods (start and stop) for resource management.

Implementation of these protocols will look as follows:

from strands.experimental.bidi.agent import BidiAgent
from strands.experimental.bidi.types import BidiAgentInput
from strands.experimental.bidi.types import BidiOutputEvent
from strands.experimental.bidi.types import BidiInput, BidiOutput


class MyBidiInput(BidiInput):
    async def start(self, agent: BidiAgent) -> None:
        # Initialize input resources or state, using agent as needed.
        return

    async def __call__(self) -> BidiAgentInput:
        # Read or generate input and return a BidiAgentInput value.
        return {"text": "Hello"}

    async def stop(self) -> None:
        # Clean up any input resources or state.
        return


class MyBidiOutput(BidiOutput):
    async def start(self, agent: BidiAgent) -> None:
        # Initialize output resources or state, using agent as needed.
        return

    async def __call__(self, event: BidiOutputEvent) -> None:
        # Process the event as needed for your application.
        print(event)

    async def stop(self) -> None:
        # Clean up any output resources or state.
        return

I/O Usage

To connect your I/O channels into the agent loop, you can pass them as arguments into the agent run() method.

import asyncio

from strands.experimental.bidi.agent import BidiAgent
from strands_tools import stop


async def main():
    # stop tool allows user to verbally stop agent execution.
    agent = BidiAgent(tools=[stop])
    await agent.run(inputs=[MyBidiInput()], outputs=[MyBidiOutput()])


asyncio.run(main())

The run() method handles the startup, execution, and shutdown of both the agent and collection of I/O channels. The inputs and outpus all run concurrently to one another, allowing for a flexible mixing and matching.

Audio I/O

Out of the box, Strands provides BidiAudioIO to help connect your microphone and speakers to the bidi-agent using PyAudio.

import asyncio

from strands.experimental.bidi.agent import BidiAgent
from strands.experimental.bidi.io import BidiAudioIO
from strands_tools import stop


async def main():
    # stop tool allows user to verbally stop agent execution.
    agent = BidiAgent(tools=[stop])
    audio_io = BidiAudioIO(input_device_index=1)

    await agent.run(
        inputs=[audio_io.input()],
        outputs=[audio_io.output()],
    )


asyncio.run(main())

This creates a voice-enabled agent that captures audio from your microphone, streams it to the model as AudioDelta inputs, and plays responses through your speakers.

Audio output also displays live transcripts, with user speech in shaded > blocks and assistant speech as plain text. The next user prompt appears when response generation finishes or is interrupted.

Configurations

Use these options to configure audio devices and buffering:

Parameter	Description	Example	Default
`input_buffer_size`	Maximum number of audio chunks to buffer from microphone before dropping oldest.	`1024`	None (unbounded)
`input_device_index`	Specific microphone device ID to use for audio input.	`1`	None (system default)
`input_frames_per_buffer`	Number of audio frames to be read per input callback (affects latency and performance).	`1024`	512
`output_buffer_size`	Maximum number of audio chunks to buffer for speaker playback before dropping oldest.	`2048`	None (unbounded)
`output_device_index`	Specific speaker device ID to use for audio output.	`2`	None (system default)
`output_frames_per_buffer`	Number of audio frames to be written per output callback (affects latency and performance).	`1024`	512

Configure voice and supported sample rates on the model. BidiAudioIO reads model.get_audio_config(), which returns separate input and output dictionaries, each containing sample_rate, channels, and format. It uses those values to configure capture and playback, so you do not need to repeat them on the I/O channel.

BidiAudioIO requires signed 16-bit little-endian PCM. Starting a device stream with another encoding raises ValueError. Use custom I/O for other encodings.

Interruption Handling

BidiAudioIO automatically handles interruptions to create natural conversational flow where users can interrupt the agent mid-response. When an interruption occurs:

The agent emits a BidiInterruptionEvent
BidiAudioIO’s internal output buffer is cleared to stop playback
The agent begins responding immediately to the new user input

Text I/O

Strands also provides BidiTextIO for terminal-based text input and output using prompt-toolkit.

import asyncio

from strands.experimental.bidi.agent import BidiAgent
from strands.experimental.bidi.io import BidiTextIO
from strands_tools import stop


async def main():
    # stop tool allows user to verbally stop agent execution.
    agent = BidiAgent(tools=[stop])
    text_io = BidiTextIO(input_prompt="> You: ")

    await agent.run(
        inputs=[text_io.input()],
        outputs=[text_io.output()],
    )


asyncio.run(main())

This creates a text-based agent that reads user input from the terminal and prints transcripts and responses to the console.

Configurations

Parameter	Description	Example	Default
`input_prompt`	Prompt text displayed when waiting for user input	`"> You: "`	`""` (blank)

WebSocket I/O

WebSockets are a common I/O channel for bidi-agents. To learn how to setup WebSockets with run(), consider the following server example:

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

from strands.experimental.bidi.agent import BidiAgent
from strands.experimental.bidi.models import OpenAIRealtimeModel

app = FastAPI()


@app.websocket("/text-chat")
async def text_chat(websocket: WebSocket) -> None:
    model = OpenAIRealtimeModel(api_key="<OPENAI_API_KEY>")
    agent = BidiAgent(model=model)

    try:
        await websocket.accept()
        await agent.run(inputs=[websocket.receive_json], outputs=[websocket.send_json])
    except* WebSocketDisconnect:
        print("client disconnected")

To start this server, you can run unvicorn server:app --reload. To interact, open a separate terminal window and run the following client script:

import asyncio
import json

import websockets


async def main():
    websocket = await websockets.connect("ws://localhost:8000/text-chat")

    input_content = {"text": "Hello, how are you?"}
    await websocket.send(json.dumps(input_content))

    while True:
        output_event = json.loads(await websocket.recv())
        if output_event["type"] == "bidi_transcript_complete":
            print(output_event["transcript"])
            break

    await websocket.close()


if __name__ == "__main__":
    asyncio.run(main())