[Python] AWS Bedrock Nova Sonic Bidirectional Stream output_stream is always None after...

I'm developing a Dockerized Python application using aws-sdk-bedrock-runtime to interact with Amazon Bedrock's Nova Sonic model (amazon.nova-sonic-v1:0) for a real-time, bidirectional speech-to-speech conversation.

Problem

My application successfully initializes the BedrockRuntimeClient and can invoke client.invoke_model_with_bidirectional_stream(). This call returns a smithy_core.aio.eventstream.DuplexEventStream object. I can successfully access the input_stream attribute of this object, and send initial control plane events (sessionStart, promptStart, system prompts) without issues.

However, the output_stream attribute of the returned DuplexEventStream object is consistently None. This prevents my application from receiving any audio or text responses from Nova Sonic, leading to an "Output stream not initialized" error when trying to process responses.

import asyncio
import json
import uuid
import warnings
import logging
import sys
from pathlib import Path
from dotenv import load_dotenv
import os

from aws_sdk_bedrock_runtime.client import BedrockRuntimeClient, InvokeModelWithBidirectionalStreamOperationInput
from aws_sdk_bedrock_runtime.config import Config, HTTPAuthSchemeResolver, SigV4AuthScheme
from aws_sdk_bedrock_runtime.models import InvokeModelWithBidirectionalStreamInputChunk, BidirectionalInputPayloadPart
from smithy_aws_core.credentials_resolvers.environment import EnvironmentCredentialsResolver

# Suppress warnings from libraries
warnings.filterwarnings("ignore")

# --- Configure Logging ---
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG) # Set to DEBUG to capture all detailed logs

# Create a console handler with a higher logging level
handler = logging.StreamHandler(sys.stdout)
handler.setLevel(logging.DEBUG) # Log DEBUG and above to console

# Create a formatter and add it to the handler
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)

# Add the handler to the logger
logger.addHandler(handler)

# Load environment variables from .env file
dotenv_path = Path('.env') # Assuming .env is in the current working directory
load_dotenv(dotenv_path=dotenv_path)

# Configuration for Nova Sonic
NOVA_SONIC_MODEL_ID = os.environ.get("NOVA_SONIC_MODEL_ID", "amazon.nova-sonic-v1:0")
ONLY_AVAILABLE_REGION_FOR_AWS_NOVA_SONIC = os.environ.get("ONLY_AVAILABLE_REGION_FOR_AWS_NOVA_SONIC", "us-east-1")


class BedrockStreamTester:
"""Tests bidirectional streaming with AWS Bedrock Nova Sonic without audio hardware."""

# Event templates (copied from original with minor formatting for direct use)
START_SESSION_EVENT = {
"event": {
"sessionStart": {
"inferenceConfiguration": {
"maxTokens": 1024,
"topP": 0.9,
"temperature": 0.7
}
}
}
}

START_PROMPT_EVENT = {
"event": {
"promptStart": {
"promptName": "", # Will be dynamically filled
"textOutputConfiguration": {
"mediaType": "text/plain"
},
"audioOutputConfiguration": {
"mediaType": "audio/lpcm",
"sampleRateHertz": 24000,
"sampleSizeBits": 16,
"channelCount": 1,
"voiceId": "matthew", # Or any available voice
"encoding": "base64",
"audioType": "SPEECH"
},
"toolUseOutputConfiguration": {
"mediaType": "application/json"
},
"toolConfiguration": {
"tools": []
}
}
}
}

TEXT_CONTENT_START_EVENT = {
"event": {
"contentStart": {
"promptName": "", # Will be dynamically filled
"contentName": "", # Will be dynamically filled
"role": "", # Will be dynamically filled (e.g., "SYSTEM")
"type": "TEXT",
"interactive": True, # Usually False for SYSTEM/ASSISTANT text
"textInputConfiguration": {
"mediaType": "text/plain"
}
}
}
}

TEXT_INPUT_EVENT = {
"event": {
"textInput": {
"promptName": "", # Will be dynamically filled
"contentName": "", # Will be dynamically filled
"content": "" # Will be dynamically filled
}
}
}

CONTENT_END_EVENT = {
"event": {
"contentEnd": {
"promptName": "", # Will be dynamically filled
"contentName": "" # Will be dynamically filled
}
}
}

