feat: touchdesigner mcp

[jeevanpillay]

TouchDesigner MCP Server

Overview

Implement MCP server for TouchDesigner to enable AI-driven interactive media, real-time visual programming, and live performance workflows.

Technical Approach

• Connection Method: Socket-based communication + Python scripting
• Dependencies: TouchDesigner Python API, OSC support
• Architecture: Real-time node-based visual programming control

Priority

• Phase: 2 (High Priority)
• Category: Interactive Media
• Connection Method: Socket/Python
• Type: Feature

Key Features

• Network operations (node creation/connection, parameter control)
• Real-time control (live performance parameters, MIDI/OSC integration)
• Media processing (video/audio input/output, effects processing)
• Generative systems (procedural content generation, data visualization)
• Hardware integration (sensor input, projection mapping control)

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
• interactive-media
• phase-2

[jeevanpillay]

TouchDesigner MCP Server Implementation Plan

Technical Approach

• Connection Method: Socket-based communication + Python scripting
• Dependencies: TouchDesigner Python API, OSC support
• Architecture: Real-time node-based visual programming control

Directory Structure

src/lightfast_mcp/servers/touchdesigner/
├── server.py          # Main TouchDesignerMCPServer implementation
├── __init__.py        # Package initialization
├── tools/             # TouchDesigner-specific tools
│   ├── __init__.py
│   ├── operators.py   # Operator (node) management
│   ├── parameters.py  # Parameter control
│   ├── networks.py    # Network operations
│   ├── media.py       # Media input/output
│   ├── performance.py # Live performance tools
│   └── export.py      # Export operations
└── utils/             # TouchDesigner utilities
    ├── __init__.py
    ├── td_bridge.py   # TouchDesigner communication bridge
    └── osc_client.py  # OSC communication

Key Features to Implement

Core Tools

@mcp.tool()
async def create_operator(op_type: str, name: str = None, parent_path: str = "/") -> str:
    """Create a new operator (node) in TouchDesigner."""

@mcp.tool()
async def connect_operators(output_op: str, input_op: str, output_index: int = 0, input_index: int = 0) -> str:
    """Connect two operators together."""

@mcp.tool()
async def set_parameter(op_path: str, parameter: str, value: any) -> str:
    """Set parameter value for an operator."""

@mcp.tool()
async def get_parameter(op_path: str, parameter: str) -> str:
    """Get parameter value from an operator."""

@mcp.tool()
async def load_media(file_path: str, op_path: str = None) -> str:
    """Load media file into TouchDesigner."""

@mcp.tool()
async def start_performance() -> str:
    """Start performance mode."""

@mcp.tool()
async def record_output(output_path: str, duration: float = 10.0) -> str:
    """Record output to file."""

@mcp.tool()
async def send_osc_message(address: str, *args) -> str:
    """Send OSC message."""

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, Union
import subprocess

