pyforge-integration.jsā¢4.21 kB
// PyForge IDE Integration for EGW MCP Server
// Exposes existing PyForge IDE sandbox as MCP tool for AI agents
import { exec } from 'child_process';
import path from 'path';
import { existsSync, writeFileSync, unlinkSync } from 'fs';
export class PyForgeIntegration {
constructor() {
// Dynamic path resolution - look for PyForge IDE in multiple locations
const possiblePaths = [
path.join(process.cwd(), 'pyforge-ide', 'PyForge-IDE-MCP-Server'),
path.join(process.cwd(), 'packages', 'shared', 'src', 'database-utils', 'PyForge-IDE-MCP-Server'),
'C:\\Users\\pc\\VScodeProjects\\EGW-NEW2\\egw_writings_mcp_server\\pyforge-ide\\PyForge-IDE-MCP-Server'
];
// Find first existing path
this.pyforgePath = possiblePaths.find(p => existsSync(p));
if (!this.pyforgePath) {
console.error('ā PyForge IDE not found in any of these locations:');
possiblePaths.forEach(p => console.error(` - ${p}`));
this.pyforgePath = possiblePaths[0]; // Default to first option for error messages
}
this.tempCounter = 0;
}
// Execute Python code in PyForge IDE environment
async executePythonCode(code) {
console.error('š PyForge IDE: Executing Python code...');
try {
// Create temporary Python file for execution
const tempFile = path.join(this.pyforgePath, `temp_${Date.now()}_${this.tempCounter++}.py`);
writeFileSync(tempFile, code);
const result = await this.runPythonCommand(`python "${tempFile}"`);
// Clean up temp file
try {
unlinkSync(tempFile);
} catch (e) {
// Ignore cleanup errors
}
return result;
} catch (error) {
console.error('ā PyForge IDE execution error:', error.message);
throw error;
}
}
// Install Python package
async installPackage(packageName) {
console.error(`š¦ PyForge IDE: Installing package ${packageName}...`);
return this.runPythonCommand(`pip install "${packageName}"`, 60000);
}
// List files in PyForge IDE directory
async listFiles() {
console.error('š PyForge IDE: Listing files...');
try {
const fs = await import('fs');
const files = fs.readdirSync(this.pyforgePath);
const fileDetails = [];
for (const file of files) {
const filePath = path.join(this.pyforgePath, file);
const stats = fs.statSync(filePath);
fileDetails.push({
name: file,
type: stats.isDirectory() ? 'directory' : 'file',
size: stats.size,
modified: stats.mtime.toISOString()
});
}
return {
success: true,
files: fileDetails,
total_count: fileDetails.length,
directory: this.pyforgePath
};
} catch (error) {
return {
success: false,
error: error.message,
files: []
};
}
}
// Run Python command with timeout and buffering
runPythonCommand(command, timeout = 30000) {
return new Promise((resolve, reject) => {
exec(command, {
cwd: this.pyforgePath,
timeout: timeout,
maxBuffer: 1024 * 1024 * 10 // 10MB buffer
}, (error, stdout, stderr) => {
if (error) {
resolve({
success: false,
error: error.message,
stdout: stdout || '',
stderr: stderr || '',
exit_code: error.code || -1
});
} else {
resolve({
success: true,
stdout: stdout || '',
stderr: stderr || '',
exit_code: 0,
execution_time: new Date().toISOString()
});
}
});
});
}
// Check if PyForge IDE is available
isAvailable() {
return existsSync(this.pyforgePath);
}
// Get PyForge IDE status
getStatus() {
return {
available: this.isAvailable(),
path: this.pyforgePath,
python_available: true, // Assume Python is available if PyForge IDE exists
features: [
'python_code_execution',
'package_installation',
'file_management',
'virtual_filesystem'
]
};
}
}