How to Build Multi-Agent Systems with OpenAI Codex SDK: Orchestrating Autonomous Coding Agents

1. What Is the OpenAI Codex SDK and How Does It Differ from the ChatGPT API?

The OpenAI Codex SDK, released in its current form in late 2025 and significantly expanded through 2026, is a purpose-built developer toolkit for orchestrating software engineering agents. Unlike the ChatGPT API, which exposes a general-purpose conversational model endpoint, the Codex SDK provides a structured abstraction layer specifically designed for autonomous code generation, execution, testing, and multi-agent coordination within software development pipelines.

Understanding the distinction is critical before you invest architectural effort in either direction. The ChatGPT API (via openai.chat.completions.create()) is a stateless request-response interface. You send a message thread, receive a completion, and manage all state, context, and orchestration yourself. The Codex SDK, by contrast, ships with built-in agent lifecycle management, sandboxed code execution environments, tool-use primitives tailored to engineering tasks, and a first-class multi-agent coordination layer.

For a deeper look at the foundational ChatGPT API patterns, see our guide on ChatGPT API integration. Here we focus exclusively on what the Codex SDK adds on top of that foundation.

The Codex SDK wraps the underlying codex-2 model family (optimized for code understanding, generation, and reasoning about software systems) while also allowing you to route specific sub-tasks to gpt-4.5 or o3 models for non-code reasoning tasks. This hybrid routing capability is one of the most powerful features for multi-agent system design, and we will explore it in depth throughout this tutorial.

For background on what the Codex platform offers at a higher level, refer to our comprehensive OpenAI Codex guide before proceeding.

” alt=”Codex SDK architecture diagram showing agent layers” />

2. Setting Up the Codex SDK: Installation, Authentication, and Configuration

Installation

The Codex SDK requires Python 3.11 or later. It is distributed via PyPI and installs cleanly into a virtual environment. The package includes the core SDK, the CLI tool for local agent testing, and optional extras for Slack integration and observability.

# Create and activate a virtual environment
python -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

# Install the core SDK
pip install openai-codex-sdk>=2.4.0

# Install with Slack integration support
pip install "openai-codex-sdk[slack]>=2.4.0"

# Install with full observability stack (OpenTelemetry exporters)
pip install "openai-codex-sdk[observability]>=2.4.0"

# Verify installation
codex --version
# Expected: openai-codex-sdk 2.4.x

Authentication

The Codex SDK uses a layered authentication model. At minimum, you need an OpenAI API key with Codex SDK access enabled on your organization account. For multi-agent deployments in team environments, you will also configure organization-scoped agent tokens that enforce role-based access control (RBAC) on which agents can access which tools and repositories.

import os
from openai_codex import CodexClient, AgentConfig

# Method 1: Environment variable (recommended for production)
# Set OPENAI_API_KEY and CODEX_ORG_TOKEN in your environment
client = CodexClient()

# Method 2: Explicit credentials (useful for testing multiple orgs)
client = CodexClient(
    api_key=os.environ["OPENAI_API_KEY"],
    org_token=os.environ["CODEX_ORG_TOKEN"],
    organization_id="org-your-org-id-here"
)

# Method 3: Config file (~/.codex/config.yaml)
# Run `codex auth login` to generate this interactively
client = CodexClient.from_config(profile="production")

Configuration File Structure

For complex multi-agent deployments, a codex.yaml project configuration file provides centralized control over agent defaults, model routing, sandbox settings, and cost limits. Place this file in your project root.

# codex.yaml
version: "2"
project:
  name: "engineering-automation"
  environment: "production"

defaults:
  model: "codex-2"
  max_tokens: 8192
  temperature: 0.2
  sandbox:
    enabled: true
    timeout_seconds: 300
    memory_mb: 2048
    network_access: false  # Disable by default for security

agents:
  coordinator:
    model: "o3"  # Use reasoning model for orchestration decisions
    max_tokens: 4096
  code_writer:
    model: "codex-2"
    sandbox:
      network_access: false
  code_reviewer:
    model: "codex-2"
    tools:
      - static_analysis
      - security_scan
  test_runner:
    model: "codex-2"
    sandbox:
      timeout_seconds: 600  # Tests may run longer

cost_limits:
  per_run_usd: 5.00
  per_agent_usd: 1.50
  daily_usd: 50.00
  alert_threshold_percent: 80

telemetry:
  enabled: true
  exporter: "otlp"
  endpoint: "http://localhost:4317"

Initializing the Client with Configuration

from openai_codex import CodexClient
from openai_codex.config import ProjectConfig

# Load from codex.yaml in current directory
config = ProjectConfig.from_file("codex.yaml")
client = CodexClient(config=config)

# Verify connectivity and permissions
health = client.health_check()
print(f"SDK Version: {health.sdk_version}")
print(f"Models Available: {health.available_models}")
print(f"Sandbox Status: {health.sandbox_status}")
print(f"Org Permissions: {health.permissions}")

For a complete walkthrough of setting up your local development environment for AI coding tools, see our guide on AI coding assistant setup.

3. Designing Agent Architectures: Coordinator, Pipeline, and Swarm Patterns

Before writing a single line of agent code, you must choose an architectural pattern that matches your task’s characteristics. The Codex SDK natively supports three primary multi-agent patterns, each with distinct tradeoffs in complexity, cost, latency, and fault tolerance.

The Coordinator Pattern

In the coordinator pattern, a single orchestrator agent receives a high-level task, decomposes it into sub-tasks, assigns each sub-task to a specialized worker agent, and synthesizes the results. The coordinator uses a reasoning-optimized model (such as o3) while workers use the code-optimized codex-2 model. This pattern is ideal when tasks have complex interdependencies that require dynamic planning.

Best for: Large feature implementation, architecture design with multiple components, complex debugging sessions where the root cause is unknown.

Tradeoff: The coordinator creates a single point of failure and adds latency because worker tasks cannot begin until the coordinator has finished its decomposition step.

The Pipeline Pattern

In the pipeline pattern, agents are arranged in a fixed sequence. The output of each agent becomes the input to the next. This is the software engineering equivalent of a Unix pipe. Each agent in the pipeline is specialized for one transformation: write code → review code → run tests → generate documentation. The pipeline pattern is deterministic, easy to monitor, and straightforward to debug because you can inspect the output at each stage.

Best for: Code review workflows, CI/CD automation, structured transformation tasks where the sequence of operations is known in advance.

Tradeoff: Pipelines are sequential by default, so total latency is the sum of all agent latencies. Parallel pipeline variants exist but require explicit fork/join logic.

The Swarm Pattern

In the swarm pattern, multiple homogeneous agents work on different parts of a problem simultaneously, without a central coordinator. A lightweight dispatcher distributes tasks from a shared queue, and a separate aggregator collects and reconciles results. Swarms excel at embarrassingly parallel problems: scanning a large codebase for security vulnerabilities, running tests across multiple modules simultaneously, or generating documentation for hundreds of functions in parallel.

Best for: Large-scale codebase analysis, bulk code generation, parallel test execution, batch processing of independent tasks.

Tradeoff: Result aggregation and conflict resolution become complex when agents produce overlapping or contradictory outputs. You must design explicit reconciliation logic.

” alt=”Multi-agent architecture patterns: coordinator, pipeline, swarm” />

