strands.experimental.bidi.io.text

Handle text input and output to and from bidi agent.

BidiOutputEvent = BidiConnectionStartEvent | BidiConnectionRestartEvent | BidiResponseStartEvent | BidiAudioStreamEvent | BidiTranscriptStreamEvent | BidiInterruptionEvent | BidiResponseCompleteEvent | BidiUsageEvent | BidiConnectionCloseEvent | BidiErrorEvent | ToolUseStreamEvent module-attribute

Union of different bidi output event types.

logger = logging.getLogger(__name__) module-attribute

BidiConnectionCloseEvent

Bases: TypedEvent

Streaming connection closed.

Parameters:

Name	Type	Description	Default
connection_id	str	Unique identifier for this streaming connection (matches BidiConnectionStartEvent).	required
reason	Literal['client_disconnect', 'timeout', 'error', 'complete', 'user_request']	Why the connection was closed.	required

Source code in strands/experimental/bidi/types/events.py

class BidiConnectionCloseEvent(TypedEvent):
    """Streaming connection closed.

    Parameters:
        connection_id: Unique identifier for this streaming connection (matches BidiConnectionStartEvent).
        reason: Why the connection was closed.
    """

    def __init__(
        self,
        connection_id: str,
        reason: Literal["client_disconnect", "timeout", "error", "complete", "user_request"],
    ):
        """Initialize connection close event."""
        super().__init__(
            {
                "type": "bidi_connection_close",
                "connection_id": connection_id,
                "reason": reason,
            }
        )

    @property
    def connection_id(self) -> str:
        """Unique identifier for this streaming connection."""
        return cast(str, self["connection_id"])

    @property
    def reason(self) -> str:
        """Why the interruption occurred."""
        return cast(str, self["reason"])

connection_id property

Unique identifier for this streaming connection.

reason property

Why the interruption occurred.

__init__(connection_id, reason)

Initialize connection close event.

Source code in strands/experimental/bidi/types/events.py

def __init__(
    self,
    connection_id: str,
    reason: Literal["client_disconnect", "timeout", "error", "complete", "user_request"],
):
    """Initialize connection close event."""
    super().__init__(
        {
            "type": "bidi_connection_close",
            "connection_id": connection_id,
            "reason": reason,
        }
    )

BidiInput

Bases: Protocol

Protocol for bidirectional input callables.

Input callables read data from a source (microphone, camera, websocket, etc.) and return events to be sent to the agent.

Source code in strands/experimental/bidi/types/io.py

@runtime_checkable
class BidiInput(Protocol):
    """Protocol for bidirectional input callables.

    Input callables read data from a source (microphone, camera, websocket, etc.)
    and return events to be sent to the agent.
    """

    async def start(self, agent: "BidiAgent") -> None:
        """Start input."""
        return

    async def stop(self) -> None:
        """Stop input."""
        return

    def __call__(self) -> Awaitable[BidiInputEvent]:
        """Read input data from the source.

        Returns:
            Awaitable that resolves to an input event (audio, text, image, etc.)
        """
        ...

__call__()

Read input data from the source.

Returns:

Type	Description
Awaitable[BidiInputEvent]	Awaitable that resolves to an input event (audio, text, image, etc.)

Source code in strands/experimental/bidi/types/io.py

def __call__(self) -> Awaitable[BidiInputEvent]:
    """Read input data from the source.

    Returns:
        Awaitable that resolves to an input event (audio, text, image, etc.)
    """
    ...

start(agent) async

Start input.

Source code in strands/experimental/bidi/types/io.py

async def start(self, agent: "BidiAgent") -> None:
    """Start input."""
    return

stop() async

Stop input.

Source code in strands/experimental/bidi/types/io.py

async def stop(self) -> None:
    """Stop input."""
    return

BidiInterruptionEvent

Bases: TypedEvent

Model generation was interrupted.

Parameters:

Name	Type	Description	Default
reason	Literal['user_speech', 'error']	Why the interruption occurred.	required

Source code in strands/experimental/bidi/types/events.py

class BidiInterruptionEvent(TypedEvent):
    """Model generation was interrupted.

    Parameters:
        reason: Why the interruption occurred.
    """

    def __init__(self, reason: Literal["user_speech", "error"]):
        """Initialize interruption event."""
        super().__init__(
            {
                "type": "bidi_interruption",
                "reason": reason,
            }
        )

    @property
    def reason(self) -> str:
        """Why the interruption occurred."""
        return cast(str, self["reason"])

reason property

Why the interruption occurred.

__init__(reason)

Initialize interruption event.

Source code in strands/experimental/bidi/types/events.py

def __init__(self, reason: Literal["user_speech", "error"]):
    """Initialize interruption event."""
    super().__init__(
        {
            "type": "bidi_interruption",
            "reason": reason,
        }
    )

BidiOutput

Bases: Protocol

Protocol for bidirectional output callables.

Output callables receive events from the agent and handle them appropriately (play audio, display text, send over websocket, etc.).

Source code in strands/experimental/bidi/types/io.py

