import os import re import subprocess import tempfile from typing import Optional import asyncio import logging """ CAUTION: Use your best judgement as deploying/running code executions on your system carries inherent risks.. """ logging.basicConfig(level=logging.INFO) class Action: def __init__(self, base_env_dir='/path/to/envs/'): self.base_env_dir = base_env_dir if not os.path.exists(self.base_env_dir): os.makedirs(self.base_env_dir) self.venv_path = self.get_next_env_name() self.create_virtual_environment() def get_next_env_name(self): pattern = re.compile(r'^OUIenv(\d{3})$') envs = os.listdir(self.base_env_dir) numbers = [] for env in envs: match = pattern.match(env) if match: numbers.append(int(match.group(1))) if numbers: next_num = max(numbers) + 1 else: next_num = 1 env_name = f'OUIenv{next_num:03d}' return os.path.join(self.base_env_dir, env_name) def create_virtual_environment(self): try: subprocess.run(['python', '-m', 'venv', self.venv_path], check=True) logging.info(f"Virtual environment created at {self.venv_path}") except subprocess.CalledProcessError as e: logging.error(f"Failed to create virtual environment: {e}") raise def execute_python_code(self, code: str) -> str: logging.info(f"Executing code in {self.venv_path}") try: with tempfile.NamedTemporaryFile(delete=True, suffix='.py') as temp: temp.write(code.encode('utf-8')) temp.flush() python_executable = os.path.join(self.venv_path, 'bin', 'python') result = subprocess.run([python_executable, temp.name], capture_output=True, text=True, timeout=10) return result.stdout + result.stderr except Exception as e: return f"Error executing code: {str(e)}" async def action( self, body: dict, __user__=None, __event_emitter__=None, __event_call__=None, ) -> Optional[dict]: logging.info("Action method started") print(f"action:{__name__}") user_valves = __user__.get("valves") if __user__ else None if not user_valves: user_valves = self.UserValves() if __event_emitter__: messages = body.get("messages", []) if not messages: await __event_emitter__( { "type": "status", "data": {"description": "No messages provided", "done": True}, } ) return None last_assistant_message = messages[-1] content = last_assistant_message.get("content", "") if user_valves.show_status: await __event_emitter__( { "type": "status", "data": {"description": "Processing your input", "done": False}, } ) if content.startswith("```python") and content.endswith("```"): code = content[9:-3].strip() # Remove the ```python and ``` markers loop = asyncio.get_event_loop() output = await loop.run_in_executor(None, self.execute_python_code, code) return {"type": "code_execution_result", "data": {"output": output}} else: if user_valves.show_status: await __event_emitter__( { "type": "status", "data": { "description": "No valid Python code detected", "done": True, }, } ) return None else: return None