Events

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 barge-ins

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

async def main():
    model = BedrockNovaSonicModel(model_id="amazon.nova-2-sonic-v1:0")

    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.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

Consume output events through agent.receive().

Text, reasoning, and transcript streams share a lifecycle: a start event, zero or more deltas, a stop event, then a block event containing the accumulated content. Match events by content_id, since streams can overlap.

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-3.8-live”)

BidiConnectionRestartEvent

Emitted when the agent restarts the model connection, on either restart path. The agent preserves the conversation history and resumes automatically. A scheduled restart fires proactively when the restart 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 restart 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 restart timing.

BidiConnectionWarningEvent

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

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

Properties:

• time_left_s: Approximate seconds until the scheduled restart

Usage:

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

BidiConnectionStopEvent

Emitted when the streaming connection is closed.

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

Properties:

• connection_id: Unique identifier for this streaming connection
• reason: Why the connection closed. Always "user_request", emitted once a cancellation requested through agent.cancel() takes effect.

Response Lifecycle Events

Response start and stop events bracket the assistant’s audio, text, reasoning, transcript, and tool requests. User transcription is independent and may arrive outside these boundaries.

BidiResponseStartEvent

Emitted before a response’s assistant audio, text, reasoning, transcript, and tool-use events.

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

Properties:

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

BidiResponseStopEvent

Emitted when response output ends.

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

Properties:

• response_id: Unique identifier for this response

Audio Events

Each assistant audio stream has start, delta, and stop events sharing a content_id. A response can contain multiple audio streams. Audio has no block event.

BidiAudioStartEvent

Marks the beginning of an audio stream before chunks arrive.

{
    "type": "bidi_audio_start",
    "content_id": "audio_123"
}

BidiAudioDeltaEvent

Emitted for each chunk of audio output. Audio is base64-encoded for JSON compatibility.

{
    "type": "bidi_audio_delta",
    "audio": "base64_encoded_audio_data...",
    "format": "pcm",
    "sample_rate": 16000,
    "channels": 1,
    "content_id": "audio_123"
}

Properties:

• audio: Base64-encoded audio chunk
• 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)
• content_id: Identifier shared by this audio stream’s start, delta, and stop events

Usage:

import base64

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

BidiAudioStopEvent

Emitted when the audio stream ends, including when stopped due to barge-in. Buffered audio may still be playing. This event carries no audio and does not end the response or its transcript.

{
    "type": "bidi_audio_stop",
    "content_id": "audio_123"
}

Text and Reasoning Events

Text events carry written assistant responses. Reasoning events carry reasoning text or thought summaries exposed by the model.

All events include content_id: str. Delta events also carry delta: str, and block events carry the accumulated text: str. Start and stop events carry only the ID.

Event class	type
BidiTextStartEvent	bidi_text_start
BidiTextDeltaEvent	bidi_text_delta
BidiTextStopEvent	bidi_text_stop
BidiTextBlockEvent	bidi_text_block
BidiReasoningStartEvent	bidi_reasoning_start
BidiReasoningDeltaEvent	bidi_reasoning_delta
BidiReasoningStopEvent	bidi_reasoning_stop
BidiReasoningBlockEvent	bidi_reasoning_block

Transcript Events

Transcript events describe user or assistant speech. Each event includes a role and content_id.

BidiTranscriptStartEvent

Identifies a transcript and reserves its message in conversation history. It contains no text and may arrive after speech has begun.

{
    "type": "bidi_transcript_start",
    "role": "user",
    "content_id": "content_123"
}

Properties:

• role: Who is speaking ("user" or "assistant")
• content_id: Identifier for this transcript

BidiTranscriptDeltaEvent

Emitted for each incremental transcript update.

{
    "type": "bidi_transcript_delta",
    "delta": "Hello",
    "role": "user",
    "content_id": "content_123"
}

Properties:

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

BidiTranscriptStopEvent

Marks the end of a user or assistant transcript stream. It carries no transcript text.

{
    "type": "bidi_transcript_stop",
    "role": "user",
    "content_id": "content_123"
}

Properties:

• role: Who spoke ("user" or "assistant")
• content_id: Identifier for this transcript

BidiTranscriptBlockEvent

Carries the accumulated transcript after its stop event.

{
    "type": "bidi_transcript_block",
    "transcript": "Hello world",
    "role": "user",
    "content_id": "content_123"
}

Properties:

• transcript: The final transcript text
• role: Who spoke ("user" or "assistant")
• content_id: Identifier for this transcript

Reading Completed Content

