# Make 2 ai vtubers talk to eachother in vtube studio With functioning token System For mouths (working code example for everyone cause i could not find this anywhere) > Source: > Published: 2026-07-07 18:46:08+00:00 ali: VTS_PORT = 8001 MOUTH_PARAM = “MouthOpen” # VTuber mouth parameter TOKEN_FILE = “vts_token.json” class VTubeStudioClient: ``` def \__init_\_(self): self.url = f"ws://localhost:{VTS_PORT}" self.authenticated = False self.token = self.load_token() self.\_auth_requested = False # Prevent repeated auth requests self.\_connect() def \_connect(self): clean_print(f"\[cyan\]Connecting to VTube Studio on {self.url}...\[/cyan\]") self.ws = websocket.WebSocketApp( self.url, on_open=self.on_open, on_message=self.on_message, on_error=self.on_error, on_close=self.on_close ) self.thread = threading.Thread(target=self.ws.run_forever, daemon=True) self.thread.start() def on_open(self, ws): clean_print("\[green\]Connected to VTube Studio API\[/green\]") time.sleep(1) self.\_auth_requested = False # Reset on new connection if self.token: self.send({ "apiName": "VTubeStudioPublicAPI", "apiVersion": "1.0", "requestID": "auth", "messageType": "AuthenticationRequest", "data": { "pluginName": "Local AI VTuber", "pluginDeveloper": "Bro77xp", "authenticationToken": self.token } }) elif not self.\_auth_requested: self.\_auth_requested = True self.send({ "apiName": "VTubeStudioPublicAPI", "apiVersion": "1.0", "requestID": "auth_token", "messageType": "AuthenticationTokenRequest", "data": { "pluginName": "Local AI VTuber", "pluginDeveloper": "Bro77xp", "pluginIcon": "" } }) def on_message(self, ws, message): try: msg = json.loads(message) mtype = msg.get("messageType", "") if mtype == "AuthenticationTokenResponse": token = msg\["data"\]\["authenticationToken"\] self.save_token(token) clean_print("\[green\]✅ Token received and saved. Please restart the script.\[/green\]") self.authenticated = True elif mtype == "AuthenticationResponse": self.authenticated = True clean_print("\[green\]Authenticated with VTube Studio!\[/green\]") except Exception as e: clean_print(f"\[red\]Error parsing message:\[/red\] {e}") def on_error(self, ws, error): clean_print(f"\[red\]VTS WebSocket error:\[/red\] {error}") def on_close(self, ws, close_status_code, close_msg): now = time.time() if not hasattr(self, '\_last_close_log') or now - self.\_last_close_log > 30: self.\_last_close_log = now clean_print("\[yellow\]VTS connection closed. Retrying in 5 seconds...\[/yellow\]") time.sleep(5) self.\_connect() def send(self, payload): try: if self.ws and self.ws.sock and self.ws.sock.connected: self.ws.send(json.dumps(payload)) else: now = time.time() if not hasattr(self, '\_last_not_connected_log') or now - self.\_last_not_connected_log > 30: self.\_last_not_connected_log = now clean_print("\[yellow\]VTS not connected; could not send message (suppressing for 30s)\[/yellow\]") except Exception as e: clean_print(f"\[red\]Error sending to VTS:\[/red\] {e}") def trigger_hotkey(self, hotkey_id, transition_duration: float = 2.0): """Trigger a hotkey with optional transition duration (seconds) for smooth animation fade.""" self.send({ "apiName": "VTubeStudioPublicAPI", "apiVersion": "1.0", "requestID": f"hotkey\_{hotkey_id}", "messageType": "HotkeyTriggerRequest", "data": { "hotkeyID": hotkey_id, "transitionDuration": transition_duration } }) def inject_parameters(self, params: dict, request_id: str = "inject"): """Inject multiple Live2D parameters in one request. params: {param_id: value, ...} e.g. {"FaceAngleX": 15.0, "EyeOpenLeft": 0.5} """ values = \[{"id": k, "value": float(v)} for k, v in params.items()\] self.send({ "apiName": "VTubeStudioPublicAPI", "apiVersion": "1.0", "requestID": request_id, "messageType": "InjectParameterDataRequest", "data": {"parameterValues": values} }) def set_mouth(self, value): self.send({ "apiName": "VTubeStudioPublicAPI", "apiVersion": "1.0", "requestID": "mouth", "messageType": "InjectParameterDataRequest", "data": { "parameterValues": \[{ "id": MOUTH_PARAM, "value": float(value) }\] } }) def save_token(self, token): with open(TOKEN_FILE, "w") as f: json.dump({"token": token}, f) def load_token(self): if os.path.exists(TOKEN_FILE): with open(TOKEN_FILE, "r") as f: return json.load(f).get("token") return None ``` # Initialize VTS client vtube = VTubeStudioClient() # Idle animation engine - autonomous head/eye/brow movement when not speaking idle_anim = IdleAnimationEngine(vtube, lambda: tts_active or tts_generating or tts_playing) hara: VTS_PORT = 8003 MOUTH_PARAM = “MouthOpen” TOKEN_FILE = “vts_token1.json” class VTubeStudioClient: ``` API_NAME = "VTubeStudioPublicAPI" API_VERSION = "1.0" def \__init_\_(self): self.url = f"ws://localhost:{VTS_PORT}" self.ws = None self.thread = None self.authenticated = False self.token = self.load_token() self.hotkeys = {} self.\_last_mouth = -1.0 self.\_last_mouth_time = 0.0 self.\_connect() try: print("Authenticated:", self.authenticated) except Exception: pass def \_connect(self): console.print(f"\[cyan\]Connecting to VTube Studio on {self.url}...\[/cyan\]") self.ws = websocket.WebSocketApp( self.url, on_open=self.on_open, on_message=self.on_message, on_error=self.on_error, on_close=self.on_close ) self.thread = threading.Thread(target=self.ws.run_forever, daemon=True) self.thread.start() def on_open(self, ws): console.print("\[green\]Connected to VTube Studio API\[/green\]") time.sleep(1) if self.token: console.print("\[cyan\]Attempting to authenticate with stored token...\[/cyan\]") self.send_request( "auth", "AuthenticationRequest", { "pluginName": "Local AI VT3.0", "pluginDeveloper": "Bro77xp", "authenticationToken": self.token } ) else: console.print("\[cyan\]No stored token found. Requesting authentication token...\[/cyan\]") self.send_request( "auth_token", "AuthenticationTokenRequest", { "pluginName": "Local AI VT3.0", "pluginDeveloper": "Bro77xp", "pluginIcon": "" } ) def on_message(self, ws, message): try: msg = json.loads(message) mtype = msg.get("messageType", "") if mtype == "AuthenticationTokenResponse": token = msg\["data"\]\["authenticationToken"\] self.save_token(token) console.print("\[green\]✅ Token received and saved. Authenticating...\[/green\]") self.send_request( "auth", "AuthenticationRequest", { "pluginName": "Local AI VT3.0", "pluginDeveloper": "Bro77xp", "authenticationToken": token } ) elif mtype == "AuthenticationResponse": self.authenticated = True console.print("\[green\]Authenticated with VTube Studio!\[/green\]") except Exception as e: console.print(f"\[red\]Error parsing message:\[/red\] {e}") def on_error(self, ws, error): console.print(f"\[red\]VTS WebSocket error:\[/red\] {error}") def on_close(self, ws, close_status_code, close_msg): console.print("\[yellow\]VTS connection closed. Retrying in 5 seconds...\[/yellow\]") time.sleep(5) self.\_connect() def save_token(self, token): with open(TOKEN_FILE, "w") as f: json.dump({"token": token}, f) def send_request( self, request_id: str, message_type: str, data: dict | None = None ): if not self.is_connected(): return False payload = { "apiName": self.API_NAME, "apiVersion": self.API_VERSION, "requestID": request_id, "messageType": message_type, } if data: payload\["data"\] = data try: self.ws.send(json.dumps(payload)) return True except Exception as e: console.print(f"\[red\]VTS Send Error:\[/red\] {e}") return False def is_connected(self): return ( self.ws and self.ws.sock and self.ws.sock.connected ) def load_token(self): try: if not os.path.exists(TOKEN_FILE): return None with open(TOKEN_FILE, "r", encoding="utf-8") as f: data = json.load(f) return data.get("token") or data.get("authenticationToken") except Exception as e: console.print(f"\[yellow\]Could not load token: {e}\[/yellow\]") return None ```