4. Implementing Single Agent Task Execution

Before composing multi-agent systems, you must understand how a single Codex agent executes a task. The fundamental unit is the Agent object, which encapsulates a model configuration, a set of tools, a sandbox environment, and an execution context.

Creating a Basic Code-Writing Agent

from openai_codex import CodexClient, Agent, AgentConfig
from openai_codex.tools import FilesystemTool, ShellTool, StaticAnalysisTool
from openai_codex.sandbox import SandboxConfig

client = CodexClient()

# Configure the sandbox environment
sandbox_config = SandboxConfig(
    base_image="python:3.11-slim",
    timeout_seconds=120,
    memory_mb=1024,
    network_access=False,
    working_directory="/workspace"
)

# Define the agent
code_writer = Agent(
    client=client,
    name="code_writer",
    model="codex-2",
    system_prompt="""You are an expert Python developer. When given a task:
1. Write clean, well-documented Python code
2. Follow PEP 8 style guidelines
3. Include type hints for all function signatures
4. Write docstrings for all public functions and classes
5. Handle edge cases and raise appropriate exceptions
Always output complete, runnable code files.""",
    tools=[
        FilesystemTool(permissions=["read", "write"], allowed_paths=["/workspace"]),
        ShellTool(allowed_commands=["python", "pip", "pytest"])
    ],
    sandbox=sandbox_config,
    max_tokens=4096,
    temperature=0.1  # Low temperature for deterministic code generation
)

# Execute a task
task_result = code_writer.execute(
    task="""Create a Python module called `data_validator.py` that:
- Validates email addresses using regex
- Validates phone numbers (US format)
- Validates dates in ISO 8601 format
- Returns detailed error messages for invalid inputs
- Includes a ValidationResult dataclass with fields: is_valid, value, errors
""",
    working_directory="/workspace/src"
)

print(f"Status: {task_result.status}")  # SUCCESS, FAILED, TIMEOUT
print(f"Files Created: {task_result.files_created}")
print(f"Execution Time: {task_result.execution_time_ms}ms")
print(f"Tokens Used: {task_result.usage.total_tokens}")
print(f"Cost: ${task_result.cost_usd:.4f}")

# Access the generated code
for file_path, content in task_result.file_contents.items():
    print(f"\n--- {file_path} ---")
    print(content)

Executing Code and Capturing Test Results

from openai_codex import Agent
from openai_codex.tools import FilesystemTool, ShellTool, PytestTool

test_runner = Agent(
    client=client,
    name="test_runner",
    model="codex-2",
    system_prompt="""You are a test execution agent. Run pytest on the provided 
test files, capture all output, and return a structured report including:
- Pass/fail counts
- Failed test names and error messages
- Code coverage percentage
- Performance metrics for slow tests (>1s)""",
    tools=[
        FilesystemTool(permissions=["read"], allowed_paths=["/workspace"]),
        PytestTool(coverage=True, timeout_per_test=30)
    ],
    sandbox=sandbox_config
)

test_result = test_runner.execute(
    task="Run all tests in /workspace/tests/ and provide a detailed report",
    context={
        "source_directory": "/workspace/src",
        "test_directory": "/workspace/tests"
    }
)

# Access structured test output
report = test_result.structured_output
print(f"Tests Passed: {report['passed']}")
print(f"Tests Failed: {report['failed']}")
print(f"Coverage: {report['coverage_percent']}%")
if report['failed_tests']:
    for test in report['failed_tests']:
        print(f"  FAIL: {test['name']} - {test['error']}")

The task_result.structured_output field is populated when the agent’s system prompt instructs it to produce JSON-structured output. The Codex SDK automatically parses and validates this output against an optional Pydantic schema you can provide via the output_schema parameter on the Agent constructor.

5. Task Delegation Between Agents

Task delegation is the mechanism by which one agent assigns work to another. In the Codex SDK, delegation is explicit and typed. The delegating agent calls a delegate() method on an AgentRegistry object, specifying the target agent by name, the task description, and any context data to pass along. The SDK handles serialization, sandboxing, and result propagation automatically.

Setting Up an Agent Registry

from openai_codex import CodexClient, Agent, AgentRegistry
from openai_codex.tools import FilesystemTool, ShellTool, GitTool, StaticAnalysisTool

client = CodexClient()

# Create specialized agents
agents = {
    "code_writer": Agent(
        client=client,
        name="code_writer",
        model="codex-2",
        system_prompt="You write production-quality Python code based on specifications.",
        tools=[FilesystemTool(permissions=["read", "write"]), ShellTool()],
        temperature=0.1
    ),
    "code_reviewer": Agent(
        client=client,
        name="code_reviewer",
        model="codex-2",
        system_prompt="""You review Python code for:
- Logic errors and bugs
- Security vulnerabilities (injection, XSS, insecure deserialization)
- Performance issues (N+1 queries, unnecessary loops, memory leaks)
- Style violations (PEP 8, naming conventions)
Return a structured review with severity levels: CRITICAL, HIGH, MEDIUM, LOW.""",
        tools=[FilesystemTool(permissions=["read"]), StaticAnalysisTool()],
        temperature=0.0
    ),
    "test_writer": Agent(
        client=client,
        name="test_writer",
        model="codex-2",
        system_prompt="You write comprehensive pytest test suites for Python code.",
        tools=[FilesystemTool(permissions=["read", "write"])],
        temperature=0.1
    ),
    "doc_writer": Agent(
        client=client,
        name="doc_writer",
        model="gpt-4.5",  # Use GPT-4.5 for prose quality
        system_prompt="You write clear, accurate technical documentation in Markdown.",
        tools=[FilesystemTool(permissions=["read", "write"])],
        temperature=0.3
    )
}

# Register all agents
registry = AgentRegistry(client=client)
for name, agent in agents.items():
    registry.register(agent)

print(f"Registered agents: {registry.list_agents()}")

Implementing a Coordinator with Delegation

from openai_codex import Agent, AgentRegistry
from openai_codex.delegation import DelegationContext, DelegationResult
from openai_codex.tools import DelegationTool
import json

def create_coordinator(registry: AgentRegistry) -> Agent:
    """Create a coordinator agent that can delegate to registered workers."""
    
    coordinator = Agent(
        client=client,
        name="coordinator",
        model="o3",  # Reasoning model for planning
        system_prompt=f"""You are a software engineering coordinator. 
You decompose complex engineering tasks and delegate to specialized agents.

Available agents: {json.dumps(registry.get_agent_capabilities())}

For each task:
1. Analyze the requirements
2. Create a delegation plan specifying which agent handles which sub-task
3. Specify dependencies between sub-tasks (which must complete before others start)
4. Define the expected output format for each delegation
5. Synthesize results into a coherent final output

Always return your plan as structured JSON before executing delegations.""",
        tools=[
            DelegationTool(registry=registry),
            FilesystemTool(permissions=["read", "write"])
        ],
        temperature=0.0,
        max_tokens=8192
    )
    return coordinator

coordinator = create_coordinator(registry)

# Execute a complex task through the coordinator
result = coordinator.execute(
    task="""Implement a REST API endpoint for user authentication:
- POST /auth/login accepting {email, password}
- JWT token generation with 24h expiry
- Bcrypt password hashing
- Rate limiting (5 attempts per minute per IP)
- Comprehensive tests
- API documentation

Use FastAPI. Follow security best practices."""
)

