strands.experimental.bidi.agent

Bidirectional Agent for real-time streaming conversations.

Provides real-time audio and text interaction through persistent streaming connections. Unlike traditional request-response patterns, this agent maintains long-running conversations where users can interrupt, provide additional input, and receive continuous responses including audio output.

Key capabilities:

• Persistent conversation connections with concurrent processing
• Real-time audio input/output streaming
• Automatic interruption detection and tool execution
• Event-driven communication with model providers

BidiAgent

class BidiAgent(LocalAgent)

Defined in: src/strands/experimental/bidi/agent/agent.py:62

Agent for bidirectional streaming conversations.

Enables real-time audio and text interaction with AI models through persistent connections. Supports concurrent tool execution and interruption handling.

__init__

def __init__(model: BidiModel | str | None = None,
             tools: list[str | AgentTool | ToolProvider] | None = None,
             system_prompt: str | list[SystemContentBlock] | None = None,
             messages: Messages | None = None,
             record_direct_tool_call: bool = True,
             load_tools_from_directory: bool = False,
             agent_id: str | None = None,
             name: str | None = None,
             description: str | None = None,
             hooks: list[HookProvider] | None = None,
             state: AgentState | dict | None = None,
             session_manager: "SessionManager[LocalAgent] | None" = None,
             tool_executor: ToolExecutor | None = None,
             **kwargs: Any)

Defined in: src/strands/experimental/bidi/agent/agent.py:71

Initialize bidirectional agent.

Arguments:

• model - BidiModel instance, string model_id, or None for default detection.
• tools - Optional list of tools with flexible format support.
• system_prompt - System prompt for conversations as a string or structured content blocks. Structured blocks are retained, while their text is passed to Bidi models as a string.
• messages - Optional conversation history to initialize with.
• record_direct_tool_call - Whether to record direct tool calls in message history.
• load_tools_from_directory - Whether to load and automatically reload tools in the ./tools/ directory.
• agent_id - Optional ID for the agent, useful for connection management and multi-agent scenarios.
• name - Name of the Agent.
• description - Description of what the Agent does.
• hooks - Optional list of hook providers to register for lifecycle events.
• state - Stateful information for the agent. Can be either an AgentState object, or a json serializable dict.
• session_manager - Manager for handling agent sessions including conversation history and state. If provided, enables session-based persistence and state management.
• tool_executor - Definition of tool execution strategy (e.g., sequential, concurrent, etc.).
• **kwargs - Additional configuration for future extensibility.

Raises:

• ValueError - If model configuration is invalid or state is invalid type.
• TypeError - If model type is unsupported.

tool

@property
def tool() -> _ToolCaller

Defined in: src/strands/experimental/bidi/agent/agent.py:198

Call tool as a function.

Returns:

ToolCaller for method-style tool execution.

Example:

agent = BidiAgent(model=model, tools=[calculator])
agent.tool.calculator(expression="2+2")

tool_names

@property
def tool_names() -> list[str]

Defined in: src/strands/experimental/bidi/agent/agent.py:213

Get a list of all registered tool names.

Returns:

Names of all tools available to this agent.

system_prompt

@property
def system_prompt() -> str | None

Defined in: src/strands/experimental/bidi/agent/agent.py:223

Get the system prompt as a string.

system_prompt

@system_prompt.setter
def system_prompt(value: str | list[SystemContentBlock] | None) -> None

Defined in: src/strands/experimental/bidi/agent/agent.py:228

Set the system prompt and retain its structured content representation.

system_prompt_content

@property
def system_prompt_content() -> list[SystemContentBlock] | None

Defined in: src/strands/experimental/bidi/agent/agent.py:233

Get the system prompt as structured content blocks.

session_id

@property
def session_id() -> str

Defined in: src/strands/experimental/bidi/agent/agent.py:238

Get the conversation session identifier.

add_hook

def add_hook(callback: HookCallback[TEvent],
             event_type: type[TEvent] | list[type[TEvent]] | None = None,
             *,
             order: float = HookOrder.DEFAULT) -> None

Defined in: src/strands/experimental/bidi/agent/agent.py:242

Register a callback function for a specific event type.

This method supports multiple call patterns:

1. add_hook(callback) - Event type inferred from callback’s type hint
2. add_hook(callback, event_type) - Event type specified explicitly
3. add_hook(callback, [TypeA, TypeB]) - Register for multiple event types

When the callback’s type hint is a union type (A | B or Union[A, B]), the callback is automatically registered for each event type in the union.

Callbacks can be either synchronous or asynchronous functions.

Arguments:

