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.

```mermaid
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:

```python
from strands.experimental.bidi import BidiAgent, BidiAgentInput
from strands.experimental.bidi.types.events import BidiOutputEvent
from strands.experimental.bidi.types.io 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.

```python
import asyncio

from strands.experimental.bidi 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](https://pypi.org/project/PyAudio/).

Installation Required

`BidiAudioIO` requires the `bidi-io` and `bidi-pyaudio` extras plus the PortAudio system library:

```bash
pip install "strands-agents[bidi,bidi-io,bidi-pyaudio]"
```

```python
import asyncio

from strands.experimental.bidi 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 in real time, 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:

1.  The agent emits a `BidiInterruptionEvent`
2.  `BidiAudioIO`’s internal output buffer is cleared to stop playback
3.  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](https://pypi.org/project/prompt-toolkit/).

Installation Required

`BidiTextIO` is included with the `bidi-io` extra:

```bash
pip install "strands-agents[bidi-io]"
```

```python
import asyncio

from strands.experimental.bidi 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:

server.py

```python
from fastapi import FastAPI, WebSocket, WebSocketDisconnect

from strands.experimental.bidi 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:

client.py

```python
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())
```

## Related pages

- [BidiAgent](/docs/user-guide/concepts/bidirectional-streaming/agent/index.md) (1 shared tag)
- [Events](/docs/user-guide/concepts/bidirectional-streaming/events/index.md) (1 shared tag)
- [Google Gemini Live](/docs/user-guide/concepts/bidirectional-streaming/models/google/index.md) (1 shared tag)
- [Interruptions](/docs/user-guide/concepts/bidirectional-streaming/interruption/index.md) (1 shared tag)
- [OpenAI Realtime](/docs/user-guide/concepts/bidirectional-streaming/models/openai/index.md) (1 shared tag)
- [Bidirectional Streaming Observability](/docs/user-guide/concepts/bidirectional-streaming/observability/index.md) (1 shared tag)
- [Bidirectional Streaming Hooks](/docs/user-guide/concepts/bidirectional-streaming/hooks/index.md) (1 shared tag)
- [Voice & Realtime Quickstart](/docs/user-guide/concepts/bidirectional-streaming/quickstart/index.md) (1 shared tag)
- [Bedrock Nova Sonic](/docs/user-guide/concepts/bidirectional-streaming/models/bedrock/index.md) (1 shared tag)
- [Bidirectional Streaming Session Management](/docs/user-guide/concepts/bidirectional-streaming/session-management/index.md) (1 shared tag)


## Implementation

### Python

- [harness-sdk/strands-py/src/strands/experimental/bidi/io/text.py](https://github.com/strands-agents/harness-sdk/blob/main/strands-py/src/strands/experimental/bidi/io/text.py)
- [harness-sdk/strands-py/src/strands/experimental/bidi/io/audio.py](https://github.com/strands-agents/harness-sdk/blob/main/strands-py/src/strands/experimental/bidi/io/audio.py)
- [harness-sdk/strands-py/src/strands/experimental/bidi/io/transcript.py](https://github.com/strands-agents/harness-sdk/blob/main/strands-py/src/strands/experimental/bidi/io/transcript.py)