consult_codex
Submit a prompt to Codex and receive formatted output as text, JSON, or code. Specify a working directory for context.
Instructions
Consult Codex in non-interactive mode with structured output.
Processes prompt and returns formatted response.
Supports text, JSON, and code extraction formats.
Args:
query: The prompt to send to Codex
directory: Working directory (required)
format: Output format - "text", "json", or "code" (default: "json")
timeout: Optional timeout in seconds (overrides env var, recommended: 60-120)
Returns:
Formatted response based on format parameterInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | ||
| directory | Yes | ||
| format | No | json | |
| timeout | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/mcp_server.py:295-296 (registration)Tool registration via @mcp.tool() decorator on consult_codex function
@mcp.tool() - src/mcp_server.py:296-423 (handler)Main handler function for the consult_codex tool - executes Codex CLI with query, supports text/json/code formats
@mcp.tool() def consult_codex( query: str, directory: str, format: str = "json", timeout: Optional[int] = None ) -> str: """ Consult Codex in non-interactive mode with structured output. Processes prompt and returns formatted response. Supports text, JSON, and code extraction formats. Args: query: The prompt to send to Codex directory: Working directory (required) format: Output format - "text", "json", or "code" (default: "json") timeout: Optional timeout in seconds (overrides env var, recommended: 60-120) Returns: Formatted response based on format parameter """ # Check if codex CLI is available if not _get_codex_command(): error_response = "Error: Codex CLI not found. Install from OpenAI" if format == "json": return json.dumps({"status": "error", "error": error_response}, indent=2) return error_response # Validate directory if not os.path.isdir(directory): error_response = f"Error: Directory does not exist: {directory}" if format == "json": return json.dumps({"status": "error", "error": error_response}, indent=2) return error_response # Validate format if format not in ["text", "json", "code"]: error_response = f"Error: Invalid format '{format}'. Must be 'text', 'json', or 'code'" # Always return JSON for invalid format errors for consistency return json.dumps({"status": "error", "error": error_response}, indent=2) # Prepare query based on format if format == "json": processed_query = _format_prompt_for_json(query) else: processed_query = query # Setup command and timeout cmd = _build_codex_exec_command() if _should_skip_git_check(): cmd.append("--skip-git-repo-check") timeout_value = timeout or _get_timeout() # Execute with timing start_time = time.time() try: result = _run_codex_command(cmd, directory, timeout_value, processed_query) execution_time = time.time() - start_time if result.returncode == 0: cleaned_output = _clean_codex_output(result.stdout) raw_response = cleaned_output if cleaned_output else "No output from Codex CLI" return _format_response(raw_response, format, execution_time, directory) else: error_response = f"Codex CLI Error: {result.stderr.strip()}" if format == "json": return json.dumps({ "status": "error", "error": error_response, "metadata": { "execution_time": execution_time, "directory": directory, "format": format } }, indent=2) return error_response except subprocess.TimeoutExpired: error_response = f"Error: Codex CLI command timed out after {timeout_value} seconds" if format == "json": return json.dumps({ "status": "error", "error": error_response, "metadata": { "timeout": timeout_value, "directory": directory, "format": format } }, indent=2) return error_response except FileNotFoundError as e: # More specific error for when codex command is not found codex_path = _get_codex_command() if _is_windows(): error_response = ( f"Error: Codex CLI not found or not executable. " f"Detected path: {codex_path or 'None'}. " f"Please ensure 'codex' is installed and in your PATH. " f"Try running 'codex --version' in Command Prompt to verify." ) else: error_response = f"Error: Codex CLI not found: {str(e)}" if format == "json": return json.dumps({ "status": "error", "error": error_response, "metadata": { "directory": directory, "format": format, "platform": platform.system() } }, indent=2) return error_response except Exception as e: error_response = f"Error executing Codex CLI: {str(e)}" if format == "json": return json.dumps({ "status": "error", "error": error_response, "metadata": { "directory": directory, "format": format, "platform": platform.system(), "exception_type": type(e).__name__ } }, indent=2) return error_response - src/mcp_server.py:297-302 (schema)Input parameters/schema for consult_codex: query (str), directory (str), format (str, default json), timeout (Optional[int])
def consult_codex( query: str, directory: str, format: str = "json", timeout: Optional[int] = None ) -> str: - src/mcp_server.py:37-65 (helper)Helper: _get_codex_command - locates the codex CLI executable path
def _get_codex_command() -> Optional[str]: """Get the codex command path with Windows compatibility. Returns: Path to codex executable or None if not found """ # First try the standard shutil.which approach codex_path = shutil.which("codex") if codex_path: return codex_path # On Windows, explicitly check for common executable extensions if _is_windows(): for ext in ['.exe', '.bat', '.cmd', '.ps1']: codex_path = shutil.which(f"codex{ext}") if codex_path: return codex_path # Also check common installation paths on Windows common_paths = [ os.path.expandvars(r'%LOCALAPPDATA%\Programs\codex\codex.exe'), os.path.expandvars(r'%APPDATA%\npm\codex.cmd'), os.path.expandvars(r'%USERPROFILE%\.cargo\bin\codex.exe'), ] for path in common_paths: if os.path.isfile(path): return path return None - src/mcp_server.py:68-114 (helper)Helper: _build_codex_exec_command - builds the command list for codex exec subprocess
def _build_codex_exec_command() -> List[str]: """Build the command list to execute codex exec. On Windows, if the codex CLI is a PowerShell script (.ps1), we need to invoke it through PowerShell explicitly. Otherwise, use the resolved path or fall back to 'codex' for PATH resolution. Returns: Command list suitable for subprocess execution """ codex_path = _get_codex_command() if codex_path and _is_windows(): # Check if it's a PowerShell script if codex_path.lower().endswith('.ps1'): # Execute PowerShell script: powershell -ExecutionPolicy Bypass -File script.ps1 exec return ['powershell', '-ExecutionPolicy', 'Bypass', '-File', codex_path, 'exec'] else: # Use the resolved path directly for .exe, .bat, .cmd return [codex_path, 'exec'] # For Unix or when codex is in PATH, use simple command return ['codex', 'exec'] def _get_timeout() -> int: """Get timeout from environment variable or default to 90 seconds. Recommended: 60-120 seconds. Values under 60 may cause hanging. """ try: return int(os.environ.get("CODEX_TIMEOUT", "90")) except ValueError: return 90 def _should_skip_git_check() -> bool: """Check if git repository check should be skipped. Reads CODEX_SKIP_GIT_CHECK environment variable. Defaults to False for security. Set to 'true' or '1' to enable. """ skip_check = os.environ.get("CODEX_SKIP_GIT_CHECK", "false").lower() return skip_check in ("true", "1", "yes") def _run_codex_command(cmd: List[str], directory: str, timeout_value: int, input_text: str) -> subprocess.CompletedProcess: