Empezar a usar la API Live con WebSockets

En este tutorial se muestra cómo conectarse a la API Live mediante WebSockets. En este tutorial, configurarás un proyecto Google Cloud para usar la API 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 WebSockets.

Antes de empezar

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

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

Para obtener instrucciones sobre cómo configurar la autenticación con ADC, consulta nuestra guía de inicio rápido.

Instalar la biblioteca WebSockets

Ejecuta lo siguiente para instalar la biblioteca websockets:

pip install websockets

Configurar variables de entorno

Define las variables de entorno del ID y la ubicación de tu proyecto. Sustituye PROJECT_ID por elGoogle Cloud ID de tu proyecto.

export GOOGLE_CLOUD_PROJECT=PROJECT_ID
export GOOGLE_CLOUD_LOCATION=global

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

Siguientes pasos