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 The Complete ChatGPT API Integration Masterclass: Building Production Applications with the Responses API in 2026“>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 Hits 10 Million Users: The Complete 2026 Guide to Getting Started with AI-Powered Development“>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 5 Best AI Writing Assistants for coding Compared u2014 Features, Pricing, Use Cases“>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 The Big Prompt Engineering Story: What July 23’s News Means for Developers“>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}]: 
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 How to Build Custom AI Agents with OpenAI’s Responses API: From Single-Turn Chat to Multi-Step Autonomous Workflows“>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  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
  
  
  

  


    

    

      
      
      
      
      

      Please leave this field empty



Infrastructure & EngineeringInsight & AnalyticsGovernance, Privacy & EthicsProduct & StrategyAdministration & ManagementFinance & AccountingSales & MarketingOperations & ProductionHuman ResourcesInformation TechnologyKnowledge Seekers & CreatorsEnthusiasts & Specialized UsersStudentHousehold ManagerInfluencerResearcher


      

        
Thank you! Please check your inbox (and spam folder) for a confirmation email. Click the link to get instant access to our 40,000+ ChatGPT Prompt Library.Check your inbox or spam folder to confirm your subscription.
        
        
        
      
    

      

    
  
  

  


    

    

      
      
      
      
      

      Please leave this field empty



Infrastructure & EngineeringInsight & AnalyticsGovernance, Privacy & EthicsProduct & StrategyAdministration & ManagementFinance & AccountingSales & MarketingOperations & ProductionHuman ResourcesInformation TechnologyKnowledge Seekers & CreatorsEnthusiasts & Specialized UsersStudentHousehold ManagerInfluencerResearcher


      

        
Thank you! Please check your inbox (and spam folder) for a confirmation email. Click the link to get instant access to our 40,000+ ChatGPT Prompt Library.Check your inbox or spam folder to confirm your subscription.
        
        
        
      
    

          
      

    
  
  

  


    

    

      
      
      
      
      

      Please leave this field empty



Infrastructure & EngineeringInsight & AnalyticsGovernance, Privacy & EthicsProduct & StrategyAdministration & ManagementFinance & AccountingSales & MarketingOperations & ProductionHuman ResourcesInformation TechnologyKnowledge Seekers & CreatorsEnthusiasts & Specialized UsersStudentHousehold ManagerInfluencerResearcher


      

        
Thank you! Please check your inbox (and spam folder) for a confirmation email. Click the link to get instant access to our 40,000+ ChatGPT Prompt Library.Check your inbox or spam folder to confirm your subscription.