Upload 10 files

language_coach.py

@@ -1,6 +1,9 @@
 import re
 import json
 import os
+import boto3
+from datetime import datetime
+from config import REGION
 
 class LanguageCoach:
     def __init__(self):
@@ -9,21 +12,24 @@ class LanguageCoach:
         self.feedback_provided = set()  # Track which issues we've already given feedback on
         self.analysis_dir = "speech_analysis"
         
-        # Make sure analytics directory exists
+        # Create analytics directory if it doesn't exist
         if not os.path.exists(self.analysis_dir):
             os.makedirs(self.analysis_dir)
         
+        # Initialize Bedrock client for Claude analysis
+        self.bedrock = boto3.client('bedrock-runtime', region_name=REGION)
+        
         # Common speech disfluencies and patterns to track
         self.pattern_categories = {
             "filler_words": ["um", "uh", "er", "like", "you know", "sort of", "kind of"],
             "repetition": [],  # Will be filled dynamically
-            "incomplete_sentences": [],  # Detected by sentence structure analysis
+            "incomplete_sentences": [],  # Detected by LLM analysis
             "extended_pauses": []  # Detected in audio analysis
         }
         
     def analyze(self, transcript, session_id=None):
         """Analyze speech patterns in the transcript"""
-        if not transcript:
+        if not transcript or transcript.strip() == "":
             return None
             
         # Convert to lowercase for pattern matching
@@ -37,7 +43,8 @@ class LanguageCoach:
                 "incomplete_sentences": 0,
                 "extended_pauses": 0,
                 "total_words": 0,
-                "total_sentences": 0
+                "total_sentences": 0,
+                "llm_analyses": []
             }
             
         # Count total words and sentences
@@ -49,8 +56,26 @@ class LanguageCoach:
             self.user_patterns[session_id]["total_words"] += len(words)
             self.user_patterns[session_id]["total_sentences"] += len(sentences)
         
-        # Analyze filler words
+        # Basic pattern analysis
+        detected_patterns = self._analyze_basic_patterns(text, words, session_id)
+        
+        # Add LLM analysis if there's enough speech content
+        if len(words) > 5 and session_id:
+            llm_patterns = self._perform_llm_analysis(transcript, session_id)
+            if llm_patterns:
+                detected_patterns.extend(llm_patterns)
+        
+        # If session_id is provided, save the analysis
+        if session_id:
+            self._save_analysis(session_id)
+            
+        return detected_patterns
+    
+    def _analyze_basic_patterns(self, text, words, session_id=None):
+        """Analyze basic speech patterns like fillers and repetitions"""
         detected_patterns = []
+        
+        # Analyze filler words
         for filler in self.pattern_categories["filler_words"]:
             filler_count = text.count(filler)
             if filler_count > 0:
@@ -91,12 +116,88 @@ class LanguageCoach:
                         "suggestion": f"You sometimes repeat '{word}' when speaking"
                     })
                     self.feedback_provided.add(pattern_key)
-        
-        # If session_id is provided, save the analysis
-        if session_id:
-            self._save_analysis(session_id)
-            
+                    
         return detected_patterns
+    
+    def _perform_llm_analysis(self, transcript, session_id):
+        """Use an LLM to analyze speech patterns more deeply"""
+        try:
+            # Prepare the prompt for LLM analysis
+            prompt = f"""Analyze the following speech transcript for language patterns and provide actionable feedback:
+
+Speech: "{transcript}"
+
+Focus on:
+1. Speech fluency (hesitations, incomplete sentences, abrupt transitions)
+2. Grammar issues (tense, agreement, word order)
+3. Vocabulary usage (appropriate word choice, variety)
+4. Pronunciation indicators (as inferred from transcription)
+5. Organization of thoughts
+
+Respond with a JSON structure containing:
+- Overall assessment (short paragraph)
+- Detailed pattern analysis (array of specific issues found)
+- Suggestions for improvement (actionable items)
+
+Format your response as valid JSON only, no preamble or explanation.
+"""
+
+            # Call Claude via Amazon Bedrock
+            response = self.bedrock.invoke_model(
+                modelId="anthropic.claude-3-sonnet-20240229-v1:0",
+                body=json.dumps({
+                    "anthropic_version": "bedrock-2023-05-31",
+                    "max_tokens": 1000,
+                    "messages": [
+                        {
+                            "role": "user",
+                            "content": prompt
+                        }
+                    ],
+                    "temperature": 0.3
+                }),
+                contentType="application/json",
+                accept="application/json"
+            )
+            
+            # Parse the response
+            response_body = json.loads(response['body'].read())
+            content = response_body.get('content', [])
+            llm_text = ''.join([item.get('text', '') for item in content if item.get('type') == 'text'])
+            
+            # Extract JSON from the response
+            try:
+                analysis = json.loads(llm_text.strip())
+                
+                # Store the analysis in the user patterns
+                if session_id:
+                    timestamp = datetime.now().isoformat()
+                    self.user_patterns[session_id]["llm_analyses"].append({
+                        "timestamp": timestamp,
+                        "transcript": transcript,
+                        "analysis": analysis
+                    })
+                
+                # Convert LLM analysis to our pattern format
+                patterns = []
+                if "detailed_pattern_analysis" in analysis:
+                    for issue in analysis["detailed_pattern_analysis"]:
+                        if isinstance(issue, dict) and "type" in issue and "description" in issue:
+                            patterns.append({
+                                "type": "llm_" + issue["type"].lower().replace(" ", "_"),
+                                "pattern": issue["description"],
+                                "suggestion": issue.get("suggestion", "Consider working on this pattern")
+                            })
+                
+                return patterns
+                
+            except json.JSONDecodeError:
+                print("LLM response did not contain valid JSON")
+                return []
+                
+        except Exception as e:
+            print(f"Error during LLM analysis: {e}")
+            return []
         
     def get_coaching_context(self, session_id):
         """Get contextual coaching information for the Nova assistant"""
@@ -116,15 +217,86 @@ class LanguageCoach:
             
         if top_repetitions:
             coaching_context += "\n- Word repetitions: " + ", ".join([f"'{w}' ({c} times)" for w, c in top_repetitions])
+        
+        # Add LLM insights if available
+        if patterns.get("llm_analyses") and len(patterns["llm_analyses"]) > 0:
+            latest_analysis = patterns["llm_analyses"][-1]["analysis"]
+            if "overall_assessment" in latest_analysis:
+                coaching_context += f"\n\nSpeech assessment: {latest_analysis['overall_assessment']}"
+                
+            if "suggestions_for_improvement" in latest_analysis:
+                suggestions = latest_analysis["suggestions_for_improvement"]
+                if isinstance(suggestions, list) and len(suggestions) > 0:
+                    coaching_context += "\n\nSuggested areas to work on:"
+                    for idx, suggestion in enumerate(suggestions[:3]):  # Limit to top 3 suggestions
+                        if isinstance(suggestion, str):
+                            coaching_context += f"\n- {suggestion}"
+                        elif isinstance(suggestion, dict) and "text" in suggestion:
+                            coaching_context += f"\n- {suggestion['text']}"
             
         # Add coaching instructions
         coaching_context += "\n\nCoaching approach:"
         coaching_context += "\n- Model clear speech without these patterns"
         coaching_context += "\n- Maintain natural conversation flow"
         coaching_context += "\n- Don't explicitly point out errors"
+        coaching_context += "\n- Rephrase user's sentences occasionally with correct patterns"
         
         return coaching_context
         
+    def get_coaching_report(self, session_id):
+        """Generate a coaching report for the user"""
+        if session_id not in self.user_patterns:
+            return "No speech data available for analysis."
+            
+        patterns = self.user_patterns[session_id]
+        
+        report = "## Speech Analysis Report\n\n"
+        
+        # Basic statistics
+        report += "### Conversation Statistics\n"
+        report += f"- Total words spoken: {patterns['total_words']}\n"
+        report += f"- Total sentences: {patterns['total_sentences']}\n"
+        
+        # Filler word usage
+        if patterns["filler_words"]:
+            report += "\n### Filler Word Usage\n"
+            for word, count in sorted(patterns["filler_words"].items(), key=lambda x: x[1], reverse=True):
+                percentage = (count / patterns['total_words']) * 100 if patterns['total_words'] > 0 else 0
+                report += f"- '{word}': {count} times ({percentage:.1f}% of total words)\n"
+        
+        # Word repetitions
+        if patterns["repetition"]:
+            report += "\n### Word Repetitions\n"
+            for word, count in sorted(patterns["repetition"].items(), key=lambda x: x[1], reverse=True)[:5]:
+                report += f"- '{word}': repeated {count} times\n"
+        
+        # LLM insights
+        if patterns.get("llm_analyses") and len(patterns["llm_analyses"]) > 0:
+            report += "\n### Language Model Analysis\n"
+            
+            # Get the latest analysis
+            latest = patterns["llm_analyses"][-1]["analysis"]
+            
+            if "overall_assessment" in latest:
+                report += f"{latest['overall_assessment']}\n\n"
+                
+            if "detailed_pattern_analysis" in latest and isinstance(latest["detailed_pattern_analysis"], list):
+                report += "#### Detailed Pattern Analysis\n"
+                for item in latest["detailed_pattern_analysis"]:
+                    if isinstance(item, dict):
+                        if "type" in item and "description" in item:
+                            report += f"- **{item['type']}**: {item['description']}\n"
+            
+            if "suggestions_for_improvement" in latest and isinstance(latest["suggestions_for_improvement"], list):
+                report += "\n#### Suggestions for Improvement\n"
+                for item in latest["suggestions_for_improvement"]:
+                    if isinstance(item, str):
+                        report += f"- {item}\n"
+                    elif isinstance(item, dict) and "text" in item:
+                        report += f"- {item['text']}\n"
+        
+        return report
+        
     def _save_analysis(self, session_id):
         """Save the speech analysis for this session"""
         analysis_file = os.path.join(self.analysis_dir, f"{session_id}_analysis.json")

nova_sonic.py

@@ -7,49 +7,99 @@ from config import MODEL_ID, REGION, INPUT_SAMPLE_RATE, BIT_DEPTH, CHANNELS, GRE
 class NovaSonicClient:
     def __init__(self):
         self.client = boto3.client('bedrock-runtime', region_name=REGION)
-        self.model_id = MODEL_ID
-        self.audio_content_type = f"application/vnd.amazon.models.audio.lpcm;sample-rate={INPUT_SAMPLE_RATE};bit-depth={BIT_DEPTH};channels={CHANNELS}"
+        self.model_id = "amazon.nova-sonic-v1:0"  # Use the official Nova Sonic model ID
         self.stream = None
-        self.buffer = b""
-        self.transcripts = []
+        self.session_active = False
         
     def start_stream(self, session_manager=None, language_coach=None):
-        """Initialize a bidirectional stream with Nova"""
+        """Initialize a bidirectional stream with Nova Sonic"""
         try:
+            # Initialize the stream
             self.stream = self.client.invoke_model_with_bidirectional_stream(
                 modelId=self.model_id,
-                contentType=self.audio_content_type,
-                accept="application/json"
+                contentType="application/vnd.amazon.eventstream",
+                accept="application/vnd.amazon.eventstream"
             )
             
-            print("Connected to Nova Sonic")
+            # Send session start event with configuration
+            self._send_session_start()
             
-            # Send a system message to set up Nova's role
+            # Send system message to configure Nova's behavior
             self._send_system_message(session_manager, language_coach)
             
-            # Send initial greeting to start the conversation
+            # Send initial greeting as a text message
             greeting_result = self.send_text(GREETING_TEXT)
             
             if session_manager and greeting_result:
                 session_manager.set_last_response(greeting_result['audio'])
                 
+            self.session_active = True
+            print("Connected to Nova Sonic")
             return greeting_result
-                
+            
         except Exception as e:
             print(f"Error starting Nova stream: {e}")
             return None
             
+    def _send_session_start(self):
+        """Send session start event to initialize the stream"""
+        start_session_event = {
+            "event": {
+                "sessionStart": {
+                    "inferenceConfiguration": {
+                        "maxTokens": 1024,
+                        "topP": 0.9,
+                        "temperature": 0.7
+                    },
+                    "audioOutputConfiguration": {
+                        "sampleRateHertz": 24000,
+                        "sampleSizeBits": 16,
+                        "channelCount": 1,
+                        "encoding": "base64",
+                        "audioType": "SPEECH"
+                    }
+                }
+            }
+        }
+        
+        try:
+            # Send the event as bytes
+            self.stream.send_chunk({
+                "chunk": {
+                    "bytes": json.dumps(start_session_event).encode('utf-8')
+                }
+            })
+            
+            # Wait for acknowledgment
+            time.sleep(0.5)
+            
+        except Exception as e:
+            print(f"Error sending session start: {e}")
+            
     def send_audio(self, audio_chunk):
