Codex Bridge

Instructions

Implementation Reference

• src/mcp_server.py:426-559 (handler)

  The primary handler function for the 'consult_codex_with_stdin' MCP tool. Decorated with @mcp.tool() for automatic registration. Executes codex CLI by piping combined stdin_content and prompt, handles platform-specific subprocess execution (Windows UTF-8), formats output as JSON/text/code, and includes comprehensive error handling with timeouts.
  @mcp.tool() def consult_codex_with_stdin( stdin_content: str, prompt: str, directory: str, format: str = "json", timeout: Optional[int] = None ) -> str: """ Consult Codex with stdin content piped to prompt - pipeline-friendly execution. Similar to 'echo "content" | codex exec "prompt"' - combines stdin with prompt. Perfect for CI/CD workflows where you pipe file contents to the AI. Args: stdin_content: Content to pipe as stdin (e.g., file contents, diff, logs) prompt: The prompt to process the stdin content 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) # Combine stdin content with prompt combined_input = f"{stdin_content}\n\n{prompt}" # Prepare query based on format if format == "json": processed_query = _format_prompt_for_json(combined_input) else: processed_query = combined_input # 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:114-167 (helper)

  Key helper function used by the tool to execute the codex subprocess command, with special handling for Windows (UTF-8 encoding, PowerShell support) and Unix, timeout, and input piping.
  def _run_codex_command(cmd: List[str], directory: str, timeout_value: int, input_text: str) -> subprocess.CompletedProcess: """Execute codex command with platform-specific handling. Args: cmd: Command list to execute directory: Working directory timeout_value: Timeout in seconds input_text: Input text to pipe to the command Returns: CompletedProcess result with stdout/stderr as strings """ # Windows-specific handling with UTF-8 encoding support if _is_windows(): # On Windows, we need to: # 1. Use encoding='utf-8' instead of text=True to avoid code page issues # 2. Set PYTHONUTF8=1 and PYTHONIOENCODING=utf-8 for consistent encoding # 3. Don't use start_new_session as it's not supported on Windows env = os.environ.copy() env['PYTHONUTF8'] = '1' env['PYTHONIOENCODING'] = 'utf-8' # Encode input as UTF-8 bytes input_bytes = input_text.encode('utf-8') if input_text else None result = subprocess.run( cmd, cwd=directory, capture_output=True, timeout=timeout_value, input=input_bytes, shell=False, env=env ) # Decode output as UTF-8 with error handling return subprocess.CompletedProcess( args=result.args, returncode=result.returncode, stdout=result.stdout.decode('utf-8', errors='replace') if result.stdout else '', stderr=result.stderr.decode('utf-8', errors='replace') if result.stderr else '' ) else: # Unix/macOS handling (original behavior) return subprocess.run( cmd, cwd=directory, capture_output=True, text=True, timeout=timeout_value, input=input_text, start_new_session=True )
• src/mcp_server.py:245-294 (helper)

  Helper for formatting the tool's output in text, JSON (with extraction), or code modes.
  def _format_response(raw_response: str, format_type: str, execution_time: float, directory: str) -> str: """Format response according to specified output format.""" if format_type == "text": return raw_response elif format_type == "json": # Try to extract JSON from response first extracted_json = _extract_json_from_response(raw_response) if extracted_json: # Wrap extracted JSON in standard structure return json.dumps({ "status": "success", "response": extracted_json, "metadata": { "execution_time": execution_time, "directory": directory, "format": "json" } }, indent=2) else: # Fallback: wrap raw response return json.dumps({ "status": "success", "response": raw_response, "metadata": { "execution_time": execution_time, "directory": directory, "format": "json" } }, indent=2) elif format_type == "code": # Extract code blocks code_blocks = re.findall(r'```(\w+)?\n(.*?)\n```', raw_response, re.DOTALL) return json.dumps({ "status": "success", "response": raw_response, "code_blocks": [{"language": lang or "text", "code": code.strip()} for lang, code in code_blocks], "metadata": { "execution_time": execution_time, "directory": directory, "format": "code" } }, indent=2) else: return raw_response
• src/mcp_server.py:201-243 (helper)

  Helper to parse JSON from Codex CLI output which may contain extra text.
  def _extract_json_from_response(response: str) -> Optional[Dict]: """Extract JSON from mixed text response using regex.""" # Clean the response to remove CLI noise lines = response.split('\n') clean_lines = [] json_started = False for line in lines: # Skip CLI headers and metadata if (line.startswith('[') and ']' in line and ('OpenAI' in line or 'workdir:' in line or 'model:' in line)): continue if line.startswith('--------'): continue if 'tokens used:' in line: continue if 'thinking' in line and line.startswith('['): continue if 'codex' in line and line.startswith('['): continue # Look for JSON content if '{' in line: json_started = True if json_started: clean_lines.append(line) clean_response = '\n'.join(clean_lines) # Try to find complete JSON objects json_pattern = r'\{(?:[^{}]|{[^{}]*})*\}' matches = re.findall(json_pattern, clean_response, re.DOTALL) for match in matches: try: parsed = json.loads(match.strip()) # Validate it has expected structure if isinstance(parsed, dict) and any(key in parsed for key in ['result', 'response', 'answer']): return parsed except json.JSONDecodeError: continue return None
• src/mcp_server.py:426-426 (registration)

  The @mcp.tool() decorator registers the function as an MCP tool named 'consult_codex_with_stdin' (function name). No separate registration needed.
  @mcp.tool()