Implementation:Googleapis Python genai AsyncLive Connect

Knowledge Sources	Googleapis_Python_genai Google GenAI SDK Docs
Domains	Real_Time_AI, Streaming, WebSocket
Last Updated	2026-02-15 14:00 GMT

Overview

Concrete tool for real-time bidirectional streaming communication with Gemini models provided by the Google Gen AI SDK.

Description

The AsyncLive class provides the Live API module for establishing WebSocket-based streaming sessions with Gemini. The AsyncSession class manages individual sessions, supporting text/audio/video input streaming, function calling, and real-time model responses. Only async operation is supported due to the WebSocket-based nature of the protocol.

Usage

Import this when building real-time conversational AI applications that need continuous bidirectional streaming, such as voice assistants, live captioning, or interactive AI agents that process audio/video in real time.

Code Reference

Source Location

• Repository: Googleapis_Python_genai
• File: google/genai/live.py
• Lines: 87-1193 (AsyncSession at L87-887, AsyncLive at L889-1193)

Signature

class AsyncLive(_api_module.BaseModule):
    @asynccontextmanager
    async def connect(
        self,
        *,
        model: str,
        config: Optional[types.LiveConnectConfigOrDict] = None,
    ) -> AsyncIterator[AsyncSession]: ...

class AsyncSession:
    async def send_client_content(
        self,
        *,
        turns: Optional[Union[types.Content, list[types.Content]]] = None,
        turn_complete: bool = True,
    ) -> None: ...

    async def send_realtime_input(
        self,
        *,
        media: Optional[types.BlobImageUnionDict] = None,
        audio: Optional[types.BlobOrDict] = None,
        video: Optional[types.BlobImageUnionDict] = None,
        text: Optional[str] = None,
    ) -> None: ...

    async def send_tool_response(
        self,
        *,
        function_responses: Union[
            types.FunctionResponseOrDict,
            Sequence[types.FunctionResponseOrDict],
        ],
    ) -> None: ...

    async def receive(self) -> AsyncIterator[types.LiveServerMessage]: ...
    async def close(self) -> None: ...

Import

from google import genai
client = genai.Client(api_key="...")
# Access via: client.aio.live

I/O Contract

Inputs

Name	Type	Required	Description
model	str	Yes	Model ID for live session (e.g. "gemini-2.0-flash-live")
config	LiveConnectConfigOrDict	No	Session configuration (tools, system instruction, generation config)
turns	Content or list[Content]	No	Client content to send to model
media/audio/video	BlobOrDict	No	Real-time media input
function_responses	FunctionResponseOrDict	For tool_response	Function execution results

Outputs

Name	Type	Description
connect() yields	AsyncSession	Live session context manager
receive() yields	LiveServerMessage	Server messages (text, audio, tool calls)

Usage Examples

Basic Text Chat

import asyncio
from google import genai

client = genai.Client(api_key="GEMINI_API_KEY")

async def main():
    async with client.aio.live.connect(model="gemini-2.0-flash-live") as session:
        await session.send_client_content(
            turns=genai.types.Content(
                role="user",
                parts=[genai.types.Part(text="Hello, how are you?")]
            )
        )

        async for message in session.receive():
            if message.server_content and message.server_content.model_turn:
                for part in message.server_content.model_turn.parts:
                    if part.text:
                        print(part.text, end="")

asyncio.run(main())

Related Pages

• Environment:Googleapis_Python_genai_Python_3_10_SDK_Runtime