Use deltas for live updates and block events for completed text, reasoning, and transcripts. This function reads completed content from a started agent:

from strands.bidi.agent import BidiAgent
from strands.bidi.types import (
    BidiReasoningBlockEvent,
    BidiTextBlockEvent,
    BidiTranscriptBlockEvent,
)

async def print_completed_content(agent: BidiAgent) -> None:
    async for event in agent.receive():
        if isinstance(event, (BidiTextBlockEvent, BidiReasoningBlockEvent)):
            print(event.text)
        elif isinstance(event, BidiTranscriptBlockEvent):
            print(f"{event.role}: {event.transcript}")

Conversation History Metadata

Streamed messages in agent.messages include BidiContentMetadata under message["metadata"]["custom"]["bidi"]:

Field	Values
kind	"text", "reasoning", "transcript"
status	"pending", "complete", "incomplete"

A start event reserves an empty message with status "pending". A stop event fills the message with accumulated content and marks it "complete". Text and transcripts use text content blocks; reasoning uses reasoningContent blocks.

If a connection ends before the stop event, the message is marked "incomplete". Text and reasoning messages retain their accumulated content. Unfinished transcripts contain [Transcript unavailable.].

Barge-in Events

Events for handling barge-in, when the user starts speaking during a model response.

BidiBargeInEvent

Signals a barge-in that stops response generation or playback, typically when the user starts speaking. The bidirectional session continues.

{
    "type": "bidi_barge_in"
}

Usage:

async for event in agent.receive():
    if event["type"] == "bidi_barge_in":
        print("Barge-in detected")
        # Audio output automatically cleared
        # Model ready for new input

Barge-in and tool interrupts

BidiBargeInEvent applies to the current response or its playback. New input typically starts another response. Agent’s human-in-the-loop interrupts pause an invocation to await input before resuming tool execution. BidiAgent does not yet support tool interrupts.

Tool Events

BidiToolUseBlocksEvent

Emitted when the model provides a complete group of tool calls. tool_uses contains the calls in provider order. A provider that emits individual calls uses a one-element list.

{
    "type": "bidi_tool_use_blocks",
    "tool_uses": [
        {
            "toolUseId": "tool_123",
            "name": "get_weather",
            "input": {"city": "Seattle"}
        },
        {
            "toolUseId": "tool_456",
            "name": "get_weather",
            "input": {"city": "Portland"}
        }
    ]
}

Each call contains toolUseId, name, and input. The agent executes the calls concurrently and sends their results together. A response can contain multiple tool-use groups.

Tools execute in the background. ToolResultEvent carries each result as it becomes available. Once the group finishes, ToolResultMessageEvent carries the history message containing all its results. Results retain their original tool-use IDs. Dispatch acknowledgements emit MessageAddedEvent hooks when appended to history.

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

Event Flow Examples

Basic Audio Conversation

import asyncio
from strands.bidi.agent import BidiAgent
from strands.bidi.io import AudioIO
from strands.bidi.models import BedrockNovaSonicModel

async def main():
    model = BedrockNovaSonicModel(model_id="amazon.nova-2-sonic-v1:0")
    agent = BidiAgent(model=model)
    audio_io = AudioIO()

    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_delta":
            print(f"Audio chunk: {len(event['audio'])} bytes")

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

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

    await agent.stop()

asyncio.run(main())

Tracking Transcript State

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

async def main():
    model = BedrockNovaSonicModel(model_id="amazon.nova-2-sonic-v1:0")

    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_block":
                print(f"{event['role']}: {event['transcript']}")

asyncio.run(main())

Tool Execution During Conversation

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

async def main():
    model = BedrockNovaSonicModel(model_id="amazon.nova-2-sonic-v1:0")
    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_block":
                print(f"{event['role']}: {event['transcript']}")

            elif event_type == "bidi_tool_use_blocks":
                for tool_use in event["tool_uses"]:
                    print(f"Using tool: {tool_use['name']}")
                    print(f"   Input: {tool_use['input']}")

asyncio.run(main())

Handling Barge-in

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

async def main():
    model = BedrockNovaSonicModel(model_id="amazon.nova-2-sonic-v1:0")

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

        barge_in_count = 0

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

            elif event["type"] == "bidi_barge_in":
                barge_in_count += 1
                print(f"\nBarge-in (#{barge_in_count})")

asyncio.run(main())

Connection Restart Handling

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

async def main():
    model = BedrockNovaSonicModel(model_id="amazon.nova-2-sonic-v1:0")  # 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"Restarting 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_block":
                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 barge-in). 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-streaming12