PROMPT_END_EVENT = {
"event": {
"promptEnd": {
"promptName": "" # Will be dynamically filled
}
}
}

SESSION_END_EVENT = {
"event": {
"sessionEnd": {}
}
}

def __init__(self, model_id: str, region: str):
self.model_id = model_id
self.region = region
self.bedrock_client = None
self.stream_response = None
self.is_active = False

self.prompt_name = str(uuid.uuid4())
self.content_name = str(uuid.uuid4())

logger.info(f"Initializing BedrockStreamTester for model: {self.model_id} in region: {self.region}")
logger.info(f"Access Key ID prefix from env: {os.environ.get('AWS_ACCESS_KEY_ID', 'N/A')[:4]}...")
logger.info(f"Secret Access Key length from env: {len(os.environ.get('AWS_SECRET_ACCESS_KEY', ''))}")
logger.info(f"Session Token from env: {'Present' if os.environ.get('AWS_SESSION_TOKEN') else 'NOT Present'}")

def _initialize_client(self):
"""Initialize the Bedrock client."""
logger.info("Initializing BedrockRuntimeClient...")
try:
config = Config(
endpoint_uri=f"https://bedrock-runtime.{self.region}.amazonaws.com",
region=self.region,
aws_credentials_identity_resolver=EnvironmentCredentialsResolver(),
http_auth_scheme_resolver=HTTPAuthSchemeResolver(),
http_auth_schemes={"aws.auth#sigv4": SigV4AuthScheme()}
)
self.bedrock_client = BedrockRuntimeClient(config=config)
logger.info("BedrockRuntimeClient initialized successfully.")
except Exception as e:
logger.critical(f"Failed to initialize BedrockRuntimeClient: {e}", exc_info=True)
raise

async def _send_event(self, event_dict: dict):
"""Helper to send a JSON event dictionary to Bedrock's bidirectional input stream."""
if not self.stream_response or not self.is_active:
logger.warning("Stream not initialized or active, cannot send event.")
return

event_json = json.dumps(event_dict)
chunk = InvokeModelWithBidirectionalStreamInputChunk(
value=BidirectionalInputPayloadPart(bytes_=event_json.encode('utf-8'))
)
try:
await self.stream_response.input_stream.send(chunk)
# Log only the event type for clarity and to avoid sensitive data
event_keys = list(event_dict['event'].keys())
logger.debug(f"Sent control plane event: {event_keys}")
except Exception as e:
logger.error(f"Error sending control plane event {list(event_dict['event'].keys())}: {e}", exc_info=True)
raise

async def start_stream(self):
"""Starts the bidirectional stream and sends initial control plane messages."""
if not self.bedrock_client:
self._initialize_client()

logger.info("Attempting to invoke model with bidirectional stream...")
try:
self.stream_response = await self.bedrock_client.invoke_model_with_bidirectional_stream(
InvokeModelWithBidirectionalStreamOperationInput(model_id=self.model_id)
)
self.is_active = True
logger.info(f"Received raw stream response object type: {type(self.stream_response)}")

# Check if input_stream and output_stream attributes exist and are not None
input_stream = getattr(self.stream_response, 'input_stream', None)
output_stream = getattr(self.stream_response, 'output_stream', None)

if input_stream is not None:
logger.info("Nova Sonic input stream obtained successfully.")
else:
logger.error("Failed to obtain Nova Sonic input stream: It is None.")
raise Exception("Input stream not initialized.")

if output_stream is not None:
logger.info("Nova Sonic output stream obtained successfully.")
else:
logger.error("Failed to obtain Nova Sonic output stream: It is None.")
raise Exception("Output stream not initialized.")

logger.info("Both input and output streams appear to be valid.")

# --- Send initial session and prompt events as per documentation ---
logger.info("Sending initial START_SESSION_EVENT...")
await self._send_event(self.START_SESSION_EVENT)

logger.info("Sending START_PROMPT_EVENT...")
prompt_event = self.START_PROMPT_EVENT.copy()
prompt_event["event"]["promptStart"]["promptName"] = self.prompt_name
await self._send_event(prompt_event)