# The coordinator will automatically delegate to code_writer, code_reviewer,
# test_writer, and doc_writer based on its planning output
print(f"Coordinator Status: {result.status}")
print(f"Delegations Made: {len(result.delegation_log)}")
for delegation in result.delegation_log:
    print(f"  → {delegation.agent_name}: {delegation.status} "
          f"({delegation.execution_time_ms}ms, ${delegation.cost_usd:.4f})")

The DelegationTool is the key primitive here. When the coordinator agent calls this tool, the SDK intercepts the call, routes it to the target agent in the registry, executes the sub-task in a separate sandbox, and returns the result to the coordinator’s context. The coordinator never directly executes code itself — it only reasons about task decomposition and result synthesis.

For advanced prompt engineering strategies to improve coordinator decision quality, see our guide on prompt engineering for developers.

6. Handling Inter-Agent Communication

Inter-agent communication in the Codex SDK goes beyond simple task delegation. Agents may need to share partial results, request clarification from other agents, broadcast state changes, or subscribe to events from a shared message bus. The SDK provides three communication primitives: direct delegation (synchronous), message passing (asynchronous), and shared state via the agent context store.

Asynchronous Message Passing

import asyncio
from openai_codex.async_client import AsyncCodexClient
from openai_codex.messaging import AgentMessageBus, Message, MessageType

async def main():
    client = AsyncCodexClient()
    bus = AgentMessageBus(client=client)
    
    # Define message handlers
    async def handle_review_request(message: Message):
        """Code reviewer handles incoming review requests."""
        code_content = message.payload["code"]
        file_path = message.payload["file_path"]
        
        review_result = await code_reviewer.execute_async(
            task=f"Review this code for issues:\n\n{code_content}",
            context={"file_path": file_path}
        )
        
        # Publish review results back to the bus
        await bus.publish(Message(
            type=MessageType.RESULT,
            sender="code_reviewer",
            recipient=message.sender,
            correlation_id=message.id,
            payload={
                "review": review_result.structured_output,
                "file_path": file_path
            }
        ))
    
    async def handle_fix_request(message: Message):
        """Bug fixer handles incoming fix requests from reviewer."""
        issues = message.payload["review"]["issues"]
        file_path = message.payload["file_path"]
        
        # Only process CRITICAL and HIGH severity issues
        critical_issues = [i for i in issues if i["severity"] in ["CRITICAL", "HIGH"]]
        
        if critical_issues:
            fix_result = await code_writer.execute_async(
                task=f"Fix these critical issues in {file_path}:\n{json.dumps(critical_issues, indent=2)}"
            )
            
            await bus.publish(Message(
                type=MessageType.RESULT,
                sender="bug_fixer",
                recipient="coordinator",
                correlation_id=message.correlation_id,
                payload={"fixes_applied": fix_result.structured_output}
            ))
    
    # Subscribe agents to message types
    bus.subscribe("code_reviewer", MessageType.REVIEW_REQUEST, handle_review_request)
    bus.subscribe("bug_fixer", MessageType.FIX_REQUEST, handle_fix_request)
    
    # Start the message bus
    await bus.start()
    
    # Publish an initial task to kick off the workflow
    await bus.publish(Message(
        type=MessageType.REVIEW_REQUEST,
        sender="coordinator",
        recipient="code_reviewer",
        payload={
            "code": open("/workspace/src/auth.py").read(),
            "file_path": "/workspace/src/auth.py"
        }
    ))
    
    # Wait for all messages to be processed
    await bus.drain(timeout_seconds=300)
    await bus.stop()

asyncio.run(main())

Shared State via Agent Context Store

from openai_codex.state import AgentContextStore, StateScope

# Initialize a shared context store
# StateScope.SESSION: persists for the duration of the current run
# StateScope.PROJECT: persists across runs (stored in Codex cloud)
# StateScope.EPHEMERAL: in-memory only, lost when process exits
store = AgentContextStore(
    client=client,
    scope=StateScope.SESSION,
    namespace="code_review_pipeline"
)

# Writer agent stores its output
async def code_writer_task(store: AgentContextStore, spec: str):
    result = await code_writer.execute_async(task=spec)
    
    # Store result for other agents to access
    await store.set("generated_code", {
        "files": result.file_contents,
        "summary": result.summary,
        "timestamp": result.completed_at.isoformat()
    })
    await store.set("pipeline_stage", "code_written")
    return result

# Reviewer agent reads writer's output
async def code_reviewer_task(store: AgentContextStore):
    # Wait for code to be available
    await store.wait_for_key("generated_code", timeout_seconds=60)
    
    generated = await store.get("generated_code")
    stage = await store.get("pipeline_stage")
    
    if stage != "code_written":
        raise RuntimeError(f"Unexpected pipeline stage: {stage}")
    
    # Review all generated files
    review_tasks = []
    for file_path, content in generated["files"].items():
        task = code_reviewer.execute_async(
            task=f"Review {file_path}:\n\n{content}"
        )
        review_tasks.append(task)
    
    reviews = await asyncio.gather(*review_tasks)
    
    # Store aggregated reviews
    await store.set("review_results", {
        file_path: review.structured_output
        for file_path, review in zip(generated["files"].keys(), reviews)
    })
    await store.set("pipeline_stage", "review_complete")
    
    return reviews

7. Multi-Agent Pipeline for Code Review

A code review pipeline is one of the most practical applications of multi-agent systems in software engineering. The following implementation creates a production-grade pipeline that takes a pull request, runs it through a series of specialized review agents, and produces a consolidated review report with actionable feedback.

Pipeline Definition

from openai_codex import CodexClient
from openai_codex.pipeline import Pipeline, PipelineStage, PipelineConfig
from openai_codex.tools import (
    GitTool, FilesystemTool, StaticAnalysisTool,
    SecurityScanTool, DependencyCheckTool, PytestTool
)
from openai_codex.models import PipelineResult
from dataclasses import dataclass
from typing import List, Dict, Any
import asyncio

client = CodexClient()

@dataclass
class PRContext:
    """Context passed through the review pipeline."""
    pr_number: int
    repo_url: str
    base_branch: str
    head_branch: str
    changed_files: List[str]
    diff_content: str
    pr_description: str

