0
# Live API
1

2
Real-time bidirectional streaming for interactive applications with support for audio, video, and function calling (Preview). The Live API enables continuous, low-latency communication with the model for conversational AI, voice assistants, and real-time collaboration tools.
3

4
## Capabilities
5

6
### Connect to Live Session
7

8
Establish a bidirectional streaming connection with the model. Returns an async context manager that yields a session for sending and receiving messages.
9

10
```python { .api }
11
class AsyncLive:
12
    """Asynchronous Live API (Preview)."""
13

14
    async def connect(
15
        self,
16
        *,
17
        model: str,
18
        config: Optional[LiveConnectConfig] = None
19
    ) -> AsyncIterator[AsyncSession]:
20
        """
21
        Connect to live server (async context manager).
22

23
        Parameters:
24
            model (str): Model identifier (e.g., 'gemini-2.0-flash-exp').
25
            config (LiveConnectConfig, optional): Connection configuration including:
26
                - system_instruction: System instruction for the session
27
                - generation_config: Generation parameters
28
                - tools: Function declarations for function calling
29
                - speech_config: Speech synthesis configuration
30

31
        Yields:
32
            AsyncSession: Active session for bidirectional communication.
33

34
        Raises:
35
            ClientError: For client errors
36
            ServerError: For server errors
37

38
        Usage:
39
            async with client.aio.live.connect(model='gemini-2.0-flash-exp') as session:
40
                # Send and receive messages
41
                pass
42
        """
43
        ...
44
```
45

46
**Usage Example - Basic Live Session:**
47

48
```python
49
import asyncio
50
from google.genai import Client
51

52
async def live_session():
53
    client = Client(api_key='YOUR_API_KEY')
54

55
    async with client.aio.live.connect(
56
        model='gemini-2.0-flash-exp'
57
    ) as session:
58
        # Send message
59
        await session.send_client_content(
60
            turns=[Content(parts=[Part(text='Hello, how are you?')])],
61
            turn_complete=True
62
        )
63

64
        # Receive responses
65
        async for message in session.receive():
66
            if message.server_content:
67
                if message.server_content.model_turn:
68
                    for part in message.server_content.model_turn.parts:
69
                        if part.text:
70
                            print(part.text, end='', flush=True)
71
                if message.server_content.turn_complete:
72
                    print()  # New line after turn completes
73
                    break
74

75
asyncio.run(live_session())
76
```
77

78
### Send Client Content
79

80
Send conversational content (text, images, etc.) to the model during a live session.
81

82
```python { .api }
83
class AsyncSession:
84
    """Live session for bidirectional streaming (Preview)."""
85

86
    async def send_client_content(
87
        self,
88
        *,
89
        turns: Optional[Union[Content, list[Content]]] = None,
90
        turn_complete: bool = False
91
    ) -> None:
92
        """
93
        Send client content to the model.
94

95
        Parameters:
96
            turns (Union[Content, list[Content]], optional): Content to send. Can be
97
                text, images, or other supported content types.
98
            turn_complete (bool): Whether this completes the current turn. If True,
99
                model will begin generating response. Default: False.
100

101
        Raises:
102
            RuntimeError: If session is closed or invalid state
103
        """
104
        ...
105
```
106

107
### Send Realtime Input
108

109
Send realtime media chunks (audio, video) for streaming input.
110

111
```python { .api }
112
class AsyncSession:
113
    """Live session for bidirectional streaming (Preview)."""
114

115
    async def send_realtime_input(
116
        self,
117
        *,
118
        media_chunks: Optional[Sequence[Blob]] = None
119
    ) -> None:
120
        """
121
        Send realtime media input (audio/video chunks).
122

123
        Parameters:
124
            media_chunks (Sequence[Blob], optional): Media chunks to send. Each Blob
125
                contains binary data with MIME type (e.g., 'audio/pcm' for raw audio).
126

127
        Raises:
128
            RuntimeError: If session is closed or invalid state
129
        """
130
        ...
131
```
132

133
**Usage Example - Audio Streaming:**
134

135
```python
136
import asyncio
137
from google.genai import Client
138
from google.genai.types import Blob
139

140
async def stream_audio():
141
    client = Client(api_key='YOUR_API_KEY')
142

143
    async with client.aio.live.connect(
144
        model='gemini-2.0-flash-exp'
145
    ) as session:
146
        # Stream audio chunks
147
        with open('audio.pcm', 'rb') as f:
148
            while chunk := f.read(4096):
149
                await session.send_realtime_input(
150
                    media_chunks=[Blob(
151
                        mime_type='audio/pcm',
152
                        data=chunk
153
                    )]
154
                )
155

156
        # Receive responses
157
        async for message in session.receive():
158
            # Process responses
159
            pass
160

161
asyncio.run(stream_audio())
162
```
163

164
### Send Tool Response
165

166
Send function execution results back to the model after it requests a function call.
167

168
```python { .api }
169
class AsyncSession:
170
    """Live session for bidirectional streaming (Preview)."""
171

172
    async def send_tool_response(
173
        self,
174
        *,
175
        function_responses: Sequence[FunctionResponse]
176
    ) -> None:
177
        """
178
        Send tool/function execution responses.
179

180
        Parameters:
181
            function_responses (Sequence[FunctionResponse]): Results from executing
182
                functions that the model requested.
183

184
        Raises:
185
            RuntimeError: If session is closed or invalid state
186
        """
187
        ...
188
```
189

190
### Receive Messages
191

192
Receive streaming messages from the model including content, function calls, and metadata.
193

194
```python { .api }
195
class AsyncSession:
196
    """Live session for bidirectional streaming (Preview)."""
197

198
    async def receive(self) -> AsyncIterator[LiveServerMessage]:
199
        """
200
        Receive messages from the model as an async iterator.
201

202
        Yields:
203
            LiveServerMessage: Server messages including:
204
                - setup_complete: Connection setup confirmation
205
                - server_content: Generated content from model
206
                - tool_call: Function call requests from model
207
                - tool_call_cancellation: Cancelled function calls
208

209
        Raises:
210
            RuntimeError: If session is closed
211
        """
212
        ...
213
```
214

215
### Close Session
216

217
Explicitly close the live session.
218

219
```python { .api }
220
class AsyncSession:
221
    """Live session for bidirectional streaming (Preview)."""
222

223
    async def close(self) -> None:
224
        """
225
        Close the live session.
226

227
        Note: Sessions are automatically closed when exiting the context manager.
228
        """
229
        ...
230
```
231

232
**Usage Example - Function Calling:**
233

234
```python
235
import asyncio
236
from google.genai import Client
237
from google.genai.types import (
238
    LiveConnectConfig,
239
    Tool,
240
    FunctionDeclaration,
241
    Schema,
242
    Type,
243
    Content,
244
    Part,
245
    FunctionResponse
246
)
247

248
async def live_with_functions():
249
    client = Client(api_key='YOUR_API_KEY')
250

251
    # Define function
252
    weather_func = FunctionDeclaration(
253
        name='get_weather',
254
        description='Get weather for a location',
255
        parameters=Schema(
256
            type=Type.OBJECT,
257
            properties={
258
                'location': Schema(type=Type.STRING)
259
            }
260
        )
261
    )
262

263
    config = LiveConnectConfig(
264
        tools=[Tool(function_declarations=[weather_func])]
265
    )
266

267
    async with client.aio.live.connect(
268
        model='gemini-2.0-flash-exp',
269
        config=config
270
    ) as session:
271
        # Ask for weather
272
        await session.send_client_content(
273
            turns=[Content(parts=[Part(text='What is the weather in Tokyo?')])],
274
            turn_complete=True
275
        )
276

277
        # Handle responses
278
        async for message in session.receive():
279
            if message.tool_call:
280
                # Execute function
281
                if message.tool_call.function_calls:
282
                    for fc in message.tool_call.function_calls:
283
                        if fc.name == 'get_weather':
284
                            # Execute and return result
285
                            await session.send_tool_response(
286
                                function_responses=[FunctionResponse(
287
                                    name='get_weather',
288
                                    response={'temperature': 22, 'condition': 'sunny'},
289
                                    id=fc.id
290
                                )]
291
                            )
292

293
            if message.server_content:
294
                if message.server_content.model_turn:
295
                    for part in message.server_content.model_turn.parts:
296
                        if part.text:
297
                            print(part.text)
298
                if message.server_content.turn_complete:
299
                    break
300

301
asyncio.run(live_with_functions())
302
```
303

304
## Types
305