logger.info("Sending SYSTEM role text content...")
system_prompt_text = "You are a friendly assistant. Keep your responses short, generally two or three sentences."
text_content_start = self.TEXT_CONTENT_START_EVENT.copy()
text_content_start["event"]["contentStart"]["promptName"] = self.prompt_name
text_content_start["event"]["contentStart"]["contentName"] = self.content_name
text_content_start["event"]["contentStart"]["role"] = "SYSTEM"
text_content_start["event"]["contentStart"]["interactive"] = False # System prompts are not interactive
await self._send_event(text_content_start)

text_input_event = self.TEXT_INPUT_EVENT.copy()
text_input_event["event"]["textInput"]["promptName"] = self.prompt_name
text_input_event["event"]["textInput"]["contentName"] = self.content_name
text_input_event["event"]["textInput"]["content"] = system_prompt_text
await self._send_event(text_input_event)

text_content_end = self.CONTENT_END_EVENT.copy()
text_content_end["event"]["contentEnd"]["promptName"] = self.prompt_name
text_content_end["event"]["contentEnd"]["contentName"] = self.content_name
await self._send_event(text_content_end)

logger.info("Initial control plane events sent. Starting to read from output stream...")

# --- Process responses from the output stream ---
# This loop will continue until the stream is closed or an error occurs
async for chunk in self.stream_response.output_stream.receive_events():
if not self.is_active:
logger.info("Stream no longer active, stopping output processing.")
break

if chunk.value and chunk.value.bytes_:
response_data = chunk.value.bytes_.decode('utf-8')
try:
json_data = json.loads(response_data)
logger.debug(f"Received output event JSON keys: {list(json_data.keys())}")
# More detailed logging for different event types
if 'event' in json_data:
event_payload = json_data['event']
if 'completionStart' in event_payload:
logger.info(f"Received completionStart event: {event_payload['completionStart']}")
elif 'contentStart' in event_payload:
logger.info(
f"Received contentStart event: Type={event_payload['contentStart'].get('type')}, Role={event_payload['contentStart'].get('role')}")
elif 'textOutput' in event_payload:
# Log a snippet or length, not full text if potentially sensitive
content = event_payload['textOutput']['content']
logger.info(
f"Received textOutput (Role={event_payload['textOutput'].get('role')}, Length={len(content)}): {content[:100]}...")
elif 'audioOutput' in event_payload:
audio_len = len(base64.b64decode(event_payload['audioOutput']['content']))
logger.info(f"Received audioOutput: {audio_len} bytes of audio.")
elif 'contentEnd' in event_payload:
logger.info(
f"Received contentEnd event: Type={event_payload['contentEnd'].get('type')}, StopReason={event_payload['contentEnd'].get('stopReason')}")
elif 'completionEnd' in event_payload:
logger.info(f"Received completionEnd event: {event_payload['completionEnd']}")
break # Exit after getting the completion end
elif 'modelStreamErrorException' in event_payload:
logger.error(f"Received ModelStreamError: {event_payload['modelStreamErrorException']}")
break
elif 'internalServerException' in event_payload:
logger.error(
f"Received InternalServerException: {event_payload['internalServerException']}")
break
else:
logger.debug(f"Received unknown event type: {list(event_payload.keys())}")
else:
logger.warning(f"Received event without 'event' key: {response_data[:100]}...")

except json.JSONDecodeError:
logger.error(
f"Failed to decode JSON from output stream. Raw data snippet: {response_data[:100]}...",
exc_info=True)
break # Stop processing on JSON decode error
else:
logger.debug("Received empty or non-byte chunk from output stream.")

logger.info("Finished reading from output stream. Closing connection.")

except Exception as e:
self.is_active = False
logger.critical(f"Failed to start or maintain stream: {e}", exc_info=True)
raise

async def close_stream(self):
"""Properly closes the bidirectional stream."""
if not self.is_active:
logger.info("Stream is not active, nothing to close.")
return

logger.info("Initiating stream graceful shutdown...")
try:
logger.info("Sending PROMPT_END_EVENT...")
prompt_end_event = self.PROMPT_END_EVENT.copy()
prompt_end_event["event"]["promptEnd"]["promptName"] = self.prompt_name
await self._send_event(prompt_end_event)

logger.info("Sending SESSION_END_EVENT...")
await self._send_event(self.SESSION_END_EVENT)