def build_code_review_pipeline() -> Pipeline:
    """Construct the complete code review pipeline."""
    
    # Stage 1: Static Analysis Agent
    static_analysis_stage = PipelineStage(
        name="static_analysis",
        agent=Agent(
            client=client,
            name="static_analyzer",
            model="codex-2",
            system_prompt="""Perform static analysis on the provided code diff.
Identify:
1. Syntax errors and type errors
2. Undefined variables and imports
3. Unreachable code
4. Complexity metrics (cyclomatic complexity > 10 is HIGH)
5. Code duplication (>20 lines duplicated is a flag)

Return JSON: {"issues": [{"file": str, "line": int, "type": str, 
"severity": str, "message": str, "suggestion": str}], 
"metrics": {"avg_complexity": float, "max_complexity": float}}""",
            tools=[StaticAnalysisTool(linters=["pylint", "mypy", "radon"])],
            temperature=0.0
        ),
        input_transformer=lambda ctx: {
            "task": f"Analyze this diff:\n\n{ctx.diff_content}",
            "context": {"files": ctx.changed_files}
        },
        output_key="static_analysis"
    )
    
    # Stage 2: Security Review Agent
    security_review_stage = PipelineStage(
        name="security_review",
        agent=Agent(
            client=client,
            name="security_reviewer",
            model="codex-2",
            system_prompt="""You are a security engineer reviewing code for vulnerabilities.
Check for:
1. SQL injection vulnerabilities
2. Command injection risks
3. Hardcoded credentials or API keys
4. Insecure deserialization
5. Missing input validation
6. OWASP Top 10 violations
7. Dependency vulnerabilities (check against CVE database)

Severity: CRITICAL (exploitable remotely), HIGH (exploitable with access),
MEDIUM (requires specific conditions), LOW (best practice violation)

Return JSON: {"vulnerabilities": [...], "risk_score": int (0-100), 
"requires_security_review": bool}""",
            tools=[
                SecurityScanTool(scanners=["bandit", "semgrep"]),
                DependencyCheckTool()
            ],
            temperature=0.0
        ),
        input_transformer=lambda ctx: {
            "task": f"Security review for PR #{ctx.pr_number}:\n\n{ctx.diff_content}",
            "context": {"pr_description": ctx.pr_description}
        },
        output_key="security_review"
    )
    
    # Stage 3: Logic Review Agent
    logic_review_stage = PipelineStage(
        name="logic_review",
        agent=Agent(
            client=client,
            name="logic_reviewer",
            model="codex-2",
            system_prompt="""Review code changes for logical correctness.
Evaluate:
1. Algorithm correctness (does it do what the PR description claims?)
2. Edge case handling (null inputs, empty collections, overflow)
3. Race conditions and thread safety issues
4. Error handling completeness (are all exceptions caught appropriately?)
5. Business logic alignment with the PR description
6. API contract correctness (request/response shapes)

Return JSON: {"logic_issues": [...], "missing_edge_cases": [...], 
"alignment_score": int (0-10), "overall_assessment": str}""",
            tools=[FilesystemTool(permissions=["read"])],
            temperature=0.1
        ),
        input_transformer=lambda ctx, prev_results: {
            "task": f"""Review logic in PR #{ctx.pr_number}:
Description: {ctx.pr_description}
Diff:
{ctx.diff_content}

Static analysis already found: {len(prev_results.get('static_analysis', {}).get('issues', []))} issues""",
        },
        output_key="logic_review",
        depends_on=["static_analysis"]  # Runs after static analysis
    )
    
    # Stage 4: Test Coverage Agent
    test_coverage_stage = PipelineStage(
        name="test_coverage",
        agent=Agent(
            client=client,
            name="test_coverage_checker",
            model="codex-2",
            system_prompt="""Analyze test coverage for code changes.
Determine:
1. Which changed functions/classes have corresponding tests
2. Which changed functions/classes are NOT tested
3. Whether existing tests cover the new code paths
4. Suggest specific test cases that should be added

Return JSON: {"coverage_gaps": [...], "suggested_tests": [...], 
"estimated_coverage_percent": float, "test_quality_score": int (0-10)}""",
            tools=[
                FilesystemTool(permissions=["read"]),
                PytestTool(coverage=True, dry_run=True)
            ],
            temperature=0.1
        ),
        input_transformer=lambda ctx: {
            "task": f"Check test coverage for changes in PR #{ctx.pr_number}:\n{ctx.diff_content}"
        },
        output_key="test_coverage"
    )
    
    # Stage 5: Report Synthesis Agent
    synthesis_stage = PipelineStage(
        name="synthesis",
        agent=Agent(
            client=client,
            name="review_synthesizer",
            model="gpt-4.5",  # Better prose for human-readable output
            system_prompt="""You synthesize multiple code review reports into a single,
actionable pull request review. 

Format the output as a GitHub PR review comment with:
1. Executive summary (2-3 sentences)
2. MUST FIX items (blocking the merge)
3. SHOULD FIX items (important but non-blocking)
4. SUGGESTIONS (optional improvements)
5. POSITIVE FEEDBACK (what was done well)
6. Overall verdict: APPROVE / REQUEST_CHANGES / COMMENT

Be specific: include file names, line numbers, and concrete fix suggestions.""",
            tools=[],
            temperature=0.3
        ),
        input_transformer=lambda ctx, prev_results: {
            "task": f"""Synthesize this review for PR #{ctx.pr_number}: "{ctx.pr_description}"
            
Static Analysis: {prev_results['static_analysis']}
Security Review: {prev_results['security_review']}
Logic Review: {prev_results['logic_review']}
Test Coverage: {prev_results['test_coverage']}"""
        },
        output_key="final_review",
        depends_on=["static_analysis", "security_review", "logic_review", "test_coverage"]
    )
    
    # Assemble the pipeline
    pipeline = Pipeline(
        client=client,
        name="code_review_pipeline",
        stages=[
            static_analysis_stage,
            security_review_stage,
            logic_review_stage,
            test_coverage_stage,
            synthesis_stage
        ],
        config=PipelineConfig(
            parallel_stages=["static_analysis", "security_review", "test_coverage"],
            max_parallel_agents=3,
            fail_fast=True,  # Stop pipeline if CRITICAL security issue found
            cost_limit_usd=3.00
        )
    )
    
    return pipeline

# Execute the pipeline
async def run_code_review(pr_context: PRContext):
    pipeline = build_code_review_pipeline()
    
    result: PipelineResult = await pipeline.execute_async(input_data=pr_context)
    
    print(f"Pipeline Status: {result.status}")
    print(f"Total Duration: {result.total_duration_ms}ms")
    print(f"Total Cost: ${result.total_cost_usd:.4f}")
    print(f"\n{'='*60}")
    print("FINAL REVIEW:")
    print(result.outputs["final_review"]["text"])
    
    # Post review to GitHub if verdict is not APPROVE
    verdict = result.outputs["final_review"]["verdict"]
    if verdict != "APPROVE":
        print(f"\nVerdict: {verdict} - Posting to GitHub PR #{pr_context.pr_number}")
    
    return result

8. Autonomous Bug-Fixing Agent Swarm

The swarm pattern shines when you need to process many independent tasks in parallel. An autonomous bug-fixing swarm takes a list of known issues (from a bug tracker, test failures, or static analysis output) and dispatches multiple agents to fix them simultaneously. The challenge lies in preventing agents from making conflicting changes to shared files and in aggregating their fixes into a coherent, mergeable patch.

Swarm Infrastructure

from openai_codex import CodexClient, Agent
from openai_codex.swarm import AgentSwarm, SwarmConfig, TaskQueue, WorkerPool
from openai_codex.tools import FilesystemTool, GitTool, ShellTool
from openai_codex.locking import FileLockManager
from dataclasses import dataclass
from typing import List, Optional
import asyncio
import uuid

@dataclass
class BugTask:
    """Represents a single bug to be fixed by a swarm agent."""
    id: str
    file_path: str
    line_number: int
    issue_type: str
    severity: str
    description: str
    suggested_fix: Optional[str] = None
    test_command: Optional[str] = None

