Events

Experimental Feature

This feature is experimental and may change in future versions. Use with caution in production environments.

You process audio, text, and tool activity as it happens by consuming bidirectional streaming events. Standard streaming uses async iterators or callbacks in a one-shot request-response pattern; bidirectional streaming uses send() and receive() for explicit control over a persistent, two-way conversation.

Event Model

Bidirectional streaming uses a different event model than standard streaming:

Standard Streaming:

• Uses stream_async() or callback handlers
• Request-response pattern (one invocation per call)
• Events flow in one direction (model → application)

Bidirectional Streaming:

• Uses send() and receive() methods
• Persistent connection (multiple turns per connection)
• Events flow in both directions (application ↔ model)
• Supports real-time audio and interruptions

import asyncio
from strands.experimental.bidi.agent import BidiAgent
from strands.experimental.bidi.models import BedrockNovaSonicModel

async def main():
    model = BedrockNovaSonicModel()

    async with BidiAgent(model=model) as agent:
        # Send input to model
        await agent.send("What is 2+2?")

        # Receive events from model
        async for event in agent.receive():
            print(f"Event: {event['type']}")

asyncio.run(main())

Input Types

Send text, streaming audio, or images with agent.send(). It accepts a string, a TextBlock, AudioDelta, or ImageBlock, or a dictionary containing exactly one text, audio_delta, or image key.

Text blocks contain complete text input, and image blocks contain complete images. Audio deltas add samples to the live input stream without explicitly ending the user’s turn.

Text

Send text input to the model.

from strands.types.content import TextBlock

await agent.send(TextBlock("What is the weather?"))

# Strings and dictionaries are also accepted:
await agent.send("What is the weather?")
await agent.send({"text": "What is the weather?"})

Audio

Send each chunk of audio samples with AudioDelta. The model configuration determines the sample rate and channel count for real-time PCM audio.

from pathlib import Path

from strands.experimental.bidi.types import AudioDelta

audio_bytes = Path("audio-chunk.pcm").read_bytes()

await agent.send(AudioDelta(format="pcm", source={"bytes": audio_bytes}))

# Or use a dictionary:
await agent.send({
    "audio_delta": {
        "format": "pcm",
        "source": {"bytes": audio_bytes},
    }
})

Image

Send image bytes using an image content block.

from strands.types.media import ImageBlock

with open("image.jpg", "rb") as f:
    image_bytes = f.read()

await agent.send(ImageBlock(format="jpeg", source={"bytes": image_bytes}))

# Or use a dictionary:
await agent.send({
    "image": {
        "format": "jpeg",
        "source": {"bytes": image_bytes},
    }
})

Output Event Types

Events received from the model via agent.receive().

Connection Lifecycle Events

Events that track the connection state throughout the conversation.

BidiConnectionStartEvent

Emitted when the streaming connection is established and ready for interaction.

{
    "type": "bidi_connection_start",
    "connection_id": "conn_abc123",
    "model": "amazon.nova-2-sonic-v1:0"
}

Properties:

• connection_id: Unique identifier for this streaming connection
• model: Model identifier (e.g., “amazon.nova-2-sonic-v1:0”, “gemini-2.0-flash-live”)

BidiConnectionRestartEvent

Emitted when the agent restarts the model connection, on either reconnect path. The agent preserves the conversation history and resumes automatically. A scheduled restart fires proactively when the reconnect timer reaches the provider’s limit; a timeout restart fires reactively after the model reports a timeout.

{
    "type": "bidi_connection_restart",
    "reason": "scheduled",
    "timeout_error": None,
    "turn_interrupted": False
}

Properties:

• reason: What triggered the restart
  • "scheduled": The reconnect timer fired ahead of the provider’s limit (the normal path)
  • "timeout": The connection timed out and the model reported it
• timeout_error: The timeout error on the reactive path; None when the reason is "scheduled"
• turn_interrupted: True when the restart cut an in-progress or owed turn. The provider replays history as context, so that turn is not answered on its own: re-prompt or notify the user when this is set.

Usage:

async for event in agent.receive():
    if event["type"] == "bidi_connection_restart":
        print(f"Connection restarting (reason={event['reason']})")
        if event["turn_interrupted"]:
            # This turn was not answered; re-prompt or notify the user.
            pass
        # Connection resumes automatically with full history.

See Connection Lifecycle for more on reconnect timing.

BidiConnectionWarningEvent

Emitted by the proactive reconnect timer shortly before a scheduled restart. Informational only: use it to surface a “reconnecting shortly” hint in a UI.

{
    "type": "bidi_connection_warning",
    "time_left_s": 8.0
}

Properties:

• time_left_s: Approximate seconds until the scheduled reconnect

Usage:

async for event in agent.receive():
    if event["type"] == "bidi_connection_warning":
        print(f"Reconnecting in ~{event['time_left_s']:.0f}s")

BidiConnectionCloseEvent

Emitted when the streaming connection is closed.

{
    "type": "bidi_connection_close",
    "connection_id": "conn_abc123",
    "reason": "user_request"
}

Properties:

• connection_id: Unique identifier for this streaming connection
• reason: Why the connection closed
  • "client_disconnect": Client disconnected
  • "timeout": Connection timed out
  • "error": Error occurred
  • "complete": Conversation completed normally
  • "user_request": User requested closure (via the SDK’s experimental stop tool or any tool that sets request_state["stop_event_loop"])

Response Lifecycle Events

Events that track individual model responses within the conversation.

BidiResponseStartEvent

Emitted when the model begins generating a response.

{
    "type": "bidi_response_start",
    "response_id": "resp_xyz789"
}

Properties:

• response_id: Unique identifier for this response (matches BidiResponseCompleteEvent)

BidiResponseCompleteEvent

Emitted when the model finishes generating a response.

{
    "type": "bidi_response_complete",
    "response_id": "resp_xyz789",
    "stop_reason": "complete"
}

Properties:

• response_id: Unique identifier for this response
• stop_reason: Why the response ended
  • "complete": Model completed its response
  • "interrupted": User interrupted the response
  • "tool_use": Model is requesting tool execution
  • "error": Error occurred during generation

Audio Events

Events for streaming audio input and output.

BidiAudioStreamEvent

Emitted when the model generates audio output. Audio is base64-encoded for JSON compatibility.

{
    "type": "bidi_audio_stream",
    "audio": "base64_encoded_audio_data...",
    "format": "pcm",
    "sample_rate": 16000,
    "channels": 1
}

Properties:

• audio: Base64-encoded audio string
• format: Audio encoding format ("pcm", "wav", "opus", "mp3")
• sample_rate: Sample rate in Hz (16000, 24000, 48000)
• channels: Number of audio channels (1 = mono, 2 = stereo)

Usage:

import base64

async for event in agent.receive():
    if event["type"] == "bidi_audio_stream":
        # Decode and play audio
        audio_bytes = base64.b64decode(event["audio"])
        play_audio(audio_bytes, sample_rate=event["sample_rate"])

Transcript Events

Events for speech-to-text transcription of both user and assistant speech.

BidiTranscriptStreamEvent

Emitted for each incremental transcript update.

{
    "type": "bidi_transcript_stream",
    "delta": "Hello",
    "role": "assistant"
}

Properties:

• delta: The incremental transcript text
• role: Who is speaking ("user" or "assistant")

BidiTranscriptCompleteEvent

Emitted once with the complete transcript for a user or assistant turn.

{
    "type": "bidi_transcript_complete",
    "transcript": "Hello world",
    "role": "assistant"
}

Properties:

• transcript: The complete transcript text
• role: Who spoke ("user" or "assistant")

Usage:

async for event in agent.receive():
    if event["type"] == "bidi_transcript_stream":
        print(event["delta"], end="", flush=True)
    elif event["type"] == "bidi_transcript_complete":
        print(f"\n{event['role']}: {event['transcript']}")

Interruption Events

Events for handling user interruptions during model responses.

BidiInterruptionEvent

Emitted when the model’s response is interrupted, typically by user speech detected via voice activity detection.

{
    "type": "bidi_interruption",
    "reason": "user_speech"
}

Properties:

• reason: Why the interruption occurred
  • "user_speech": User started speaking (most common)
  • "error": Error caused interruption

Usage:

async for event in agent.receive():
    if event["type"] == "bidi_interruption":
        print(f"Interrupted by {event['reason']}")
        # Audio output automatically cleared
        # Model ready for new input

BidiInterruptionEvent vs Human-in-the-Loop Interrupts

BidiInterruptionEvent is different from human-in-the-loop (HIL) interrupts. BidiInterruptionEvent is emitted when the model detects user speech during audio conversations and automatically stops generating the current response. HIL interrupts pause agent execution to request human approval or input before continuing, typically used for tool execution approval. BidiInterruptionEvent is automatic and audio-specific, while HIL interrupts are programmatic and require explicit handling.

Tool Events

Events for tool execution during conversations. Bidirectional streaming reuses the standard ToolUseStreamEvent from Strands.

ToolUseStreamEvent

Emitted when the model requests tool execution. See Tools Overview for details.

{
    "type": "tool_use_stream",
    "current_tool_use": {
        "toolUseId": "tool_123",
        "name": "notebook",
        "input": {"expression": "2+2"}
    }
}

Properties:

• current_tool_use: Information about the tool being used
  • toolUseId: Unique ID for this tool use
  • name: Name of the tool
  • input: Tool input parameters

Tools execute automatically in the background and results are sent back to the model without blocking the conversation.

Usage Events

Events for tracking token consumption across different modalities.

BidiUsageEvent

Emitted periodically to report token usage with modality breakdown.

{
    "type": "bidi_usage",
    "inputTokens": 150,
    "outputTokens": 75,
    "totalTokens": 225,
    "modality_details": [
        {"modality": "text", "input_tokens": 100, "output_tokens": 50},
        {"modality": "audio", "input_tokens": 50, "output_tokens": 25}
    ]
}

Properties:

• inputTokens: Total tokens used for all input modalities
• outputTokens: Total tokens used for all output modalities
• totalTokens: Sum of input and output tokens
• modality_details: Optional list of token usage per modality
• cacheReadInputTokens: Optional tokens read from cache
• cacheWriteInputTokens: Optional tokens written to cache

Error Events

Events for error handling during conversations.

BidiErrorEvent

Emitted when an error occurs during the session.

{
    "type": "bidi_error",
    "message": "Connection failed",
    "code": "ConnectionError",
    "details": {"retry_after": 5}
}

Properties:

• message: Human-readable error message
• code: Error code (exception class name)
• details: Optional additional error context
• error: The original exception (accessible via property, not in JSON)

Usage:

async for event in agent.receive():
    if event["type"] == "bidi_error":
        print(f"Error: {event['message']}")
        # Access original exception if needed
        if hasattr(event, 'error'):
            raise event.error

Event Flow Examples

Basic Audio Conversation

import asyncio
from strands.experimental.bidi.agent import BidiAgent
from strands.experimental.bidi.io import BidiAudioIO
from strands.experimental.bidi.models import BedrockNovaSonicModel

async def main():
    model = BedrockNovaSonicModel()
    agent = BidiAgent(model=model)
    audio_io = BidiAudioIO()

    await agent.start()

    # Process events from audio conversation
    async for event in agent.receive():
        if event["type"] == "bidi_connection_start":
            print(f"Connected to {event['model']}")

        elif event["type"] == "bidi_response_start":
            print(f"Response starting: {event['response_id']}")

        elif event["type"] == "bidi_audio_stream":
            print(f"Audio chunk: {len(event['audio'])} bytes")

        elif event["type"] == "bidi_transcript_complete":
            print(f"{event['role']}: {event['transcript']}")

        elif event["type"] == "bidi_response_complete":
            print(f"Response complete: {event['stop_reason']}")

    await agent.stop()

