Realtime audio

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,
)

from tenxgraph.prebuilt.agent import AudioAgent
from tenxgraph.core.realtime.base import INPUT_SAMPLE_RATE, OUTPUT_SAMPLE_RATE

import asyncio

from tenxgraph.core.realtime import LiveInputQueue
from tenxgraph.prebuilt.agent import AudioAgent


def get_time() -> str:
    """Return the current time as a string."""
    from datetime import datetime

    return datetime.now().isoformat(timespec="seconds")


agent = AudioAgent(
    model="gemini-2.5-flash",  # use the Gemini Live model your account has access to
    system_prompt=[{"role": "system", "content": "You are a helpful assistant."}],
    tools=[get_time],
)
app = agent.compile()


async def main() -> None:
    queue = LiveInputQueue()
    queue.send_text("What time is it?")
    queue.close()  # closing the queue ends the session once the provider goes idle

    async for event in app.arealtime(queue, config={"thread_id": "demo"}):
        if event.type == "output_transcript" and event.finished:
            print("Model:", event.text)
        elif event.type == "error":
            print("Error:", event.message)


asyncio.run(main())

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,
    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: Any,
)

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

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]

from tenxgraph.core.realtime import LiveInputQueue

queue = LiveInputQueue(maxsize=100)
queue.send_audio(b"\x00\x00" * 1600, sample_rate=16000)  # 100 ms of PCM16 silence
queue.send_text("Hello")
queue.send_activity_end()  # Manual VAD only: mark end of user turn
queue.close()


async def drain() -> None:
    async for frame in queue:  # yields LiveInput frames until the close sentinel
        print(frame.kind, frame.sample_rate)

LiveInputQueue(maxsize: int = 0)

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

from tenxgraph.core.realtime import RealtimeConfig, VADConfig

config = RealtimeConfig(
    model="gemini-2.5-flash",
    response_modalities=["AUDIO"],
    voice="Puck",
    vad=VADConfig(enabled=True),
)

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,
)

from tenxgraph.core.realtime import VADConfig

vad = VADConfig(
    enabled=False,  # Manual VAD
    start_sensitivity="HIGH",
)

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,
)

from tenxgraph.core.realtime import ReconnectConfig

reconnect = ReconnectConfig(
    base_delay=1.0,
    max_delay=30.0,
    max_attempts=10,
)

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

from tenxgraph.core.realtime import RealtimeEvent

async for event in app.arealtime(queue):
    if event.type == "audio_delta":
        speaker.write(event.data)
    elif event.type == "input_transcript":
        print(f"User: {event.text}" if event.finished else f"User (partial): {event.text}")
    elif event.type == "error":
        if event.fatal:
            break

type: Literal["audio_delta"]
data: bytes        # PCM16 chunk
sample_rate: int   # 24000

type: Literal["input_transcript"]
text: str
finished: bool     # True on final chunk

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

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

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

type: Literal["turn_complete"]

type: Literal["interrupted"]

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

type: Literal["go_away"]
time_left: str | None    # Provider duration string, e.g. "5s"

type: Literal["agent_changed"]
author: str

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

from tenxgraph.core.realtime import RealtimeClient

from tenxgraph.core.realtime import GeminiLiveClient, normalize_message

from tenxgraph.prebuilt.agent import AudioAgent


# Inject a custom client per connection (for tests or pre-built credentials).
def client_factory() -> GeminiLiveClient:
    return GeminiLiveClient(api_key="YOUR_API_KEY")  # placeholder, load from your secret store


app = AudioAgent("gemini-2.5-flash", realtime_client_factory=client_factory).compile()

from tenxgraph.utils.callbacks import CallbackManager, GraphLifecycleHook

class MyHook(GraphLifecycleHook):
    async def on_graph_start(self, context, state):
        print("Session started")
        return None  # None keeps the current state

    async def on_turn_end(self, context, state, turn_index):
        print(f"Turn {turn_index} complete")
        return None

cb = CallbackManager()
cb.register_lifecycle_hook(MyHook())
app = AudioAgent("gemini-2.5-flash").compile(callback_manager=cb)