@runtime_checkable
class BidiOutput(Protocol):
    """Protocol for bidirectional output callables.

    Output callables receive events from the agent and handle them appropriately
    (play audio, display text, send over websocket, etc.).
    """

    async def start(self, agent: "BidiAgent") -> None:
        """Start output."""
        return

    async def stop(self) -> None:
        """Stop output."""
        return

    def __call__(self, event: BidiOutputEvent) -> Awaitable[None]:
        """Process output events from the agent.

        Args:
            event: Output event from the agent (audio, text, tool calls, etc.)
        """
        ...

__call__(event)

Process output events from the agent.

Parameters:

Name	Type	Description	Default
event	BidiOutputEvent	Output event from the agent (audio, text, tool calls, etc.)	required

Source code in strands/experimental/bidi/types/io.py

def __call__(self, event: BidiOutputEvent) -> Awaitable[None]:
    """Process output events from the agent.

    Args:
        event: Output event from the agent (audio, text, tool calls, etc.)
    """
    ...

start(agent) async

Start output.

Source code in strands/experimental/bidi/types/io.py

async def start(self, agent: "BidiAgent") -> None:
    """Start output."""
    return

stop() async

Stop output.

Source code in strands/experimental/bidi/types/io.py

async def stop(self) -> None:
    """Stop output."""
    return

BidiTextIO

Handle text input and output to and from bidi agent.

Accepts input from stdin and outputs to stdout.

Source code in strands/experimental/bidi/io/text.py

class BidiTextIO:
    """Handle text input and output to and from bidi agent.

    Accepts input from stdin and outputs to stdout.
    """

    def __init__(self, **config: Any) -> None:
        """Initialize I/O.

        Args:
            **config: Optional I/O configurations.

                - input_prompt (str): Input prompt to display on screen (default: blank)
        """
        self._config = config

    def input(self) -> _BidiTextInput:
        """Return text processing BidiInput."""
        return _BidiTextInput(self._config)

    def output(self) -> _BidiTextOutput:
        """Return text processing BidiOutput."""
        return _BidiTextOutput()

__init__(**config)

Initialize I/O.

Parameters:

Name	Type	Description	Default
**config	Any	Optional I/O configurations. input_prompt (str): Input prompt to display on screen (default: blank)	{}

Source code in strands/experimental/bidi/io/text.py

def __init__(self, **config: Any) -> None:
    """Initialize I/O.

    Args:
        **config: Optional I/O configurations.

            - input_prompt (str): Input prompt to display on screen (default: blank)
    """
    self._config = config

input()

Return text processing BidiInput.

Source code in strands/experimental/bidi/io/text.py

def input(self) -> _BidiTextInput:
    """Return text processing BidiInput."""
    return _BidiTextInput(self._config)

output()

Return text processing BidiOutput.

Source code in strands/experimental/bidi/io/text.py

def output(self) -> _BidiTextOutput:
    """Return text processing BidiOutput."""
    return _BidiTextOutput()

BidiTextInputEvent

Bases: TypedEvent

Text input event for sending text to the model.

Used for sending text content through the send() method.

Parameters:

Name	Type	Description	Default
text	str	The text content to send to the model.	required
role	Role	The role of the message sender (default: "user").	'user'

Source code in strands/experimental/bidi/types/events.py

class BidiTextInputEvent(TypedEvent):
    """Text input event for sending text to the model.

    Used for sending text content through the send() method.

    Parameters:
        text: The text content to send to the model.
        role: The role of the message sender (default: "user").
    """

    def __init__(self, text: str, role: Role = "user"):
        """Initialize text input event."""
        super().__init__(
            {
                "type": "bidi_text_input",
                "text": text,
                "role": role,
            }
        )

    @property
    def text(self) -> str:
        """The text content to send to the model."""
        return cast(str, self["text"])

    @property
    def role(self) -> Role:
        """The role of the message sender."""
        return cast(Role, self["role"])

role property

The role of the message sender.

text property

The text content to send to the model.

__init__(text, role='user')

Initialize text input event.

Source code in strands/experimental/bidi/types/events.py

def __init__(self, text: str, role: Role = "user"):
    """Initialize text input event."""
    super().__init__(
        {
            "type": "bidi_text_input",
            "text": text,
            "role": role,
        }
    )

BidiTranscriptStreamEvent

Bases: ModelStreamEvent

Audio transcription streaming (user or assistant speech).

Supports incremental transcript updates for providers that send partial transcripts before the final version.

Parameters:

Name	Type	Description	Default
delta	ContentBlockDelta	The incremental transcript change (ContentBlockDelta).	required
text	str	The delta text (same as delta content for convenience).	required
role	Role	Who is speaking ("user" or "assistant").	required
is_final	bool	Whether this is the final/complete transcript.	required
current_transcript	str | None	The accumulated transcript text so far (None for first delta).	None

Source code in strands/experimental/bidi/types/events.py

class BidiTranscriptStreamEvent(ModelStreamEvent):
    """Audio transcription streaming (user or assistant speech).

    Supports incremental transcript updates for providers that send partial
    transcripts before the final version.

    Parameters:
        delta: The incremental transcript change (ContentBlockDelta).
        text: The delta text (same as delta content for convenience).
        role: Who is speaking ("user" or "assistant").
        is_final: Whether this is the final/complete transcript.
        current_transcript: The accumulated transcript text so far (None for first delta).
    """

    def __init__(
        self,
        delta: ContentBlockDelta,
        text: str,
        role: Role,
        is_final: bool,
        current_transcript: str | None = None,
    ):
        """Initialize transcript stream event."""
        super().__init__(
            {
                "type": "bidi_transcript_stream",
                "delta": delta,
                "text": text,
                "role": role,
                "is_final": is_final,
                "current_transcript": current_transcript,
            }
        )

    @property
    def delta(self) -> ContentBlockDelta:
        """The incremental transcript change."""
        return cast(ContentBlockDelta, self["delta"])

    @property
    def text(self) -> str:
        """The text content to send to the model."""
        return cast(str, self["text"])

    @property
    def role(self) -> Role:
        """The role of the message sender."""
        return cast(Role, self["role"])

    @property
    def is_final(self) -> bool:
        """Whether this is the final/complete transcript."""
        return cast(bool, self["is_final"])

    @property
    def current_transcript(self) -> str | None:
        """The accumulated transcript text so far."""
        return cast(str | None, self.get("current_transcript"))

current_transcript property

The accumulated transcript text so far.

delta property

The incremental transcript change.

is_final property

Whether this is the final/complete transcript.

role property

The role of the message sender.

text property

The text content to send to the model.

__init__(delta, text, role, is_final, current_transcript=None)

Initialize transcript stream event.

Source code in strands/experimental/bidi/types/events.py

def __init__(
    self,
    delta: ContentBlockDelta,
    text: str,
    role: Role,
    is_final: bool,
    current_transcript: str | None = None,
):
    """Initialize transcript stream event."""
    super().__init__(
        {
            "type": "bidi_transcript_stream",
            "delta": delta,
            "text": text,
            "role": role,
            "is_final": is_final,
            "current_transcript": current_transcript,
        }
    )

_BidiTextInput

Bases: BidiInput

Handle text input from user.

Source code in strands/experimental/bidi/io/text.py

class _BidiTextInput(BidiInput):
    """Handle text input from user."""

    def __init__(self, config: dict[str, Any]) -> None:
        """Extract configs and setup prompt session."""
        prompt = config.get("input_prompt", "")
        self._session: PromptSession = PromptSession(prompt)

    async def __call__(self) -> BidiTextInputEvent:
        """Read user input from stdin."""
        text = await self._session.prompt_async()
        return BidiTextInputEvent(text.strip(), role="user")

__call__() async

Read user input from stdin.

Source code in strands/experimental/bidi/io/text.py

async def __call__(self) -> BidiTextInputEvent:
    """Read user input from stdin."""
    text = await self._session.prompt_async()
    return BidiTextInputEvent(text.strip(), role="user")

__init__(config)

Extract configs and setup prompt session.

Source code in strands/experimental/bidi/io/text.py

def __init__(self, config: dict[str, Any]) -> None:
    """Extract configs and setup prompt session."""
    prompt = config.get("input_prompt", "")
    self._session: PromptSession = PromptSession(prompt)

_BidiTextOutput

Bases: BidiOutput

Handle text output from bidi agent.

Source code in strands/experimental/bidi/io/text.py

class _BidiTextOutput(BidiOutput):
    """Handle text output from bidi agent."""

    async def __call__(self, event: BidiOutputEvent) -> None:
        """Print text events to stdout."""
        if isinstance(event, BidiInterruptionEvent):
            logger.debug("reason=<%s> | text output interrupted", event["reason"])
            print("interrupted")

        elif isinstance(event, BidiConnectionCloseEvent):
            if event.reason == "user_request":
                print("user requested connection close using the stop_conversation tool.")
                logger.debug("connection_id=<%s> | user requested connection close", event.connection_id)
        elif isinstance(event, BidiTranscriptStreamEvent):
            text = event["text"]
            is_final = event["is_final"]
            role = event["role"]

            logger.debug(
                "role=<%s>, is_final=<%s>, text_length=<%d> | text transcript received",
                role,
                is_final,
                len(text),
            )

            if not is_final:
                text = f"Preview: {text}"

            print(text)

__call__(event) async

Print text events to stdout.

Source code in strands/experimental/bidi/io/text.py

async def __call__(self, event: BidiOutputEvent) -> None:
    """Print text events to stdout."""
    if isinstance(event, BidiInterruptionEvent):
        logger.debug("reason=<%s> | text output interrupted", event["reason"])
        print("interrupted")

    elif isinstance(event, BidiConnectionCloseEvent):
        if event.reason == "user_request":
            print("user requested connection close using the stop_conversation tool.")
            logger.debug("connection_id=<%s> | user requested connection close", event.connection_id)
    elif isinstance(event, BidiTranscriptStreamEvent):
        text = event["text"]
        is_final = event["is_final"]
        role = event["role"]

        logger.debug(
            "role=<%s>, is_final=<%s>, text_length=<%d> | text transcript received",
            role,
            is_final,
            len(text),
        )

        if not is_final:
            text = f"Preview: {text}"

        print(text)