Implementation:Openai Openai agents python CodexTool Pattern
| Knowledge Sources | |
|---|---|
| Domains | Code Generation, CLI Integration, Experimental Tools |
| Last Updated | 2026-02-11 00:00 GMT |
Overview
Demonstrates running the experimental Codex CLI as an agent tool via the codex_tool() factory, with comprehensive streaming event handling covering reasoning, command execution, MCP tool calls, file changes, web searches, and todo lists.
Description
The CodexTool pattern wraps the Codex CLI as a subprocess-based tool that can be invoked by an agent. The codex_tool() factory function accepts configuration for sandbox mode, thread options, turn options, and a streaming callback. It returns a tool instance that, when called by the agent, spawns a Codex CLI process and streams events back to the caller.
The ThreadOptions class configures the Codex session with parameters including the model ("gpt-5.2-codex"), reasoning effort level, network access, web search capability, and approval policy. The TurnOptions class provides per-turn settings such as idle_timeout_seconds to abort if the CLI becomes unresponsive. The sandbox_mode parameter ("workspace-write" in this example) controls the Codex CLI's file system access level.
The streaming callback (on_stream) receives CodexToolStreamEvent payloads containing typed events. Thread-level events include ThreadStartedEvent, ThreadErrorEvent. Turn-level events include TurnStartedEvent, TurnCompletedEvent (with usage stats), and TurnFailedEvent. Item-level events wrap specific activity types: ReasoningItem for model reasoning text, CommandExecutionItem for shell commands with status and output, McpToolCallItem for MCP server interactions, FileChangeItem for file modifications, WebSearchItem for web queries, TodoListItem for task tracking, and ErrorItem for errors. Each item event comes in ItemStartedEvent, ItemUpdatedEvent, or ItemCompletedEvent wrappers.
This tool is marked as experimental and its API may change before general availability.
Usage
Use this pattern when you need an agent to perform complex multi-step coding tasks that benefit from a full CLI environment, including running shell commands, accessing MCP servers, modifying files, and performing web searches. It is suitable for automated code review, test coverage improvement, workspace analysis, and skill-based development workflows.
Code Reference
Source Location
- Repository: Openai_Openai_agents_python
- File: examples/tools/codex.py
- Lines: 1-163
Signature
codex_tool(
sandbox_mode="workspace-write",
default_thread_options=ThreadOptions(
model="gpt-5.2-codex",
model_reasoning_effort="low",
network_access_enabled=True,
web_search_enabled=False,
approval_policy="never",
),
default_turn_options=TurnOptions(
idle_timeout_seconds=60,
),
on_stream=on_codex_stream,
)
Import
from agents import Agent, Runner, gen_trace_id, trace
from agents.extensions.experimental.codex import (
CodexToolStreamEvent,
CommandExecutionItem,
ErrorItem,
FileChangeItem,
ItemCompletedEvent,
ItemStartedEvent,
ItemUpdatedEvent,
McpToolCallItem,
ReasoningItem,
ThreadErrorEvent,
ThreadOptions,
ThreadStartedEvent,
TodoListItem,
TurnCompletedEvent,
TurnFailedEvent,
TurnOptions,
TurnStartedEvent,
WebSearchItem,
codex_tool,
)
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| sandbox_mode | str |
Yes | Codex CLI sandbox mode: "workspace-write", "read-only", etc.
|
| default_thread_options | ThreadOptions |
No | Session-level configuration including model, reasoning effort, network access, and approval policy |
| default_thread_options.model | str |
No | Model ID for the Codex CLI (e.g., "gpt-5.2-codex")
|
| default_thread_options.model_reasoning_effort | str |
No | Reasoning effort level: "low", "medium", "high"
|
| default_thread_options.network_access_enabled | bool |
No | Whether the Codex CLI can access the network |
| default_thread_options.web_search_enabled | bool |
No | Whether web search is available to the CLI |
| default_thread_options.approval_policy | str |
No | Approval policy for CLI operations: "never", "always"
|
| default_turn_options | TurnOptions |
No | Per-turn configuration such as idle timeout |
| default_turn_options.idle_timeout_seconds | int |
No | Seconds to wait before aborting if no events arrive |
| on_stream | Callable[[CodexToolStreamEvent], Awaitable[None]] |
No | Async callback for streaming events from the Codex CLI |
Outputs
| Name | Type | Description |
|---|---|---|
| result.final_output | str |
The agent's final text response incorporating Codex CLI results |
| CodexToolStreamEvent.event | Event |
Typed event from the Codex CLI stream (see event types below) |
Stream Event Types
| Event Type | Description |
|---|---|
ThreadStartedEvent |
Emitted when a Codex thread begins, includes thread_id
|
ThreadErrorEvent |
Emitted on stream-level errors, includes message
|
TurnStartedEvent |
Emitted when a new turn begins |
TurnCompletedEvent |
Emitted when a turn finishes, includes usage statistics
|
TurnFailedEvent |
Emitted when a turn fails, includes error.message
|
ReasoningItem |
Model reasoning text |
CommandExecutionItem |
Shell command with command, status, and aggregated_output
|
McpToolCallItem |
MCP tool invocation with server, tool, and status
|
FileChangeItem |
File modification with changes and status
|
WebSearchItem |
Web search with query
|
TodoListItem |
Task list with items
|
ErrorItem |
Error with message
|
Usage Examples
Running Codex CLI as an Agent Tool
import asyncio
from agents import Agent, Runner, gen_trace_id, trace
from agents.extensions.experimental.codex import (
CodexToolStreamEvent,
CommandExecutionItem,
ReasoningItem,
ThreadStartedEvent,
ThreadOptions,
TurnCompletedEvent,
TurnOptions,
codex_tool,
)
async def on_codex_stream(payload: CodexToolStreamEvent) -> None:
event = payload.event
if isinstance(event, ThreadStartedEvent):
print(f"Thread started: {event.thread_id}")
elif isinstance(event, TurnCompletedEvent):
print(f"Turn completed, usage: {event.usage}")
elif hasattr(event, "item"):
item = event.item
if isinstance(item, ReasoningItem):
print(f"Reasoning: {item.text}")
elif isinstance(item, CommandExecutionItem):
print(f"Command: {item.command} | Status: {item.status}")
async def main():
agent = Agent(
name="Codex Agent",
instructions="Use the codex tool to inspect the workspace and answer the question.",
tools=[
codex_tool(
sandbox_mode="workspace-write",
default_thread_options=ThreadOptions(
model="gpt-5.2-codex",
model_reasoning_effort="low",
network_access_enabled=True,
approval_policy="never",
),
default_turn_options=TurnOptions(idle_timeout_seconds=60),
on_stream=on_codex_stream,
)
],
)
trace_id = gen_trace_id()
with trace("Codex tool example", trace_id=trace_id):
result = await Runner.run(
agent,
"Use $openai-knowledge skill to fetch the latest realtime model name.",
)
print(result.final_output)
asyncio.run(main())