Skip to content

Voice & Realtime Quickstart

This quickstart guide shows you how to create your first bidirectional streaming agent for real-time audio and text conversations. You’ll learn how to set up audio I/O, handle streaming events, use tools during conversations, and work with different model providers.

After completing this guide, you can build voice assistants, interactive chatbots, multi-modal applications, and integrate bidirectional streaming with web servers or custom I/O channels.

Before starting, ensure you have:

  • Python 3.10+ installed (3.12+ required for Bedrock Nova Sonic)
  • Audio hardware (microphone and speakers) for voice conversations
  • Model provider credentials configured (AWS, OpenAI, or Google)

Bidirectional streaming is included in the Strands Agents SDK as an experimental feature. Install the SDK with bidirectional streaming support:

To install all bidirectional streaming providers and portable I/O dependencies:

Terminal window
pip install "strands-agents[bidi-all]"

This includes all 3 supported providers (Bedrock Nova Sonic, Google Gemini Live, and OpenAI Realtime), BidiTextIO, and audio processing. Local microphone and speaker access also requires the bidi-pyaudio extra and the PortAudio system library.

You can also install support for specific providers:

Terminal window
# With local microphone and speaker I/O
pip install "strands-agents[bidi,bidi-io,bidi-pyaudio]"
# With terminal text I/O
pip install "strands-agents[bidi,bidi-io]"
Terminal window
brew install portaudio
pip install "strands-agents[bidi-all,bidi-pyaudio]"

Bidirectional streaming supports multiple model providers. Choose one based on your needs:

Nova Sonic is Amazon’s bidirectional streaming model. Configure AWS credentials:

Terminal window
export AWS_ACCESS_KEY_ID=your_access_key
export AWS_SECRET_ACCESS_KEY=your_secret_key
export AWS_DEFAULT_REGION=us-east-1

Enable Nova Sonic model access in the Amazon Bedrock console.

Now let’s create a simple voice-enabled agent that can have real-time conversations:

import asyncio
from strands.experimental.bidi import BidiAgent, BidiAudioIO
from strands.experimental.bidi.models import BedrockNovaSonicModel
# Create a bidirectional streaming model
model = BedrockNovaSonicModel()
# Create the agent
agent = BidiAgent(
model=model,
system_prompt="You are a helpful voice assistant. Keep responses concise and natural."
)
# Setup audio I/O for microphone and speakers
audio_io = BidiAudioIO()
# Run the conversation
async def main():
await agent.run(
inputs=[audio_io.input()],
outputs=[audio_io.output()]
)
asyncio.run(main())

And that’s it! We now have a voice-enabled agent that can:

  • Listen to your voice through the microphone
  • Process speech in real-time
  • Respond with natural voice output
  • Display live user and assistant transcripts
  • Handle interruptions when you start speaking

BidiAudioIO.output() displays user and assistant transcripts while audio plays through the speakers. User speech appears in shaded > blocks and assistant speech appears as plain text.

The run() method runs indefinitely by default. The simplest way to stop conversations is using Ctrl+C:

import asyncio
from strands.experimental.bidi import BidiAgent, BidiAudioIO
from strands.experimental.bidi.models import BedrockNovaSonicModel
async def main():
model = BedrockNovaSonicModel()
agent = BidiAgent(model=model)
audio_io = BidiAudioIO()
try:
# Runs indefinitely until interrupted
await agent.run(
inputs=[audio_io.input()],
outputs=[audio_io.output()]
)
except asyncio.CancelledError:
print("\nConversation cancelled by user")
finally:
# stop() should only be called after run() exits
await agent.stop()
asyncio.run(main())

Just like standard Strands agents, bidirectional agents can use tools during conversations:

import asyncio
from strands import tool
from strands.experimental.bidi import BidiAgent, BidiAudioIO
from strands.experimental.bidi.models import BedrockNovaSonicModel
from strands_tools import calculator, current_time
# Define a custom tool
@tool
def get_weather(location: str) -> str:
"""
Get the current weather for a location.
Args:
location: City name or location
Returns:
Weather information
"""
# In a real application, call a weather API
return f"The weather in {location} is sunny and 72°F"
# Create agent with tools
model = BedrockNovaSonicModel()
agent = BidiAgent(
model=model,
tools=[calculator, current_time, get_weather],
system_prompt="You are a helpful assistant with access to tools."
)
audio_io = BidiAudioIO()
async def main():
await agent.run(
inputs=[audio_io.input()],
outputs=[audio_io.output()]
)
asyncio.run(main())

You can now ask questions like:

  • “What time is it?”
  • “Calculate 25 times 48”
  • “What’s the weather in San Francisco?”

The agent automatically determines when to use tools and executes them concurrently without blocking the conversation.

Strands supports three bidirectional streaming providers:

Each provider has different features, timeout limits, and audio quality. See the individual provider documentation for detailed configuration options.

Choose supported audio settings on the model and device buffering on the I/O channel:

import asyncio
from strands.experimental.bidi import BidiAgent, BidiAudioIO
from strands.experimental.bidi.models import GoogleGeminiLiveModel
# Configure model audio settings
model = GoogleGeminiLiveModel(
audio={"input": {"sample_rate": 48000}},
voice="Puck",
)
# Configure I/O buffer settings
audio_io = BidiAudioIO(
input_buffer_size=10, # Max input queue size
output_buffer_size=20, # Max output queue size
input_frames_per_buffer=512, # Input chunk size
output_frames_per_buffer=512 # Output chunk size
)
agent = BidiAgent(model=model)
async def main():
await agent.run(
inputs=[audio_io.input()],
outputs=[audio_io.output()]
)
asyncio.run(main())

BidiAudioIO reads the model’s resolved input and output formats through get_audio_config(). You do not need to repeat rates or channel counts on the I/O channel.

Bidirectional agents automatically handle interruptions when users start speaking:

import asyncio
from strands.experimental.bidi import BidiAgent, BidiAudioIO
from strands.experimental.bidi.models import BedrockNovaSonicModel
from strands.experimental.bidi.types.events import BidiInterruptionEvent
model = BedrockNovaSonicModel()
agent = BidiAgent(model=model)
audio_io = BidiAudioIO()
async def main():
await agent.start()
# Start receiving events
async for event in agent.receive():
if isinstance(event, BidiInterruptionEvent):
print(f"User interrupted: {event.reason}")
# Audio output automatically cleared
# Model stops generating
# Ready for new input
asyncio.run(main())

Interruptions are detected via voice activity detection (VAD) and handled automatically:

  1. User starts speaking
  2. Model stops generating
  3. Audio output buffer cleared
  4. Model ready for new input

If you need more control over the agent lifecycle, you can manually call start() and stop():

import asyncio
from strands.experimental.bidi import BidiAgent
from strands.experimental.bidi.models import BedrockNovaSonicModel
from strands.experimental.bidi.types.events import BidiResponseCompleteEvent
async def main():
model = BedrockNovaSonicModel()
agent = BidiAgent(model=model)
# Manually start the agent
await agent.start()
try:
await agent.send("What is Python?")
async for event in agent.receive():
if isinstance(event, BidiResponseCompleteEvent):
break
finally:
# Always stop after exiting receive loop
await agent.stop()
asyncio.run(main())

See Controlling Conversation Lifecycle for more patterns and best practices.

Use the stop tool from strands_tools to allow users to end conversations naturally. The stop tool sets request_state["stop_event_loop"], which the agent loop checks to trigger a graceful shutdown:

import asyncio
from strands.experimental.bidi import BidiAgent, BidiAudioIO
from strands.experimental.bidi.models import BedrockNovaSonicModel
from strands_tools import stop
model = BedrockNovaSonicModel()
agent = BidiAgent(
model=model,
tools=[stop],
system_prompt="You are a helpful assistant. When the user says 'stop conversation', use the stop tool."
)
audio_io = BidiAudioIO()
async def main():
await agent.run(
inputs=[audio_io.input()],
outputs=[audio_io.output()]
)
# Conversation ends when user says "stop conversation"
asyncio.run(main())

You can also create custom stop tools using the request_state["stop_event_loop"] flag:

from strands import tool
@tool
def end_session(request_state: dict) -> str:
request_state["stop_event_loop"] = True
return "Goodbye!"

The agent will gracefully close the connection when any tool sets request_state["stop_event_loop"] = True.

To enable debug logs in your agent, configure the strands logger:

import asyncio
import logging
from strands.experimental.bidi import BidiAgent, BidiAudioIO
from strands.experimental.bidi.models import BedrockNovaSonicModel
# Enable debug logs
logging.getLogger("strands").setLevel(logging.DEBUG)
logging.basicConfig(
format="%(levelname)s | %(name)s | %(message)s",
handlers=[logging.StreamHandler()]
)
model = BedrockNovaSonicModel()
agent = BidiAgent(model=model)
audio_io = BidiAudioIO()
async def main():
await agent.run(
inputs=[audio_io.input()],
outputs=[audio_io.output()]
)
asyncio.run(main())

Debug logs show:

  • Connection lifecycle events
  • Audio buffer operations
  • Tool execution details
  • Event processing flow

BidiAudioIO uses PyAudio, which does not support echo cancellation. A headset is required to prevent audio feedback loops.

If you don’t hear audio:

# List available audio devices
import pyaudio
p = pyaudio.PyAudio()
for i in range(p.get_device_count()):
info = p.get_device_info_by_index(i)
print(f"{i}: {info['name']}")
# Specify output device explicitly
audio_io = BidiAudioIO(output_device_index=2)

If the agent doesn’t respond to speech:

# Specify input device explicitly
audio_io = BidiAudioIO(input_device_index=1)
# Check system permissions (macOS)
# System Preferences → Security & Privacy → Microphone

If you experience frequent disconnections:

# Use OpenAI for longer timeout (60 min vs Nova's 8 min)
from strands.experimental.bidi.models import OpenAIRealtimeModel
model = OpenAIRealtimeModel()
# Or handle restarts gracefully
from strands.experimental.bidi import BidiConnectionRestartEvent
async for event in agent.receive():
if isinstance(event, BidiConnectionRestartEvent):
print("Reconnecting...")
continue

Ready to learn more? Check out these resources: