MCP Odoo Shell

""" Odoo Shell Manager ================== Manages a persistent Odoo shell subprocess with threaded I/O handling. """ import os import queue import subprocess import threading import time from typing import Dict, Any, Optional class OdooShellManager: """ Manages a persistent Odoo shell subprocess. This class handles the lifecycle of an Odoo shell process, including: - Starting the shell with appropriate configuration - Managing input/output communication through queues - Executing code and capturing results - Handling process lifecycle :param odoo_bin_path: Path to the odoo-bin executable :type odoo_bin_path: str :param addons_path: Comma-separated list of addon directories :type addons_path: str :param db_name: Name of the Odoo database to connect to :type db_name: str :param config_file: Optional path to Odoo configuration file :type config_file: Optional[str] """ def __init__(self, odoo_bin_path: str, addons_path: str, db_name: str, config_file: Optional[str] = None): self.odoo_bin_path = odoo_bin_path self.addons_path = addons_path self.db_name = db_name self.config_file: Optional[str] = config_file self.process: Optional[subprocess.Popen] = None self.input_queue: queue.Queue[str] = queue.Queue() self.output_queue: queue.Queue[str] = queue.Queue() self.session_vars: Dict[str, Any] = {} def start_shell(self) -> None: """ Start the Odoo shell subprocess. Constructs the command line arguments and starts the Odoo shell process with appropriate configuration. Also starts input/output handling threads and waits for the initial shell prompt. :raises TimeoutError: If the shell doesn't start within the expected timeframe """ cmd = [ self.odoo_bin_path, 'shell', '--addons-path', self.addons_path, '--database', self.db_name, '--no-http' ] if self.config_file: cmd.extend(['--config', self.config_file]) self.process = subprocess.Popen( cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, universal_newlines=True ) # Start threads to handle input/output threading.Thread(target=self._input_thread, daemon=True).start() threading.Thread(target=self._output_thread, daemon=True).start() # Wait for initial prompt self._wait_for_prompt() def _input_thread(self) -> None: """ Handle input to Odoo shell. Runs in a separate thread to send code from the input queue to the Odoo shell process stdin. """ while self.process and self.process.poll() is None: try: code = self.input_queue.get(timeout=1) if code: self.process.stdin.write(code + '\n') self.process.stdin.flush() except queue.Empty: continue def _output_thread(self) -> None: """ Handle output from Odoo shell. Runs in a separate thread to read output from the Odoo shell process stdout and place it in the output queue for consumption. """ buffer = [] while self.process and self.process.poll() is None: try: char = self.process.stdout.read(1) if char: buffer.append(char) if char == '\n' or len(buffer) > 1000: output = ''.join(buffer) self.output_queue.put(output) buffer = [] except Exception as e: print(f"Error reading shell output: {e}") break def _wait_for_prompt(self) -> None: """ Wait for the Odoo shell prompt to appear. Monitors the output queue until a recognizable shell prompt is detected, indicating that the shell is ready to accept commands. :raises TimeoutError: If no prompt is detected within 10 seconds """ output = "" while True: try: chunk = self.output_queue.get(timeout=10) output += chunk # Look for the typical Odoo shell prompt if ">>>" in output or "In [" in output: break except queue.Empty: raise TimeoutError("Timeout waiting for Odoo shell prompt") def execute_code(self, code: str, timeout: int = 30) -> str: """ Execute code in the Odoo shell and return output. Sends the provided code to the Odoo shell process and collects the resulting output until a new prompt appears or timeout occurs. :param code: Python code to execute in the Odoo shell context :type code: str :param timeout: Maximum time to wait for execution completion in seconds :type timeout: int :return: The output from executing the code :rtype: str :raises TimeoutError: If execution doesn't complete within timeout """ if not self.process or self.process.poll() is not None: self.start_shell() # Clear output queue while not self.output_queue.empty(): self.output_queue.get() # Send code self.input_queue.put(code) # Collect output output_lines = [] start_time = time.time() while True: try: chunk = self.output_queue.get(timeout=1) output_lines.append(chunk) # Check if we got a new prompt (indicating completion) if (">>>" in chunk or "In [" in chunk) and len(output_lines) > 1: break # Timeout check if time.time() - start_time > timeout: raise TimeoutError(f"Command execution timed out after {timeout} seconds") except queue.Empty: if time.time() - start_time > timeout: raise TimeoutError(f"Command execution timed out after {timeout} seconds") continue return ''.join(output_lines).strip() def stop(self) -> None: """ Stop the Odoo shell. Terminates the Odoo shell process and waits for it to exit cleanly. """ if self.process: self.process.terminate() self.process.wait() self.process = None def __enter__(self): """Context manager entry.""" self.start_shell() return self def __exit__(self, exc_type, exc_val, exc_tb): """Context manager exit.""" self.stop()