Bidirectional Streaming Agent

[mehtarac]

Description

This PR introduces bidirectional streaming capabilities to Strands SDK, enabling real-time voice and audio conversations with AI models through persistent streaming connections.

Overview

Bidirectional streaming moves beyond traditional request-response patterns by maintaining long-running conversations where users can interrupt, provide continuous input, and receive real-time audio responses. This implementation is marked as experimental as we refine the API based on user feedback and evolving model capabilities.

Key Features:

• Real-time audio I/O streaming with PyAudio integration
• Automatic interruption detection that clears audio buffers when users speak
• Concurrent tool execution during active conversations
• Multi-modal input support for text, audio, and images
• Provider-agnostic event system with strongly-typed, JSON-serializable events

Implementation Details

Core Components:

• BidiAgent - Main agent class with start(), send(), receive(), stop() lifecycle methods
• _BidiAgentLoop - Event processing engine handling model events and tool execution with connection restart logic
• BidiModel - Model interface for bidirectional model providers
• BidiInput/BidiOutput - Pluggable I/O channel abstractions

Model Providers:

• BidiNovaSonicModel - AWS Bedrock Nova Sonic with complex event sequencing
• BidiGeminiLiveModel - Google Gemini Live using official SDK
• BidiOpenAIRealtimeModel - OpenAI Realtime API via WebSocket

I/O Handlers:

• BidiAudioIO - PyAudio-based microphone/speaker handling with buffering
• BidiTextIO - Terminal-based text input/output

Usage Example

import asyncio from strands.experimental.bidi import BidiAgent from strands.experimental.bidi.models import BidiNovaSonicModel from strands.experimental.bidi.io import BidiAudioIO, BidiTextIO from strands_tools import calculator async def main(): model = BidiNovaSonicModel() agent = BidiAgent(model=model, tools=[calculator]) audio_io = BidiAudioIO() text_io = BidiTextIO() await agent.run( inputs=[audio_io.input()], outputs=[audio_io.output(), text_io.output()] ) asyncio.run(main()) 

This is a new experimental feature under strands.experimental.bidi.

Related Issues

#217

Documentation PR

Type of Change

New feature

Testing

How have you tested the change? Verify that the changes do not break functionality or introduce warnings in consuming repositories: agents-docs, agents-tools, agents-cli

• [] I ran hatch run prepare
• I ran hatch run bidi:prepare: This is done to isolate the bidirectional streaming environment which needs Python 3.12+

Checklist

• [ x] I have read the CONTRIBUTING document
• I have added any necessary tests that prove my fix is effective or my feature works
• I have updated the documentation accordingly
• I have added an appropriate example to the documentation to outline the feature, or no new docs are needed
• My changes generate no new warnings
• Any dependent changes have been merged and published

---

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

[codecov]

Codecov Report

❌ Patch coverage is 90.00000% with 14 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
src/strands/types/session.py	54.54%	2 Missing and 3 partials ⚠️
src/strands/session/repository_session_manager.py	90.62%	1 Missing and 2 partials ⚠️
src/strands/session/session_manager.py	72.72%	3 Missing ⚠️
src/strands/tools/_caller.py	66.66%	0 Missing and 1 partial ⚠️
src/strands/tools/executors/_executor.py	95.23%	0 Missing and 1 partial ⚠️
src/strands/types/_events.py	85.71%	1 Missing ⚠️

📢 Thoughts on this report? Let us know!