Codex Bridge

Instructions

Implementation Reference

• src/mcp_server.py:345-452 (handler)

  The handler function decorated with @mcp.tool(), which defines the tool logic, input schema via type hints and docstring, and registers it with the MCP server. It executes the codex CLI with combined stdin_content and prompt, handles errors, formatting, and 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 = ["codex", "exec"] 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 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 } }, indent=2) return error_response
• src/mcp_server.py:76-110 (helper)

  Core helper function that runs the codex CLI subprocess with cross-platform support, capturing output and handling timeouts. Used by consult_codex_with_stdin.
  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 """ # Windows-specific handling if _is_windows(): # On Windows, don't use start_new_session as it's not supported return subprocess.run( cmd, cwd=directory, capture_output=True, text=True, timeout=timeout_value, input=input_text, shell=False ) 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:113-130 (helper)

  Helper to clean Codex CLI output by removing warnings and empty lines.
  def _clean_codex_output(output: str) -> str: """Clean irrelevant messages from Codex CLI output.""" if not output: return output # Filter out common CLI noise and warnings lines = output.split('\n') cleaned_lines = [] for line in lines: # Skip lines that contain irrelevant CLI messages if (line.strip().startswith("Warning:") or line.strip().startswith("Note:") or line.strip() == ""): continue cleaned_lines.append(line) return '\n'.join(cleaned_lines).strip()
• src/mcp_server.py:145-186 (helper)

  Helper to extract structured JSON from the raw Codex response.
  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:189-237 (helper)

  Helper to format the final response in text, json, or code modes, including metadata.
  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