Files
reaper-control/mcp/server.py

357 lines
10 KiB
Python

"""
FL Studio MCP Server — FastMCP server with MIDI SysEx transport.
Sends nibble-encoded JSON commands to FL Studio via Windows MIDI Services loopback.
FL Studio controller script receives via OnSysEx() and calls FL Studio API.
"""
from __future__ import annotations
import sys
from mcp.server.fastmcp import FastMCP
from protocol.transport import MidiTransport
mcp = FastMCP("fl-studio-mcp")
transport = MidiTransport()
def _log(msg: str) -> None:
"""Print a log message to stderr."""
print(f"[fl-studio-mcp] {msg}", file=sys.stderr)
# ─── Transport Tools ────────────────────────────────────────────────────────────
@mcp.tool()
def play() -> str:
"""Start FL Studio playback."""
try:
transport.send_command("start_playback")
return "Playback started"
except Exception as e:
return f"Error: {e}"
@mcp.tool()
def stop() -> str:
"""Stop FL Studio playback."""
try:
transport.send_command("stop_playback")
return "Playback stopped"
except Exception as e:
return f"Error: {e}"
@mcp.tool()
def record() -> str:
"""Start recording in FL Studio."""
try:
transport.send_command("start_recording")
return "Recording started"
except Exception as e:
return f"Error: {e}"
@mcp.tool()
def set_tempo(bpm: float) -> str:
"""Set FL Studio tempo (40-999 BPM)."""
try:
transport.send_command("set_tempo", {"tempo": float(bpm)})
return f"Tempo set to {bpm} BPM"
except Exception as e:
return f"Error: {e}"
@mcp.tool()
def set_time_signature(numerator: int = 4, denominator: int = 4) -> str:
"""Set time signature."""
try:
transport.send_command("set_time_signature", {
"numerator": int(numerator),
"denominator": int(denominator),
})
return f"Time signature set to {numerator}/{denominator}"
except Exception as e:
return f"Error: {e}"
# ─── Channel Tools ────────────────────────────────────────────────────────────
@mcp.tool()
def select_channel(channel_index: int) -> str:
"""Select a channel in the channel rack (0-based index)."""
try:
transport.send_command("select_channel", {"channel_index": int(channel_index)})
return f"Channel {channel_index} selected"
except Exception as e:
return f"Error: {e}"
@mcp.tool()
def mute_channel(channel_index: int, muted: bool = True) -> str:
"""Mute or unmute a channel."""
try:
transport.send_command("mute_channel", {
"channel_index": int(channel_index),
"muted": bool(muted),
})
return f"Channel {channel_index} muted={muted}"
except Exception as e:
return f"Error: {e}"
@mcp.tool()
def solo_channel(channel_index: int) -> str:
"""Solo a channel."""
try:
transport.send_command("solo_channel", {"channel_index": int(channel_index)})
return f"Channel {channel_index} soloed"
except Exception as e:
return f"Error: {e}"
@mcp.tool()
def set_channel_volume(channel_index: int, volume: float) -> str:
"""Set channel volume (0.0-1.0)."""
try:
transport.send_command("set_channel_volume", {
"channel_index": int(channel_index),
"volume": float(volume),
})
return f"Channel {channel_index} volume={volume}"
except Exception as e:
return f"Error: {e}"
@mcp.tool()
def set_channel_pan(channel_index: int, pan: float) -> str:
"""Set channel pan (-1.0 to 1.0)."""
try:
transport.send_command("set_channel_pan", {
"channel_index": int(channel_index),
"pan": float(pan),
})
return f"Channel {channel_index} pan={pan}"
except Exception as e:
return f"Error: {e}"
@mcp.tool()
def note_on(channel_index: int, note: int, velocity: int = 100) -> str:
"""Send a note on event."""
try:
transport.send_command("note_on", {
"channel_index": int(channel_index),
"note": int(note),
"velocity": int(velocity),
})
return f"Note on: ch={channel_index} note={note} vel={velocity}"
except Exception as e:
return f"Error: {e}"
@mcp.tool()
def note_off(channel_index: int, note: int) -> str:
"""Send a note off event."""
try:
transport.send_command("note_off", {
"channel_index": int(channel_index),
"note": int(note),
})
return f"Note off: ch={channel_index} note={note}"
except Exception as e:
return f"Error: {e}"
@mcp.tool()
def stop_all_notes() -> str:
"""Stop all playing notes (panic)."""
try:
transport.send_command("stop_all_notes")
return "All notes stopped"
except Exception as e:
return f"Error: {e}"
# ─── Mixer Tools ──────────────────────────────────────────────────────────────
@mcp.tool()
def set_mixer_volume(track_index: int, volume: float) -> str:
"""Set mixer track volume (0.0-1.25)."""
try:
transport.send_command("set_mixer_volume", {
"track_index": int(track_index),
"volume": float(volume),
})
return f"Mixer track {track_index} volume={volume}"
except Exception as e:
return f"Error: {e}"
@mcp.tool()
def set_mixer_pan(track_index: int, pan: float) -> str:
"""Set mixer track pan (-1.0 to 1.0)."""
try:
transport.send_command("set_mixer_pan", {
"track_index": int(track_index),
"pan": float(pan),
})
return f"Mixer track {track_index} pan={pan}"
except Exception as e:
return f"Error: {e}"
@mcp.tool()
def mute_mixer_track(track_index: int, muted: bool = True) -> str:
"""Mute or unmute a mixer track."""
try:
transport.send_command("mute_mixer_track", {
"track_index": int(track_index),
"muted": bool(muted),
})
return f"Mixer track {track_index} muted={muted}"
except Exception as e:
return f"Error: {e}"
@mcp.tool()
def solo_mixer_track(track_index: int) -> str:
"""Solo a mixer track."""
try:
transport.send_command("solo_mixer_track", {"track_index": int(track_index)})
return f"Mixer track {track_index} soloed"
except Exception as e:
return f"Error: {e}"
@mcp.tool()
def rename_mixer_track(track_index: int, name: str) -> str:
"""Rename a mixer track."""
try:
transport.send_command("set_mixer_track_name", {
"track_index": int(track_index),
"name": str(name),
})
return f"Mixer track {track_index} renamed to '{name}'"
except Exception as e:
return f"Error: {e}"
# ─── Pattern Tools ────────────────────────────────────────────────────────────
@mcp.tool()
def select_pattern(pattern_index: int) -> str:
"""Select a pattern by index."""
try:
transport.send_command("select_pattern", {"pattern_index": int(pattern_index)})
return f"Pattern {pattern_index} selected"
except Exception as e:
return f"Error: {e}"
@mcp.tool()
def create_pattern(name: str) -> str:
"""Create a new pattern with the given name."""
try:
transport.send_command("set_pattern_name", {
"pattern_index": 0, # Will create new
"name": str(name),
})
return f"Pattern '{name}' created"
except Exception as e:
return f"Error: {e}"
@mcp.tool()
def rename_pattern(pattern_index: int, name: str) -> str:
"""Rename a pattern."""
try:
transport.send_command("set_pattern_name", {
"pattern_index": int(pattern_index),
"name": str(name),
})
return f"Pattern {pattern_index} renamed to '{name}'"
except Exception as e:
return f"Error: {e}"
# ─── UI Tools ─────────────────────────────────────────────────────────────────
@mcp.tool()
def show_channel_rack() -> str:
"""Show the FL Studio channel rack window."""
try:
transport.send_command("show_channel_rack")
return "Channel rack shown"
except Exception as e:
return f"Error: {e}"
@mcp.tool()
def show_mixer() -> str:
"""Show the FL Studio mixer window."""
try:
transport.send_command("show_mixer")
return "Mixer shown"
except Exception as e:
return f"Error: {e}"
@mcp.tool()
def show_piano_roll() -> str:
"""Show the FL Studio piano roll window."""
try:
transport.send_command("show_piano_roll")
return "Piano roll shown"
except Exception as e:
return f"Error: {e}"
@mcp.tool()
def show_playlist() -> str:
"""Show the FL Studio playlist window."""
try:
transport.send_command("show_playlist")
return "Playlist shown"
except Exception as e:
return f"Error: {e}"
# ─── Meta Tools ────────────────────────────────────────────────────────────────
@mcp.tool()
def ping() -> str:
"""Ping the FL Studio MCP server to verify connectivity."""
try:
transport.send_command("ping", {"ts": 1})
return "pong"
except Exception as e:
return f"Error: {e}"
@mcp.tool()
def list_ports() -> dict[str, list[str]]:
"""List all available MIDI input and output ports."""
return MidiTransport.list_ports()
@mcp.tool()
def get_session_info() -> dict:
"""Get FL Studio session information (transport state)."""
return {
"server": "fl-studio-mcp",
"transport": "midi_sysex",
"loopback": "FL_MCP",
}
if __name__ == "__main__":
mcp.run(transport="stdio")