Realtime

The realtime subsystem provides live, audio-to-audio sessions backed by Gemini Live. It is a separate runtime from the turn-based invoke/stream path: the provider owns the turn loop, and 10xGraph wraps it with a duplex pump, tool dispatch, transcript persistence, and automatic reconnection.

Install extra

pip install "10xgraph[realtime]"

Provider SDK imports are lazy. Importing tenxgraph.core.realtime does not load google-genai until a session opens.

Import paths

# Public API surface
from tenxgraph.core.realtime import (
    # Input queue
    LiveInputQueue, LiveInput, LiveInputKind,
    # Config types
    RealtimeConfig, VADConfig, ReconnectConfig,
    # Event union and all event classes
    RealtimeEvent,
    AudioDeltaEvent, InputTranscriptEvent, OutputTranscriptEvent,
    ToolCallEvent, ToolResultEvent, TurnCompleteEvent, InterruptedEvent,
    SessionUpdateEvent, GoAwayEvent, AgentChangedEvent, ErrorEvent,
    # Provider client (Gemini)
    RealtimeClient, GeminiLiveClient, normalize_message,
)

# Prebuilt agent
from tenxgraph.prebuilt.agent import AudioAgent

# Audio format constants
from tenxgraph.core.realtime.base import INPUT_SAMPLE_RATE, OUTPUT_SAMPLE_RATE

INPUT_SAMPLE_RATE = 16000 (Hz); OUTPUT_SAMPLE_RATE = 24000 (Hz).

AudioAgent

Prebuilt realtime agent builder. Mirrors ReactAgent’s construction surface and wraps a LiveAgent as the graph root. Compile it once and drive sessions with CompiledGraph.arealtime().

from tenxgraph.prebuilt.agent import AudioAgent

Constructor

AudioAgent(
    model: str,
    state: StateT | None = None,
    context_manager: BaseContextManager | None = None,
    publisher: BasePublisher | list[BasePublisher] | None = None,
    id_generator: BaseIDGenerator = DefaultIDGenerator(),
    container: Any | None = None,
    *,
    realtime_config: RealtimeConfig | None = None,
    system_prompt: list[dict[str, Any]] | None = None,
    tools: Iterable[Callable] | None = None,
    client: Any = None,                        # MCP client
    pass_user_info_to_mcp: bool = False,
    skills: SkillConfig | None = None,
    memory: MemoryConfig | None = None,
    realtime_client_factory: Callable[[], RealtimeClient] | None = None,
    live_node_name: str = "LIVE",
    **agent_kwargs,
)

A ToolNode is only created when tools or client is provided; otherwise the live agent runs without tools.

compile()

app = AudioAgent(...).compile(
    checkpointer: BaseCheckpointer | None = None,
    store: BaseStore | None = None,
    callback_manager: CallbackManager | None = None,
    shutdown_timeout: float = 30.0,
) -> CompiledGraph

compile() does not accept media_store, interrupt_before, or interrupt_after. Realtime media is sent frame-by-frame through LiveInputQueue and is never offloaded.

CompiledGraph.arealtime / realtime

async def arealtime(
    input_queue: LiveInputQueue,
    config: dict[str, Any] | None = None,
    state: AgentState | None = None,
) -> AsyncIterator[RealtimeEvent]

def realtime(
    input_queue: LiveInputQueue,
    config: dict[str, Any] | None = None,
    state: AgentState | None = None,
) -> Generator[RealtimeEvent]

arealtime is an async generator; use it from async contexts. realtime is a synchronous wrapper that drives a private event loop and raises if called from inside a running event loop.

Config keys:

Forcing rules:

• A graph containing a LiveAgent must use arealtime() / realtime(). Calling invoke, ainvoke, stream, or astream raises RuntimeError.
• arealtime() requires exactly one LiveAgent in the graph. Zero or more than one raises.
• realtime() raises if called from a thread with a running event loop.

LiveInputQueue

Non-blocking upstream input queue. Feed it from any context including audio capture callbacks on other threads.

LiveInputQueue(maxsize: int = 0)

