MCP Python Interpreter

Instructions

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'."
• mcp_python_interpreter/server.py:56-120 (helper)

  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 }
• mcp_python_interpreter/server.py:259-306 (helper)

  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)
• mcp_python_interpreter/server.py:124-129 (helper)

  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]