feat: ableton live mcp

[jeevanpillay]

Ableton Live MCP Server

Overview

Implement MCP server for Ableton Live to enable AI-driven music production, live performance, and audio processing workflows.

Technical Approach

• Connection Method: TCP/UDP via Max for Live or Python Live API
• Dependencies: Max for Live, Python Live API, OSC support
• Architecture: Network-based communication with Live's API

Priority

• Phase: 1 (High Priority)
• Category: Audio Production
• Connection Method: OSC/Max for Live
• Type: Feature

Key Features

• Session management (scene triggering, clip launching, transport control)
• Track operations (track creation, device management, routing)
• MIDI control (MIDI mapping, note generation, CC automation)
• Audio processing (effect parameter control, audio routing)
• Live performance (set management, scene automation, recording)
• Device control (instrument/effect parameter manipulation)

Implementation Requirements

• Follow BaseServer architecture pattern
• Integrate with existing lightfast-mcp ecosystem
• Support both stdio and streamable-http transports
• Comprehensive error handling and logging
• Security validation for all operations

Related Issues

Part of the broader creative applications integration roadmap for 40+ applications.

Labels

• enhancement
• audio-production
• phase-1

[jeevanpillay]

Ableton Live MCP Server Implementation Plan

Technical Approach

• Connection Method: TCP/UDP via Max for Live or Python Live API
• Dependencies: Max for Live, Python Live API, OSC support
• Architecture: Network-based communication with Live's API

Directory Structure

src/lightfast_mcp/servers/ableton/
├── server.py          # Main AbletonMCPServer implementation
├── __init__.py        # Package initialization
├── tools/             # Ableton Live-specific tools
│   ├── __init__.py
│   ├── session.py     # Session management
│   ├── tracks.py      # Track operations
│   ├── clips.py       # Clip manipulation
│   ├── devices.py     # Device control
│   ├── transport.py   # Transport controls
│   └── midi.py        # MIDI operations
└── utils/             # Ableton utilities
    ├── __init__.py
    ├── live_helpers.py # Helper functions
    └── osc_bridge.py   # OSC communication bridge

Key Features to Implement

Core Tools

@mcp.tool()
async def play_session() -> str:
    """Start/stop session playback."""

@mcp.tool()
async def create_track(track_type: str = "audio", name: str = None) -> str:
    """Create a new audio or MIDI track."""

@mcp.tool()
async def load_clip(track_index: int, clip_slot: int, file_path: str) -> str:
    """Load audio/MIDI clip into specified slot."""

@mcp.tool()
async def trigger_clip(track_index: int, clip_slot: int) -> str:
    """Trigger clip playback."""

@mcp.tool()
async def add_device(track_index: int, device_name: str) -> str:
    """Add audio device/instrument to track."""

@mcp.tool()
async def set_device_parameter(track_index: int, device_index: int, parameter: str, value: float) -> str:
    """Set device parameter value."""

@mcp.tool()
async def record_clip(track_index: int, clip_slot: int, length_bars: int = 4) -> str:
    """Record new clip into specified slot."""

@mcp.tool()
async def set_tempo(bpm: float) -> str:
    """Set session tempo."""

Implementation Example

Server Class

from lightfast_mcp.core.base_server import BaseServer
from lightfast_mcp.core.server_config import ServerConfig
import socket
import json
import asyncio
from typing import Dict, Any, List
import threading