All send methods are synchronous (put_nowait), so the input side keeps accepting audio while the model is still generating. That is the precondition for barge-in. Feed one queue per session.

maxsize=0 (the default) means unbounded. With a positive maxsize, a frame that arrives when the queue is full is dropped and logged at warning level rather than blocking the producer. Sends after close() are dropped silently (logged at DEBUG).

Methods

closed is a read-only property reporting whether close() has been called.

Image input

send_image is a Python-SDK capability. Gemini Live accepts still images and video as individual frames, so send video as a stream of frames; roughly 1 fps is the model’s effective ceiling. mime_type must be an image type the provider supports and defaults to image/jpeg.

queue.send_image(jpeg_bytes, mime_type="image/jpeg")

The API server’s /v1/graph/live WebSocket bridge forwards audio and text only, so image input is not available over that socket. Use the SDK path when you need it.

LiveInput

The dataclass frames enqueued by send_*. Not normally constructed directly.

@dataclass
class LiveInput:
    kind: LiveInputKind   # "audio" | "text" | "image" | "activity_start" | "activity_end" | "close"
    data: bytes | None = None
    text: str | None = None
    sample_rate: int = 16000
    mime_type: str | None = None

RealtimeConfig

Per-session configuration for a realtime session.

from tenxgraph.core.realtime import RealtimeConfig

RealtimeConfig(
    model: str,
    response_modalities: list[Literal["AUDIO", "TEXT"]] = ["AUDIO"],
    voice: str | None = None,
    system_instruction: str | None = None,
    input_audio_transcription: bool = True,
    output_audio_transcription: bool = True,
    vad: VADConfig = VADConfig(),
    reconnect: ReconnectConfig = ReconnectConfig(),
    context_window_compression: bool = False,
    session_resumption: bool = True,
    tools: list[Any] | None = None,
    tools_tags: list[str] | None = None,
)

response_modalities is validated: it must contain exactly one entry. Passing ["AUDIO", "TEXT"] raises ValueError.

VADConfig

Voice-activity detection settings. Disable for push-to-talk (manual activity) workflows.

from tenxgraph.core.realtime import VADConfig

VADConfig(
    enabled: bool = True,
    start_sensitivity: str | None = None,
    end_sensitivity: str | None = None,
    prefix_padding_ms: int | None = None,
    silence_duration_ms: int | None = None,
)

ReconnectConfig

Reconnect and backoff policy for dropped sessions.

from tenxgraph.core.realtime import ReconnectConfig

ReconnectConfig(
    base_delay: float = 0.5,
    max_delay: float = 10.0,
    max_attempts: int = 5,
)

Reconnect rules:

• go_away (planned provider rotation): reconnect immediately with no backoff, attempts counter is not incremented.
• Transient drop / receive error: attempt n waits min(base_delay * 2^(n-1), max_delay) seconds, up to max_attempts. After the cap, a fatal ErrorEvent(code="reconnect_failed") is emitted and the session ends.
• Any successful receive resets the attempts counter to 0.

RealtimeEvent

Discriminated union keyed on type. All events are Pydantic models.

from tenxgraph.core.realtime import RealtimeEvent

AudioDeltaEvent

type: Literal["audio_delta"]
data: bytes        # PCM16 chunk at OUTPUT_SAMPLE_RATE (24000 Hz)
sample_rate: int   # always 24000

A chunk of model audio output. Write it to a speaker or file.

InputTranscriptEvent

type: Literal["input_transcript"]
text: str
finished: bool     # True on the final chunk; text carries the complete transcript

Transcript of the user’s speech. The provider streams partial chunks (finished=False) as they are transcribed. On finished=True, text carries the complete transcript for the turn. Persisted to the checkpointer thread when finished.

OutputTranscriptEvent

type: Literal["output_transcript"]
text: str
finished: bool

Transcript of the model’s speech. Same streaming/finished semantics as InputTranscriptEvent. Persisted to the checkpointer thread when finished.

ToolCallEvent

type: Literal["tool_call"]
id: str
name: str
args: dict[str, Any]