class TouchDesignerMCPServer(BaseServer):
    SERVER_TYPE = "touchdesigner"
    REQUIRED_APPS = ["TouchDesigner"]
    
    def __init__(self, config: ServerConfig):
        super().__init__(config)
        self.td_socket = None
        self.osc_client = None
        self.host = config.connection.get("host", "localhost")
        self.port = config.connection.get("port", 7000)
        self.osc_port = config.connection.get("osc_port", 7001)
        self.td_path = config.connection.get("td_path", "C:/Program Files/Derivative/TouchDesigner/bin/TouchDesigner.exe")
        
    async def _check_application_availability(self) -> bool:
        """Check if TouchDesigner is running and accessible."""
        try:
            await self._connect_to_td()
            result = await self._send_td_command("op('/').name")
            return result is not None
        except Exception:
            return False
    
    async def _connect_to_td(self):
        """Establish connection to TouchDesigner."""
        if self.td_socket is None:
            try:
                self.td_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                self.td_socket.connect((self.host, self.port))
                
                # Setup OSC client
                try:
                    from pythonosc import udp_client
                    self.osc_client = udp_client.SimpleUDPClient(self.host, self.osc_port)
                except ImportError:
                    self.logger.warning("python-osc not available")
                    
            except ConnectionRefusedError:
                self.logger.error("Could not connect to TouchDesigner. Ensure TD is running with TCP server enabled.")
                raise
    
    async def _send_td_command(self, command: str) -> Any:
        """Send Python command to TouchDesigner and get response."""
        await self._connect_to_td()
        
        try:
            # Send command
            message = json.dumps({"command": command}) + "\n"
            self.td_socket.send(message.encode())
            
            # Receive response
            response = self.td_socket.recv(4096).decode()
            return json.loads(response) if response else None
            
        except Exception as e:
            self.logger.error(f"TouchDesigner command failed: {e}")
            return None
    
    def _register_tools(self):
        """Register all TouchDesigner-specific tools."""
        self.mcp.tool()(self.create_operator)
        self.mcp.tool()(self.connect_operators)
        self.mcp.tool()(self.set_parameter)
        self.mcp.tool()(self.get_parameter)
        self.mcp.tool()(self.load_media)
        self.mcp.tool()(self.start_performance)
        self.mcp.tool()(self.record_output)
        self.mcp.tool()(self.send_osc_message)
        self.mcp.tool()(self.get_network_info)
        self.mcp.tool()(self.delete_operator)
        self.mcp.tool()(self.copy_operator)
    
    async def create_operator(self, op_type: str, name: str = None, parent_path: str = "/") -> str:
        """Create a new operator (node) in TouchDesigner."""
        try:
            # Generate unique name if not provided
            if not name:
                name = f"{op_type}1"
            
            command = f"""
parent = op('{parent_path}')
if parent:
    new_op = parent.create({op_type}OP, '{name}')
    result = new_op.path if new_op else None
else:
    result = None
result
"""
            
            result = await self._send_td_command(command)
            
            if result:
                return f"Created {op_type} operator: {result}"
            else:
                raise Exception(f"Failed to create operator {op_type}")
                
        except Exception as e:
            return f"Error creating operator: {str(e)}"
    
    async def connect_operators(self, output_op: str, input_op: str, 
                              output_index: int = 0, input_index: int = 0) -> str:
        """Connect two operators together."""
        try:
            command = f"""
output_op = op('{output_op}')
input_op = op('{input_op}')

if output_op and input_op:
    try:
        input_op.inputConnectors[{input_index}].connect(output_op.outputConnectors[{output_index}])
        result = True
    except:
        result = False
else:
    result = False
result
"""
            
            result = await self._send_td_command(command)
            
            if result:
                return f"Connected {output_op} output {output_index} to {input_op} input {input_index}"
            else:
                raise Exception("Failed to connect operators")
                
        except Exception as e:
            return f"Error connecting operators: {str(e)}"
    
    async def set_parameter(self, op_path: str, parameter: str, value: Union[str, int, float, bool]) -> str:
        """Set parameter value for an operator."""
        try:
            # Handle different value types
            if isinstance(value, str):
                value_str = f"'{value}'"
            else:
                value_str = str(value)
            
            command = f"""
op_node = op('{op_path}')
if op_node and hasattr(op_node.par, '{parameter}'):
    op_node.par.{parameter} = {value_str}
    result = op_node.par.{parameter}.eval()
else:
    result = None
result
"""
            
            result = await self._send_td_command(command)
            
            if result is not None:
                return f"Set {op_path}.{parameter} = {value}"
            else:
                raise Exception(f"Parameter {parameter} not found on {op_path}")
                
        except Exception as e:
            return f"Error setting parameter: {str(e)}"
    
    async def get_parameter(self, op_path: str, parameter: str) -> str:
        """Get parameter value from an operator."""
        try:
            command = f"""
op_node = op('{op_path}')
if op_node and hasattr(op_node.par, '{parameter}'):
    result = op_node.par.{parameter}.eval()
else:
    result = None
result
"""
            
            result = await self._send_td_command(command)
            
            if result is not None:
                return f"{op_path}.{parameter} = {result}"
            else:
                raise Exception(f"Parameter {parameter} not found on {op_path}")
                
        except Exception as e:
            return f"Error getting parameter: {str(e)}"
    
    async def load_media(self, file_path: str, op_path: str = None) -> str:
        """Load media file into TouchDesigner."""
        try:
            if not op_path:
                # Create a new movie file in operator
                op_path = "/moviefilein1"
                await self.create_operator("moviefilein", "moviefilein1")
            
            # Set the file parameter
            await self.set_parameter(op_path, "file", file_path)
            
            return f"Loaded media {file_path} into {op_path}"
            
        except Exception as e:
            return f"Error loading media: {str(e)}"
    
    async def start_performance(self) -> str:
        """Start performance mode."""
        try:
            command = """
# Enable performance mode
ui.performMode = True
result = ui.performMode
result
"""
            
            result = await self._send_td_command(command)
            
            if result:
                return "Performance mode started"
            else:
                return "Failed to start performance mode"
                
        except Exception as e:
            return f"Error starting performance: {str(e)}"
    
    async def record_output(self, output_path: str, duration: float = 10.0) -> str:
        """Record output to file."""
        try:
            command = f"""
# Create movie file out operator if it doesn't exist
movie_out = op('/moviefileout1')
if not movie_out:
    movie_out = op('/').create(moviefileoutOP, 'moviefileout1')

# Set recording parameters
movie_out.par.file = '{output_path}'
movie_out.par.record = True

# Schedule stop recording
run("op('/moviefileout1').par.record = False", delayFrames=int({duration} * 60))

result = True
result
"""
            
            result = await self._send_td_command(command)
            
            if result:
                return f"Started recording to {output_path} for {duration} seconds"
            else:
                raise Exception("Failed to start recording")
                
        except Exception as e:
            return f"Error recording output: {str(e)}"
    
    async def send_osc_message(self, address: str, *args) -> str:
        """Send OSC message."""
        try:
            if self.osc_client:
                self.osc_client.send_message(address, args)
                return f"Sent OSC message to {address} with args: {args}"
            else:
                # Fallback: use TouchDesigner's OSC out
                command = f"""
osc_out = op('/oscout1')
if not osc_out:
    osc_out = op('/').create(oscoutOP, 'oscout1')

osc_out.sendOSC('{address}', {list(args)})
result = True
result
"""
                
                result = await self._send_td_command(command)
                return f"Sent OSC message via TouchDesigner: {address}"
                
        except Exception as e:
            return f"Error sending OSC message: {str(e)}"
    
    async def get_network_info(self, network_path: str = "/") -> str:
        """Get information about a network."""
        try:
            command = f"""
network = op('{network_path}')
if network:
    info = {{
        'name': network.name,
        'path': network.path,
        'children_count': len(network.children),
        'children': [child.name for child in network.children[:10]]  # Limit to first 10
    }}
    result = info
else:
    result = None
result
"""
            
            result = await self._send_td_command(command)
            
            if result:
                return json.dumps(result, indent=2)
            else:
                raise Exception(f"Network {network_path} not found")
                
        except Exception as e:
            return f"Error getting network info: {str(e)}"
    
    async def delete_operator(self, op_path: str) -> str:
        """Delete an operator."""
        try:
            command = f"""
op_node = op('{op_path}')
if op_node:
    op_node.destroy()
    result = True
else:
    result = False
result
"""
            
            result = await self._send_td_command(command)
            
            if result:
                return f"Deleted operator: {op_path}"
            else:
                raise Exception(f"Operator {op_path} not found")
                
        except Exception as e:
            return f"Error deleting operator: {str(e)}"
    
    async def copy_operator(self, source_path: str, dest_parent: str, new_name: str = None) -> str:
        """Copy an operator to a new location."""
        try:
            if not new_name:
                new_name = f"copy_{source_path.split('/')[-1]}"
            
            command = f"""
source_op = op('{source_path}')
dest_parent_op = op('{dest_parent}')

if source_op and dest_parent_op:
    copied_op = source_op.copy(dest_parent_op, name='{new_name}')
    result = copied_op.path if copied_op else None
else:
    result = None
result
"""
            
            result = await self._send_td_command(command)
            
            if result:
                return f"Copied {source_path} to {result}"
            else:
                raise Exception("Failed to copy operator")
                
        except Exception as e:
            return f"Error copying operator: {str(e)}"

