Bidirectional Streaming Agent (#1276)

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

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.

---------

Co-authored-by: Murat Kaan Meral <murmeral@amazon.nl>
Co-authored-by: Patrick Gray <pgrayy@amazon.com>

Author: 3 people

.github/workflows/test-lint.yml

@@ -59,6 +59,20 @@ jobs:
         uses: actions/setup-python@v6
         with:
           python-version: ${{ matrix.python-version }}
+      - name: Install system audio dependencies (Linux)
+        if: matrix.os-name == 'linux'
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y portaudio19-dev libasound2-dev
+      - name: Install system audio dependencies (macOS)
+        if: matrix.os-name == 'macOS'
+        run: |
+          brew install portaudio
+      - name: Install system audio dependencies (Windows)
+        if: matrix.os-name == 'windows'
+        run: |
+          # Windows typically has audio libraries available by default
+          echo "Windows audio dependencies handled by PyAudio wheels"
       - name: Install dependencies
         run: |
           pip install --no-cache-dir hatch
@@ -89,6 +103,11 @@ jobs:
           python-version: '3.10'
           cache: 'pip'
 
+      - name: Install system audio dependencies (Linux)
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y portaudio19-dev libasound2-dev
+
       - name: Install dependencies
         run: |
           pip install --no-cache-dir hatch
@@ -97,3 +116,4 @@ jobs:
         id: lint
         run: hatch fmt --linter --check
         continue-on-error: false
+

.gitignore

@@ -13,3 +13,4 @@ dist
 repl_state
 .kiro
 uv.lock
+.audio_cache

README.md

@@ -197,6 +197,74 @@ agent("What is the square root of 1764")
 
 It's also available on GitHub via [strands-agents/tools](https://github.com/strands-agents/tools).
 
+### Bidirectional Streaming
+
+> **⚠️ Experimental Feature**: Bidirectional streaming is currently in experimental status. APIs may change in future releases as we refine the feature based on user feedback and evolving model capabilities.
+
+Build real-time voice and audio conversations with persistent streaming connections. Unlike traditional request-response patterns, bidirectional streaming maintains long-running conversations where users can interrupt, provide continuous input, and receive real-time audio responses. Get started with your first BidiAgent by following the [Quickstart](https://strandsagents.com/latest/documentation/docs/user-guide/concepts/experimental/bidirectional-streaming/quickstart) guide. 
+
+**Supported Model Providers:**
+- Amazon Nova Sonic (`amazon.nova-sonic-v1:0`)
+- Google Gemini Live (`gemini-2.5-flash-native-audio-preview-09-2025`)
+- OpenAI Realtime API (`gpt-realtime`)
+
+**Quick Example:**
+
+```python
+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.experimental.bidi.tools import stop_conversation
+from strands_tools import calculator
+
+async def main():
+    # Create bidirectional agent with audio model
+    model = BidiNovaSonicModel()
+    agent = BidiAgent(model=model, tools=[calculator, stop_conversation])
+
+    # Setup audio and text I/O
+    audio_io = BidiAudioIO()
+    text_io = BidiTextIO()
+
+    # Run with real-time audio streaming
+    # Say "stop conversation" to gracefully end the conversation
+    await agent.run(
+        inputs=[audio_io.input()],
+        outputs=[audio_io.output(), text_io.output()]
+    )
+
+if __name__ == "__main__":
+    asyncio.run(main())
+```
+
+**Configuration Options:**
+
+```python
+# Configure audio settings
+model = BidiNovaSonicModel(
+    provider_config={
+        "audio": {
+            "input_rate": 16000,
+            "output_rate": 16000,
+            "voice": "matthew"
+        },
+        "inference": {
+            "max_tokens": 2048,
+            "temperature": 0.7
+        }
+    }
+)
+
+# Configure I/O devices
+audio_io = BidiAudioIO(
+    input_device_index=0,  # Specific microphone
+    output_device_index=1,  # Specific speaker
+    input_buffer_size=10,
+    output_buffer_size=10
+)
+```
+
 ## Documentation
 
 For detailed guidance & examples, explore our documentation:

pyproject.toml

@@ -69,7 +69,18 @@ a2a = [
     "fastapi>=0.115.12,<1.0.0",
     "starlette>=0.46.2,<1.0.0",
 ]
+
+bidi = [
+    "aws_sdk_bedrock_runtime; python_version>='3.12'",
+    "prompt_toolkit>=3.0.0,<4.0.0",
+    "pyaudio>=0.2.13,<1.0.0",
+    "smithy-aws-core>=0.0.1; python_version>='3.12'",
+]
+bidi-gemini = ["google-genai>=1.32.0,<2.0.0"]
+bidi-openai = ["websockets>=15.0.0,<16.0.0"]
+
 all = ["strands-agents[a2a,anthropic,docs,gemini,litellm,llamaapi,mistral,ollama,openai,writer,sagemaker,otel]"]
+bidi-all = ["strands-agents[a2a,bidi,bidi-gemini,bidi-openai,docs,otel]"]
 
 dev = [
     "commitizen>=4.4.0,<5.0.0",
@@ -104,9 +115,10 @@ features = ["all"]
 dependencies = [
   "mypy>=1.15.0,<2.0.0",
   "ruff>=0.13.0,<0.14.0",
-  # Include required pacakge dependencies for mypy
+  # Include required package dependencies for mypy
   "strands-agents @ {root:uri}",
 ]
+python = "3.10"
 
 # Define static-analysis scripts so we can include mypy as part of the linting check
 [tool.hatch.envs.hatch-static-analysis.scripts]
@@ -118,7 +130,7 @@ format-fix = [
 ]
 lint-check = [
     "ruff check",
-    "mypy -p src"
+    "mypy ./src"
 ]
 lint-fix = [
     "ruff check --fix"
@@ -192,11 +204,16 @@ warn_no_return = true
 warn_unreachable = true
 follow_untyped_imports = true
 ignore_missing_imports = false
+exclude = ["src/strands/experimental/bidi"]
 
+[[tool.mypy.overrides]]
+module = ["strands.experimental.bidi.*"]
+follow_imports = "skip"
 
 [tool.ruff]
 line-length = 120
 include = ["examples/**/*.py", "src/**/*.py", "tests/**/*.py", "tests_integ/**/*.py"]
+exclude = ["src/strands/experimental/bidi/**/*.py", "tests/strands/experimental/bidi/**/*.py", "tests_integ/bidi/**/*.py"]
 
 [tool.ruff.lint]
 select = [
@@ -219,6 +236,7 @@ convention = "google"
 [tool.pytest.ini_options]
 testpaths = ["tests"]
 asyncio_default_fixture_loop_scope = "function"
+addopts = "--ignore=tests/strands/experimental/bidi --ignore=tests_integ/bidi"
 
 
 [tool.coverage.run]
@@ -227,6 +245,7 @@ source = ["src"]
 context = "thread"
 parallel = true
 concurrency = ["thread", "multiprocessing"]
+omit = ["src/strands/experimental/bidi/*"]
 
 [tool.coverage.report]
 show_missing = true
@@ -256,3 +275,48 @@ style = [
     ["text", ""],
     ["disabled", "fg:#858585 italic"]
 ]
+
+# =========================
+# Bidi development configs
+# =========================
+
+[tool.hatch.envs.bidi]
+dev-mode = true
+features = ["dev", "bidi-all"]
+installer = "uv"
+
+[tool.hatch.envs.bidi.scripts]
+prepare = [
+    "hatch run bidi-lint:format-fix",
+    "hatch run bidi-lint:quality-fix",
+    "hatch run bidi-lint:type-check",
+    "hatch run bidi-test:test-cov",
+]
+
+[tools.hatch.envs.bidi-lint]
+template = "bidi"
+
+[tool.hatch.envs.bidi-lint.scripts]
+format-check = "format-fix --check"
+format-fix = "ruff format {args} --target-version py312 ./src/strands/experimental/bidi/**/*.py"
+quality-check = "ruff check {args} --target-version py312 ./src/strands/experimental/bidi/**/*.py"
+quality-fix = "quality-check --fix"
+type-check = "mypy {args} --python-version 3.12 ./src/strands/experimental/bidi/**/*.py"
+
+[tool.hatch.envs.bidi-test]
+template = "bidi"
+
+[tool.hatch.envs.bidi-test.scripts]
+test = "pytest {args} tests/strands/experimental/bidi"
+test-cov = """
+test \
+    --cov=strands.experimental.bidi \
+    --cov-config= \
+    --cov-branch \
+    --cov-report=term-missing \
+    --cov-report=xml:build/coverage/bidi-coverage.xml \
+    --cov-report=html:build/coverage/bidi-html
+"""
+
+[[tool.hatch.envs.bidi-test.matrix]]
+python = ["3.13", "3.12"]

src/strands/experimental/bidi/__init__.py

@@ -0,0 +1,78 @@
+"""Bidirectional streaming package."""
+
+import sys
+
+if sys.version_info < (3, 12):
+    raise ImportError("bidi only supported for >= Python 3.12")
+
+# Main components - Primary user interface
+# Re-export standard agent events for tool handling
+from ...types._events import (
+    ToolResultEvent,
+    ToolStreamEvent,
+    ToolUseStreamEvent,
+)
+from .agent.agent import BidiAgent
+
+# IO channels - Hardware abstraction
+from .io.audio import BidiAudioIO
+
+# Model interface (for custom implementations)
+from .models.model import BidiModel
+from .models.nova_sonic import BidiNovaSonicModel
+
+# Built-in tools
+from .tools import stop_conversation
+
+# Event types - For type hints and event handling
+from .types.events import (
+    BidiAudioInputEvent,
+    BidiAudioStreamEvent,
+    BidiConnectionCloseEvent,
+    BidiConnectionStartEvent,
+    BidiErrorEvent,
+    BidiImageInputEvent,
+    BidiInputEvent,
+    BidiInterruptionEvent,
+    BidiOutputEvent,
+    BidiResponseCompleteEvent,
+    BidiResponseStartEvent,
+    BidiTextInputEvent,
+    BidiTranscriptStreamEvent,
+    BidiUsageEvent,
+    ModalityUsage,
+)
+
+__all__ = [
+    # Main interface
+    "BidiAgent",
+    # IO channels
+    "BidiAudioIO",
+    # Model providers
+    "BidiNovaSonicModel",
+    # Built-in tools
+    "stop_conversation",
+    # Input Event types
+    "BidiTextInputEvent",
+    "BidiAudioInputEvent",
+    "BidiImageInputEvent",
+    "BidiInputEvent",
+    # Output Event types
+    "BidiConnectionStartEvent",
+    "BidiConnectionCloseEvent",
+    "BidiResponseStartEvent",
+    "BidiResponseCompleteEvent",
+    "BidiAudioStreamEvent",
+    "BidiTranscriptStreamEvent",
+    "BidiInterruptionEvent",
+    "BidiUsageEvent",
+    "ModalityUsage",
+    "BidiErrorEvent",
+    "BidiOutputEvent",
+    # Tool Event types (reused from standard agent)
+    "ToolUseStreamEvent",
+    "ToolResultEvent",
+    "ToolStreamEvent",
+    # Model interface
+    "BidiModel",
+]

src/strands/experimental/bidi/_async/__init__.py

@@ -0,0 +1,29 @@
+"""Utilities for async operations."""
+
+from typing import Awaitable, Callable
+
+from ._task_pool import _TaskPool
+
+__all__ = ["_TaskPool"]
+
+
+async def stop_all(*funcs: Callable[..., Awaitable[None]]) -> None:
+    """Call all stops in sequence and aggregate errors.
+
+    A failure in one stop call will not block subsequent stop calls.
+
+    Args:
+        funcs: Stop functions to call in sequence.
+
+    Raises:
+        ExceptionGroup: If any stop function raises an exception.
+    """
+    exceptions = []
+    for func in funcs:
+        try:
+            await func()
+        except Exception as exception:
+            exceptions.append(exception)
+
+    if exceptions:
+        raise ExceptionGroup("failed stop sequence", exceptions)

src/strands/experimental/bidi/_async/_task_pool.py

@@ -0,0 +1,43 @@
+"""Manage pool of active async tasks.
+
+This is particularly useful for cancelling multiple tasks at once.
+"""
+
+import asyncio
+from typing import Any, Coroutine
+
+
+class _TaskPool:
+    """Manage pool of active async tasks."""
+
+    def __init__(self) -> None:
+        """Setup task container."""
+        self._tasks: set[asyncio.Task] = set()
+
+    def __len__(self) -> int:
+        """Number of active tasks."""
+        return len(self._tasks)
+
+    def create(self, coro: Coroutine[Any, Any, Any]) -> asyncio.Task:
+        """Create async task.
+
+        Adds a clean up callback to run after task completes.
+
+        Returns:
+            The created task.
+        """
+        task = asyncio.create_task(coro)
+        task.add_done_callback(lambda task: self._tasks.remove(task))
+
+        self._tasks.add(task)
+        return task
+
+    async def cancel(self) -> None:
+        """Cancel all active tasks in pool."""
+        for task in self._tasks:
+            task.cancel()
+
+        try:
+            await asyncio.gather(*self._tasks)
+        except asyncio.CancelledError:
+            pass

src/strands/experimental/bidi/agent/__init__.py

@@ -0,0 +1,5 @@
+"""Bidirectional agent for real-time streaming conversations."""
+
+from .agent import BidiAgent
+
+__all__ = ["BidiAgent"]