class AbletonMCPServer(BaseServer):
    SERVER_TYPE = "ableton"
    REQUIRED_APPS = ["Ableton Live"]
    
    def __init__(self, config: ServerConfig):
        super().__init__(config)
        self.live_socket = None
        self.osc_client = None
        self.host = config.connection.get("host", "localhost")
        self.port = config.connection.get("port", 9000)
        self.use_max_for_live = config.connection.get("use_max_for_live", True)
        
    async def _check_application_availability(self) -> bool:
        """Check if Ableton Live is running and accessible."""
        try:
            if self.use_max_for_live:
                await self._connect_to_live()
                result = await self._send_live_command("live_set get tempo")
                return result is not None
            else:
                # Alternative: Check if Live process is running
                import subprocess
                result = subprocess.run(["pgrep", "-f", "Ableton"], capture_output=True)
                return result.returncode == 0
        except Exception:
            return False
    
    async def _connect_to_live(self):
        """Establish connection to Ableton Live via Max for Live."""
        if self.live_socket is None:
            self.live_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            # Setup OSC client for communication
            try:
                from pythonosc import udp_client
                self.osc_client = udp_client.SimpleUDPClient(self.host, self.port)
            except ImportError:
                self.logger.warning("python-osc not available, using basic UDP")
    
    async def _send_live_command(self, command: str, *args) -> Any:
        """Send command to Ableton Live and get response."""
        await self._connect_to_live()
        
        if self.osc_client:
            # Send OSC message
            self.osc_client.send_message(f"/live/{command}", args)
        else:
            # Send raw UDP message
            message = json.dumps({"command": command, "args": args})
            self.live_socket.sendto(message.encode(), (self.host, self.port))
        
        # Wait for response (simplified - would need proper OSC response handling)
        try:
            response, addr = self.live_socket.recvfrom(1024)
            return json.loads(response.decode())
        except:
            return None
    
    def _register_tools(self):
        """Register all Ableton Live-specific tools."""
        self.mcp.tool()(self.play_session)
        self.mcp.tool()(self.stop_session)
        self.mcp.tool()(self.create_track)
        self.mcp.tool()(self.load_clip)
        self.mcp.tool()(self.trigger_clip)
        self.mcp.tool()(self.stop_clip)
        self.mcp.tool()(self.add_device)
        self.mcp.tool()(self.set_device_parameter)
        self.mcp.tool()(self.record_clip)
        self.mcp.tool()(self.set_tempo)
        self.mcp.tool()(self.get_session_info)
        self.mcp.tool()(self.arm_track)
        self.mcp.tool()(self.set_track_volume)
    
    async def play_session(self) -> str:
        """Start session playback."""
        try:
            result = await self._send_live_command("song", "start_playing")
            return "Session playback started"
        except Exception as e:
            return f"Error starting playback: {str(e)}"
    
    async def stop_session(self) -> str:
        """Stop session playback."""
        try:
            result = await self._send_live_command("song", "stop_playing")
            return "Session playback stopped"
        except Exception as e:
            return f"Error stopping playback: {str(e)}"
    
    async def create_track(self, track_type: str = "audio", name: str = None) -> str:
        """Create a new audio or MIDI track."""
        try:
            track_name = name or f"{track_type}_track"
            
            if track_type.lower() == "audio":
                result = await self._send_live_command("song", "create_audio_track", -1)
            elif track_type.lower() == "midi":
                result = await self._send_live_command("song", "create_midi_track", -1)
            else:
                raise ValueError(f"Invalid track type: {track_type}")
            
            # Set track name if provided
            if name:
                track_count = await self._send_live_command("song", "get", "tracks", "length")
                await self._send_live_command("song", "get", "tracks", track_count-1, "set", "name", name)
            
            return f"Created {track_type} track: {track_name}"
            
        except Exception as e:
            return f"Error creating track: {str(e)}"
    
    async def load_clip(self, track_index: int, clip_slot: int, file_path: str) -> str:
        """Load audio/MIDI clip into specified slot."""
        try:
            result = await self._send_live_command(
                "song", "get", "tracks", track_index, 
                "get", "clip_slots", clip_slot, 
                "set", "clip", "file_path", file_path
            )
            
            return f"Loaded clip from {file_path} into track {track_index}, slot {clip_slot}"
            
        except Exception as e:
            return f"Error loading clip: {str(e)}"
    
    async def trigger_clip(self, track_index: int, clip_slot: int) -> str:
        """Trigger clip playback."""
        try:
            result = await self._send_live_command(
                "song", "get", "tracks", track_index,
                "get", "clip_slots", clip_slot,
                "fire"
            )
            
            return f"Triggered clip at track {track_index}, slot {clip_slot}"
            
        except Exception as e:
            return f"Error triggering clip: {str(e)}"
    
    async def stop_clip(self, track_index: int, clip_slot: int) -> str:
        """Stop clip playback."""
        try:
            result = await self._send_live_command(
                "song", "get", "tracks", track_index,
                "get", "clip_slots", clip_slot,
                "stop"
            )
            
            return f"Stopped clip at track {track_index}, slot {clip_slot}"
            
        except Exception as e:
            return f"Error stopping clip: {str(e)}"
    
    async def add_device(self, track_index: int, device_name: str) -> str:
        """Add audio device/instrument to track."""
        try:
            result = await self._send_live_command(
                "song", "get", "tracks", track_index,
                "get", "devices",
                "add_device_by_name", device_name
            )
            
            return f"Added device '{device_name}' to track {track_index}"
            
        except Exception as e:
            return f"Error adding device: {str(e)}"
    
    async def set_device_parameter(self, track_index: int, device_index: int, 
                                 parameter: str, value: float) -> str:
        """Set device parameter value."""
        try:
            result = await self._send_live_command(
                "song", "get", "tracks", track_index,
                "get", "devices", device_index,
                "get", "parameters", parameter,
                "set", "value", value
            )
            
            return f"Set {parameter} to {value} on device {device_index}, track {track_index}"
            
        except Exception as e:
            return f"Error setting device parameter: {str(e)}"
    
    async def record_clip(self, track_index: int, clip_slot: int, length_bars: int = 4) -> str:
        """Record new clip into specified slot."""
        try:
            # Arm track for recording
            await self._send_live_command(
                "song", "get", "tracks", track_index,
                "set", "arm", 1
            )
            
            # Start recording
            result = await self._send_live_command(
                "song", "get", "tracks", track_index,
                "get", "clip_slots", clip_slot,
                "create_clip", length_bars
            )
            
            return f"Started recording clip on track {track_index}, slot {clip_slot} for {length_bars} bars"
            
        except Exception as e:
            return f"Error recording clip: {str(e)}"
    
    async def set_tempo(self, bpm: float) -> str:
        """Set session tempo."""
        try:
            result = await self._send_live_command("song", "set", "tempo", bpm)
            return f"Set tempo to {bpm} BPM"
            
        except Exception as e:
            return f"Error setting tempo: {str(e)}"
    
    async def get_session_info(self) -> str:
        """Get information about the current session."""
        try:
            tempo = await self._send_live_command("song", "get", "tempo")
            is_playing = await self._send_live_command("song", "get", "is_playing")
            track_count = await self._send_live_command("song", "get", "tracks", "length")
            
            info = {
                "tempo": tempo,
                "is_playing": is_playing,
                "track_count": track_count,
                "current_song_time": await self._send_live_command("song", "get", "current_song_time")
            }
            
            return str(info)
            
        except Exception as e:
            return f"Error getting session info: {str(e)}"
    
    async def arm_track(self, track_index: int, armed: bool = True) -> str:
        """Arm/disarm track for recording."""
        try:
            result = await self._send_live_command(
                "song", "get", "tracks", track_index,
                "set", "arm", 1 if armed else 0
            )
            
            status = "armed" if armed else "disarmed"
            return f"Track {track_index} {status} for recording"
            
        except Exception as e:
            return f"Error arming track: {str(e)}"
    
    async def set_track_volume(self, track_index: int, volume: float) -> str:
        """Set track volume (0.0 to 1.0)."""
        try:
            # Clamp volume to valid range
            volume = max(0.0, min(1.0, volume))
            
            result = await self._send_live_command(
                "song", "get", "tracks", track_index,
                "get", "mixer_device",
                "get", "volume",
                "set", "value", volume
            )
            
            return f"Set track {track_index} volume to {volume}"
            
        except Exception as e:
            return f"Error setting track volume: {str(e)}"

