run_python_code
Execute Python code with flexible execution modes including inline for speed and subprocess for isolation. Use to run Python scripts, test code snippets, or maintain session state across executions.
Instructions
Execute Python code with flexible execution modes.
Args:
code: Python code to execute
execution_mode: Execution mode - "inline" (default, fast, in-process) or "subprocess" (isolated)
session_id: Session ID for inline mode to maintain state across executions
environment: Python environment name (only for subprocess mode)
save_as: Optional filename to save the code before execution
timeout: Maximum execution time in seconds (only enforced for subprocess mode)
Returns:
Execution result with output
Execution modes:
- "inline" (default): Executes code in the current process. Fast and reliable,
maintains session state. Use for most code execution tasks.
- "subprocess": Executes code in a separate Python process. Use when you need
environment isolation or a different Python environment.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| code | Yes | ||
| execution_mode | No | inline | |
| session_id | No | default | |
| environment | No | system | |
| save_as | No | ||
| timeout | No |
Implementation Reference
- mcp_python_interpreter/server.py:513-625 (handler)The primary handler function for the 'run_python_code' tool. Decorated with @mcp.tool() for automatic registration and schema inference from signature/docstring. Handles both 'inline' REPL execution (persistent sessions) and 'subprocess' isolated execution.
@mcp.tool() async def run_python_code( code: str, execution_mode: str = "inline", session_id: str = "default", environment: str = "system", save_as: Optional[str] = None, timeout: int = 300 ) -> str: """ Execute Python code with flexible execution modes. Args: code: Python code to execute execution_mode: Execution mode - "inline" (default, fast, in-process) or "subprocess" (isolated) session_id: Session ID for inline mode to maintain state across executions environment: Python environment name (only for subprocess mode) save_as: Optional filename to save the code before execution timeout: Maximum execution time in seconds (only enforced for subprocess mode) Returns: Execution result with output Execution modes: - "inline" (default): Executes code in the current process. Fast and reliable, maintains session state. Use for most code execution tasks. - "subprocess": Executes code in a separate Python process. Use when you need environment isolation or a different Python environment. """ # Save code if requested if save_as: save_path = WORKING_DIR / save_as if not save_path.suffix == '.py': save_path = save_path.with_suffix('.py') try: save_path.parent.mkdir(parents=True, exist_ok=True) with open(save_path, 'w', encoding='utf-8') as f: f.write(code) except Exception as e: return f"Error saving code to file: {str(e)}" # Execute based on mode if execution_mode == "inline": # In-process execution (default, fast, no subprocess issues) try: session = get_session(session_id) result = session.execute(code, timeout) # Store in history session.history.append({ "code": code, "stdout": result["stdout"], "stderr": result["stderr"], "status": result["status"] }) output = f"Execution in session '{session_id}' (inline mode)" if save_as: output += f" (saved to {save_as})" output += ":\n\n" if result["status"] == 0: output += "--- Output ---\n" output += result["stdout"] if result["stdout"] else "(No output)\n" else: output += "--- Error ---\n" output += result["stderr"] if result["stderr"] else "(No error message)\n" if result["stdout"]: output += "\n--- Output ---\n" output += result["stdout"] return output except Exception as e: return f"Error in inline execution: {str(e)}\n{traceback.format_exc()}" elif execution_mode == "subprocess": # Subprocess execution (for environment isolation) environments = get_python_environments() if environment == "default" and not any(e["name"] == "default" for e in environments): environment = "system" env = next((e for e in environments if e["name"] == environment), None) if not env: return f"Environment '{environment}' not found. Available: {', '.join(e['name'] for e in environments)}" result = await execute_python_code_subprocess(code, env["path"], str(WORKING_DIR), timeout) output = f"Execution in '{environment}' environment (subprocess mode)" if save_as: output += f" (saved to {save_as})" output += ":\n\n" if result["status"] == 0: output += "--- Output ---\n" output += result["stdout"] if result["stdout"] else "(No output)\n" else: output += f"--- Error (status code: {result['status']}) ---\n" output += result["stderr"] if result["stderr"] else "(No error message)\n" if result["stdout"]: output += "\n--- Output ---\n" output += result["stdout"] return output else: return f"Unknown execution mode: {execution_mode}. Use 'inline' or 'subprocess'." - Core helper class for managing REPL sessions in 'inline' execution mode, providing persistent namespace and execution history.
class ReplSession: """Manages a Python REPL session with persistent state.""" def __init__(self): self.locals = { "__builtins__": builtins, "__name__": "__main__", "__doc__": None, "__package__": None, } self.history = [] def execute(self, code: str, timeout: Optional[int] = None) -> Dict[str, Any]: """ Execute Python code in this session. Args: code: Python code to execute timeout: Optional timeout (not enforced for inline execution) Returns: Dict with stdout, stderr, result, and status """ stdout_capture = StringIO() stderr_capture = StringIO() # Save original streams old_stdout, old_stderr = sys.stdout, sys.stderr sys.stdout, sys.stderr = stdout_capture, stderr_capture result_value = None status = 0 try: # Change to working directory for execution old_cwd = os.getcwd() os.chdir(WORKING_DIR) try: # Try to evaluate as expression first try: result_value = eval(code, self.locals) if result_value is not None: print(repr(result_value)) except SyntaxError: # If not an expression, execute as statement exec(code, self.locals) except Exception: traceback.print_exc() status = 1 finally: os.chdir(old_cwd) finally: # Restore original streams sys.stdout, sys.stderr = old_stdout, old_stderr return { "stdout": stdout_capture.getvalue(), "stderr": stderr_capture.getvalue(), "result": result_value, "status": status } - Helper function for executing Python code in isolated subprocess mode, used when execution_mode='subprocess'.
async def execute_python_code_subprocess( code: str, python_path: Optional[str] = None, working_dir: Optional[str] = None, timeout: int = 300 ) -> Dict[str, Any]: """Execute Python code via subprocess (for environment isolation).""" if python_path is None: python_path = DEFAULT_PYTHON_PATH temp_file = None try: fd, temp_file = tempfile.mkstemp(suffix='.py', text=True) try: with os.fdopen(fd, 'w', encoding='utf-8') as f: f.write(code) f.flush() os.fsync(f.fileno()) except Exception as e: os.close(fd) raise e if sys.platform == "win32": await asyncio.sleep(0.05) temp_file = os.path.abspath(temp_file) if working_dir: working_dir = os.path.abspath(working_dir) result = await run_subprocess_async( [python_path, temp_file], cwd=working_dir, timeout=timeout ) return result finally: if temp_file: try: if sys.platform == "win32": await asyncio.sleep(0.05) if os.path.exists(temp_file): os.unlink(temp_file) except Exception as e: print(f"Warning: Could not delete temp file {temp_file}: {e}", file=sys.stderr) - Utility to retrieve or initialize REPL sessions.
def get_session(session_id: str = "default") -> ReplSession: """Get or create a REPL session.""" if session_id not in _sessions: _sessions[session_id] = ReplSession() return _sessions[session_id]