-        """Send an audio chunk to Nova and get response"""
-        if not self.stream:
-            print("Stream not initialized. Call start_stream() first.")
+        """Send an audio chunk to Nova Sonic and get response"""
+        if not self.stream or not self.session_active:
+            print("Stream not initialized or session not active")
             return None
             
         try:
-            # Send the audio chunk
+            # Encode the audio as base64
+            audio_base64 = base64.b64encode(audio_chunk).decode('utf-8')
+            
+            # Create the audio event
+            audio_event = {
+                "event": {
+                    "audioInput": {
+                        "data": audio_base64,
+                        "endpointResponse": True
+                    }
+                }
+            }
+            
+            # Send the audio event
             self.stream.send_chunk({
                 "chunk": {
-                    "bytes": audio_chunk
+                    "bytes": json.dumps(audio_event).encode('utf-8')
                 }
             })
             
@@ -57,24 +107,29 @@ class NovaSonicClient:
             audio_data = b""
             transcript = ""
             
-            response_ready = False
-            while not response_ready:
+            # Process response chunks until we get the final one
+            response_complete = False
+            while not response_complete:
                 chunk = self.stream.read_chunk()
                 if "chunk" in chunk:
-                    payload = json.loads(chunk["chunk"]["bytes"].decode())
-                    
-                    # Extract audio if present
-                    if "output_audio" in payload:
-                        audio_base64 = payload["output_audio"]
-                        audio_data += base64.b64decode(audio_base64)
-                    
-                    # Extract transcript if present
-                    if "transcript" in payload:
-                        transcript = payload["transcript"]
-                    
-                    # Check if this is the last chunk
-                    if "is_final" in payload and payload["is_final"]:
-                        response_ready = True
+                    try:
+                        payload = json.loads(chunk["chunk"]["bytes"].decode('utf-8'))
+                        
+                        # Check for audio output
+                        if "event" in payload and "audioOutput" in payload["event"]:
+                            audio_base64 = payload["event"]["audioOutput"]["data"]
+                            audio_data += base64.b64decode(audio_base64)
+                            
+                        # Check for transcript
+                        if "event" in payload and "transcript" in payload["event"]:
+                            transcript = payload["event"]["transcript"]["text"]
+                            
+                        # Check if this is the end of the response
+                        if "event" in payload and "endpointResponse" in payload["event"]:
+                            response_complete = True
+                            
+                    except Exception as e:
+                        print(f"Error processing response chunk: {e}")
                         
             return {
                 "audio": audio_data,
@@ -87,20 +142,25 @@ class NovaSonicClient:
             
     def send_text(self, text_message):
         """Send a text message instead of audio and get response"""
-        if not self.stream:
-            print("Stream not initialized. Call start_stream() first.")
+        if not self.stream or not self.session_active:
+            print("Stream not initialized or session not active")
             return None
             
         try:
-            # Create the text message payload
-            message_payload = {
-                "message": text_message
+            # Create the text input event
+            text_event = {
+                "event": {
+                    "textInput": {
+                        "text": text_message,
+                        "endpointResponse": True
+                    }
+                }
             }
             
-            # Send the message
+            # Send the text event
             self.stream.send_chunk({
                 "chunk": {
-                    "bytes": json.dumps(message_payload).encode()
+                    "bytes": json.dumps(text_event).encode('utf-8')
                 }
             })
             
@@ -108,24 +168,29 @@ class NovaSonicClient:
             audio_data = b""
             transcript = ""
             
-            response_ready = False
-            while not response_ready:
+            # Process response chunks until we get the final one
+            response_complete = False
+            while not response_complete:
                 chunk = self.stream.read_chunk()
                 if "chunk" in chunk:
-                    payload = json.loads(chunk["chunk"]["bytes"].decode())
-                    
-                    # Extract audio if present
-                    if "output_audio" in payload:
-                        audio_base64 = payload["output_audio"]
-                        audio_data += base64.b64decode(audio_base64)
-                    
-                    # Extract transcript if present
-                    if "response" in payload:
-                        transcript = payload["response"]
-                    
-                    # Check if this is the last chunk
-                    if "is_final" in payload and payload["is_final"]:
-                        response_ready = True
+                    try:
+                        payload = json.loads(chunk["chunk"]["bytes"].decode('utf-8'))
+                        
+                        # Check for audio output
+                        if "event" in payload and "audioOutput" in payload["event"]:
+                            audio_base64 = payload["event"]["audioOutput"]["data"]
+                            audio_data += base64.b64decode(audio_base64)
+                            
+                        # Check for transcript
+                        if "event" in payload and "transcript" in payload["event"]:
+                            transcript = payload["event"]["transcript"]["text"]
+                            
+                        # Check if this is the end of the response
+                        if "event" in payload and "endpointResponse" in payload["event"]:
+                            response_complete = True
+                            
+                    except Exception as e:
+                        print(f"Error processing response chunk: {e}")
                         
             return {
                 "audio": audio_data,
@@ -139,35 +204,41 @@ class NovaSonicClient:
     def _send_system_message(self, session_manager=None, language_coach=None):
         """Send a system message to configure Nova's behavior"""
         # Build system message with context from session and language coach
-        system_message = {
-            "role": "system",
-            "content": (
-                "You are Nova, a friendly and supportive conversation partner. "
-                "Your goal is to engage in natural, flowing conversation that feels human and authentic. "
-                "Respond thoughtfully and maintain the conversation context. "
-                "Keep your responses concise and conversational. "
-                "Ask open-ended questions to encourage the user to practice speaking. "
-                "Never mention that you are an AI unless explicitly asked."
-            )
-        }
+        system_content = (
+            "You are Nova, a friendly and supportive conversation partner. "
+            "Your goal is to engage in natural, flowing conversation that feels human and authentic. "
+            "Respond thoughtfully and maintain the conversation context. "
+            "Keep your responses concise and conversational. "
+            "Ask open-ended questions to encourage the user to practice speaking. "
+            "Never mention that you are an AI unless explicitly asked."
+        )
         
         # Add conversation history context if available
         if session_manager:
             conversation_context = session_manager.get_conversation_context()
             if conversation_context:
-                system_message["content"] += f"\n\nConversation history:\n{conversation_context}"
+                system_content += f"\n\nConversation history:\n{conversation_context}"
                 
         # Add language coaching context if available
         if language_coach and session_manager:
             coaching_context = language_coach.get_coaching_context(session_manager.session_id)
             if coaching_context:
-                system_message["content"] += f"\n\n{coaching_context}"
+                system_content += f"\n\n{coaching_context}"
+        
+        # Create the system event
+        system_event = {
+            "event": {
+                "systemInput": {
+                    "content": system_content
+                }
+            }
+        }
         
-        # Send the system message
         try:
+            # Send the system event
             self.stream.send_chunk({
                 "chunk": {
-                    "bytes": json.dumps(system_message).encode()
+                    "bytes": json.dumps(system_event).encode('utf-8')
                 }
             })
             
@@ -181,7 +252,24 @@ class NovaSonicClient:
         """Close the streaming connection"""
         if self.stream:
             try:
+                # Send session end event
+                end_event = {
+                    "event": {
+                        "sessionEnd": {}
+                    }
+                }
+                
+                self.stream.send_chunk({
+                    "chunk": {
+                        "bytes": json.dumps(end_event).encode('utf-8')
+                    }
+                })
+                
+                # Close the stream
                 self.stream.done()
+                self.session_active = False
                 print("Nova stream closed")
+                
             except Exception as e:
                 print(f"Error closing stream: {e}")
+EOF < /dev/null

nova_sonic_tool_use.py

@@ -0,0 +1,882 @@
+import os
+import asyncio
+import base64
+import json
+import uuid
+import warnings
+import pyaudio
+import pytz
+import random
+import hashlib
+import datetime
+import time
+import inspect
+from aws_sdk_bedrock_runtime.client import BedrockRuntimeClient, InvokeModelWithBidirectionalStreamOperationInput
+from aws_sdk_bedrock_runtime.models import InvokeModelWithBidirectionalStreamInputChunk, BidirectionalInputPayloadPart
+from aws_sdk_bedrock_runtime.config import Config, HTTPAuthSchemeResolver, SigV4AuthScheme
+from smithy_aws_core.credentials_resolvers.environment import EnvironmentCredentialsResolver
+
+# Suppress warnings
+warnings.filterwarnings("ignore")
+
+# Audio configuration
+INPUT_SAMPLE_RATE = 16000
+OUTPUT_SAMPLE_RATE = 24000
+CHANNELS = 1
+FORMAT = pyaudio.paInt16
+CHUNK_SIZE = 1024  # Number of frames per buffer
+
+# Debug mode flag
+DEBUG = False
+
+def debug_print(message):
+    """Print only if debug mode is enabled"""
+    if DEBUG:
+        functionName = inspect.stack()[1].function
+        if  functionName == 'time_it' or functionName == 'time_it_async':
+            functionName = inspect.stack()[2].function
+        print('{:%Y-%m-%d %H:%M:%S.%f}'.format(datetime.datetime.now())[:-3] + ' ' + functionName + ' ' + message)
+
+def time_it(label, methodToRun):
+    start_time = time.perf_counter()
+    result = methodToRun()
+    end_time = time.perf_counter()
+    debug_print(f"Execution time for {label}: {end_time - start_time:.4f} seconds")
+    return result
+
+async def time_it_async(label, methodToRun):
+    start_time = time.perf_counter()
+    result = await methodToRun()
+    end_time = time.perf_counter()
+    debug_print(f"Execution time for {label}: {end_time - start_time:.4f} seconds")
+    return result
+
+class BedrockStreamManager:
+    """Manages bidirectional streaming with AWS Bedrock using asyncio"""
+    
+    # Event templates
+    START_SESSION_EVENT = '''{
+        "event": {
+            "sessionStart": {
+            "inferenceConfiguration": {
+                "maxTokens": 1024,
+                "topP": 0.9,
+                "temperature": 0.7
+                }
+            }
+        }
+    }'''
+
+    CONTENT_START_EVENT = '''{
+        "event": {
+            "contentStart": {
+            "promptName": "%s",
+            "contentName": "%s",
+            "type": "AUDIO",
+            "interactive": true,
+            "role": "USER",
+            "audioInputConfiguration": {
+                "mediaType": "audio/lpcm",
+                "sampleRateHertz": 16000,
+                "sampleSizeBits": 16,
+                "channelCount": 1,
+                "audioType": "SPEECH",
+                "encoding": "base64"
+                }
+            }
+        }
+    }'''
+
+    AUDIO_EVENT_TEMPLATE = '''{
+        "event": {
+            "audioInput": {
+            "promptName": "%s",
+            "contentName": "%s",
+            "content": "%s"
+            }
+        }
+    }'''
+
+    TEXT_CONTENT_START_EVENT = '''{
+        "event": {
+            "contentStart": {
+            "promptName": "%s",
+            "contentName": "%s",
+            "type": "TEXT",
+            "role": "%s",
+            "interactive": true,
+                "textInputConfiguration": {
+                    "mediaType": "text/plain"
+                }
+            }
+        }
+    }'''
+
+    TEXT_INPUT_EVENT = '''{
+        "event": {
+            "textInput": {
+            "promptName": "%s",
+            "contentName": "%s",
+            "content": "%s"
+            }
+        }
+    }'''
+
+    TOOL_CONTENT_START_EVENT = '''{
+        "event": {
+            "contentStart": {
+                "promptName": "%s",
+                "contentName": "%s",
+                "interactive": false,
+                "type": "TOOL",
+                "role": "TOOL",
+                "toolResultInputConfiguration": {
+                    "toolUseId": "%s",
+                    "type": "TEXT",
+                    "textInputConfiguration": {
+                        "mediaType": "text/plain"
+                    }
+                }
+            }
+        }
+    }'''
+
+    CONTENT_END_EVENT = '''{
+        "event": {
+            "contentEnd": {
+            "promptName": "%s",
+            "contentName": "%s"
+            }
+        }
+    }'''
+
+    PROMPT_END_EVENT = '''{
+        "event": {
+            "promptEnd": {
+            "promptName": "%s"
+            }
+        }
+    }'''
+
+    SESSION_END_EVENT = '''{
+        "event": {
+            "sessionEnd": {}
+        }
+    }'''
+    
+    def start_prompt(self):
+        """Create a promptStart event"""
+        get_default_tool_schema = json.dumps({
+            "type": "object",
+            "properties": {},
+            "required": []
+        })
+
+        get_order_tracking_schema = json.dumps({
+            "type": "object",
+            "properties": {
+                "orderId": {
+                    "type": "string",
+                    "description": "The order number or ID to track"
+                },
+                "requestNotifications": {
+                    "type": "boolean",
+                    "description": "Whether to set up notifications for this order",
+                    "default": False
+                }
+            },
+            "required": ["orderId"]
+        })
+
+        
+        prompt_start_event = {
+            "event": {
+                "promptStart": {
+                    "promptName": self.prompt_name,
+                    "textOutputConfiguration": {
+                        "mediaType": "text/plain"
+                    },
+                    "audioOutputConfiguration": {
+                        "mediaType": "audio/lpcm",
+                        "sampleRateHertz": 24000,
+                        "sampleSizeBits": 16,
+                        "channelCount": 1,
+                        "voiceId": "matthew",
+                        "encoding": "base64",
+                        "audioType": "SPEECH"
+                    },
+                    "toolUseOutputConfiguration": {
+                        "mediaType": "application/json"
+                    },
+                    "toolConfiguration": {
+                        "tools": [
+                            {
+                                "toolSpec": {
+                                    "name": "getDateAndTimeTool",
+                                    "description": "get information about the current date and time",
+                                    "inputSchema": {
+                                        "json": get_default_tool_schema
+                                    }
+                                }
+                            },
+                            {
+                                "toolSpec": {
+                                    "name": "trackOrderTool",
+                                    "description": "Retrieves real-time order tracking information and detailed status updates for customer orders by order ID. Provides estimated delivery dates. Use this tool when customers ask about their order status or delivery timeline.",
+                                    "inputSchema": {
+                                    "json": get_order_tracking_schema
+                                    }
+                                }
+                            }
+                        ]
+                    }
+                }
+            }
+        }
+        
+        return json.dumps(prompt_start_event)
+    
+    def tool_result_event(self, content_name, content, role):
+        """Create a tool result event"""
+
+        if isinstance(content, dict):
+            content_json_string = json.dumps(content)
+        else:
+            content_json_string = content
+            
+        tool_result_event = {
+            "event": {
+                "toolResult": {
+                    "promptName": self.prompt_name,
+                    "contentName": content_name,
+                    "content": content_json_string
+                }
+            }
+        }
+        return json.dumps(tool_result_event)
+   
+    def __init__(self, model_id='amazon.nova-sonic-v1:0', region='us-east-1'):
+        """Initialize the stream manager."""
+        self.model_id = model_id
+        self.region = region
+        
+        # Replace RxPy subjects with asyncio queues
+        self.audio_input_queue = asyncio.Queue()
+        self.audio_output_queue = asyncio.Queue()
+        self.output_queue = asyncio.Queue()
+        
+        self.response_task = None
+        self.stream_response = None
+        self.is_active = False
+        self.barge_in = False
+        self.bedrock_client = None
+        
+        # Audio playback components
+        self.audio_player = None
+        
+        # Text response components
+        self.display_assistant_text = False
+        self.role = None
+
+        # Session information
+        self.prompt_name = str(uuid.uuid4())
+        self.content_name = str(uuid.uuid4())
+        self.audio_content_name = str(uuid.uuid4())
+        self.toolUseContent = ""
+        self.toolUseId = ""
+        self.toolName = ""
+
+    def _initialize_client(self):
+        """Initialize the Bedrock client."""
+        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)
+    
+    async def initialize_stream(self):
+        """Initialize the bidirectional stream with Bedrock."""
+        if not self.bedrock_client:
+            self._initialize_client()
+        
+        try:
+            self.stream_response = await time_it_async("invoke_model_with_bidirectional_stream", lambda : self.bedrock_client.invoke_model_with_bidirectional_stream( InvokeModelWithBidirectionalStreamOperationInput(model_id=self.model_id)))
+            self.is_active = True
+            default_system_prompt = "You are a friend. The user and you will engage in a spoken dialog exchanging the transcripts of a natural real-time conversation." \
+            "When reading order numbers, please read each digit individually, separated by pauses. For example, order #1234 should be read as 'order number one-two-three-four' rather than 'order number one thousand two hundred thirty-four'."
+            
+            # Send initialization events
+            prompt_event = self.start_prompt()
+            text_content_start = self.TEXT_CONTENT_START_EVENT % (self.prompt_name, self.content_name, "SYSTEM")
+            text_content = self.TEXT_INPUT_EVENT % (self.prompt_name, self.content_name, default_system_prompt)
+            text_content_end = self.CONTENT_END_EVENT % (self.prompt_name, self.content_name)
+            
+            init_events = [self.START_SESSION_EVENT, prompt_event, text_content_start, text_content, text_content_end]
+            
+            for event in init_events:
+                await self.send_raw_event(event)
+                # Small delay between init events
+                await asyncio.sleep(0.1)
+            
+            # Start listening for responses
+            self.response_task = asyncio.create_task(self._process_responses())
+            
+            # Start processing audio input
+            asyncio.create_task(self._process_audio_input())
+            
+            # Wait a bit to ensure everything is set up
+            await asyncio.sleep(0.1)
+            
+            debug_print("Stream initialized successfully")
+            return self
+        except Exception as e:
+            self.is_active = False
+            print(f"Failed to initialize stream: {str(e)}")
+            raise
+    
+    async def send_raw_event(self, event_json):
+        """Send a raw event JSON to the Bedrock stream."""
+        if not self.stream_response or not self.is_active:
+            debug_print("Stream not initialized or closed")
+            return
+       
+        event = InvokeModelWithBidirectionalStreamInputChunk(
+            value=BidirectionalInputPayloadPart(bytes_=event_json.encode('utf-8'))
+        )
+        
+        try:
+            await self.stream_response.input_stream.send(event)
+            # For debugging large events, you might want to log just the type
+            if DEBUG:
+                if len(event_json) > 200:
+                    event_type = json.loads(event_json).get("event", {}).keys()
+                    debug_print(f"Sent event type: {list(event_type)}")
+                else:
+                    debug_print(f"Sent event: {event_json}")
+        except Exception as e:
+            debug_print(f"Error sending event: {str(e)}")
+            if DEBUG:
+                import traceback
+                traceback.print_exc()
+    
+    async def send_audio_content_start_event(self):
+        """Send a content start event to the Bedrock stream."""
+        content_start_event = self.CONTENT_START_EVENT % (self.prompt_name, self.audio_content_name)
+        await self.send_raw_event(content_start_event)
+    
+    async def _process_audio_input(self):
+        """Process audio input from the queue and send to Bedrock."""
+        while self.is_active:
+            try:
+                # Get audio data from the queue
+                data = await self.audio_input_queue.get()
+                
+                audio_bytes = data.get('audio_bytes')
+                if not audio_bytes:
+                    debug_print("No audio bytes received")
+                    continue
+                
+                # Base64 encode the audio data
+                blob = base64.b64encode(audio_bytes)
+                audio_event = self.AUDIO_EVENT_TEMPLATE % (
+                    self.prompt_name, 
+                    self.audio_content_name, 
+                    blob.decode('utf-8')
+                )
+                
+                # Send the event
+                await self.send_raw_event(audio_event)
+                
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                debug_print(f"Error processing audio: {e}")
+                if DEBUG:
+                    import traceback
+                    traceback.print_exc()
+    
+    def add_audio_chunk(self, audio_bytes):
+        """Add an audio chunk to the queue."""
+        self.audio_input_queue.put_nowait({
+            'audio_bytes': audio_bytes,
+            'prompt_name': self.prompt_name,
+            'content_name': self.audio_content_name
+        })
+    
+    async def send_audio_content_end_event(self):
+        """Send a content end event to the Bedrock stream."""
+        if not self.is_active:
+            debug_print("Stream is not active")
+            return
+        
+        content_end_event = self.CONTENT_END_EVENT % (self.prompt_name, self.audio_content_name)
+        await self.send_raw_event(content_end_event)
+        debug_print("Audio ended")
+    
+    async def send_tool_start_event(self, content_name):
+        """Send a tool content start event to the Bedrock stream."""
+        content_start_event = self.TOOL_CONTENT_START_EVENT % (self.prompt_name, content_name, self.toolUseId)
+        debug_print(f"Sending tool start event: {content_start_event}")  
+        await self.send_raw_event(content_start_event)
+
+    async def send_tool_result_event(self, content_name, tool_result):
+        """Send a tool content event to the Bedrock stream."""
+        # Use the actual tool result from processToolUse
+        tool_result_event = self.tool_result_event(content_name=content_name, content=tool_result, role="TOOL")
+        debug_print(f"Sending tool result event: {tool_result_event}")
+        await self.send_raw_event(tool_result_event)
+    
+    async def send_tool_content_end_event(self, content_name):
+        """Send a tool content end event to the Bedrock stream."""
+        tool_content_end_event = self.CONTENT_END_EVENT % (self.prompt_name, content_name)
+        debug_print(f"Sending tool content event: {tool_content_end_event}")
+        await self.send_raw_event(tool_content_end_event)
+    
+    async def send_prompt_end_event(self):
+        """Close the stream and clean up resources."""
+        if not self.is_active:
+            debug_print("Stream is not active")
+            return
+        
+        prompt_end_event = self.PROMPT_END_EVENT % (self.prompt_name)
+        await self.send_raw_event(prompt_end_event)
+        debug_print("Prompt ended")
+        
+    async def send_session_end_event(self):
+        """Send a session end event to the Bedrock stream."""
+        if not self.is_active:
+            debug_print("Stream is not active")
+            return
+
+        await self.send_raw_event(self.SESSION_END_EVENT)
+        self.is_active = False
+        debug_print("Session ended")
+    
+    async def _process_responses(self):
+        """Process incoming responses from Bedrock."""
+        try:            
+            while self.is_active:
+                try:
+                    output = await self.stream_response.await_output()
+                    result = await output[1].receive()
+                    if result.value and result.value.bytes_:
+                        try:
+                            response_data = result.value.bytes_.decode('utf-8')
+                            json_data = json.loads(response_data)
+                            
+                            # Handle different response types
+                            if 'event' in json_data:
+                                if 'contentStart' in json_data['event']:
+                                    debug_print("Content start detected")
+                                    content_start = json_data['event']['contentStart']
+                                    # set role
+                                    self.role = content_start['role']
+                                    # Check for speculative content
+                                    if 'additionalModelFields' in content_start:
+                                        try:
+                                            additional_fields = json.loads(content_start['additionalModelFields'])
+                                            if additional_fields.get('generationStage') == 'SPECULATIVE':
+                                                debug_print("Speculative content detected")
+                                                self.display_assistant_text = True
+                                            else:
+                                                self.display_assistant_text = False
+                                        except json.JSONDecodeError:
+                                            debug_print("Error parsing additionalModelFields")
+                                elif 'textOutput' in json_data['event']:
+                                    text_content = json_data['event']['textOutput']['content']
+                                    role = json_data['event']['textOutput']['role']
+                                    # Check if there is a barge-in
+                                    if '{ "interrupted" : true }' in text_content:
+                                        debug_print("Barge-in detected. Stopping audio output.")
+                                        self.barge_in = True
+
+                                    if (self.role == "ASSISTANT" and self.display_assistant_text):
+                                        print(f"Assistant: {text_content}")
+                                    elif (self.role == "USER"):
+                                        print(f"User: {text_content}")
+
+                                elif 'audioOutput' in json_data['event']:
+                                    audio_content = json_data['event']['audioOutput']['content']
+                                    audio_bytes = base64.b64decode(audio_content)
+                                    await self.audio_output_queue.put(audio_bytes)
+                                elif 'toolUse' in json_data['event']:
+                                    self.toolUseContent = json_data['event']['toolUse']
+                                    self.toolName = json_data['event']['toolUse']['toolName']
+                                    self.toolUseId = json_data['event']['toolUse']['toolUseId']
+                                    debug_print(f"Tool use detected: {self.toolName}, ID: {self.toolUseId}")
+                                elif 'contentEnd' in json_data['event'] and json_data['event'].get('contentEnd', {}).get('type') == 'TOOL':
+                                    debug_print("Processing tool use and sending result")
+                                    toolResult = await self.processToolUse(self.toolName, self.toolUseContent)
+                                    toolContent = str(uuid.uuid4())
+                                    await self.send_tool_start_event(toolContent)
+                                    await self.send_tool_result_event(toolContent, toolResult)
+                                    await self.send_tool_content_end_event(toolContent)
+                                
+                                elif 'completionEnd' in json_data['event']:
+                                    # Handle end of conversation, no more response will be generated
+                                    print("End of response sequence")
+                            
+                            # Put the response in the output queue for other components
+                            await self.output_queue.put(json_data)
+                        except json.JSONDecodeError:
+                            await self.output_queue.put({"raw_data": response_data})
+                except StopAsyncIteration:
+                    # Stream has ended
+                    break
+                except Exception as e:
+                   # Handle ValidationException properly
+                    if "ValidationException" in str(e):
+                        error_message = str(e)
+                        print(f"Validation error: {error_message}")
+                    else:
+                        print(f"Error receiving response: {e}")
+                    break
+                    
+        except Exception as e:
+            print(f"Response processing error: {e}")
+        finally:
+            self.is_active = False
+
+    async def processToolUse(self, toolName, toolUseContent):
+        """Return the tool result"""
+        tool = toolName.lower()
+        debug_print(f"Tool Use Content: {toolUseContent}")
+        
+        if tool == "getdateandtimetool":
+            # Get current date in PST timezone
+            pst_timezone = pytz.timezone("America/Los_Angeles")
+            pst_date = datetime.datetime.now(pst_timezone)
+            
+            return {
+                "formattedTime": pst_date.strftime("%I:%M %p"),
+                "date": pst_date.strftime("%Y-%m-%d"),
+                "year": pst_date.year,
+                "month": pst_date.month,
+                "day": pst_date.day,
+                "dayOfWeek": pst_date.strftime("%A").upper(),
+                "timezone": "PST"
+            }
+        
+        elif tool == "trackordertool":
+
+            # Extract order ID from toolUseContent
+            content = toolUseContent.get("content", {})
+            content_data = json.loads(content)
+            order_id = content_data.get("orderId", "")
+            request_notifications = toolUseContent.get("requestNotifications", False)
+            
+            # Convert order_id to string if it's an integer
+            if isinstance(order_id, int):
+                order_id = str(order_id)
+            # Validate order ID format
+            if not order_id or not isinstance(order_id, str):
+                return {
+                    "error": "Invalid order ID format",
+                    "orderStatus": "",
+                    "estimatedDelivery": "",
+                    "lastUpdate": ""
+                }
+            
+            # Create deterministic randomness based on order ID
+            # This ensures the same order ID always returns the same status
+            seed = int(hashlib.md5(order_id.encode(), usedforsecurity=False).hexdigest(), 16) % 10000
+            random.seed(seed)
+            
+            # Possible statuses with appropriate weights
+            statuses = [
+                "Order received", 
+                "Processing", 
+                "Preparing for shipment",
+                "Shipped",
+                "In transit", 
+                "Out for delivery",
+                "Delivered",
+                "Delayed"
+            ]
+            
+            weights = [10, 15, 15, 20, 20, 10, 5, 3]
+            
+            # Select a status based on the weights
+            status = random.choices(statuses, weights=weights, k=1)[0]
+            
+            # Generate a realistic estimated delivery date
+            today = datetime.datetime.now()
+            # Handle estimated delivery date based on status
+            if status == "Delivered":
+                # For delivered items, delivery date is in the past
+                delivery_days = -random.randint(0, 3)
+                estimated_delivery = (today + datetime.timedelta(days=delivery_days)).strftime("%Y-%m-%d")
+            elif status == "Out for delivery":
+                # For out for delivery, delivery is today
+                estimated_delivery = today.strftime("%Y-%m-%d")
+            else:
+                # For other statuses, delivery is in the future
+                delivery_days = random.randint(1, 10)
+                estimated_delivery = (today + datetime.timedelta(days=delivery_days)).strftime("%Y-%m-%d")
+
+            # Handle notification request if enabled
+            notification_message = ""
+            if request_notifications and status != "Delivered":
+                notification_message = f"You will receive notifications for order {order_id}"
+
+            # Return comprehensive tracking information
+            tracking_info = {
+                "orderStatus": status,
+                "orderNumber": order_id,
+                "notificationStatus": notification_message
+            }
+
+            # Add appropriate fields based on status
+            if status == "Delivered":
+                tracking_info["deliveredOn"] = estimated_delivery
+            elif status == "Out for delivery":
+                tracking_info["expectedDelivery"] = "Today"
+            else:
+                tracking_info["estimatedDelivery"] = estimated_delivery
+
+            # Add location information based on status
+            if status == "In transit":
+                tracking_info["currentLocation"] = "Distribution Center"
+            elif status == "Delivered":
+                tracking_info["deliveryLocation"] = "Front Door"
+                
+            # Add additional info for delayed status
+            if status == "Delayed":
+                tracking_info["additionalInfo"] = "Weather delays possible"
+                
+            return tracking_info
+    
+    async def close(self):
+        """Close the stream properly."""
+        if not self.is_active:
+            return
+       
+        self.is_active = False
+        if self.response_task and not self.response_task.done():
+            self.response_task.cancel()
+
+        await self.send_audio_content_end_event()
+        await self.send_prompt_end_event()
+        await self.send_session_end_event()
+
+        if self.stream_response:
+            await self.stream_response.input_stream.close()
+
+class AudioStreamer:
+    """Handles continuous microphone input and audio output using separate streams."""
+    
+    def __init__(self, stream_manager):
+        self.stream_manager = stream_manager
+        self.is_streaming = False
+        self.loop = asyncio.get_event_loop()
+
+        # Initialize PyAudio
+        debug_print("AudioStreamer Initializing PyAudio...")
+        self.p = time_it("AudioStreamerInitPyAudio", pyaudio.PyAudio)
+        debug_print("AudioStreamer PyAudio initialized")
+
+        # Initialize separate streams for input and output
+        # Input stream with callback for microphone
+        debug_print("Opening input audio stream...")
+        self.input_stream = time_it("AudioStreamerOpenAudio", lambda  : self.p.open(
+            format=FORMAT,
+            channels=CHANNELS,
+            rate=INPUT_SAMPLE_RATE,
+            input=True,
+            frames_per_buffer=CHUNK_SIZE,
+            stream_callback=self.input_callback
+        ))
+        debug_print("input audio stream opened")
+
+        # Output stream for direct writing (no callback)
+        debug_print("Opening output audio stream...")
+        self.output_stream = time_it("AudioStreamerOpenAudio", lambda  : self.p.open(
+            format=FORMAT,
+            channels=CHANNELS,
+            rate=OUTPUT_SAMPLE_RATE,
+            output=True,
+            frames_per_buffer=CHUNK_SIZE
+        ))
+
+        debug_print("output audio stream opened")
+
+    def input_callback(self, in_data, frame_count, time_info, status):
+        """Callback function that schedules audio processing in the asyncio event loop"""
+        if self.is_streaming and in_data:
+            # Schedule the task in the event loop
+            asyncio.run_coroutine_threadsafe(
+                self.process_input_audio(in_data), 
+                self.loop
+            )
+        return (None, pyaudio.paContinue)
+
+    async def process_input_audio(self, audio_data):
+        """Process a single audio chunk directly"""
+        try:
+            # Send audio to Bedrock immediately
+            self.stream_manager.add_audio_chunk(audio_data)
+        except Exception as e:
+            if self.is_streaming:
+                print(f"Error processing input audio: {e}")
+    
+    async def play_output_audio(self):
+        """Play audio responses from Nova Sonic"""
+        while self.is_streaming:
+            try:
+                # Check for barge-in flag
+                if self.stream_manager.barge_in:
+                    # Clear the audio queue
+                    while not self.stream_manager.audio_output_queue.empty():
+                        try:
+                            self.stream_manager.audio_output_queue.get_nowait()
+                        except asyncio.QueueEmpty:
+                            break
+                    self.stream_manager.barge_in = False
+                    # Small sleep after clearing
+                    await asyncio.sleep(0.05)
+                    continue
+                
+                # Get audio data from the stream manager's queue
+                audio_data = await asyncio.wait_for(
+                    self.stream_manager.audio_output_queue.get(),
+                    timeout=0.1
+                )
+                
+                if audio_data and self.is_streaming:
+                    # Write directly to the output stream in smaller chunks
+                    chunk_size = CHUNK_SIZE  # Use the same chunk size as the stream
+                    
+                    # Write the audio data in chunks to avoid blocking too long
+                    for i in range(0, len(audio_data), chunk_size):
+                        if not self.is_streaming:
+                            break
+                        
+                        end = min(i + chunk_size, len(audio_data))
+                        chunk = audio_data[i:end]
+                        
+                        # Create a new function that captures the chunk by value
+                        def write_chunk(data):
+                            return self.output_stream.write(data)
+                        
+                        # Pass the chunk to the function
+                        await asyncio.get_event_loop().run_in_executor(None, write_chunk, chunk)
+                        
+                        # Brief yield to allow other tasks to run
+                        await asyncio.sleep(0.001)
+                    
+            except asyncio.TimeoutError:
+                # No data available within timeout, just continue
+                continue
+            except Exception as e:
+                if self.is_streaming:
+                    print(f"Error playing output audio: {str(e)}")
+                    import traceback
+                    traceback.print_exc()
+                await asyncio.sleep(0.05)
+    
+    async def start_streaming(self):
+        """Start streaming audio."""
+        if self.is_streaming:
+            return
+        
+        print("Starting audio streaming. Speak into your microphone...")
+        print("Press Enter to stop streaming...")
+        
+        # Send audio content start event
+        await time_it_async("send_audio_content_start_event", lambda : self.stream_manager.send_audio_content_start_event())
+        
+        self.is_streaming = True
+        
+        # Start the input stream if not already started
+        if not self.input_stream.is_active():
+            self.input_stream.start_stream()
+        
+        # Start processing tasks
+        #self.input_task = asyncio.create_task(self.process_input_audio())
+        self.output_task = asyncio.create_task(self.play_output_audio())
+        
+        # Wait for user to press Enter to stop
+        await asyncio.get_event_loop().run_in_executor(None, input)
+        
+        # Once input() returns, stop streaming
+        await self.stop_streaming()
+    
+    async def stop_streaming(self):
+        """Stop streaming audio."""
+        if not self.is_streaming:
+            return
+            
+        self.is_streaming = False
+
+        # Cancel the tasks
+        tasks = []
+        if hasattr(self, 'input_task') and not self.input_task.done():
+            tasks.append(self.input_task)
+        if hasattr(self, 'output_task') and not self.output_task.done():
+            tasks.append(self.output_task)
+        for task in tasks:
+            task.cancel()
+        if tasks:
+            await asyncio.gather(*tasks, return_exceptions=True)
+        # Stop and close the streams
+        if self.input_stream:
+            if self.input_stream.is_active():
+                self.input_stream.stop_stream()
+            self.input_stream.close()
+        if self.output_stream:
+            if self.output_stream.is_active():
+                self.output_stream.stop_stream()
+            self.output_stream.close()
+        if self.p:
+            self.p.terminate()
+        
+        await self.stream_manager.close() 
+
+
+async def main(debug=False):
+    """Main function to run the application."""
+    global DEBUG
+    DEBUG = debug
+
+    # Create stream manager
+    stream_manager = BedrockStreamManager(model_id='amazon.nova-sonic-v1:0', region='us-east-1')
+
+    # Create audio streamer
+    audio_streamer = AudioStreamer(stream_manager)
+
+    # Initialize the stream
+    await time_it_async("initialize_stream", stream_manager.initialize_stream)
+
+    try:
+        # This will run until the user presses Enter
+        await audio_streamer.start_streaming()
+        
+    except KeyboardInterrupt:
+        print("Interrupted by user")
+    finally:
+        # Clean up
+        await audio_streamer.stop_streaming()
+        
+
+if __name__ == "__main__":
+    import argparse
+    
+    parser = argparse.ArgumentParser(description='Nova Sonic Python Streaming')
+    parser.add_argument('--debug', action='store_true', help='Enable debug mode')
+    args = parser.parse_args()
+    # Set your AWS credentials here or use environment variables
+    # os.environ['AWS_ACCESS_KEY_ID'] = "AWS_ACCESS_KEY_ID"
+    # os.environ['AWS_SECRET_ACCESS_KEY'] = "AWS_SECRET_ACCESS_KEY"
+    # os.environ['AWS_DEFAULT_REGION'] = "us-east-1"
+
+    # Run the main function
+    try:
+        asyncio.run(main(debug=args.debug))
+    except Exception as e:
+        print(f"Application error: {e}")
+        if args.debug:
+            import traceback
+            traceback.print_exc()