Very simple app listening to Voxta websocket

app.py

import asyncio
import json
import base64
import aiohttp
import os
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Authentication (optional)
AUTH = os.getenv('AUTH')
AUTH_B64 = base64.b64encode(AUTH.encode('utf-8')).decode() if AUTH else None

# WebSocket hostname
WS_HOST = os.getenv('WS_HOST')
if not WS_HOST:
    print("Error: Environment variable WS_HOST is not defined")
    exit(1)

WS_URL = f"wss://{WS_HOST}/hub"

RECORD_SEPARATOR = '\x1e'

async def send_message(websocket, message):
    """Send a message to the WebSocket"""
    await websocket.send_str(json.dumps(message) + RECORD_SEPARATOR)

async def connect_to_websocket():
    """Connect to WebSocket and handle communication"""
    headers = {}
    if AUTH_B64:
        headers["Authorization"] = f"Basic {AUTH_B64}"
    
    async with aiohttp.ClientSession() as session:
        async with session.ws_connect(WS_URL, headers=headers) as websocket:
            print("Connected to WebSocket")
            
            # Initial handshake
            await send_message(websocket, {"protocol": "json", "version": 1})
            await websocket.receive()  # Wait for handshake response
            
            # Authentication
            auth_msg = {
                "arguments": [{
                    "$type": "authenticate",
                    "client": "SimpleClient",
                    "clientVersion": "1.0",
                    "scope": ["role:app"],
                    "capabilities": {"audioInput": "None", "audioOutput": "None"}
                }],
                "target": "SendMessage",
                "type": 1
            }
            await send_message(websocket, auth_msg)
            
            try:
                while True:
                    msg = await websocket.receive()
                    
                    if msg.type in (aiohttp.WSMsgType.CLOSE, aiohttp.WSMsgType.CLOSED):
                        break
                        
                    if msg.type == aiohttp.WSMsgType.TEXT:
                        try:
                            data = json.loads(msg.data)
                            print(f"Received: {json.dumps(data, indent=2)}")
                        except json.JSONDecodeError:
                            print(f"Raw message: {msg.data}")
            finally:
                pass

if __name__ == "__main__":
    asyncio.run(connect_to_websocket())