async def create_bug_fixing_swarm(
    bug_tasks: List[BugTask],
    repo_path: str,
    max_workers: int = 5
) -> AgentSwarm:
    """Create and configure a bug-fixing agent swarm."""
    
    client = CodexClient()
    
    # File lock manager prevents multiple agents from editing the same file
    lock_manager = FileLockManager(
        lock_timeout_seconds=120,
        deadlock_detection=True
    )
    
    # Worker agent factory - creates identical worker agents
    def create_worker_agent(worker_id: int) -> Agent:
        return Agent(
            client=client,
            name=f"bug_fixer_{worker_id}",
            model="codex-2",
            system_prompt=f"""You are bug-fixing agent #{worker_id}.
You fix one specific bug at a time. For each bug:
1. Read the affected file(s) to understand context
2. Implement the minimal fix that resolves the issue
3. Do NOT refactor unrelated code
4. Do NOT change function signatures unless required by the bug
5. Add a comment above your fix: # BUG FIX [{worker_id}]: <brief description>
6. Run the test command if provided to verify your fix
7. If the fix causes test failures, revert and report inability to fix

Return JSON: {{"fixed": bool, "files_modified": [str], 
"fix_description": str, "tests_passed": bool, "error": str | null}}""",
            tools=[
                FilesystemTool(
                    permissions=["read", "write"],
                    allowed_paths=[repo_path],
                    lock_manager=lock_manager  # Prevent concurrent file access
                ),
                ShellTool(allowed_commands=["python", "pytest", "git"]),
                GitTool(permissions=["diff", "add"])
            ],
            temperature=0.05,
            max_tokens=4096
        )
    
    # Create worker pool
    worker_pool = WorkerPool(
        workers=[create_worker_agent(i) for i in range(max_workers)],
        max_concurrent_tasks=max_workers
    )
    
    # Create task queue from bug list
    task_queue = TaskQueue()
    for bug in bug_tasks:
        # Group bugs by file to prevent conflicts
        task_queue.enqueue(
            task_id=bug.id,
            task_data={
                "task": f"""Fix this bug:
File: {bug.file_path}
Line: {bug.line_number}
Type: {bug.issue_type}
Severity: {bug.severity}
Description: {bug.description}
Suggested Fix: {bug.suggested_fix or 'Determine appropriate fix'}
Test Command: {bug.test_command or 'Run relevant tests'}""",
                "file_path": bug.file_path,
                "priority": 1 if bug.severity == "CRITICAL" else 2
            },
            priority=1 if bug.severity == "CRITICAL" else 2
        )
    
    swarm = AgentSwarm(
        client=client,
        worker_pool=worker_pool,
        task_queue=task_queue,
        config=SwarmConfig(
            max_retries_per_task=2,
            retry_on_failure=True,
            file_conflict_strategy="queue",  # Queue conflicting tasks, don't skip
            cost_limit_usd=10.00,
            progress_callback=lambda progress: print(
                f"Progress: {progress.completed}/{progress.total} "
                f"({progress.success_rate:.1%} success rate)"
            )
        )
    )
    
    return swarm

# Execute the swarm
async def run_bug_fixing_swarm(bugs: List[BugTask], repo_path: str):
    swarm = await create_bug_fixing_swarm(bugs, repo_path, max_workers=5)
    
    print(f"Starting swarm with {len(bugs)} bugs to fix...")
    swarm_result = await swarm.execute_async()
    
    print(f"\nSwarm Complete:")
    print(f"  Total Bugs: {swarm_result.total_tasks}")
    print(f"  Fixed: {swarm_result.successful_tasks}")
    print(f"  Failed: {swarm_result.failed_tasks}")
    print(f"  Skipped (conflicts): {swarm_result.skipped_tasks}")
    print(f"  Total Cost: ${swarm_result.total_cost_usd:.4f}")
    print(f"  Duration: {swarm_result.total_duration_ms / 1000:.1f}s")
    
    # Show failures for investigation
    for failed in swarm_result.failed_task_results:
        print(f"\n  FAILED: {failed.task_id}")
        print(f"    Error: {failed.error}")
        print(f"    Agent: {failed.agent_name}")
    
    return swarm_result

Conflict Detection and Resolution in the Swarm

from openai_codex.swarm import ConflictDetector, ConflictResolutionStrategy
from openai_codex.diff import DiffMerger, MergeConflict

async def resolve_swarm_conflicts(swarm_result) -> dict:
    """Merge all agent fixes and resolve any conflicts."""
    
    conflict_detector = ConflictDetector()
    diff_merger = DiffMerger(strategy=ConflictResolutionStrategy.SEMANTIC)
    
    # Collect all diffs produced by swarm agents
    all_diffs = {
        task_id: result.git_diff
        for task_id, result in swarm_result.successful_task_results.items()
        if result.git_diff
    }
    
    # Detect conflicts between diffs
    conflicts = conflict_detector.detect(all_diffs)
    
    if not conflicts:
        print("No conflicts detected. All fixes can be applied cleanly.")
        merged_diff = diff_merger.merge_all(all_diffs)
        return {"status": "clean", "merged_diff": merged_diff}
    
    print(f"Detected {len(conflicts)} conflicts:")
    
    resolved_diffs = dict(all_diffs)
    conflict_resolutions = []
    
    for conflict in conflicts:
        print(f"  Conflict: {conflict.file_path} "
              f"(tasks: {conflict.task_id_a} vs {conflict.task_id_b})")
        
        # Use a dedicated resolution agent for each conflict
        resolution_agent = Agent(
            client=client,
            name="conflict_resolver",
            model="o3",  # Reasoning model for conflict resolution
            system_prompt="""You resolve merge conflicts between two code fixes.
Given two overlapping diffs for the same file, produce a single unified diff
that correctly applies both fixes without breaking either. 
Preserve the intent of both fixes.
Return the merged diff in unified diff format.""",
            tools=[],
            temperature=0.0
        )
        
        resolution = await resolution_agent.execute_async(
            task=f"""Resolve conflict in {conflict.file_path}:

FIX A (Task {conflict.task_id_a}):
{conflict.diff_a}

FIX B (Task {conflict.task_id_b}):
{conflict.diff_b}

Original context:
{conflict.original_context}"""
        )
        
        resolved_diffs[f"resolved_{conflict.file_path}"] = resolution.text
        conflict_resolutions.append({
            "file": conflict.file_path,
            "resolution": resolution.text
        })
    
    final_merged = diff_merger.merge_all(resolved_diffs)
    
    return {
        "status": "resolved",
        "conflicts_found": len(conflicts),
        "conflicts_resolved": len(conflict_resolutions),
        "merged_diff": final_merged,
        "resolutions": conflict_resolutions
    }

9. Result Aggregation and Conflict Resolution

When multiple agents produce outputs that must be combined into a single coherent result, you need a principled aggregation strategy. The Codex SDK provides built-in aggregators for common patterns, and you can implement custom aggregation logic for domain-specific needs.

Voting Aggregator for Consensus Decisions

from openai_codex.aggregation import VotingAggregator, WeightedAggregator, AggregationResult
from openai_codex import Agent
from typing import List, Dict, Any