# Close input stream first
if self.stream_response and hasattr(self.stream_response,
'input_stream') and self.stream_response.input_stream is not None:
await self.stream_response.input_stream.close()
logger.info("Input stream closed.")
else:
logger.warning("Input stream was already None or not available during close.")

# Attempt to gracefully close the output stream by ensuring it's consumed.
# In a real app, the consumer task would handle this. Here, we force it.
if self.stream_response and hasattr(self.stream_response,
'output_stream') and self.stream_response.output_stream is not None:
logger.info("Attempting to consume remaining output stream events before closing...")
try:
# Give it a very short moment for any final events to arrive after sending end signals
await asyncio.sleep(0.1)
async for _ in self.stream_response.output_stream.receive_events():
logger.debug("Consumed a remaining event from output stream during close.")
except Exception as e:
logger.warning(f"Error consuming remaining output stream events: {e}")
finally:
await self.stream_response.output_stream.close()
logger.info("Output stream closed.")
else:
logger.warning("Output stream was already None or not available during close.")

except Exception as e:
logger.error(f"Error during graceful stream shutdown: {e}", exc_info=True)
finally:
self.is_active = False
logger.info("Stream resources cleaned up.")


async def main():
logger.info("Starting Nova Sonic stream test application.")
tester = BedrockStreamTester(
model_id=NOVA_SONIC_MODEL_ID,
region=ONLY_AVAILABLE_REGION_FOR_AWS_NOVA_SONIC
)
try:
await tester.start_stream()
except Exception as e:
logger.critical(f"Application terminated due to an error during stream test: {e}", exc_info=True)
finally:
await tester.close_stream()
logger.info("Nova Sonic stream test application finished.")


if __name__ == "__main__":
# This ensures dotenv is loaded before main() attempts to access env vars
asyncio.run(main())


Observed Behavior (from detailed logs):

2025-07-10 09:43:59,414 - __main__ - INFO - Initializing BedrockStreamTester for model: amazon.nova-sonic-v1:0 in region: us-east-1
2025-07-10 09:43:59,425 - __main__ - INFO - BedrockRuntimeClient initialized successfully.
2025-07-10 09:43:59,425 - __main__ - INFO - Attempting to invoke model with bidirectional stream...
2025-07-10 09:43:59,436 - __main__ - INFO - Received raw stream response object type: <class 'smithy_core.aio.eventstream.DuplexEventStream'>
2025-07-10 09:43:59,436 - __main__ - INFO - Nova Sonic input stream obtained successfully.
2025-07-10 09:43:59,436 - __main__ - ERROR - Failed to obtain Nova Sonic output stream: It is None.
Exception: Output stream not initialized.


Context & Troubleshooting Steps Already Taken:

• 
  Network Connectivity: Confirmed the Docker container has outbound internet access to bedrock-runtime.us-east-1.amazonaws.com on port 443 via curl -vvv. The curl command successfully established a TLSv1.3 HTTP/2 connection and received a 404 UnknownOperationException (expected, as it was not a valid API request). This rules out basic network, DNS, or firewall issues.
  
  
• 
  AWS Credentials: Environment variables AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY are correctly loaded and are valid for other AWS operations (e.g., aws sts get-caller-identity works). AWS_SESSION_TOKEN is not present, as expected for long-term credentials.
  
  
• 
  IAM Policy: My IAM policy includes bedrock:InvokeModelWithBidirectionalStream and bedrock:InvokeModel actions for arn:aws:bedrock:us-east-1::foundation-model/* and specifically arn:aws:bedrock:us-east-1::foundation-model/amazon.nova-sonic-v1:0.
  
  
• 
  Bedrock Model Access (Opt-in): I have confirmed in the AWS Bedrock console that "Amazon Nova Sonic" in the US East (N. Virginia) region shows "Access granted".
  
  
• 
  SDK Version: aws-sdk-bedrock-runtime is updated to the latest available version.
  
  
• 
  Time Sync: Time synchronization on the Docker host is accurate.
  

Question

What could cause the output_stream to be None after invoke_model_with_bidirectional_stream successfully returns a DuplexEventStream object and the input_stream is available, especially given that general network access and account-level model opt-in seem to be configured correctly? Is there a subtle IAM permission missing specifically for the output part of the bidirectional stream, or a known issue with the aws-sdk-bedrock-runtime library that could cause this?

Continue reading...