Implementing a Custom Code Interpreter for LibreChat using HuggingFace Spaces

LibreChat is a powerful open-source AI chat platform that supports various AI models and extensions. This guide will walk you through implementing a complete code execution system using HuggingFace Spaces with SSE (Server-Sent Events) for real-time communication with LibreChat’s MCP (Model Context Protocol) infrastructure.

Overview

We’ll create a code interpreter that:

Executes Python, JavaScript, and shell code safely
Deploys as a private HuggingFace Space
Uses SSE for real-time communication with LibreChat
Maintains persistent sessions with variable state
Handles file uploads and visualizations

Architecture

Our implementation consists of:

HuggingFace Space with Gradio + MCP SSE Server - Handles code execution and SSE communication
LibreChat MCP Configuration - Connects to the HF Space via SSE
Authentication - Uses HF tokens for private space access

HuggingFace Space Implementation

Main Application (`app.py`)

#!/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
    )

Requirements File (`requirements.txt`)

gradio>=4.0.0
matplotlib>=3.7.0
numpy>=1.24.0
pandas>=2.0.0

Space Configuration (`README.md`)

---
title: LibreChat Code Interpreter
emoji: 🐍
colorFrom: blue
colorTo: green
sdk: gradio
sdk_version: 4.0.0
app_file: app.py
pinned: false
license: mit
private: true
---
 
# LibreChat Code Interpreter
 
A secure code execution environment for LibreChat using MCP (Model Context Protocol).
 
## Features
 
- Execute Python, JavaScript, and Bash code safely
- Matplotlib visualization support
- Session management for persistent context
- Security restrictions and sandboxing
- SSE integration with LibreChat
 
## Security
 
This space implements multiple security measures:
- Code analysis for dangerous patterns
- Execution timeouts (30 seconds)
- Output length limits
- Blocked dangerous imports and commands
- Isolated execution environment
 
## Usage
 
This space is designed to work with LibreChat's MCP integration. Configure your LibreChat instance to connect via SSE.

Space Secrets Configuration

In your HuggingFace Space settings, you may want to add these secrets:

EXECUTION_TIMEOUT: Maximum execution time (default: 30)
MAX_OUTPUT_LENGTH: Maximum output length (default: 10000)
MAX_SESSIONS: Maximum concurrent sessions (default: 10)

LibreChat Configuration

Update `librechat.yaml`

version: 1.2.4
 
# Your existing configuration...
 
mcpServers:
  code_interpreter:
    type: sse
    url: https://YOUR_USERNAME-librechat-code-interpreter.hf.space/gradio_api/mcp/sse
    headers:
      Authorization: "Bearer ${HF_TOKEN}"
      Content-Type: "application/json"
    serverInstructions: |
      LibreChat Code Interpreter Instructions:
      
      Available Functions:
      - execute_python_code(code, session_id=None): Execute Python code with matplotlib support
      - execute_javascript_code(code, session_id=None): Execute JavaScript with Node.js
      - execute_bash_command(command, session_id=None): Execute safe bash commands
      - create_execution_session(): Create persistent session for stateful execution
      - get_execution_history(session_id): View execution history
      - get_system_info(): Get available packages and system information
      
      Security Features:
      - 30-second execution timeout
      - Blocked dangerous imports/commands
      - Output length limits (10KB)
      - Isolated execution environment
      
      Visualization Support:
      - Matplotlib plots automatically captured as base64 images
      - Use plt.show() to generate visualizations
      - Supports numpy, pandas for data processing
      
      Session Management:
      - Create sessions for persistent variable state
      - Sessions auto-expire after 1 hour
      - Maximum 10 concurrent sessions per space
      
      Usage Tips:
      1. Always use create_execution_session() for multi-step code execution
      2. Sessions maintain variable state between executions
      3. Use get_system_info() to check available packages
      4. Matplotlib plots are automatically captured and returned as base64

Environment Variables

Add to your LibreChat .env file:

# HuggingFace token for private space access
HF_TOKEN=hf_your_token_here

Deployment Steps

1. Create HuggingFace Space

Go to HuggingFace Spaces
Choose “Gradio” as the SDK
Set the space to Private
Name it something like librechat-code-interpreter

2. Upload Files

Upload these files to your space:

app.py (main application)
requirements.txt (dependencies)
README.md (space configuration)

3. Configure LibreChat

Get your HuggingFace token from Settings > Access Tokens
Update your librechat.yaml with the space URL and token
Add the HF_TOKEN to your environment variables

4. Test the Integration

Start LibreChat
Create a new conversation
Ask the AI to execute some code:

Execute this Python code:
```python
import matplotlib.pyplot as plt
import numpy as np

x = np.linspace(0, 2*np.pi, 100)
y = np.sin(x)

plt.plot(x, y)
plt.title('Sine Wave')
plt.show()

print("Hello from LibreChat Code Interpreter!")


## Advanced Features

### Session Persistence Example

```python
# Create a session first
session_result = create_execution_session()
session_id = json.loads(session_result)["session_id"]