TouchDesigner TCP Server Setup Script

# td_tcp_server.py - Script to run inside TouchDesigner
import socket
import threading
import json

class TouchDesignerTCPServer:
    def __init__(self, host='localhost', port=7000):
        self.host = host
        self.port = port
        self.server_socket = None
        self.running = False
        
    def start_server(self):
        """Start the TCP server."""
        self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.server_socket.bind((self.host, self.port))
        self.server_socket.listen(5)
        self.running = True
        
        print(f"TouchDesigner TCP Server started on {self.host}:{self.port}")
        
        while self.running:
            try:
                client_socket, address = self.server_socket.accept()
                client_thread = threading.Thread(
                    target=self.handle_client, 
                    args=(client_socket, address)
                )
                client_thread.daemon = True
                client_thread.start()
            except Exception as e:
                if self.running:
                    print(f"Server error: {e}")
                break
    
    def handle_client(self, client_socket, address):
        """Handle client connection."""
        print(f"Client connected from {address}")
        
        try:
            while self.running:
                data = client_socket.recv(4096).decode()
                if not data:
                    break
                
                try:
                    message = json.loads(data.strip())
                    command = message.get('command', '')
                    
                    # Execute Python command in TouchDesigner context
                    result = eval(command)
                    
                    response = json.dumps({'result': result, 'error': None})
                    client_socket.send(response.encode())
                    
                except Exception as e:
                    error_response = json.dumps({'result': None, 'error': str(e)})
                    client_socket.send(error_response.encode())
                    
        except Exception as e:
            print(f"Client handler error: {e}")
        finally:
            client_socket.close()
            print(f"Client {address} disconnected")
    
    def stop_server(self):
        """Stop the TCP server."""
        self.running = False
        if self.server_socket:
            self.server_socket.close()