async def multi_agent_security_audit(code_file: str) -> AggregationResult:
    """Run multiple security agents and aggregate their findings by consensus."""
    
    # Three independent security agents with different specializations
    security_agents = [
        Agent(client=client, name="owasp_agent", model="codex-2",
              system_prompt="You specialize in OWASP Top 10 vulnerabilities."),
        Agent(client=client, name="crypto_agent", model="codex-2",
              system_prompt="You specialize in cryptographic vulnerabilities and key management."),
        Agent(client=client, name="injection_agent", model="codex-2",
              system_prompt="You specialize in injection attacks: SQL, command, LDAP, XPath.")
    ]
    
    code_content = open(code_file).read()
    
    # Run all agents in parallel
    audit_tasks = [
        agent.execute_async(
            task=f"Audit this code for security vulnerabilities:\n\n{code_content}",
            output_schema={
                "type": "object",
                "properties": {
                    "vulnerabilities": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": {
                                "id": {"type": "string"},
                                "type": {"type": "string"},
                                "severity": {"type": "string"},
                                "line": {"type": "integer"},
                                "description": {"type": "string"},
                                "cve": {"type": "string"}
                            }
                        }
                    }
                }
            }
        )
        for agent in security_agents
    ]
    
    results = await asyncio.gather(*audit_tasks)
    
    # Aggregate using voting: only include vulnerabilities found by 2+ agents
    aggregator = VotingAggregator(
        min_votes=2,  # Require consensus from at least 2 of 3 agents
        deduplication_key="type",  # Deduplicate by vulnerability type
        merge_strategy="highest_severity"  # When merging, keep highest severity
    )
    
    aggregated = aggregator.aggregate(
        results=[r.structured_output for r in results],
        agent_names=[a.name for a in security_agents]
    )
    
    print(f"Individual findings: {aggregated.total_individual_findings}")
    print(f"Consensus findings (2+ agents): {aggregated.consensus_count}")
    print(f"Unique findings (1 agent only): {aggregated.unique_count}")
    print(f"Confidence score: {aggregated.confidence:.2%}")
    
    return aggregated

# Weighted aggregation when agents have different expertise levels
async def weighted_code_quality_assessment(code_file: str) -> AggregationResult:
    """Aggregate quality scores from agents with different weights."""
    
    agents_and_weights = [
        (senior_dev_agent, 0.5),    # Senior dev opinion weighted highest
        (style_checker_agent, 0.3), # Style matters but less than logic
        (perf_agent, 0.2)           # Performance is important but context-dependent
    ]
    
    code_content = open(code_file).read()
    
    results = await asyncio.gather(*[
        agent.execute_async(task=f"Rate code quality (0-10):\n\n{code_content}")
        for agent, _ in agents_and_weights
    ])
    
    aggregator = WeightedAggregator(
        weights=[weight for _, weight in agents_and_weights],
        numeric_fields=["quality_score", "maintainability_score", "readability_score"]
    )
    
    return aggregator.aggregate([r.structured_output for r in results])

For more advanced patterns around orchestrating complex agent workflows, see our deep dive on AI agent workflows.

10. Using @Codex in Slack for Team Workflows

The Codex SDK ships with a first-class Slack integration that enables engineering teams to trigger multi-agent workflows directly from their communication platform. The @Codex bot can be mentioned in any channel to initiate tasks, and it reports progress and results back to the thread in real time.

Installing and Configuring the Slack Integration

pip install "openai-codex-sdk[slack]>=2.4.0"

from openai_codex.slack import CodexSlackBot, SlackBotConfig, SlackCommand
from openai_codex import AgentRegistry

# Configure the Slack bot
bot_config = SlackBotConfig(
    bot_token=os.environ["SLACK_BOT_TOKEN"],
    app_token=os.environ["SLACK_APP_TOKEN"],
    signing_secret=os.environ["SLACK_SIGNING_SECRET"],
    
    # Which Slack channels the bot is active in
    allowed_channels=["#engineering", "#code-review", "#devops"],
    
    # Map Slack user IDs to Codex permission levels
    user_permissions={
        "U123456": "admin",       # Can trigger any agent, modify configs
        "U789012": "developer",   # Can trigger code agents, cannot modify configs
        "U345678": "viewer"       # Can only view agent outputs, cannot trigger
    },
    
    # Cost limit per Slack-triggered run
    per_request_cost_limit_usd=2.00,
    
    # Require confirmation for expensive operations
    confirmation_threshold_usd=1.00
)

# Register agents with the Slack bot
registry = AgentRegistry(client=client)
registry.register(code_writer)
registry.register(code_reviewer)
registry.register(bug_fixer)

bot = CodexSlackBot(config=bot_config, registry=registry)

# Define custom Slack commands
@bot.command("/codex-review")
async def handle_review_command(command: SlackCommand):
    """Handle /codex-review <PR_URL> command."""
    pr_url = command.text.strip()
    
    if not pr_url.startswith("https://github.com/"):
        await command.respond("Please provide a valid GitHub PR URL.")
        return
    
    # Post initial acknowledgment
    await command.respond(
        f"🤖 Starting code review for {pr_url}...",
        in_thread=True
    )
    
    # Extract PR details and run pipeline
    pr_context = await extract_pr_context(pr_url)
    pipeline = build_code_review_pipeline()
    
    # Stream progress updates to Slack thread
    async for progress_update in pipeline.execute_stream(pr_context):
        if progress_update.stage_completed:
            await command.respond(
                f"✅ {progress_update.stage_name} complete "
                f"({progress_update.duration_ms}ms)",
                in_thread=True
            )
    
    # Post final review
    result = pipeline.last_result
    await command.respond(
        blocks=format_review_as_slack_blocks(result.outputs["final_review"]),
        in_thread=True
    )

@bot.mention_handler
async def handle_mention(mention: SlackCommand):
    """Handle @Codex mentions in channels."""
    message_text = mention.text.lower()
    
    if "fix" in message_text and "bug" in message_text:
        # Route to bug fixing workflow
        await mention.respond("🔧 Initiating bug analysis...")
        # Extract issue details from message context
        issue_details = await extract_issue_from_message(mention)
        await run_targeted_bug_fix(issue_details, mention)
        
    elif "review" in message_text:
        await mention.respond("👀 Starting code review workflow...")
        # Extract code from message attachments or linked files
        
    elif "explain" in message_text:
        # Route to explanation agent (no code execution needed)
        explanation_agent = Agent(
            client=client,
            name="explainer",
            model="gpt-4.5",
            system_prompt="Explain code clearly to engineering teams."
        )
        result = await explanation_agent.execute_async(task=mention.text)
        await mention.respond(result.text, in_thread=True)

# Start the bot
if __name__ == "__main__":
    bot.start()  # Uses Socket Mode for real-time event handling

Configuring Slack Workflow Triggers

Beyond slash commands and mentions, the Codex Slack integration supports workflow triggers that activate agents based on channel events. For example, you can configure an agent to automatically run a security scan whenever a message containing a GitHub PR link is posted to #code-review.

from openai_codex.slack import ChannelEventTrigger, TriggerCondition

# Auto-trigger security scan on PR links
pr_link_trigger = ChannelEventTrigger(
    name="auto_security_scan",
    condition=TriggerCondition(
        event_type="message",
        channels=["#code-review"],
        contains_pattern=r"github\.com/[^/]+/[^/]+/pull/\d+",
        user_permission_required="developer"
    ),
    action=lambda event: run_security_scan_for_pr(
        pr_url=extract_github_url(event.text)
    ),
    cooldown_seconds=60  # Don't trigger again for same PR within 60s
)

