livekit.plugins.aws.experimental.realtime

class *RealtimeModel*(*,

```
High-level entry point that conforms to the LiveKit RealtimeModel interface.
The object is very light-weight-– it mainly stores default inference options and spawns a RealtimeSession when session() is invoked.
Instantiate a new RealtimeModel.
Args
*`voice`* : `VOICE_ID | NotGiven`
Preferred voice id for Sonic TTS output. Falls back to "tiffany".
*`temperature`* : `float | NotGiven`
Sampling temperature (0-1). Defaults to DEFAULT_TEMPERATURE.
*`top_p`* : `float | NotGiven`
Nucleus sampling probability mass. Defaults to DEFAULT_TOP_P.
*`max_tokens`* : `int | NotGiven`
Upper bound for tokens emitted by the model. Defaults to DEFAULT_MAX_TOKENS.
*`tool_choice`* : `llm.ToolChoice | None | NotGiven`
Strategy for tool invocation ("auto", "required", or explicit function).
*`region`* : `str | NotGiven`
AWS region of the Bedrock runtime endpoint.
Ancestors
• livekit.agents.llm.realtime.RealtimeModel
Methods
`def *session*(self) ‑> livekit.plugins.aws.experimental.realtime.realtime_model.RealtimeSession`
Expand source code
```

class *RealtimeSession*(realtime_model: RealtimeModel)

```
Bidirectional streaming session against the Nova Sonic Bedrock runtime.
The session owns two asynchronous tasks:
1. _process_audio_input – pushes user mic audio and tool results to Bedrock.
2. _process_responses – receives server events from Bedrock and converts them into LiveKit abstractions such as llm.MessageGeneration.
A set of helper handlers (_handle_*) transform the low-level Bedrock JSON payloads into higher-level application events and keep _ResponseGeneration state in sync.
Create and wire-up a new realtime session.
Args
*`realtime_model`* : `RealtimeModel`
Parent model instance that stores static inference options and the Smithy Bedrock client configuration.
Ancestors
• livekit.agents.llm.realtime.RealtimeSession abc.ABC <https://docs.livekit.io/reference/python/v1/livekit/rtc/event_emitter.html#livekit.rtc.event_emitter.EventEmitter|EventEmitter> typing.Generic
Instance variables
`prop *chat_ctx* : llm.ChatContext`
Expand source code
```

prop *tools* : llm.ToolContext

```
Methods
`async def *aclose*(self) ‑> None`
Expand source code
```

def *clear_audio*(self) ‑> None

```
`def *commit_audio*(self) ‑> None`
Expand source code
```

def *emit_generation_event*(self) ‑> None

```
Publish a llm.GenerationCreatedEvent to external subscribers.
`def *generate_reply*(self, *, instructions: NotGivenOr[str] = NOT_GIVEN) ‑> _asyncio.Future[livekit.agents.llm.realtime.GenerationCreatedEvent]`
Expand source code
```

async def *initialize_streams*(self, is_restart: bool = False) ‑> None

```
Open the Bedrock bidirectional stream and spawn background worker tasks.
This coroutine is idempotent and can be invoked again when recoverable errors (e.g. timeout, throttling) require a fresh session.
Args
*`is_restart`* : `bool`, optional
Marks whether we are re-initialising an existing session after an error. Defaults to False.
`def *interrupt*(self) ‑> None`
Expand source code
```

def *push_audio*(self, frame: rtc.AudioFrame) ‑> None

```
Enqueue an incoming mic rtc.AudioFrame for transcription.
`def *push_video*(self, frame: rtc.VideoFrame) ‑> None`
Expand source code
```

def *truncate*(self,

```
`async def *update_chat_ctx*(self, chat_ctx: llm.ChatContext) ‑> None`
Expand source code
```

async def *update_instructions*(self, instructions: str) ‑> None

```
Injects the system prompt at the start of the session.
`def *update_options*(self, *, tool_choice: NotGivenOr[llm.ToolChoice | None] = NOT_GIVEN) ‑> None`
Expand source code
```

async def *update_tools*(self, tools: list[llm.FunctionTool | llm.RawFunctionTool | Any]) ‑> None