#!/usr/bin/env python3
"""
OSC-based MCP Server for REAPER
This module provides an MCP server implementation using python-osc to communicate with REAPER
"""
import json
import sys
import os
import time
import traceback
import logging
from pythonosc import udp_client
from pythonosc.dispatcher import Dispatcher
from pythonosc.osc_server import BlockingOSCUDPServer
from mcp.server.fastmcp import FastMCP
import threading
# Set up logging
logger = logging.getLogger("reaper_mcp.osc_server")
# REAPER Action IDs
ACTION_NEW_PROJECT = 40023
ACTION_INSERT_TRACK = 40001
ACTION_SELECT_TRACK_1 = 40939
ACTION_SELECT_TRACK_2 = 40940
ACTION_SELECT_TRACK_3 = 40941
ACTION_SELECT_TRACK_4 = 40942
ACTION_SELECT_TRACK_5 = 40943
ACTION_SELECT_TRACK_6 = 40944
ACTION_SELECT_TRACK_7 = 40945
ACTION_SELECT_TRACK_8 = 40946
ACTION_INSERT_MIDI_ITEM = 40214
ACTION_SET_TEMPO = 40364
ACTION_SAVE_PROJECT = 40022
class ReaperOSCServer:
"""
REAPER OSC Server implementation for MCP
"""
def __init__(self, host="127.0.0.1", send_port=8000, receive_port=9000):
"""
Initialize the REAPER OSC Server
Args:
host (str): The IP address of the REAPER instance
send_port (int): The port REAPER listens on for OSC messages
receive_port (int): The port we listen on for REAPER responses
"""
self.host = host
self.send_port = send_port
self.receive_port = receive_port
# Create the MCP server
self.mcp = FastMCP("ReaperOSCMCP")
# Create OSC client for sending commands to REAPER
self.client = udp_client.SimpleUDPClient(self.host, self.send_port)
# Global variables to store project state
self.current_project = {
"name": "Untitled",
"path": "",
"tracks": []
}
# Set up OSC dispatcher
self.dispatcher = Dispatcher()
self.dispatcher.map("/track/*/name", self.handle_track_info)
self.dispatcher.map("/project/name", self.handle_project_info)
self.dispatcher.map("/project/path", self.handle_project_info)
# Start OSC server in a separate thread
self.osc_thread = threading.Thread(target=self.start_osc_server, daemon=True)
self.osc_thread.start()
# Register MCP tools
self.register_mcp_tools()
# Request initial project info
self.request_project_info()
def start_osc_server(self):
"""Start the OSC server to listen for REAPER responses"""
server = BlockingOSCUDPServer((self.host, self.receive_port), self.dispatcher)
logger.info(f"OSC server listening on {self.host}:{self.receive_port}")
server.serve_forever()
def handle_track_info(self, address, *args):
"""Handle track info messages from REAPER"""
logger.debug(f"Received track info: {address} {args}")
# Update our track info cache
track_idx = int(address.split('/')[-1])
if track_idx >= len(self.current_project["tracks"]):
self.current_project["tracks"].extend([{} for _ in range(track_idx - len(self.current_project["tracks"]) + 1)])
self.current_project["tracks"][track_idx]["name"] = args[0] if args else f"Track {track_idx+1}"
def handle_project_info(self, address, *args):
"""Handle project info messages from REAPER"""
logger.debug(f"Received project info: {address} {args}")
if address == "/project/name" and args:
self.current_project["name"] = args[0]
elif address == "/project/path" and args:
self.current_project["path"] = args[0]
def request_project_info(self):
"""Request project information from REAPER"""
self.client.send_message("/project/name/get", None)
self.client.send_message("/project/path/get", None)
self.client.send_message("/track/count/get", None)
time.sleep(0.1) # Give REAPER time to respond
def refresh_track_list(self):
"""Refresh the track list from REAPER"""
self.client.send_message("/track/count/get", None)
time.sleep(0.1) # Wait for response
# Request info for each track
for i in range(len(self.current_project["tracks"])):
self.client.send_message(f"/track/{i}/name/get", None)
time.sleep(0.1) # Wait for responses
def select_track(self, track_index):
"""Select a track by index using action IDs"""
if track_index == 0:
self.client.send_message("/action", [ACTION_SELECT_TRACK_1])
elif track_index == 1:
self.client.send_message("/action", [ACTION_SELECT_TRACK_2])
elif track_index == 2:
self.client.send_message("/action", [ACTION_SELECT_TRACK_3])
elif track_index == 3:
self.client.send_message("/action", [ACTION_SELECT_TRACK_4])
elif track_index == 4:
self.client.send_message("/action", [ACTION_SELECT_TRACK_5])
elif track_index == 5:
self.client.send_message("/action", [ACTION_SELECT_TRACK_6])
elif track_index == 6:
self.client.send_message("/action", [ACTION_SELECT_TRACK_7])
elif track_index == 7:
self.client.send_message("/action", [ACTION_SELECT_TRACK_8])
else:
# For tracks beyond 8, we'll use the OSC method (less reliable)
self.client.send_message(f"/track/{track_index}/select", [1])
time.sleep(0.2) # Wait for REAPER to process
def register_mcp_tools(self):
"""Register all MCP tools"""
@self.mcp.tool()
def create_project(name, template=None):
"""Creates a new REAPER project."""
try:
# Send action to create new project
self.client.send_message("/action", [ACTION_NEW_PROJECT])
time.sleep(0.5) # Give REAPER time to create the project
# Set project name by saving it
if name:
save_path = os.path.join(os.path.expanduser("~/Documents/REAPER Projects"), f"{name}.rpp")
self.client.send_message("/action", [ACTION_SAVE_PROJECT])
time.sleep(0.3)
# We can't directly set the save path via OSC, but we can update our state
self.current_project["name"] = name
self.current_project["path"] = save_path
# Request updated project info
self.request_project_info()
return {"success": True, "message": f"Created project: {name}"}
except Exception as e:
logger.error(f"Error creating project: {e}")
return {"success": False, "error": str(e)}
@self.mcp.tool()
def create_track(name=None):
"""Creates a new track in the current project."""
try:
# Use action ID to create new track (more reliable)
self.client.send_message("/action", [ACTION_INSERT_TRACK])
time.sleep(0.5) # Give REAPER time to create the track
# Get track count and assume new track is at the end
self.client.send_message("/track/count/get", None)
time.sleep(0.2)
# Set track name if provided
if name and self.current_project["tracks"]:
track_index = len(self.current_project["tracks"]) - 1
self.client.send_message(f"/track/{track_index}/name", [name])
self.current_project["tracks"][track_index]["name"] = name
# Refresh track list
self.refresh_track_list()
return {"success": True, "track_index": len(self.current_project["tracks"]) - 1}
except Exception as e:
logger.error(f"Error creating track: {e}")
return {"success": False, "error": str(e)}
@self.mcp.tool()
def list_tracks():
"""Lists all tracks in the current project."""
try:
# Refresh track information
self.refresh_track_list()
tracks = []
for i, track in enumerate(self.current_project["tracks"]):
tracks.append({
"index": i,
"name": track.get("name", f"Track {i+1}"),
"is_selected": False # We don't have this info via OSC by default
})
return {"success": True, "tracks": tracks}
except Exception as e:
logger.error(f"Error listing tracks: {e}")
return {"success": False, "error": str(e)}
@self.mcp.tool()
def add_midi_note(track_index, note, start_time, duration, velocity=100):
"""Adds a MIDI note to a track."""
try:
# First, make sure we have a MIDI item at the right position
# Select the track using action IDs (more reliable)
self.select_track(track_index)
# Insert new MIDI item using action ID
self.client.send_message("/action", [ACTION_INSERT_MIDI_ITEM])
time.sleep(0.5)
# We can't directly add MIDI notes via OSC
# This is a limitation of the OSC implementation
return {"success": True, "message": f"Created MIDI item on track {track_index}. Note: Adding specific MIDI notes requires ReaScript or external MIDI file import."}
except Exception as e:
logger.error(f"Error adding MIDI note: {e}")
return {"success": False, "error": str(e)}
@self.mcp.tool()
def get_project_info():
"""Gets information about the current project."""
try:
# Request updated project info
self.request_project_info()
self.refresh_track_list()
return {
"success": True,
"name": self.current_project["name"],
"path": self.current_project["path"],
"track_count": len(self.current_project["tracks"]),
"selected_track_count": 0 # We don't have this info via OSC by default
}
except Exception as e:
logger.error(f"Error getting project info: {e}")
return {"success": False, "error": str(e)}
def run(self, transport='stdio'):
"""Run the MCP server"""
logger.info("Starting OSC-based REAPER MCP Server...")
try:
# Use specified transport for MCP protocol
self.mcp.run(transport=transport)
except Exception as e:
logger.error(f"Error running MCP server: {e}")
traceback.print_exc(file=sys.stderr)
sys.exit(1)
def create_server(host="127.0.0.1", send_port=8000, receive_port=9000):
"""
Create and return a REAPER OSC Server instance
Args:
host (str): The IP address of the REAPER instance
send_port (int): The port REAPER listens on for OSC messages
receive_port (int): The port we listen on for REAPER responses
Returns:
ReaperOSCServer: The REAPER OSC Server instance
"""
return ReaperOSCServer(host, send_port, receive_port)
def main():
"""Main entry point for the REAPER OSC Server"""
# Set up logging
logging.basicConfig(level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
stream=sys.stderr)
# Create and run the server
server = create_server()
server.run()
if __name__ == "__main__":
main()