# Execute code in the session
code1 = """
import numpy as np
data = np.array([1, 2, 3, 4, 5])
mean_value = np.mean(data)
print(f"Mean: {mean_value}")
"""

result1 = execute_python_code(code1, session_id)

# Continue in the same session - variables persist
code2 = """
# 'data' and 'mean_value' are still available
std_value = np.std(data)
print(f"Standard deviation: {std_value}")
print(f"Data range: {np.max(data) - np.min(data)}")
"""

result2 = execute_python_code(code2, session_id)

Visualization Examples

# Multiple plot types
execute_python_code("""
import matplotlib.pyplot as plt
import numpy as np
 
fig, ((ax1, ax2), (ax3, ax4)) = plt.subplots(2, 2, figsize=(12, 8))
 
# Line plot
x = np.linspace(0, 10, 100)
ax1.plot(x, np.sin(x))
ax1.set_title('Line Plot')
 
# Scatter plot
x2 = np.random.randn(50)
y2 = np.random.randn(50)
ax2.scatter(x2, y2, alpha=0.6)
ax2.set_title('Scatter Plot')
 
# Bar chart
categories = ['A', 'B', 'C', 'D']
values = [23, 45, 56, 78]
ax3.bar(categories, values)
ax3.set_title('Bar Chart')
 
# Histogram
data = np.random.normal(0, 1, 1000)
ax4.hist(data, bins=30, alpha=0.7)
ax4.set_title('Histogram')
 
plt.tight_layout()
plt.show()
""")

Security Considerations

The implementation includes several security measures:

Code Analysis: Scans for dangerous imports and patterns
Command Blocking: Prevents execution of dangerous bash commands
Execution Limits: 30-second timeout and output size limits
Isolation: Code runs in temporary directories
Session Management: Automatic cleanup of old sessions
Private Space: Access controlled via HuggingFace tokens

Monitoring and Maintenance

Checking Space Logs

Monitor your HuggingFace Space logs for:

Execution errors
Security violations
Performance issues
Session management

Usage Limits

HuggingFace Spaces have resource limits:

CPU time
Memory usage
Storage space
Network bandwidth

Monitor these in your Space settings and upgrade if needed.

Troubleshooting

Common Issues

404 Error: Ensure space is running and URL is correct
Authentication Error: Check HF_TOKEN is valid and has access
Timeout Issues: Code execution exceeds 30-second limit
Import Errors: Package not available in the space environment

Debug Steps

Test the space directly in the browser
Check the space logs for errors
Verify LibreChat configuration
Test with simple code first

Conclusion

This implementation provides LibreChat with a powerful, secure code execution capability using HuggingFace Spaces. The SSE-based architecture ensures real-time communication while maintaining security through sandboxing and access controls.

Key benefits:

Secure: Multiple security layers and isolation
Scalable: Leverages HuggingFace infrastructure
Visual: Full matplotlib support with automatic plot capture
Persistent: Session management for stateful execution
Private: Token-based access control
Cost-effective: Uses HuggingFace’s free tier for small workloads

Erick Santana

Explorer

Implementing a Custom Code Interpreter for LibreChat using HuggingFace Spaces

Overview

Architecture

HuggingFace Space Implementation

Main Application (`app.py`)

Requirements File (`requirements.txt`)

Space Configuration (`README.md`)

Space Secrets Configuration

LibreChat Configuration

Update `librechat.yaml`

Environment Variables

Deployment Steps

1. Create HuggingFace Space

2. Upload Files

3. Configure LibreChat

4. Test the Integration

Visualization Examples

Security Considerations

Monitoring and Maintenance

Checking Space Logs

Usage Limits

Troubleshooting

Common Issues

Debug Steps

Conclusion

Graph View

Table of Contents

Backlinks

Erick Santana

Explorer

Implementing a Custom Code Interpreter for LibreChat using HuggingFace Spaces

Overview

Architecture

HuggingFace Space Implementation

Main Application (app.py)

Requirements File (requirements.txt)

Space Configuration (README.md)

Space Secrets Configuration

LibreChat Configuration

Update librechat.yaml

Environment Variables

Deployment Steps

1. Create HuggingFace Space

2. Upload Files

3. Configure LibreChat

4. Test the Integration

Visualization Examples

Security Considerations

Monitoring and Maintenance

Checking Space Logs

Usage Limits

Troubleshooting

Common Issues

Debug Steps

Conclusion

Graph View

Table of Contents

Backlinks

Main Application (`app.py`)

Requirements File (`requirements.txt`)

Space Configuration (`README.md`)

Update `librechat.yaml`