bot.register_trigger(pr_link_trigger)

11. Admin Tools for Managing Agent Permissions

In team environments, controlling what agents can access and execute is as important as the agents themselves. The Codex SDK provides a comprehensive RBAC system for agent permissions, an audit log for all agent actions, and administrative tools for managing agent configurations across an organization.

Defining Permission Policies

from openai_codex.admin import PermissionPolicy, AgentRole, ResourcePermission
from openai_codex.admin import PolicyEnforcer, AuditLogger

# Define roles with specific permissions
roles = {
    "read_only_agent": AgentRole(
        name="read_only_agent",
        permissions=[
            ResourcePermission(resource="filesystem", actions=["read"]),
            ResourcePermission(resource="git", actions=["diff", "log", "show"]),
            ResourcePermission(resource="shell", actions=[]),  # No shell access
        ]
    ),
    "code_writer_agent": AgentRole(
        name="code_writer_agent",
        permissions=[
            ResourcePermission(
                resource="filesystem",
                actions=["read", "write"],
                constraints={"allowed_paths": ["/workspace/src", "/workspace/tests"]}
            ),
            ResourcePermission(
                resource="shell",
                actions=["execute"],
                constraints={"allowed_commands": ["python", "pip", "pytest"]}
            ),
            ResourcePermission(
                resource="git",
                actions=["diff", "add", "commit"],
                constraints={"max_files_per_commit": 10}
            )
        ]
    ),
    "deployment_agent": AgentRole(
        name="deployment_agent",
        permissions=[
            ResourcePermission(resource="filesystem", actions=["read"]),
            ResourcePermission(
                resource="shell",
                actions=["execute"],
                constraints={
                    "allowed_commands": ["docker", "kubectl", "helm"],
                    "require_approval_for": ["kubectl delete", "helm uninstall"]
                }
            ),
            ResourcePermission(
                resource="secrets",
                actions=["read"],
                constraints={"allowed_secret_names": ["DEPLOY_KEY", "REGISTRY_TOKEN"]}
            )
        ]
    )
}

# Create and enforce policies
policy_enforcer = PolicyEnforcer(
    roles=roles,
    default_role="read_only_agent",  # Agents default to minimal permissions
    audit_logger=AuditLogger(
        output="file",
        file_path="/var/log/codex/agent_audit.jsonl",
        include_payloads=True,  # Log full tool call payloads
        retention_days=90
    )
)

# Assign roles to agents
policy_enforcer.assign_role("code_writer", "code_writer_agent")
policy_enforcer.assign_role("code_reviewer", "read_only_agent")
policy_enforcer.assign_role("deployment_agent", "deployment_agent")

# Attach enforcer to client - all agents automatically inherit policies
client.set_policy_enforcer(policy_enforcer)

Approval Workflows for High-Risk Operations

from openai_codex.admin import ApprovalWorkflow, ApprovalRequest, ApprovalChannel

# Configure approval workflow for dangerous operations
approval_workflow = ApprovalWorkflow(
    client=client,
    approval_channel=ApprovalChannel.SLACK,
    slack_channel="#devops-approvals",
    approvers=["U123456", "U234567"],  # Slack user IDs of authorized approvers
    approval_timeout_seconds=3600,  # 1 hour to approve
    
    # Define which operations require approval
    requires_approval=[
        "kubectl delete",
        "helm uninstall",
        "git push --force",
        "DROP TABLE",  # SQL operations matching this pattern
        "rm -rf"
    ]
)

# Register with the policy enforcer
policy_enforcer.set_approval_workflow(approval_workflow)

# When an agent attempts a restricted operation, it automatically pauses
# and sends an approval request to Slack before proceeding

Viewing and Managing Active Agents

from openai_codex.admin import AgentManager

manager = AgentManager(client=client)

# List all currently running agents
active_agents = await manager.list_running()
# Illustrative admin API: not part of the documented Codex SDK
for agent in active_agents:
    print(f"{agent.id}: task={agent.current_task!r}, elapsed={agent.elapsed_seconds:.1f}s")

if active_agents:
    await manager.stop(active_agents[0].id)

The preceding block is illustrative, like the rest of section 11: the openai_codex.admin module, AgentManager, ApprovalWorkflow and the policy classes are not documented Codex SDK interfaces. The official Codex SDK documentation describes Codex and AsyncCodex clients, threads created with thread_start(), execution through run(), and sandbox presets. The next sections implement monitoring, budgets and error handling in your own orchestrator.

12. Monitoring Agent Execution

Monitoring should tell you which agent is running, how long it has been running, whether it finished, and what error ended the run. Keep that operational record outside the model response: a response may describe success while a later validation, file operation or deployment step fails.

The wrapper below uses the documented asynchronous thread flow and records lifecycle data in application-owned metrics, logs and run records. It needs the prometheus_client package (pip install prometheus-client); start_http_server serves the metrics over HTTP on the port set by METRICS_PORT, 9090 by default.

import asyncio
import logging
import os
import time
import uuid
from dataclasses import dataclass
from typing import Optional

from openai_codex import AsyncCodex, Sandbox
from prometheus_client import Counter, Gauge, Histogram, start_http_server

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("codex_orchestrator")

RUNS = Counter(
    "codex_agent_runs_total",
    "Agent runs by final status",
    ["agent", "status"],
)
DURATION = Histogram(
    "codex_agent_duration_seconds",
    "Agent execution duration",
    ["agent"],
)
ACTIVE = Gauge(
    "codex_agents_active",
    "Currently active agent runs",
    ["agent"],
)


@dataclass
class RunRecord:
    run_id: str
    agent: str
    started_at: float
    ended_at: Optional[float] = None
    status: str = "running"
    error: Optional[str] = None


class MonitoredCodexRunner:
    def __init__(self, model: Optional[str] = None) -> None:
        self.model = model
        self.runs: dict[str, RunRecord] = {}

    async def run(
        self,
        agent: str,
        prompt: str,
        sandbox: Sandbox = Sandbox.workspace_write,
    ) -> str:
        run_id = str(uuid.uuid4())
        record = RunRecord(run_id, agent, time.time())
        self.runs[run_id] = record
        started = time.monotonic()
        ACTIVE.labels(agent).inc()
        logger.info("agent_started agent=%s run_id=%s", agent, run_id)

        try:
            async with AsyncCodex() as codex:
                options = {"sandbox": sandbox}
                if self.model:
                    options["model"] = self.model
                thread = await codex.thread_start(**options)
                result = await thread.run(prompt)

            record.status = "succeeded"
            RUNS.labels(agent, "succeeded").inc()
            return result.final_response
        except asyncio.CancelledError:
            record.status = "cancelled"
            RUNS.labels(agent, "cancelled").inc()
            logger.warning("agent_cancelled agent=%s run_id=%s", agent, run_id)
            raise
        except Exception as exc:
            record.status = "failed"
            record.error = str(exc)
            RUNS.labels(agent, "failed").inc()
            logger.exception(
                "agent_failed agent=%s run_id=%s", agent, run_id
            )
            raise
        finally:
            record.ended_at = time.time()
            DURATION.labels(agent).observe(time.monotonic() - started)
            ACTIVE.labels(agent).dec()
            logger.info(
                "agent_finished agent=%s run_id=%s status=%s",
                agent,
                run_id,
                record.status,
            )


