Codex Bridge

Instructions

Implementation Reference

• src/mcp_server.py:561-561 (registration)

  Registration of the consult_codex_batch tool using the @mcp.tool() decorator.
  @mcp.tool()
• src/mcp_server.py:562-580 (schema)

  Input schema defined by type hints and documentation for queries (list of dicts), directory (str), format (str), returning str (JSON).
  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:561-712 (handler)

  Full implementation of the consult_codex_batch handler, including validation of inputs, batch processing of queries using Codex CLI, error handling, timeouts, and consolidated JSON response.
  @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 = _build_codex_exec_command() 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 FileNotFoundError as e: codex_path = _get_codex_command() if _is_windows(): error_msg = ( f"Codex CLI not found or not executable. " f"Detected path: {codex_path or 'None'}. " f"Please ensure 'codex' is installed and in your PATH." ) else: error_msg = f"Codex CLI not found: {str(e)}" results.append({ "status": "error", "index": i, "query": query[:100] + "..." if len(query) > 100 else query, "error": error_msg, "metadata": { "platform": platform.system() } }) 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)