306
```python { .api }
307
from typing import Optional, Union, Sequence, AsyncIterator, Any
308
from enum import Enum
309

310
# Configuration types
311
class LiveConnectConfig:
312
    """
313
    Configuration for live connection.
314

315
    Attributes:
316
        system_instruction (Union[str, Content], optional): System instruction.
317
        generation_config (GenerationConfig, optional): Generation parameters.
318
        tools (list[Tool], optional): Function declarations.
319
        speech_config (SpeechConfig, optional): Speech synthesis configuration.
320
        response_modalities (list[str], optional): Desired response modalities ('TEXT', 'AUDIO').
321
    """
322
    system_instruction: Optional[Union[str, Content]] = None
323
    generation_config: Optional[GenerationConfig] = None
324
    tools: Optional[list[Tool]] = None
325
    speech_config: Optional[SpeechConfig] = None
326
    response_modalities: Optional[list[str]] = None
327

328
class SpeechConfig:
329
    """
330
    Speech synthesis configuration.
331

332
    Attributes:
333
        voice_config (VoiceConfig, optional): Voice settings.
334
    """
335
    voice_config: Optional[VoiceConfig] = None
336

337
class VoiceConfig:
338
    """
339
    Voice configuration.
340

341
    Attributes:
342
        prebuilt_voice_config (PrebuiltVoiceConfig, optional): Prebuilt voice.
343
    """
344
    prebuilt_voice_config: Optional[PrebuiltVoiceConfig] = None
345

346
class PrebuiltVoiceConfig:
347
    """
348
    Prebuilt voice settings.
349

350
    Attributes:
351
        voice_name (str, optional): Voice name (e.g., 'Aoede', 'Charon').
352
    """
353
    voice_name: Optional[str] = None
354

355
# Message types
356
class LiveServerMessage:
357
    """
358
    Message from server in live session.
359

360
    Attributes:
361
        setup_complete (LiveServerSetupComplete, optional): Setup confirmation.
362
        server_content (LiveServerContent, optional): Generated content.
363
        tool_call (LiveServerToolCall, optional): Function call request.
364
        tool_call_cancellation (LiveServerToolCallCancellation, optional): Cancelled call.
365
    """
366
    setup_complete: Optional[LiveServerSetupComplete] = None
367
    server_content: Optional[LiveServerContent] = None
368
    tool_call: Optional[LiveServerToolCall] = None
369
    tool_call_cancellation: Optional[LiveServerToolCallCancellation] = None
370

371
class LiveServerSetupComplete:
372
    """Setup complete confirmation."""
373
    pass
374

375
class LiveServerContent:
376
    """
377
    Server content message.
378

379
    Attributes:
380
        model_turn (Content, optional): Model's turn content.
381
        turn_complete (bool, optional): Whether turn is complete.
382
        interrupted (bool, optional): Whether turn was interrupted.
383
        grounding_metadata (GroundingMetadata, optional): Grounding information.
384
    """
385
    model_turn: Optional[Content] = None
386
    turn_complete: Optional[bool] = None
387
    interrupted: Optional[bool] = None
388
    grounding_metadata: Optional[GroundingMetadata] = None
389

390
class LiveServerToolCall:
391
    """
392
    Tool call request from server.
393

394
    Attributes:
395
        function_calls (list[FunctionCall], optional): Requested function calls.
396
    """
397
    function_calls: Optional[list[FunctionCall]] = None
398

399
class LiveServerToolCallCancellation:
400
    """
401
    Tool call cancellation.
402

403
    Attributes:
404
        ids (list[str], optional): IDs of cancelled calls.
405
    """
406
    ids: Optional[list[str]] = None
407

408
# Core types (shared with other capabilities)
409
class Content:
410
    """Content container."""
411
    parts: list[Part]
412
    role: Optional[str] = None
413

414
class Part:
415
    """Content part."""
416
    text: Optional[str] = None
417
    inline_data: Optional[Blob] = None
418
    function_call: Optional[FunctionCall] = None
419
    function_response: Optional[FunctionResponse] = None
420

421
class Blob:
422
    """Binary data."""
423
    mime_type: str
424
    data: bytes
425

426
class FunctionCall:
427
    """Function call from model."""
428
    name: str
429
    args: dict[str, Any]
430
    id: Optional[str] = None
431

432
class FunctionResponse:
433
    """Function execution result."""
434
    name: str
435
    response: dict[str, Any]
436
    id: Optional[str] = None
437

438
class Tool:
439
    """Tool with function declarations."""
440
    function_declarations: Optional[list[FunctionDeclaration]] = None
441

442
class FunctionDeclaration:
443
    """Function definition."""
444
    name: str
445
    description: str
446
    parameters: Optional[Schema] = None
447

448
class Schema:
449
    """JSON schema."""
450
    type: Type
451
    properties: Optional[dict[str, Schema]] = None
452

453
class Type(Enum):
454
    """JSON schema types."""
455
    OBJECT = 'OBJECT'
456
    STRING = 'STRING'
457
    NUMBER = 'NUMBER'
458
    INTEGER = 'INTEGER'
459
    BOOLEAN = 'BOOLEAN'
460
    ARRAY = 'ARRAY'
461

462
class GenerationConfig:
463
    """Generation configuration."""
464
    temperature: Optional[float] = None
465
    top_p: Optional[float] = None
466
    top_k: Optional[int] = None
467
    max_output_tokens: Optional[int] = None
468

469
class GroundingMetadata:
470
    """Grounding metadata."""
471
    pass
472
```
473