asyncio.run(main())

Tracking Transcript State

import asyncio
from strands.experimental.bidi.agent import BidiAgent
from strands.experimental.bidi.models import BedrockNovaSonicModel

async def main():
    model = BedrockNovaSonicModel()

    async with BidiAgent(model=model) as agent:
        await agent.send("Tell me about Python")

        async for event in agent.receive():
            if event["type"] == "bidi_transcript_complete":
                print(f"{event['role']}: {event['transcript']}")

asyncio.run(main())

Tool Execution During Conversation

import asyncio
from strands.experimental.bidi.agent import BidiAgent
from strands.experimental.bidi.models import BedrockNovaSonicModel
from strands.vended_tools import notebook

async def main():
    model = BedrockNovaSonicModel()
    agent = BidiAgent(model=model, tools=[notebook])

    async with agent as agent:
        await agent.send('Create a notebook named "ideas" and add three project ideas.')

        async for event in agent.receive():
            event_type = event["type"]

            if event_type == "bidi_transcript_complete":
                print(f"{event['role']}: {event['transcript']}")

            elif event_type == "tool_use_stream":
                tool_use = event["current_tool_use"]
                print(f"Using tool: {tool_use['name']}")
                print(f"   Input: {tool_use['input']}")

            elif event_type == "bidi_response_complete":
                if event["stop_reason"] == "tool_use":
                    print("   Tool executing in background...")

asyncio.run(main())

Handling Interruptions

import asyncio
from strands.experimental.bidi.agent import BidiAgent
from strands.experimental.bidi.models import BedrockNovaSonicModel

async def main():
    model = BedrockNovaSonicModel()

    async with BidiAgent(model=model) as agent:
        await agent.send("Tell me a long story about space exploration")

        interruption_count = 0

        async for event in agent.receive():
            if event["type"] == "bidi_transcript_complete":
                print(f"{event['role']}: {event['transcript']}")

            elif event["type"] == "bidi_interruption":
                interruption_count += 1
                print(f"\nInterrupted (#{interruption_count})")

            elif event["type"] == "bidi_response_complete":
                if event["stop_reason"] == "interrupted":
                    print(f"Response interrupted {interruption_count} times")

asyncio.run(main())

Connection Restart Handling

import asyncio
from strands.experimental.bidi.agent import BidiAgent
from strands.experimental.bidi.models import BedrockNovaSonicModel

async def main():
    model = BedrockNovaSonicModel()  # 8-minute timeout

    async with BidiAgent(model=model) as agent:
        # Continuous conversation that handles restarts
        async for event in agent.receive():
            if event["type"] == "bidi_connection_warning":
                print(f"Reconnecting in ~{event['time_left_s']:.0f}s")

            elif event["type"] == "bidi_connection_restart":
                print(f"Connection restarting (reason={event['reason']})")
                if event["turn_interrupted"]:
                    print("   Last turn was not answered; re-prompt if needed")
                # History is preserved and the connection resumes automatically.

            elif event["type"] == "bidi_connection_start":
                print(f"Connected to {event['model']}")

            elif event["type"] == "bidi_transcript_complete":
                print(f"{event['role']}: {event['transcript']}")

asyncio.run(main())

Hook Events

Hook events are a separate concept from streaming events. While streaming events flow through agent.receive() during conversations, hook events are callbacks that trigger at specific lifecycle points (like initialization, message added, or interruption). Hook events allow you to inject custom logic for cross-cutting concerns like logging, analytics, and session persistence without processing the event stream directly.

For details on hook events and usage patterns, see the Hooks documentation.

Tags

bidi-streaming11