AbletonMCP

"""Ableton Live connection management.""" from dataclasses import dataclass import json import socket from typing import Any, Dict from ..utils.logging import get_logger logger = get_logger("AbletonMCPServer") @dataclass class AbletonConnection: """Manages connection to Ableton Live Remote Script via socket.""" host: str port: int sock: socket.socket = None def connect(self) -> bool: """Connect to the Ableton Remote Script socket server""" if self.sock: return True try: self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.connect((self.host, self.port)) logger.info(f"Connected to Ableton at {self.host}:{self.port}") return True except Exception as e: logger.error(f"Failed to connect to Ableton: {str(e)}") self.sock = None return False def disconnect(self) -> None: """Disconnect from the Ableton Remote Script""" if self.sock: try: self.sock.close() except Exception as e: logger.error(f"Error disconnecting from Ableton: {str(e)}") finally: self.sock = None def receive_full_response(self, sock: socket.socket, buffer_size: int = 8192) -> bytes: """Receive the complete response, potentially in multiple chunks""" chunks = [] sock.settimeout(15.0) # Increased timeout for operations that might take longer try: while True: try: chunk = sock.recv(buffer_size) if not chunk: if not chunks: raise Exception("Connection closed before receiving any data") break chunks.append(chunk) # Check if we've received a complete JSON object try: data = b"".join(chunks) json.loads(data.decode("utf-8")) logger.info(f"Received complete response ({len(data)} bytes)") return data except json.JSONDecodeError: # Incomplete JSON, continue receiving continue except TimeoutError: logger.warning("Socket timeout during chunked receive") break except (ConnectionError, BrokenPipeError, ConnectionResetError) as e: logger.error(f"Socket connection error during receive: {str(e)}") raise except Exception as e: logger.error(f"Error during receive: {str(e)}") raise # If we get here, we either timed out or broke out of the loop if chunks: data = b"".join(chunks) logger.info(f"Returning data after receive completion ({len(data)} bytes)") try: json.loads(data.decode("utf-8")) return data except json.JSONDecodeError: raise Exception("Incomplete JSON response received") else: raise Exception("No data received") def send_command(self, command_type: str, params: dict[str, Any] = None) -> dict[str, Any]: """Send a command to Ableton and return the response""" if not self.sock and not self.connect(): raise ConnectionError("Not connected to Ableton") command = {"type": command_type, "params": params or {}} # Check if this is a state-modifying command is_modifying_command = command_type in [ "create_midi_track", "create_audio_track", "set_track_name", "create_clip", "add_notes_to_clip", "set_clip_name", "set_tempo", "fire_clip", "stop_clip", "set_device_parameter", "start_playback", "stop_playback", "load_instrument_or_effect", ] try: logger.info(f"Sending command: {command_type} with params: {params}") # Send the command self.sock.sendall(json.dumps(command).encode("utf-8")) logger.info("Command sent, waiting for response...") # For state-modifying commands, add a small delay to give Ableton time to process if is_modifying_command: import time time.sleep(0.1) # 100ms delay # Set timeout based on command type timeout = 15.0 if is_modifying_command else 10.0 self.sock.settimeout(timeout) # Receive the response response_data = self.receive_full_response(self.sock) logger.info(f"Received {len(response_data)} bytes of data") # Parse the response response = json.loads(response_data.decode("utf-8")) logger.info(f"Response parsed, status: {response.get('status', 'unknown')}") if response.get("status") == "error": logger.error(f"Ableton error: {response.get('message')}") raise Exception(response.get("message", "Unknown error from Ableton")) # For state-modifying commands, add another small delay after receiving response if is_modifying_command: import time time.sleep(0.1) # 100ms delay return response.get("result", {}) except TimeoutError: logger.error("Socket timeout while waiting for response from Ableton") self.sock = None raise Exception("Timeout waiting for Ableton response") except (ConnectionError, BrokenPipeError, ConnectionResetError) as e: logger.error(f"Socket connection error: {str(e)}") self.sock = None raise Exception(f"Connection to Ableton lost: {str(e)}") except json.JSONDecodeError as e: logger.error(f"Invalid JSON response from Ableton: {str(e)}") if "response_data" in locals() and response_data: raw_data = response_data[:200] logger.error(f"Raw response (first 200 bytes): {raw_data!r}") self.sock = None raise Exception(f"Invalid response from Ableton: {str(e)}") except Exception as e: logger.error(f"Error communicating with Ableton: {str(e)}") self.sock = None raise Exception(f"Communication error with Ableton: {str(e)}") # Global connection for resources _ableton_connection = None def get_ableton_connection() -> AbletonConnection: """Get or create a persistent Ableton connection""" global _ableton_connection if _ableton_connection is not None: try: # Test the connection with a simple ping # We'll try to send an empty message, which should fail if the connection is dead # but won't affect Ableton if it's alive _ableton_connection.sock.settimeout(1.0) _ableton_connection.sock.sendall(b"") return _ableton_connection except Exception as e: logger.warning(f"Existing connection is no longer valid: {str(e)}") try: _ableton_connection.disconnect() except: pass _ableton_connection = None # Connection doesn't exist or is invalid, create a new one if _ableton_connection is None: # Try to connect up to 3 times with a short delay between attempts max_attempts = 3 for attempt in range(1, max_attempts + 1): try: logger.info(f"Connecting to Ableton (attempt {attempt}/{max_attempts})...") _ableton_connection = AbletonConnection(host="localhost", port=9877) if _ableton_connection.connect(): logger.info("Created new persistent connection to Ableton") # Validate connection with a simple command try: # Get session info as a test _ableton_connection.send_command("get_session_info") logger.info("Connection validated successfully") return _ableton_connection except Exception as e: logger.error(f"Connection validation failed: {str(e)}") _ableton_connection.disconnect() _ableton_connection = None # Continue to next attempt else: _ableton_connection = None except Exception as e: logger.error(f"Connection attempt {attempt} failed: {str(e)}") if _ableton_connection: _ableton_connection.disconnect() _ableton_connection = None # Wait before trying again, but only if we have more attempts left if attempt < max_attempts: import time time.sleep(1.0) # If we get here, all connection attempts failed if _ableton_connection is None: logger.error("Failed to connect to Ableton after multiple attempts") raise Exception("Could not connect to Ableton. Make sure the Remote Script is running.") return _ableton_connection def disconnect_global_connection() -> None: """Disconnect the global connection""" global _ableton_connection if _ableton_connection: logger.info("Disconnecting from Ableton") _ableton_connection.disconnect() _ableton_connection = None