• callback - The callback function to invoke when events of this type occur.
• event_type - The class type(s) of events this callback should handle. Can be a single type, a list of types, or None to infer from the callback’s first parameter type hint. If a list is provided, the callback is registered for each type in the list.
• order - Execution priority. Lower values execute first. Use a HookOrder constant such as SDK_FIRST (-100), DEFAULT (0), MODEL_ROUTING (50), or SDK_LAST (100).

Raises:

• ValueError - If event_type is not provided and cannot be inferred from the callback’s type hints, or if the event_type list is empty.

start

async def start(invocation_state: dict[str, Any] | None = None) -> None

Defined in: src/strands/experimental/bidi/agent/agent.py:277

Start a persistent bidirectional conversation connection.

Initializes the streaming connection and starts background tasks for processing model events, tool execution, and connection management.

Arguments:

• invocation_state - Optional context shared by reference with tools and hooks until stop(), including across connection restarts. Tools access it through ToolContext.invocation_state. Defaults to a new empty dictionary.

Raises:

RuntimeError: If agent already started.

Example:

await agent.start(invocation_state=\{
    "user_id": "user_123",
    "session_id": "session_456",
    "database": db_connection,
})

send

async def send(input_data: BidiAgentInput) -> None

Defined in: src/strands/experimental/bidi/agent/agent.py:308

Send content to the model.

A string is shorthand for a text block. Image blocks contain complete images. Audio deltas append samples to the live input stream without explicitly ending the user’s turn.

Arguments:

• input_data - Can be:

  • str: Text message from user
  • TextBlock, AudioDelta, or ImageBlock: Text, streaming audio, or image input
  • BidiContentBlockData: A dictionary containing one text or image key
  • BidiContentDeltaData: A dictionary containing one audio_delta key

Raises:

• RuntimeError - If start has not been called.
• TypeError - If the input has an unsupported type or invalid input arguments.
• ValueError - If the input dictionary does not contain exactly one text, audio_delta, or image key.

Example:

await agent.send(“Hello”) await agent.send(AudioDelta(format=“pcm”, source={“bytes”: audio_bytes})) await agent.send({“audio_delta”: {“format”: “pcm”, “source”: {“bytes”: audio_bytes}}})

receive

async def receive() -> AsyncGenerator[BidiOutputEvent, None]

Defined in: src/strands/experimental/bidi/agent/agent.py:358

Receive events from the model including audio, text, and tool calls.

Yields:

Model output events processed by background tasks including audio output, text responses, tool calls, and connection updates.

Raises:

• RuntimeError - If start has not been called.

stop

async def stop() -> None

Defined in: src/strands/experimental/bidi/agent/agent.py:374

End the conversation connection and cleanup all resources.

Terminates the streaming connection, cancels background tasks, and closes the connection to the model provider.

__aenter__

async def __aenter__(
        invocation_state: dict[str, Any] | None = None) -> "BidiAgent"

Defined in: src/strands/experimental/bidi/agent/agent.py:383

Async context manager entry point.

Automatically starts the bidirectional connection when entering the context.

Arguments:

• invocation_state - Optional context to pass to tools during execution. This allows passing custom data (user_id, session_id, database connections, etc.) that tools can access via their invocation_state parameter.

Returns:

Self for use in the context.

__aexit__

async def __aexit__(*_: Any) -> None

Defined in: src/strands/experimental/bidi/agent/agent.py:400

Async context manager exit point.

Automatically ends the connection and cleans up resources including when exiting the context, regardless of whether an exception occurred.

run

async def run(inputs: list[BidiInput],
              outputs: list[BidiOutput],
              invocation_state: dict[str, Any] | None = None) -> None

Defined in: src/strands/experimental/bidi/agent/agent.py:409

Run the agent using provided IO channels for bidirectional communication.

Arguments:

• inputs - Input callables to read data from a source
• outputs - Output callables to receive events from the agent
• invocation_state - Optional context shared by reference with tools and hooks for the duration of run(), including across connection restarts. Tools access it through ToolContext.invocation_state. Defaults to a new empty dictionary.

Example:

# Using model defaults:
model = BedrockNovaSonicModel()
audio_io = BidiAudioIO()
agent = BidiAgent(model=model, tools=[calculator])
await agent.run(
    inputs=[audio_io.input()],
    outputs=[audio_io.output()],
    invocation_state=\{"user_id": "user_123"}
)

# Using custom audio config:
model = BedrockNovaSonicModel(
    audio=\{
        "input": \{"sample_rate": 16000},
        "output": \{"sample_rate": 24000},
    }
)
audio_io = BidiAudioIO()
agent = BidiAgent(model=model, tools=[calculator])
await agent.run(
    inputs=[audio_io.input()],
    outputs=[audio_io.output()],
)