Max for Live Bridge Script

// ableton_mcp_bridge.amxd (Max for Live device)
{
    "patcher": {
        "boxes": [
            {
                "box": {
                    "maxclass": "newobj",
                    "text": "udpreceive 9000",
                    "id": "obj-1"
                }
            },
            {
                "box": {
                    "maxclass": "newobj", 
                    "text": "live.path",
                    "id": "obj-2"
                }
            },
            {
                "box": {
                    "maxclass": "newobj",
                    "text": "live.object",
                    "id": "obj-3"
                }
            }
        ],
        "lines": [
            {
                "patchline": {
                    "source": ["obj-1", 0],
                    "destination": ["obj-2", 0]
                }
            }
        ]
    }
}

Configuration Example

# config/servers.yaml
ableton:
  enabled: true
  transport: stdio
  connection:
    host: localhost
    port: 9000
    use_max_for_live: true
    timeout: 30
  features:
    - session_management
    - track_operations
    - clip_manipulation
    - device_control
    - midi_operations
    - live_performance

Entry Point Script

# src/lightfast_mcp/servers/ableton_mcp_server.py
#!/usr/bin/env python3
"""Ableton Live MCP Server entry point."""

import asyncio
from lightfast_mcp.servers.ableton.server import AbletonMCPServer
from lightfast_mcp.core.server_config import ServerConfig

async def main():
    config = ServerConfig.from_env("ableton")
    server = AbletonMCPServer(config)
    await server.run()

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

Setup Requirements

1. Ableton Live Installation: Ensure Ableton Live is installed with Max for Live
2. Max for Live Bridge: Install the MCP bridge Max for Live device
3. OSC Configuration: Configure OSC communication in Live preferences
4. Python Dependencies: Install python-osc for OSC communication

This implementation provides comprehensive Ableton Live automation through MCP, enabling AI-driven music production and live performance workflows.