add code interpreter service files

python_service/app.py

@@ -0,0 +1,577 @@
+#!/usr/bin/env python3
+"""
+LibreChat Custom Code Interpreter - HuggingFace Space
+A secure code execution server using Gradio + MCP with SSE support
+"""
+ 
+import json
+import subprocess
+import tempfile
+import os
+import sys
+import uuid
+import time
+import base64
+import io
+from pathlib import Path
+from typing import Dict, Any, Optional, List
+import gradio as gr
+import matplotlib
+matplotlib.use('Agg')  # Use non-interactive backend
+import matplotlib.pyplot as plt
+import numpy as np
+import pandas as pd
+ 
+class SecureCodeExecutor:
+    def __init__(self):
+        self.sessions = {}
+        self.max_execution_time = 30
+        self.max_output_length = 10000
+        self.allowed_languages = ["python", "javascript", "bash"]
+        
+        # Security: List of blocked commands/imports
+        self.blocked_imports = [
+            'subprocess', 'os', 'sys', 'shutil', 'glob', 'pickle', 
+            'marshal', 'imp', 'importlib', '__import__'
+        ]
+        self.blocked_bash_commands = [
+            'rm', 'sudo', 'chmod', 'chown', 'dd', 'mkfs', 'fdisk',
+            'curl', 'wget', 'ssh', 'scp', 'nc', 'netcat'
+        ]
+ 
+    def create_session(self) -> str:
+        """Create a new execution session"""
+        session_id = str(uuid.uuid4())[:8]  # Shorter ID for HF
+        self.sessions[session_id] = {
+            'created_at': time.time(),
+            'variables': {},
+            'history': [],
+            'files': {}
+        }
+        return session_id
+ 
+    def cleanup_old_sessions(self):
+        """Remove sessions older than 1 hour"""
+        current_time = time.time()
+        old_sessions = [
+            sid for sid, session in self.sessions.items()
+            if current_time - session['created_at'] > 3600
+        ]
+        for sid in old_sessions:
+            del self.sessions[sid]
+ 
+    def is_code_safe(self, code: str, language: str) -> tuple[bool, str]:
+        """Check if code is safe to execute"""
+        if language == "python":
+            # Check for blocked imports
+            for blocked in self.blocked_imports:
+                if blocked in code:
+                    return False, f"Blocked import/function: {blocked}"
+            
+            # Check for dangerous patterns
+            dangerous_patterns = ['exec(', 'eval(', 'open(', 'file(', '__']
+            for pattern in dangerous_patterns:
+                if pattern in code:
+                    return False, f"Dangerous pattern detected: {pattern}"
+        
+        elif language == "bash":
+            # Check for blocked commands
+            for blocked in self.blocked_bash_commands:
+                if blocked in code.lower():
+                    return False, f"Blocked command: {blocked}"
+        
+        return True, ""
+ 
+    def execute_python_code(self, code: str, session_id: Optional[str] = None) -> Dict[str, Any]:
+        """Execute Python code with visualization support"""
+        # Security check
+        is_safe, reason = self.is_code_safe(code, "python")
+        if not is_safe:
+            return {
+                "success": False,
+                "stdout": "",
+                "stderr": f"Security violation: {reason}",
+                "execution_time": time.time()
+            }
+ 
+        # Prepare execution environment
+        setup_code = '''
+import matplotlib
+matplotlib.use('Agg')
+import matplotlib.pyplot as plt
+import numpy as np
+import pandas as pd
+import json
+import math
+import random
+import base64
+import io
+from datetime import datetime, timedelta
+ 
+# Custom print function to capture output
+_output_buffer = []
+_original_print = print
+def print(*args, **kwargs):
+    _output_buffer.append(' '.join(str(arg) for arg in args))
+ 
+# Function to save plots as base64
+def save_current_plot():
+    if plt.get_fignums():  # Check if there are any figures
+        buffer = io.BytesIO()
+        plt.savefig(buffer, format='png', bbox_inches='tight', dpi=100)
+        buffer.seek(0)
+        plot_data = buffer.getvalue()
+        buffer.close()
+        return base64.b64encode(plot_data).decode()
+    return None
+'''
+ 
+        # Combine setup and user code
+        full_code = setup_code + "\n" + code + "\n"
+        
+        # Add plot capture if plotting commands detected
+        if any(cmd in code for cmd in ['plt.', 'plot(', 'scatter(', 'bar(', 'hist(']):
+            full_code += "\n_plot_data = save_current_plot()\nif _plot_data: _output_buffer.append('PLOT_DATA:' + _plot_data)\n"
+ 
+        try:
+            # Create temporary file
+            with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
+                f.write(full_code)
+                temp_file = f.name
+ 
+            # Execute with timeout
+            result = subprocess.run(
+                [sys.executable, temp_file],
+                capture_output=True,
+                text=True,
+                timeout=self.max_execution_time,
+                cwd=tempfile.gettempdir()
+            )
+ 
+            # Process output
+            stdout = result.stdout
+            stderr = result.stderr
+            plot_data = None
+ 
+            # Extract plot data if present
+            if 'PLOT_DATA:' in stdout:
+                lines = stdout.split('\n')
+                clean_lines = []
+                for line in lines:
+                    if line.startswith('PLOT_DATA:'):
+                        plot_data = line.replace('PLOT_DATA:', '')
+                    else:
+                        clean_lines.append(line)
+                stdout = '\n'.join(clean_lines)
+ 
+            # Limit output length
+            if len(stdout) > self.max_output_length:
+                stdout = stdout[:self.max_output_length] + "\n... (output truncated)"
+ 
+            execution_result = {
+                "success": result.returncode == 0,
+                "stdout": stdout.strip(),
+                "stderr": stderr.strip() if stderr else "",
+                "execution_time": time.time(),
+                "return_code": result.returncode
+            }
+ 
+            if plot_data:
+                execution_result["plot"] = plot_data
+ 
+            return execution_result
+ 
+        except subprocess.TimeoutExpired:
+            return {
+                "success": False,
+                "stdout": "",
+                "stderr": "Execution timed out (30s limit)",
+                "execution_time": time.time()
+            }
+        except Exception as e:
+            return {
+                "success": False,
+                "stdout": "",
+                "stderr": str(e),
+                "execution_time": time.time()
+            }
+        finally:
+            if 'temp_file' in locals():
+                try:
+                    os.unlink(temp_file)
+                except:
+                    pass
+ 
+    def execute_javascript_code(self, code: str, session_id: Optional[str] = None) -> Dict[str, Any]:
+        """Execute JavaScript code using Node.js"""
+        # Security check
+        is_safe, reason = self.is_code_safe(code, "javascript")
+        if not is_safe:
+            return {
+                "success": False,
+                "stdout": "",
+                "stderr": f"Security violation: {reason}",
+                "execution_time": time.time()
+            }
+ 
+        try:
+            with tempfile.NamedTemporaryFile(mode='w', suffix='.js', delete=False) as f:
+                f.write(code)
+                temp_file = f.name
+ 
+            result = subprocess.run(
+                ['node', temp_file],
+                capture_output=True,
+                text=True,
+                timeout=self.max_execution_time
+            )
+ 
+            stdout = result.stdout
+            if len(stdout) > self.max_output_length:
+                stdout = stdout[:self.max_output_length] + "\n... (output truncated)"
+ 
+            return {
+                "success": result.returncode == 0,
+                "stdout": stdout.strip(),
+                "stderr": result.stderr.strip() if result.stderr else "",
+                "execution_time": time.time(),
+                "return_code": result.returncode
+            }
+ 
+        except subprocess.TimeoutExpired:
+            return {
+                "success": False,
+                "stdout": "",
+                "stderr": "Execution timed out (30s limit)",
+                "execution_time": time.time()
+            }
+        except Exception as e:
+            return {
+                "success": False,
+                "stdout": "",
+                "stderr": str(e),
+                "execution_time": time.time()
+            }
+        finally:
+            if 'temp_file' in locals():
+                try:
+                    os.unlink(temp_file)
+                except:
+                    pass
+ 
+    def execute_bash_command(self, command: str, session_id: Optional[str] = None) -> Dict[str, Any]:
+        """Execute bash commands with security restrictions"""
+        # Security check
+        is_safe, reason = self.is_code_safe(command, "bash")
+        if not is_safe:
+            return {
+                "success": False,
+                "stdout": "",
+                "stderr": f"Security violation: {reason}",
+                "execution_time": time.time()
+            }
+ 
+        try:
+            result = subprocess.run(
+                command,
+                shell=True,
+                capture_output=True,
+                text=True,
+                timeout=self.max_execution_time,
+                cwd=tempfile.gettempdir()
+            )
+ 
+            stdout = result.stdout
+            if len(stdout) > self.max_output_length:
+                stdout = stdout[:self.max_output_length] + "\n... (output truncated)"
+ 
+            return {
+                "success": result.returncode == 0,
+                "stdout": stdout.strip(),
+                "stderr": result.stderr.strip() if result.stderr else "",
+                "execution_time": time.time(),
+                "return_code": result.returncode
+            }
+ 
+        except subprocess.TimeoutExpired:
+            return {
+                "success": False,
+                "stdout": "",
+                "stderr": "Command timed out (30s limit)",
+                "execution_time": time.time()
+            }
+        except Exception as e:
+            return {
+                "success": False,
+                "stdout": "",
+                "stderr": str(e),
+                "execution_time": time.time()
+            }
+ 
+    def execute_code(self, code: str, language: str = "python", session_id: Optional[str] = None) -> str:
+        """Main execution function - returns JSON for MCP compatibility"""
+        # Cleanup old sessions periodically
+        if len(self.sessions) > 10:
+            self.cleanup_old_sessions()
+ 
+        if language not in self.allowed_languages:
+            return json.dumps({
+                "success": False,
+                "error": f"Language '{language}' not supported. Allowed: {', '.join(self.allowed_languages)}"
+            })
+ 
+        # Create session if needed
+        if session_id and session_id not in self.sessions:
+            session_id = self.create_session()
+        elif not session_id:
+            session_id = self.create_session()
+ 
+        # Execute based on language
+        if language == "python":
+            result = self.execute_python_code(code, session_id)
+        elif language == "javascript":
+            result = self.execute_javascript_code(code, session_id)
+        elif language == "bash":
+            result = self.execute_bash_command(code, session_id)
+        else:
+            result = {
+                "success": False,
+                "error": f"Execution handler for {language} not implemented"
+            }
+ 
+        # Store in session history
+        if session_id in self.sessions:
+            self.sessions[session_id]['history'].append({
+                'code': code,
+                'language': language,
+                'result': result,
+                'timestamp': time.time()
+            })
+ 
+        result['session_id'] = session_id
+        return json.dumps(result, indent=2)
+ 
+# Global executor instance
+executor = SecureCodeExecutor()
+ 
+# MCP Functions
+def execute_python_code(code: str, session_id: str = None) -> str:
+    """
+    Execute Python code safely with visualization support.
+    
+    Args:
+        code (str): Python code to execute
+        session_id (str, optional): Session ID for persistent context
+    
+    Returns:
+        str: JSON string with execution results
+    """
+    return executor.execute_code(code, "python", session_id)
+ 
+def execute_javascript_code(code: str, session_id: str = None) -> str:
+    """
+    Execute JavaScript code using Node.js.
+    
+    Args:
+        code (str): JavaScript code to execute
+        session_id (str, optional): Session ID for persistent context
+    
+    Returns:
+        str: JSON string with execution results
+    """
+    return executor.execute_code(code, "javascript", session_id)
+ 
+def execute_bash_command(command: str, session_id: str = None) -> str:
+    """
+    Execute bash commands with security restrictions.
+    
+    Args:
+        command (str): Bash command to execute
+        session_id (str, optional): Session ID for persistent context
+    
+    Returns:
+        str: JSON string with execution results
+    """
+    return executor.execute_code(command, "bash", session_id)
+ 
+def create_execution_session() -> str:
+    """
+    Create a new execution session for maintaining state.
+    
+    Returns:
+        str: JSON string containing new session ID
+    """
+    session_id = executor.create_session()
+    return json.dumps({"session_id": session_id, "created_at": time.time()})
+ 
+def list_execution_sessions() -> str:
+    """
+    List all active execution sessions.
+    
+    Returns:
+        str: JSON string containing session information
+    """
+    return json.dumps({
+        "sessions": list(executor.sessions.keys()),
+        "count": len(executor.sessions),
+        "timestamp": time.time()
+    })
+ 
+def get_execution_history(session_id: str) -> str:
+    """
+    Get execution history for a specific session.
+    
+    Args:
+        session_id (str): Session ID to get history for
+    
+    Returns:
+        str: JSON string containing execution history
+    """
+    if session_id not in executor.sessions:
+        return json.dumps({"error": "Session not found"})
+    
+    return json.dumps({
+        "session_id": session_id,
+        "history": executor.sessions[session_id]['history'],
+        "created_at": executor.sessions[session_id]['created_at']
+    })
+ 
+def get_system_info() -> str:
+    """
+    Get system information and available packages.
+    
+    Returns:
+        str: JSON string containing system information
+    """
+    return json.dumps({
+        "python_version": sys.version,
+        "available_packages": [
+            "numpy", "pandas", "matplotlib", "json", "math", 
+            "random", "datetime", "base64", "io"
+        ],
+        "execution_limits": {
+            "max_time": executor.max_execution_time,
+            "max_output": executor.max_output_length
+        },
+        "supported_languages": executor.allowed_languages
+    })
+ 
+# Gradio Interface
+def gradio_execute_code(code: str, language: str, session_id: str = ""):
+    """Gradio interface for code execution"""
+    if not session_id:
+        session_id = None
+    
+    result_json = executor.execute_code(code, language.lower(), session_id)
+    result = json.loads(result_json)
+    
+    output = ""
+    if result.get("success"):
+        if result.get("stdout"):
+            output += f"Output:\n{result['stdout']}\n\n"
+        if result.get("stderr"):
+            output += f"Warnings:\n{result['stderr']}\n\n"
+        if result.get("plot"):
+            output += f"Plot generated (base64): {result['plot'][:100]}...\n\n"
+    else:
+        output += f"Error:\n{result.get('stderr', result.get('error', 'Unknown error'))}\n\n"
+    
+    output += f"Session ID: {result.get('session_id', 'N/A')}"
+    
+    return output
+ 
+# Create Gradio interface
+with gr.Blocks(title="LibreChat Code Interpreter") as demo:
+    gr.Markdown("# LibreChat Code Interpreter")
+    gr.Markdown("Execute Python, JavaScript, and Bash code safely through MCP integration.")
+    
+    with gr.Row():
+        with gr.Column():
+            code_input = gr.Textbox(
+                placeholder="Enter your code here...",
+                lines=10,
+                label="Code"
+            )
+            language_dropdown = gr.Dropdown(
+                choices=["Python", "JavaScript", "Bash"],
+                value="Python",
+                label="Language"
+            )
+            session_input = gr.Textbox(
+                placeholder="Optional: Session ID for persistent context",
+                label="Session ID"
+            )
+            execute_btn = gr.Button("Execute Code", variant="primary")
+        
+        with gr.Column():
+            output_display = gr.Textbox(
+                lines=15,
+                label="Execution Result",
+                interactive=False
+            )
+    
+    # Examples
+    gr.Markdown("## Examples")
+    
+    example_python = gr.Code("""
+import matplotlib.pyplot as plt
+import numpy as np
+ 
+# Generate sample data
+x = np.linspace(0, 10, 100)
+y = np.sin(x) * np.exp(-x/10)
+ 
+# Create plot
+plt.figure(figsize=(10, 6))
+plt.plot(x, y, 'b-', linewidth=2, label='Damped Sine Wave')
+plt.title('Example Visualization')
+plt.xlabel('X values')
+plt.ylabel('Y values')
+plt.legend()
+plt.grid(True, alpha=0.3)
+plt.show()
+ 
+print("Visualization created successfully!")
+""", language="python", label="Python Example with Visualization")
+    
+    example_js = gr.Code("""
+// Data processing example
+const data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
+ 
+const sum = data.reduce((acc, val) => acc + val, 0);
+const mean = sum / data.length;
+const variance = data.reduce((acc, val) => acc + Math.pow(val - mean, 2), 0) / data.length;
+const stdDev = Math.sqrt(variance);
+ 
+console.log(`Dataset: [${data.join(', ')}]`);
+console.log(`Sum: ${sum}`);
+console.log(`Mean: ${mean}`);
+console.log(`Standard Deviation: ${stdDev.toFixed(3)}`);
+ 
+// JSON processing
+const result = {
+    dataset: data,
+    statistics: {
+        sum, mean, variance, stdDev
+    },
+    timestamp: new Date().toISOString()
+};
+ 
+console.log('\\nResult:');
+console.log(JSON.stringify(result, null, 2));
+""", language="javascript", label="JavaScript Example")
+    
+    execute_btn.click(
+        fn=gradio_execute_code,
+        inputs=[code_input, language_dropdown, session_input],
+        outputs=[output_display]
+    )
+ 
+if __name__ == "__main__":
+    # Launch with MCP server enabled
+    demo.launch(
+        mcp_server=True,
+        share=False,
+        server_name="0.0.0.0",
+        server_port=7860
+    )

python_service/requirements.txt

@@ -0,0 +1,6 @@
+gradio>=4.0.0
+matplotlib>=3.7.0
+numpy>=1.24.0
+pandas>=2.0.0
+#plotly>=5.15.0
+#narwhal>=0.1.0