# example requires websocket-client library: # pip install websocket-client import os import json import websocket url = "ws://api.digislash.com/v1/realtime?model=gpt-4o-realtime-preview" headers = [ "Authorization: Bearer sk-xxx" , "OpenAI-Beta: realtime=v1" ] def decode_unicode_string(s): """Decode Unicode so non-ASCII text is readable""" try: if isinstance(s, bytes): return s.decode('utf-8') return s.encode('latin1').decode('utf-8') except Exception as e: print(f"Decode failed: {e}") return s def on_error(ws, error): print("\n=== Error ===") print(f"{error}") def on_close(ws, close_status_code, close_msg): print("\n=== Connection closed ===") print(f"Status: {close_status_code}") print(f"Message: {close_msg}") def on_open(ws): print("\n=== Connected ===") event = { "type": "response.create", "response": { "modalities": ["text"], "instructions": "Please assist the user." } } print("\n=== Sending ===") print(json.dumps(event, ensure_ascii=False, indent=2)) ws.send(json.dumps(event)) def on_message(ws, message): print("\n=== Message received ===") try: data = json.loads(message) if 'error' in data and 'message' in data['error']: data['error']['message'] = decode_unicode_string(data['error']['message']) print(json.dumps(data, ensure_ascii=False, indent=2)) if data.get('type') == 'error': print("\n=== Error summary ===") print(f"Type: {data['error'].get('type', 'unknown')}") print(f"Code: {data['error'].get('code', 'unknown')}") error_message = decode_unicode_string(data['error'].get('message', 'unknown')) print(f"Message: {error_message}") except json.JSONDecodeError as e: print(f"Failed to parse message: {message}") print(f"Error: {e}") # Enable websocket logging only for key info websocket.enableTrace(False) ws = websocket.WebSocketApp( url, header=headers, on_open=on_open, on_message=on_message, on_error=on_error, on_close=on_close ) print("\n=== Connecting ===") print(f"URL: {url}") ws.run_forever()
json
