Merge pull request #20 from mkmeral/bidi-event-types

Event Types

Author: mehtarac

pyproject.toml

@@ -63,15 +63,15 @@ bidirectional-streaming-nova = [
 ]
 bidirectional-streaming-openai = [
     "pyaudio>=0.2.13",
-    "websockets>=12.0,<14.0",
+    "websockets>=14.0,<16.0",
 ]
 bidirectional-streaming = [
     "pyaudio>=0.2.13",
     "rx>=3.2.0",
     "smithy-aws-core>=0.0.1",
     "pytz",
     "aws_sdk_bedrock_runtime",
-    "websockets>=12.0,<14.0",
+    "websockets>=14.0,<16.0",
 ]
 otel = ["opentelemetry-exporter-otlp-proto-http>=1.30.0,<2.0.0"]
 docs = [

src/strands/experimental/bidirectional_streaming/__init__.py

@@ -15,16 +15,29 @@
 from .models.openai import BidiOpenAIRealtimeModel
 
 # Event types - For type hints and event handling
-from .types.bidirectional_streaming import (
-    AudioInputEvent,
-    AudioOutputEvent,
-    BidirectionalStreamEvent,
-    ImageInputEvent,
-    InterruptionDetectedEvent,
-    TextInputEvent,
-    TextOutputEvent,
-    UsageMetricsEvent,
-    VoiceActivityEvent,
+from .types.events import (
+    BidiAudioInputEvent,
+    BidiAudioStreamEvent,
+    BidiConnectionCloseEvent,
+    BidiConnectionStartEvent,
+    BidiErrorEvent,
+    BidiImageInputEvent,
+    BidiInputEvent,
+    BidiInterruptionEvent,
+    ModalityUsage,
+    BidiUsageEvent,
+    BidiOutputEvent,
+    BidiResponseCompleteEvent,
+    BidiResponseStartEvent,
+    BidiTextInputEvent,
+    BidiTranscriptStreamEvent,
+)
+
+# Re-export standard agent events for tool handling
+from ...types._events import (
+    ToolResultEvent,
+    ToolStreamEvent,
+    ToolUseStreamEvent,
 )
 
 __all__ = [
@@ -37,16 +50,30 @@
     "BidiNovaSonicModel",
     "BidiOpenAIRealtimeModel",
 
-    # Event types
-    "AudioInputEvent",
-    "AudioOutputEvent",
-    "ImageInputEvent",
-    "TextInputEvent",
-    "TextOutputEvent",
-    "InterruptionDetectedEvent",
-    "BidirectionalStreamEvent",
-    "VoiceActivityEvent",
-    "UsageMetricsEvent",
+    # 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/bidirectional_streaming/agent/agent.py

@@ -27,19 +27,22 @@
 from ....types.content import Message, Messages
 from ....types.tools import ToolResult, ToolUse, AgentTool
 
-from ..event_loop.bidirectional_event_loop import BidirectionalConnection
+from ..event_loop.bidirectional_event_loop import (
+    BidirectionalConnection,
+    start_bidirectional_connection,
+    stop_bidirectional_connection,
+)
 from ..models.bidirectional_model import BidiModel
 from ..models.novasonic import BidiNovaSonicModel
-from ..types.bidirectional_streaming import AudioInputEvent, BidirectionalStreamEvent, ImageInputEvent
+from ..types.agent import BidiAgentInput
+from ..types.events import BidiAudioInputEvent, BidiImageInputEvent, BidiTextInputEvent, BidiInputEvent, BidiOutputEvent
 from ..types import BidiIO
 from ....experimental.tools import ToolProvider
 
 logger = logging.getLogger(__name__)
 
 _DEFAULT_AGENT_NAME = "Strands Agents"
 _DEFAULT_AGENT_ID = "default"
-# Type alias for cleaner send() method signature
-BidirectionalInput = str | AudioInputEvent | ImageInputEvent
 
 
 class BidiAgent:
@@ -250,47 +253,81 @@ async def start(self) -> None:
             raise ValueError("Conversation already active. Call end() first.")
 
         logger.debug("Conversation start - initializing connection")
+        self._agent_loop = await start_bidirectional_connection(self)
 
-        # Create model session and event loop directly
-        await self.model.start(
-            system_prompt=self.system_prompt, tools=self.tool_registry.get_all_tool_specs(), messages=self.messages
-        )
-
-        self._agent_loop = BidirectionalConnection(model=self.model, agent=self)
-        await self._agent_loop.start()
-
-        logger.debug("Conversation ready")
-
-    async def send(self, input_data: BidirectionalInput) -> None:
-        """Send input to the model (text or audio).
-
-        Unified method for sending both text and audio input to the model during
-        an active conversation connection. User input is automatically added to
-        conversation history for complete message tracking.
-
+    async def send(self, input_data: BidiAgentInput) -> None:
+        """Send input to the model (text, audio, image, or event dict).
+        
+        Unified method for sending text, audio, and image input to the model during
+        an active conversation session. Accepts TypedEvent instances or plain dicts
+        (e.g., from WebSocket clients) which are automatically reconstructed.
+        
         Args:
-            input_data: String for text, AudioInputEvent for audio, or ImageInputEvent for images.
-
+            input_data: Can be:
+                - str: Text message from user
+                - BidiAudioInputEvent: Audio data with format/sample rate
+                - BidiImageInputEvent: Image data with MIME type
+                - dict: Event dictionary (will be reconstructed to TypedEvent)
+            
         Raises:
-            ValueError: If no active connection or invalid input type.
+            ValueError: If no active session or invalid input type.
+            
+        Example:
+            await agent.send("Hello")
+            await agent.send(BidiAudioInputEvent(audio="base64...", format="pcm", ...))
+            await agent.send({"type": "bidirectional_text_input", "text": "Hello", "role": "user"})
         """
         self._validate_active_connection()
 
+        # Handle string input
         if isinstance(input_data, str):
             # Add user text message to history
             user_message: Message = {"role": "user", "content": [{"text": input_data}]}
 
             self.messages.append(user_message)
 
             logger.debug("Text sent: %d characters", len(input_data))
-            # Create TextInputEvent for send()
-            text_event = {"text": input_data, "role": "user"}
+            # Create BidiTextInputEvent for send()
+            text_event = BidiTextInputEvent(text=input_data, role="user")
             await self._agent_loop.model.send(text_event)
-        else:
-            # For audio, image, or any other input - let model handle it
+            return
+        
+        # Handle BidiInputEvent instances
+        # Check this before dict since TypedEvent inherits from dict
+        if isinstance(input_data, BidiInputEvent):
             await self._agent_loop.model.send(input_data)
+            return
+        
+        # Handle plain dict - reconstruct TypedEvent for WebSocket integration
+        if isinstance(input_data, dict) and "type" in input_data:
+            event_type = input_data["type"]
+            if event_type == "bidi_text_input":
+                input_event = BidiTextInputEvent(text=input_data["text"], role=input_data["role"])
+            elif event_type == "bidi_audio_input":
+                input_event = BidiAudioInputEvent(
+                    audio=input_data["audio"],
+                    format=input_data["format"],
+                    sample_rate=input_data["sample_rate"],
+                    channels=input_data["channels"]
+                )
+            elif event_type == "bidi_image_input":
+                input_event = BidiImageInputEvent(
+                    image=input_data["image"],
+                    mime_type=input_data["mime_type"]
+                )
+            else:
+                raise ValueError(f"Unknown event type: {event_type}")
+            
+            # Send the reconstructed TypedEvent
+            await self._agent_loop.model.send(input_event)
+            return
+        
+        # If we get here, input type is invalid
+        raise ValueError(
+            f"Input must be a string, BidiInputEvent (BidiTextInputEvent/BidiAudioInputEvent/BidiImageInputEvent), or event dict with 'type' field, got: {type(input_data)}"
+        )
 
-    async def receive(self) -> AsyncIterable[BidirectionalStreamEvent]:
+    async def receive(self) -> AsyncIterable[BidiOutputEvent]:
         """Receive events from the model including audio, text, and tool calls.
 
         Yields model output events processed by background tasks including audio output,
@@ -301,9 +338,11 @@ async def receive(self) -> AsyncIterable[BidirectionalStreamEvent]:
         """
         while self.active:
             try:
-                event = await self._output_queue.get()
+                # Use a timeout to periodically check if we should stop
+                event = await asyncio.wait_for(self._output_queue.get(), timeout=0.5)
                 yield event
             except asyncio.TimeoutError:
+                # Timeout allows us to check self.active periodically
                 continue
 
     async def stop(self) -> None:
@@ -313,7 +352,7 @@ async def stop(self) -> None:
         closes the connection to the model provider.
         """
         if self._agent_loop:
-            await self._agent_loop.stop()
+            await stop_bidirectional_connection(self._agent_loop)
             self._agent_loop = None
 
     async def __aenter__(self) -> "BidiAgent":

src/strands/experimental/bidirectional_streaming/event_loop/bidirectional_event_loop.py

@@ -223,7 +223,9 @@ async def _handle_interruption(session: BidirectionalConnection) -> None:
         try:
             while True:
                 event = session.agent._output_queue.get_nowait()
-                if event.get("audioOutput"):
+                # Check for audio events
+                event_type = event.get("type", "")
+                if event_type == "bidi_audio_stream":
                     audio_cleared += 1
                 else:
                     # Keep non-audio events
@@ -267,35 +269,38 @@ async def _process_model_events(session: BidirectionalConnection) -> None:
 
             strands_event = provider_event
 
-            # Handle interruption detection (provider converts raw patterns to interruptionDetected)
-            if strands_event.get("interruptionDetected"):
+            # Get event type
+            event_type = strands_event.get("type", "")
+            
+            # Handle interruption detection
+            if event_type == "bidi_interruption":
                 logger.debug("Interruption forwarded")
                 await _handle_interruption(session)
                 # Forward interruption event to agent for application-level handling
                 await session.agent._output_queue.put(strands_event)
                 continue
 
             # Queue tool requests for concurrent execution
-            if strands_event.get("toolUse"):
-                tool_name = strands_event["toolUse"].get("name")
-                logger.debug("Tool usage detected: %s", tool_name)
-                await session.tool_queue.put(strands_event["toolUse"])
+            # Check for ToolUseStreamEvent (standard agent event)
+            if event_type == "tool_use_stream":
+                tool_use = strands_event.get("current_tool_use")
+                if tool_use:
+                    tool_name = tool_use.get("name")
+                    logger.debug("Tool usage detected: %s", tool_name)
+                    await session.tool_queue.put(tool_use)
+                # Forward ToolUseStreamEvent to output queue for client visibility
+                await session.agent._output_queue.put(strands_event)
                 continue
 
-            # Send output events to Agent for receive() method
-            if strands_event.get("audioOutput") or strands_event.get("textOutput"):
-                await session.agent._output_queue.put(strands_event)
+            # Send all output events to Agent for receive() method
+            await session.agent._output_queue.put(strands_event)
 
-            # Update Agent conversation history using existing patterns
-            if strands_event.get("messageStop"):
-                logger.debug("Message added to history")
-                session.agent.messages.append(strands_event["messageStop"]["message"])
-            
-            # Handle user audio transcripts - add to message history
-            if strands_event.get("textOutput") and strands_event["textOutput"].get("role") == "user":
-                user_transcript = strands_event["textOutput"]["text"]
-                if user_transcript.strip():  # Only add non-empty transcripts
-                    user_message = {"role": "user", "content": user_transcript}
+            # Update Agent conversation history for user transcripts
+            if event_type == "bidi_transcript_stream":
+                role = strands_event.get("role")
+                text = strands_event.get("text", "")
+                if role == "user" and text.strip():
+                    user_message = {"role": "user", "content": text}
                     session.agent.messages.append(user_message)
                     logger.debug("User transcript added to history")
 
@@ -434,14 +439,19 @@ async def _execute_tool_with_strands(session: BidirectionalConnection, tool_use:
                 tool_result = tool_event.tool_result
                 tool_use_id = tool_result.get("toolUseId")
 
-                # Send result through send() method
-                await session.model.send(tool_result)
-                logger.debug("Tool result sent: %s", tool_use_id)
+                # Send ToolResultEvent through send() method to model
+                await session.model.send(tool_event)
+                logger.debug("Tool result sent to model: %s", tool_use_id)
+                
+                # Also forward ToolResultEvent to output queue for client visibility
+                await session.agent._output_queue.put(tool_event)
+                logger.debug("Tool result sent to client: %s", tool_use_id)
 
             # Handle streaming events if needed later
             elif isinstance(tool_event, ToolStreamEvent):
                 logger.debug("Tool stream event: %s", tool_event)
-                pass
+                # Forward tool stream events to output queue
+                await session.agent._output_queue.put(tool_event)
 
         # Add tool result message to conversation history
         if tool_results:
@@ -464,14 +474,14 @@ async def _execute_tool_with_strands(session: BidirectionalConnection, tool_use:
     except Exception as e:
         logger.error("Tool execution error: %s - %s", tool_name, str(e))
 
-        # Send error result 
+        # Send error result wrapped in ToolResultEvent
         error_result: ToolResult = {
             "toolUseId": tool_id,
             "status": "error",
             "content": [{"text": f"Error: {str(e)}"}]
         }
         try:
-            await session.model.send(error_result)
+            await session.model.send(ToolResultEvent(error_result))
             logger.debug("Error result sent: %s", tool_id)
         except Exception as send_error:
             logger.error("Failed to send error result: %s - %s", tool_id, str(send_error))