Comienza a usar la API de Live con WebSockets

En este instructivo, se muestra cómo conectarse a la API de Live con WebSockets. En este instructivo, configurarás un proyecto Google Cloud para usar la API de Live con WebSockets, enviarás un archivo de audio al modelo y recibirás audio como respuesta.

Para obtener más información sobre WebSockets, consulta la documentación de la API de WebSockets.

Antes de comenzar

Antes de enviar solicitudes, debes configurar la autenticación con Vertex AI. Puedes configurar la autenticación con una clave de API o con credenciales predeterminadas de la aplicación (ADC).

En este instructivo, la forma más rápida de comenzar es usar una clave de API:

Si deseas obtener instrucciones para configurar la autenticación con las ADC, consulta nuestra guía de inicio rápido.

Instala la biblioteca de WebSockets

Ejecuta el siguiente comando para instalar la biblioteca de websockets:

pip install websockets

Configura variables de entorno

Establece variables de entorno para el ID y la ubicación de tu proyecto. Reemplaza PROJECT_ID por el ID del proyecto deGoogle Cloud .

export GOOGLE_CLOUD_PROJECT=PROJECT_ID
export GOOGLE_CLOUD_LOCATION=global

Cómo iniciar una sesión de audio

En el siguiente ejemplo, se establece una sesión, se transmite audio desde un archivo y se imprime el tamaño de los fragmentos de audio recibidos en la respuesta.

import asyncio
import websockets
import json
import base64
import os
import sys

# Replace the [PROJECT_ID] and [LOCATION] with your Google Cloud Project ID and location.
PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT")
LOCATION = os.environ.get("GOOGLE_CLOUD_LOCATION")

# Authentication
token_list = !gcloud auth application-default print-access-token
ACCESS_TOKEN = token_list[0]

# Configuration
MODEL_ID = "gemini-live-2.5-flash-preview-native-audio-09-2025"

# Construct the WSS URL
HOST = f"{LOCATION}-aiplatform.googleapis.com"
path = "google.cloud.aiplatform.v1.LlmBidiService/BidiGenerateContent"
URI = f"wss://{HOST}/ws/{path}"
MODEL_RESOURCE = f"projects/{PROJECT_ID}/locations/{LOCATION}/publishers/google/models/{MODEL_ID}"


async def main():
   headers = {"Authorization": f"Bearer {ACCESS_TOKEN}"}
  
   async with websockets.connect(URI, additional_headers=headers) as ws:
       print("Session established.")

       # Send Setup (Handshake)
       await ws.send(json.dumps({
           "setup": {
               "model": MODEL_RESOURCE,
               "generation_config": { "response_modalities": ["AUDIO"] }
           }
       }))

       # Define Tasks
       async def send_audio():
           # Download sample if missing
           if not os.path.exists("input.wav"):
               !wget -q https://storage.googleapis.com/cloud-samples-data/generative-ai/audio/where_the_nearest_train_station_is.wav -O input.wav

           with open("input.wav", "rb") as f:
               while chunk := f.read(1024):
                   msg = {
                       "realtime_input": {
                           "media_chunks": [{
                               "mime_type": "audio/pcm;rate=16000",
                               "data": base64.b64encode(chunk).decode("utf-8")
                           }]
                       }
                   }
                   await ws.send(json.dumps(msg))
                   await asyncio.sleep(0.01)
           print("Done sending audio.")

       async def receive_audio():
           async for msg in ws:
               data = json.loads(msg)
               try:
                   parts = data["serverContent"]["modelTurn"]["parts"]
                   for part in parts:
                       if "inlineData" in part:
                           b64_audio = part["inlineData"]["data"]
                           print(f"Received chunk: {len(b64_audio)} bytes")
               except KeyError:
                   pass

               if data.get("serverContent", {}).get("turnComplete"):
                   print("Turn complete. Exiting.")
                   break

       # Run Together
       await asyncio.gather(send_audio(), receive_audio())

if __name__ == "__main__":
   # Check if running in Jupyter/Colab
   if "ipykernel" in sys.modules:
       await main()
   else:
       asyncio.run(main())

¿Qué sigue?