async def main() -> None:
    start_http_server(int(os.getenv("METRICS_PORT", "9090")))
    runner = MonitoredCodexRunner(model=os.getenv("CODEX_MODEL"))
    response = await runner.run(
        agent="reviewer",
        prompt="Review the current workspace and report correctness risks.",
        sandbox=Sandbox.read_only,
    )
    print(response)


if __name__ == "__main__":
    asyncio.run(main())

MonitoredCodexRunner.runs keeps one RunRecord per run, with status, timestamps and any error, so you can serve it from an admin endpoint or write it to your database. The Prometheus metrics cover the aggregate view: runs by status, durations and active runs per agent.

Use stable, low-cardinality metric labels. An agent role such as reviewer is suitable; a run identifier, repository URL, or task text is not. Put unique identifiers in structured logs and correlate them with your own run record.

Prometheus metrics show what your orchestrator observed, but they do not replace account-level usage information. The Codex pricing and usage documentation states that the Codex command-line interface (CLI) /status command shows remaining limits. It also links to the Codex usage dashboard. Check both when local concurrency appears normal but new runs are being constrained.

13. Cost Management Across Multiple Agents

Do not estimate a multi-agent run from prompt length alone. OpenAI states that Codex usage depends on model choice, context, reasoning, tool use, retrieval, and caching. Task size and complexity, as well as whether work runs locally or in the cloud, also affect usage. Measure completed workflows and enforce limits at the orchestration boundary rather than assuming every worker consumes the same amount.

Authentication mode also changes the available workflow. When Codex is used with an API key, usage is billed according to API pricing and cloud-based features such as Slack are unavailable. The @Codex Slack workflow in section 10 therefore assumes ChatGPT-plan sign-in; with API-key authentication you would need to build your own Slack integration. These details are documented on the official Codex pricing page.

Fast mode is not a free latency switch: OpenAI documents that it consumes limits at higher rates. Make it an explicit configuration choice for a specific workflow instead of enabling it globally.

Apply several independent controls. Bound concurrency so a large queue cannot start every agent at once. Give each run a wall-clock deadline and a maximum number of orchestration turns. Keep tasks tightly scoped, select a smaller configured model for routine subtasks when it produces a useful result, and give review-only agents Sandbox.read_only rather than write access.

import asyncio
import os
from collections.abc import Sequence

from openai_codex import AsyncCodex, Sandbox


async def run_bounded(
    tasks: Sequence[str],
    max_concurrency: int = 3,
    timeout_seconds: float = 180.0,
) -> list[str | BaseException]:
    semaphore = asyncio.Semaphore(max_concurrency)
    model = os.getenv("CODEX_MODEL")

    async with AsyncCodex() as codex:
        async def run_one(task: str) -> str:
            async with semaphore:
                options = {"sandbox": Sandbox.read_only}
                if model:
                    options["model"] = model
                thread = await codex.thread_start(**options)
                result = await asyncio.wait_for(
                    thread.run(task),
                    timeout=timeout_seconds,
                )
                return result.final_response

        results = await asyncio.gather(
            *(run_one(task) for task in tasks),
            return_exceptions=True,
        )

    return list(results)

The sample needs no turn counter because every task makes exactly one thread.run() call. If a workflow continues a thread across multiple turns, increment an application-owned counter before each call and reject the next call when it would exceed the run’s configured turn budget.

Inspect each returned item and classify exceptions as timed out, canceled, or failed before persisting the batch result. Because return_exceptions=True keeps one failure from terminating collection, successful sibling results remain available for validation or aggregation.

A semaphore limits simultaneous work, while asyncio.wait_for() limits the duration of each task. For a queue-backed service, also cap queued work per repository or team so one large request cannot crowd out unrelated work. Track accepted and validated results, not just the number of tasks started.

14. Error Handling and Fallback Strategies

Failures should remain local to the agent that caused them. Wrap each invocation with asyncio.wait_for(), record timeouts separately from other failures, and cancel dependent work when its required input is unavailable. Independent agents may continue, but the final aggregator must know which inputs are missing.

Retry only failures your application has classified as transient, such as a temporary transport interruption. Use exponential backoff with jitter and a small attempt limit. Catch a generic Exception at the SDK boundary, preserve the original exception in logs, and avoid inventing SDK-specific exception classes.

import asyncio
import random
from collections.abc import Awaitable, Callable
from typing import TypeVar

T = TypeVar("T")


async def retry_transient(
    operation: Callable[[], Awaitable[T]],
    is_transient: Callable[[Exception], bool],
    attempts: int = 3,
) -> T:
    for attempt in range(attempts):
        try:
            return await operation()
        except Exception as exc:
            if not is_transient(exc) or attempt == attempts - 1:
                raise

            delay = (2 ** attempt) + random.uniform(0.0, 0.5)
            await asyncio.sleep(delay)

    raise RuntimeError("retry loop ended unexpectedly")

Do not blindly retry non-idempotent work. A task that edits files, creates a pull request, sends a message, or starts a deployment may have completed its side effect before the caller lost the response. Assign an operation identifier and check for the existing side effect before executing the operation again.

• Isolate writes: give each writing agent its own Git branch or worktree. Validate the diff and discard the worktree when the run fails.
• Stop on conflicts: if agents modify overlapping lines, require a reconciliation step or human review instead of choosing a patch silently.
• Reduce scope safely: if a specialist times out, use a single-agent fallback only for low-risk work that remains useful with reduced coverage.
• Fail closed: send credentials, destructive commands, production changes, repeated validation failures, and unresolved conflicts to a human reviewer.

A nonempty final_response shows that a thread returned text. Treat that text as a proposed result until separate checks establish that generated code compiles, tests pass, patches apply, and required approvals exist.

15. Conclusion and Next Steps

A maintainable multi-agent coding system is an orchestrator around narrowly scoped Codex threads. It controls concurrency, selects a sandbox, records state transitions, validates artifacts outside the model, and places approval checks before risky side effects. Coordinator, pipeline, and swarm patterns remain useful designs, but the application must implement them using documented SDK primitives.

Keep model selection in configuration. OpenAI states that GPT-5.5 retires from ChatGPT, ChatGPT Work, and Codex on all plans on October 14, 2026, while the OpenAI API is not affected. Before selecting or migrating a model, check the official Codex models page and confirm that the configured model is available to the account running the workflow.

• Replace illustrative administration: inventory the illustrative openai_codex.admin, permission-policy, and approval-workflow calls in section 11. Replace them with application-owned authorization, tracking, and approval components around documented thread operations.
• Add a continuous integration entry point: connect one bounded workflow through the documented Codex GitHub Action or non-interactive mode, then make its validation result visible in the build.
• Create replay tests: save representative task specifications and expected invariants, then rerun them after prompt, model, permission, or orchestration changes.
• Gate write access: implement identity checks, approval records, and exact diff or command previews before enabling repository writes or deployment tools.

Start by replacing one read-only illustrative workflow with documented AsyncCodex, thread_start(), and run() calls. Add metrics and validation before expanding it to concurrent writers.