forked from openai/openai-agents-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathcodex.py
More file actions
165 lines (147 loc) · 5.81 KB
/
Copy pathcodex.py
File metadata and controls
165 lines (147 loc) · 5.81 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
import asyncio
from datetime import datetime
from agents import Agent, Runner, gen_trace_id, trace
# This tool is still in experimental phase and the details could be changed until being GAed.
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,
)
# This example runs the Codex CLI via the Codex tool wrapper.
# You can configure the CLI path with CODEX_PATH or CodexOptions(codex_path_override="...").
# codex_tool accepts options as keyword arguments or a plain dict.
# For example: codex_tool(sandbox_mode="read-only") or codex_tool({"sandbox_mode": "read-only"}).
async def on_codex_stream(payload: CodexToolStreamEvent) -> None:
event = payload.event
if isinstance(event, ThreadStartedEvent):
log(f"codex thread started: {event.thread_id}")
return
if isinstance(event, TurnStartedEvent):
log("codex turn started")
return
if isinstance(event, TurnCompletedEvent):
usage = event.usage
log(f"codex turn completed, usage: {usage}")
return
if isinstance(event, TurnFailedEvent):
error = event.error.message
log(f"codex turn failed: {error}")
return
if isinstance(event, ThreadErrorEvent):
log(f"codex stream error: {event.message}")
return
if not isinstance(event, ItemStartedEvent | ItemUpdatedEvent | ItemCompletedEvent):
return
item = event.item
if isinstance(item, ReasoningItem):
text = item.text
log(f"codex reasoning ({event.type}): {text}")
return
if isinstance(item, CommandExecutionItem):
command = item.command
output = item.aggregated_output
output_preview = output[-200:] if isinstance(output, str) else ""
status = item.status
log(f"codex command {event.type}: {command} | status={status} | output={output_preview}")
return
if isinstance(item, McpToolCallItem):
server = item.server
tool = item.tool
status = item.status
log(f"codex mcp {event.type}: {server}.{tool} | status={status}")
return
if isinstance(item, FileChangeItem):
changes = item.changes
status = item.status
log(f"codex file change {event.type}: {status} | {changes}")
return
if isinstance(item, WebSearchItem):
log(f"codex web search {event.type}: {item.query}")
return
if isinstance(item, TodoListItem):
items = item.items
log(f"codex todo list {event.type}: {len(items)} items")
return
if isinstance(item, ErrorItem):
log(f"codex error {event.type}: {item.message}")
def _timestamp() -> str:
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def log(message: str) -> None:
timestamp = _timestamp()
lines = str(message).splitlines() or [""]
for line in lines:
print(f"{timestamp} {line}")
async def main() -> None:
agent = Agent(
name="Codex Agent",
instructions=(
"Use the codex tool to inspect the workspace in read-only mode and answer the question. "
"When skill names, which usually starts with `$`, are mentioned, "
"you must rely on the codex tool to use the skill and answer the question.\n\n"
"When you send the final answer, you must include the following info at the end:\n\n"
"Run `codex resume <thread_id>` to continue the codex session."
),
tools=[
# Run local Codex CLI as a sub process
codex_tool(
sandbox_mode="read-only",
default_thread_options=ThreadOptions(
# You can pass a Codex instance to customize CLI details
# codex=Codex(executable_path="/path/to/codex", base_url="..."),
model="gpt-5.4",
model_reasoning_effort="low",
network_access_enabled=True,
web_search_enabled=False,
approval_policy="never", # We'll update this example once the HITL is implemented
),
default_turn_options=TurnOptions(
# Abort Codex CLI if no events arrive within this many seconds.
idle_timeout_seconds=60,
),
on_stream=on_codex_stream,
)
],
)
trace_id = gen_trace_id()
log(f"View trace: https://platform.openai.com/traces/trace?trace_id={trace_id}")
with trace("Codex tool example", trace_id=trace_id):
log("Using the Codex tool to inspect pyproject.toml and summarize Python requirements...")
result = await Runner.run(
agent,
(
"Inspect pyproject.toml in this repository and summarize the supported Python "
"version plus the main local test command. Do not modify any files."
),
)
log(result.final_output)
# Use local inspection in read-only mode.
log(
"Using the Codex tool to inspect AGENTS.md and summarize the local verification workflow..."
)
result = await Runner.run(
agent,
(
"Inspect AGENTS.md and summarize the mandatory local verification commands for this "
"repository. Do not modify any files or suggest code changes."
),
)
log(result.final_output)
# (A read-only summary of the local verification workflow will be displayed.)
if __name__ == "__main__":
asyncio.run(main())