# Global server instance
td_server = TouchDesignerTCPServer()

# Start server in background thread
server_thread = threading.Thread(target=td_server.start_server)
server_thread.daemon = True
server_thread.start()

Configuration Example

# config/servers.yaml
touchdesigner:
  enabled: true
  transport: stdio
  connection:
    host: localhost
    port: 7000
    osc_port: 7001
    td_path: "C:/Program Files/Derivative/TouchDesigner/bin/TouchDesigner.exe"
    timeout: 30
  features:
    - operator_management
    - parameter_control
    - media_processing
    - live_performance
    - osc_communication
    - real_time_control

Entry Point Script

# src/lightfast_mcp/servers/touchdesigner_mcp_server.py
#!/usr/bin/env python3
"""TouchDesigner MCP Server entry point."""

import asyncio
from lightfast_mcp.servers.touchdesigner.server import TouchDesignerMCPServer
from lightfast_mcp.core.server_config import ServerConfig

async def main():
    config = ServerConfig.from_env("touchdesigner")
    server = TouchDesignerMCPServer(config)
    await server.run()

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

Setup Requirements

1. TouchDesigner Installation: Ensure TouchDesigner is installed and licensed
2. TCP Server Setup: Load and run the TCP server script in TouchDesigner
3. OSC Configuration: Configure OSC input/output in TouchDesigner
4. Python Dependencies: Install python-osc for OSC communication
5. Network Configuration: Ensure proper network access between MCP server and TouchDesigner

This implementation provides comprehensive TouchDesigner automation through MCP, enabling AI-driven interactive media creation and live performance workflows.