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](/docs/user-guide/sdk/streaming/index.md):

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

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

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

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

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

```python
{
    "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.

```python
{
    "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:**

```python
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](/docs/user-guide/sdk/bidi/agent/index.md#connection-restart) 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.

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

**Properties:**

-   `time_left_s`: Approximate seconds until the scheduled restart

**Usage:**

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

```python
{
    "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.

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

**Properties:**

-   `response_id`: Unique identifier for this response (matches `BidiResponseStopEvent`)

#### BidiResponseStopEvent

Emitted when response output ends.

```python
{
    "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.

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

#### BidiAudioDeltaEvent

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

```python
{
    "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:**

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

```python
{
    "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.

```python
{
    "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.

```python
{
    "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.

```python
{
    "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.

```python
{
    "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:

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

```python
{
    "type": "bidi_barge_in"
}
```

**Usage:**

```python
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](/docs/user-guide/sdk/interrupts/index.md) 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.

```python
{
    "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.

```python
{
    "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

```python
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

```python
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

```python
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

```python
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

```python
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](/docs/user-guide/sdk/bidi/hooks/index.md) documentation.

## Related pages

- [Barge-in](/docs/user-guide/sdk/bidi/barge-in/index.md) (1 shared tag)
- [BidiAgent](/docs/user-guide/sdk/bidi/agent/index.md) (1 shared tag)
- [Build a realtime voice agent](/docs/user-guide/sdk/bidi/index.md) (1 shared tag)
- [Google Gemini Live](/docs/user-guide/sdk/bidi/models/google/index.md) (1 shared tag)
- [I/O Streams](/docs/user-guide/sdk/bidi/io/index.md) (1 shared tag)
- [Interrupts](/docs/user-guide/sdk/bidi/interrupts/index.md) (1 shared tag)
- [OpenAI Realtime](/docs/user-guide/sdk/bidi/models/openai/index.md) (1 shared tag)
- [Session Management](/docs/user-guide/sdk/bidi/session-management/index.md) (1 shared tag)
- [Bidirectional Streaming Observability](/docs/user-guide/sdk/bidi/observability/index.md) (1 shared tag)
- [Bidirectional Streaming Hooks](/docs/user-guide/sdk/bidi/hooks/index.md) (1 shared tag)


## Implementation

### Python

- [harness-sdk/strands-py/src/strands/bidi/types/agent.py](https://github.com/strands-agents/harness-sdk/blob/main/strands-py/src/strands/bidi/types/agent.py)
- [harness-sdk/strands-py/src/strands/bidi/types/content.py](https://github.com/strands-agents/harness-sdk/blob/main/strands-py/src/strands/bidi/types/content.py)
- [harness-sdk/strands-py/src/strands/bidi/types/media.py](https://github.com/strands-agents/harness-sdk/blob/main/strands-py/src/strands/bidi/types/media.py)
- [harness-sdk/strands-py/src/strands/types/content.py](https://github.com/strands-agents/harness-sdk/blob/main/strands-py/src/strands/types/content.py)
- [harness-sdk/strands-py/src/strands/bidi/types/events.py](https://github.com/strands-agents/harness-sdk/blob/main/strands-py/src/strands/bidi/types/events.py)
- [harness-sdk/strands-py/src/strands/bidi/types/io.py](https://github.com/strands-agents/harness-sdk/blob/main/strands-py/src/strands/bidi/types/io.py)
- [harness-sdk/strands-py/src/strands/types/media.py](https://github.com/strands-agents/harness-sdk/blob/main/strands-py/src/strands/types/media.py)