Codex Bridge

Instructions

Implementation Reference

• src/mcp_server.py:455-588 (handler)

  The core handler function for the 'consult_codex_batch' tool. Decorated with @mcp.tool() for registration. Handles batch processing of multiple Codex queries, validates inputs, executes each via codex CLI using helper functions, extracts JSON where possible, and returns consolidated JSON results with metadata.
  @mcp.tool() def consult_codex_batch( queries: List[Dict[str, Union[str, int]]], directory: str, format: str = "json" ) -> str: """ Consult multiple Codex queries in batch - perfect for CI/CD automation. Processes multiple prompts and returns consolidated JSON output. Each query can have individual timeout and format preferences. Args: queries: List of query dictionaries with keys: 'query' (required), 'timeout' (optional) directory: Working directory (required) format: Output format - currently only "json" supported for batch Returns: JSON array with all results """ # Check if codex CLI is available if not _get_codex_command(): return json.dumps({ "status": "error", "error": "Codex CLI not found. Install from OpenAI" }, indent=2) # Validate directory if not os.path.isdir(directory): return json.dumps({ "status": "error", "error": f"Directory does not exist: {directory}" }, indent=2) # Validate queries if not queries or not isinstance(queries, list): return json.dumps({ "status": "error", "error": "Queries must be a non-empty list" }, indent=2) # Force JSON format for batch processing format = "json" results = [] for i, query_item in enumerate(queries): if not isinstance(query_item, dict) or 'query' not in query_item: results.append({ "status": "error", "error": f"Query {i+1}: Must be a dictionary with 'query' key", "index": i }) continue query = str(query_item.get('query', '')) query_timeout = query_item.get('timeout', _get_timeout()) if isinstance(query_timeout, str): try: query_timeout = int(query_timeout) except ValueError: query_timeout = _get_timeout() # Process individual query processed_query = _format_prompt_for_json(query) cmd = ["codex", "exec"] if _should_skip_git_check(): cmd.append("--skip-git-repo-check") start_time = time.time() try: result = _run_codex_command(cmd, directory, query_timeout, 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" # Try to extract JSON from response extracted_json = _extract_json_from_response(raw_response) results.append({ "status": "success", "index": i, "query": query[:100] + "..." if len(query) > 100 else query, # Truncate long queries "response": extracted_json if extracted_json else raw_response, "metadata": { "execution_time": execution_time, "timeout": query_timeout } }) else: results.append({ "status": "error", "index": i, "query": query[:100] + "..." if len(query) > 100 else query, "error": f"Codex CLI Error: {result.stderr.strip()}", "metadata": { "execution_time": execution_time, "timeout": query_timeout } }) except subprocess.TimeoutExpired: results.append({ "status": "error", "index": i, "query": query[:100] + "..." if len(query) > 100 else query, "error": f"Query timed out after {query_timeout} seconds", "metadata": { "timeout": query_timeout } }) except Exception as e: results.append({ "status": "error", "index": i, "query": query[:100] + "..." if len(query) > 100 else query, "error": f"Error executing query: {str(e)}", "metadata": {} }) # Return consolidated results return json.dumps({ "status": "completed", "total_queries": len(queries), "successful": len([r for r in results if r["status"] == "success"]), "failed": len([r for r in results if r["status"] == "error"]), "results": results, "metadata": { "directory": directory, "format": format } }, indent=2)
• src/mcp_server.py:456-474 (schema)

  Input schema defined by function signature and docstring. Expects list of dicts with 'query' and optional 'timeout', directory, and format. Returns JSON with batch results.
  def consult_codex_batch( queries: List[Dict[str, Union[str, int]]], directory: str, format: str = "json" ) -> str: """ Consult multiple Codex queries in batch - perfect for CI/CD automation. Processes multiple prompts and returns consolidated JSON output. Each query can have individual timeout and format preferences. Args: queries: List of query dictionaries with keys: 'query' (required), 'timeout' (optional) directory: Working directory (required) format: Output format - currently only "json" supported for batch Returns: JSON array with all results """
• src/mcp_server.py:76-111 (helper)

  Key helper function used by the batch handler to execute individual codex CLI commands with cross-platform support and timeout handling.
  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:145-187 (helper)

  Helper function to extract structured JSON from Codex CLI output, used in batch processing for better parsing.
  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