The provider is requesting a tool invocation. 10xGraph dispatches this automatically through the ToolNode and returns the result before the model continues. Emitted before tool execution for observability.

ToolResultEvent

type: Literal["tool_result"]
id: str
result: Any

Emitted after a tool finishes, for observability. The result has already been sent back to the model.

TurnCompleteEvent

type: Literal["turn_complete"]

The model finished generating a turn (audio and transcription complete).

InterruptedEvent

type: Literal["interrupted"]

Barge-in: the user spoke while the model was talking. Flush any queued audio playback. Partial transcripts for the interrupted turn are discarded.

SessionUpdateEvent

type: Literal["session_update"]
resumption_handle: str | None

The provider issued a session-resumption handle. Stored in the checkpointer thread metadata automatically when a checkpointer is provided.

GoAwayEvent

type: Literal["go_away"]
time_left: str | None    # provider duration string (e.g. "5s"); verbatim

The provider will close the socket soon (planned rotation). The runtime reconnects immediately with the cached resumption handle; no intervention is required from the caller.

AgentChangedEvent

type: Literal["agent_changed"]
author: str

The active agent or persona changed. Reserved for future multi-agent persona swap.

ErrorEvent

type: Literal["error"]
code: str | None
message: str
fatal: bool

A normalized provider error. Non-fatal errors are transient; the session continues. Fatal errors (fatal=True) end the session; code="reconnect_failed" means reconnect attempts were exhausted.

RealtimeClient (Protocol)

Provider-neutral protocol that all provider clients implement. Not used directly in application code; use AudioAgent instead.

from tenxgraph.core.realtime import RealtimeClient

GeminiLiveClient

The Gemini Live provider client. Import path: tenxgraph.core.realtime.GeminiLiveClient. Used by LiveAgent internally; inject a custom factory via AudioAgent(realtime_client_factory=...) for testing.

from tenxgraph.core.realtime import GeminiLiveClient, normalize_message

normalize_message converts Gemini wire messages to RealtimeEvent objects.

GraphLifecycleHook integration

Realtime sessions fire lifecycle hooks via GraphLifecycleHook. Register via CallbackManager.register_lifecycle_hook and pass callback_manager to compile().

from tenxgraph.utils.callbacks import CallbackManager, GraphLifecycleHook
from tenxgraph.core.state import AgentState

class SessionAuditHook(GraphLifecycleHook):
    async def on_graph_start(self, ctx, state: AgentState) -> AgentState:
        print("session started")
        return state

    async def on_graph_end(self, ctx, state, messages, total_steps: int) -> None:
        print(f"session ended after {total_steps} turns")

    async def on_turn_start(self, ctx, state: AgentState, turn_index: int) -> AgentState:
        print(f"turn {turn_index} starting")
        return state

    async def on_turn_end(self, ctx, state: AgentState, turn_index: int) -> AgentState:
        print(f"turn {turn_index} complete")
        return state

cb = CallbackManager()
cb.register_lifecycle_hook(SessionAuditHook())

app = AudioAgent(MODEL, ...).compile(callback_manager=cb)

Hook semantics in realtime:

These hooks are no-ops for invoke and stream. Tool/MCP before/after/error callbacks fire as usual. No AI-invocation callback or input-validator pass runs in realtime.

API server WebSocket bridge

agentflow api exposes ws://<host>/v1/graph/live when the configured graph contains a LiveAgent.

Protocol reference

Auth: RequirePermission("graph", "stream"). Browser WebSocket clients can pass the token as a ?token= query parameter.

Frame 1 (client -> server): JSON control (init)

{
  "thread_id": "abc",
  "model": "gemini-live-2.5-flash-preview",
  "voice": "Puck",
  "modalities": ["AUDIO"],
  "vad": {"enabled": true}
}

All fields are optional. Present fields override the agent’s build-time RealtimeConfig for this session. A new thread_id is generated if absent.

Subsequent upstream frames:

Downstream frames:

Image/video input is SDK-only. The WebSocket bridge does not forward image frames.

WebSocket close codes

Common errors