strands.experimental.bidi.agent.loop

Agent loop.

The agent loop handles the events received from the model and executes tools when given a tool use request.

_BidiAgentLoop

class _BidiAgentLoop()

Agent loop.

Attributes:

_agent - BidiAgent instance to loop.
_started - Flag if agent loop has started.
_task_pool - Track active async tasks created in loop.
_event_queue - Queue model and tool call events for receiver.
_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.
_send_gate - Gate the sending of events to the model. Blocks when agent is reseting the model connection after timeout.

def __init__(agent: "BidiAgent") -> None

Initialize members of the agent loop.

Note, before receiving events from the loop, the user must call start.

Arguments:

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

Start the agent loop.

The agent model is started as part of this call.

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.

Raises:

async def stop() -> None

Stop the agent loop.

async def send(event: BidiInputEvent | ToolResultEvent) -> None

Send model event.

Additionally, add text input to messages array.

Arguments:

Raises:

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

Receive model and tool call events.

Returns:

Model and tool call events.

Raises: