mirror of
https://github.com/openai/codex.git
synced 2026-10-18 17:18:52 +00:00
Add Python SDK public API and examples (#14446)
## TL;DR WIP esp the examples Thin the Python SDK public surface so the wrapper layer returns canonical app-server generated models directly. - keeps `Codex` / `AsyncCodex` / `Thread` / `Turn` and input helpers, but removes alias-only type layers and custom result models - `metadata` now returns `InitializeResponse` and `run()` returns the generated app-server `Turn` - updates docs, examples, notebook, and tests to use canonical generated types and regenerates `v2_all.py` against current schema - keeps the pinned runtime-package integration flow and real integration coverage ## Validation - `PYTHONPATH=sdk/python/src python3 -m pytest sdk/python/tests` - `GH_TOKEN="$(gh auth token)" RUN_REAL_CODEX_TESTS=1 PYTHONPATH=sdk/python/src python3 -m pytest sdk/python/tests -rs` --------- Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Shaqayeq
@@ -0,0 +1,38 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import (
|
||||
assistant_text_from_turn,
|
||||
ensure_local_sdk_src,
|
||||
find_turn_by_id,
|
||||
runtime_config,
|
||||
server_label,
|
||||
)
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
import asyncio
|
||||
|
||||
from codex_app_server import AsyncCodex, TextInput
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async with AsyncCodex(config=runtime_config()) as codex:
|
||||
print("Server:", server_label(codex.metadata))
|
||||
|
||||
thread = await codex.thread_start(model="gpt-5.4", config={"model_reasoning_effort": "high"})
|
||||
turn = await thread.turn(TextInput("Say hello in one sentence."))
|
||||
result = await turn.run()
|
||||
persisted = await thread.read(include_turns=True)
|
||||
persisted_turn = find_turn_by_id(persisted.thread.turns, result.id)
|
||||
|
||||
print("Status:", result.status)
|
||||
print("Text:", assistant_text_from_turn(persisted_turn))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,28 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import (
|
||||
assistant_text_from_turn,
|
||||
ensure_local_sdk_src,
|
||||
find_turn_by_id,
|
||||
runtime_config,
|
||||
server_label,
|
||||
)
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
from codex_app_server import Codex, TextInput
|
||||
|
||||
with Codex(config=runtime_config()) as codex:
|
||||
print("Server:", server_label(codex.metadata))
|
||||
|
||||
thread = codex.thread_start(model="gpt-5.4", config={"model_reasoning_effort": "high"})
|
||||
result = thread.turn(TextInput("Say hello in one sentence.")).run()
|
||||
persisted = thread.read(include_turns=True)
|
||||
persisted_turn = find_turn_by_id(persisted.thread.turns, result.id)
|
||||
print("Status:", result.status)
|
||||
print("Text:", assistant_text_from_turn(persisted_turn))
|
||||
@@ -0,0 +1,43 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import (
|
||||
assistant_text_from_turn,
|
||||
ensure_local_sdk_src,
|
||||
find_turn_by_id,
|
||||
runtime_config,
|
||||
)
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
import asyncio
|
||||
|
||||
from codex_app_server import AsyncCodex, TextInput
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async with AsyncCodex(config=runtime_config()) as codex:
|
||||
thread = await codex.thread_start(model="gpt-5.4", config={"model_reasoning_effort": "high"})
|
||||
turn = await thread.turn(TextInput("Give 3 bullets about SIMD."))
|
||||
result = await turn.run()
|
||||
persisted = await thread.read(include_turns=True)
|
||||
persisted_turn = find_turn_by_id(persisted.thread.turns, result.id)
|
||||
|
||||
print("thread_id:", thread.id)
|
||||
print("turn_id:", result.id)
|
||||
print("status:", result.status)
|
||||
if result.error is not None:
|
||||
print("error:", result.error)
|
||||
print("text:", assistant_text_from_turn(persisted_turn))
|
||||
print(
|
||||
"persisted.items.count:",
|
||||
0 if persisted_turn is None else len(persisted_turn.items or []),
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,34 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import (
|
||||
assistant_text_from_turn,
|
||||
ensure_local_sdk_src,
|
||||
find_turn_by_id,
|
||||
runtime_config,
|
||||
)
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
from codex_app_server import Codex, TextInput
|
||||
|
||||
with Codex(config=runtime_config()) as codex:
|
||||
thread = codex.thread_start(model="gpt-5.4", config={"model_reasoning_effort": "high"})
|
||||
result = thread.turn(TextInput("Give 3 bullets about SIMD.")).run()
|
||||
persisted = thread.read(include_turns=True)
|
||||
persisted_turn = find_turn_by_id(persisted.thread.turns, result.id)
|
||||
|
||||
print("thread_id:", thread.id)
|
||||
print("turn_id:", result.id)
|
||||
print("status:", result.status)
|
||||
if result.error is not None:
|
||||
print("error:", result.error)
|
||||
print("text:", assistant_text_from_turn(persisted_turn))
|
||||
print(
|
||||
"persisted.items.count:",
|
||||
0 if persisted_turn is None else len(persisted_turn.items or []),
|
||||
)
|
||||
@@ -0,0 +1,63 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import (
|
||||
assistant_text_from_turn,
|
||||
ensure_local_sdk_src,
|
||||
find_turn_by_id,
|
||||
runtime_config,
|
||||
)
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
import asyncio
|
||||
|
||||
from codex_app_server import AsyncCodex, TextInput
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async with AsyncCodex(config=runtime_config()) as codex:
|
||||
thread = await codex.thread_start(model="gpt-5.4", config={"model_reasoning_effort": "high"})
|
||||
turn = await thread.turn(TextInput("Explain SIMD in 3 short bullets."))
|
||||
|
||||
event_count = 0
|
||||
saw_started = False
|
||||
saw_delta = False
|
||||
completed_status = "unknown"
|
||||
|
||||
async for event in turn.stream():
|
||||
event_count += 1
|
||||
if event.method == "turn/started":
|
||||
saw_started = True
|
||||
print("stream.started")
|
||||
continue
|
||||
if event.method == "item/agentMessage/delta":
|
||||
delta = getattr(event.payload, "delta", "")
|
||||
if delta:
|
||||
if not saw_delta:
|
||||
print("assistant> ", end="", flush=True)
|
||||
print(delta, end="", flush=True)
|
||||
saw_delta = True
|
||||
continue
|
||||
if event.method == "turn/completed":
|
||||
completed_status = getattr(event.payload.turn.status, "value", str(event.payload.turn.status))
|
||||
|
||||
if saw_delta:
|
||||
print()
|
||||
else:
|
||||
persisted = await thread.read(include_turns=True)
|
||||
persisted_turn = find_turn_by_id(persisted.thread.turns, turn.id)
|
||||
final_text = assistant_text_from_turn(persisted_turn).strip() or "[no assistant text]"
|
||||
print("assistant>", final_text)
|
||||
|
||||
print("stream.started.seen:", saw_started)
|
||||
print("stream.completed:", completed_status)
|
||||
print("events.count:", event_count)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,55 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import (
|
||||
assistant_text_from_turn,
|
||||
ensure_local_sdk_src,
|
||||
find_turn_by_id,
|
||||
runtime_config,
|
||||
)
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
from codex_app_server import Codex, TextInput
|
||||
|
||||
with Codex(config=runtime_config()) as codex:
|
||||
thread = codex.thread_start(model="gpt-5.4", config={"model_reasoning_effort": "high"})
|
||||
turn = thread.turn(TextInput("Explain SIMD in 3 short bullets."))
|
||||
|
||||
event_count = 0
|
||||
saw_started = False
|
||||
saw_delta = False
|
||||
completed_status = "unknown"
|
||||
|
||||
for event in turn.stream():
|
||||
event_count += 1
|
||||
if event.method == "turn/started":
|
||||
saw_started = True
|
||||
print("stream.started")
|
||||
continue
|
||||
if event.method == "item/agentMessage/delta":
|
||||
delta = getattr(event.payload, "delta", "")
|
||||
if delta:
|
||||
if not saw_delta:
|
||||
print("assistant> ", end="", flush=True)
|
||||
print(delta, end="", flush=True)
|
||||
saw_delta = True
|
||||
continue
|
||||
if event.method == "turn/completed":
|
||||
completed_status = getattr(event.payload.turn.status, "value", str(event.payload.turn.status))
|
||||
|
||||
if saw_delta:
|
||||
print()
|
||||
else:
|
||||
persisted = thread.read(include_turns=True)
|
||||
persisted_turn = find_turn_by_id(persisted.thread.turns, turn.id)
|
||||
final_text = assistant_text_from_turn(persisted_turn).strip() or "[no assistant text]"
|
||||
print("assistant>", final_text)
|
||||
|
||||
print("stream.started.seen:", saw_started)
|
||||
print("stream.completed:", completed_status)
|
||||
print("events.count:", event_count)
|
||||
@@ -0,0 +1,26 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import ensure_local_sdk_src, runtime_config, server_label
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
import asyncio
|
||||
|
||||
from codex_app_server import AsyncCodex
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async with AsyncCodex(config=runtime_config()) as codex:
|
||||
print("server:", server_label(codex.metadata))
|
||||
models = await codex.models()
|
||||
print("models.count:", len(models.data))
|
||||
print("models:", ", ".join(model.id for model in models.data[:5]) or "[none]")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,18 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import ensure_local_sdk_src, runtime_config, server_label
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
from codex_app_server import Codex
|
||||
|
||||
with Codex(config=runtime_config()) as codex:
|
||||
print("server:", server_label(codex.metadata))
|
||||
models = codex.models()
|
||||
print("models.count:", len(models.data))
|
||||
print("models:", ", ".join(model.id for model in models.data[:5]) or "[none]")
|
||||
@@ -0,0 +1,34 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import assistant_text_from_turn, ensure_local_sdk_src, find_turn_by_id, runtime_config
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
import asyncio
|
||||
|
||||
from codex_app_server import AsyncCodex, TextInput
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async with AsyncCodex(config=runtime_config()) as codex:
|
||||
original = await codex.thread_start(model="gpt-5.4", config={"model_reasoning_effort": "high"})
|
||||
|
||||
first_turn = await original.turn(TextInput("Tell me one fact about Saturn."))
|
||||
_ = await first_turn.run()
|
||||
print("Created thread:", original.id)
|
||||
|
||||
resumed = await codex.thread_resume(original.id)
|
||||
second_turn = await resumed.turn(TextInput("Continue with one more fact."))
|
||||
second = await second_turn.run()
|
||||
persisted = await resumed.read(include_turns=True)
|
||||
persisted_turn = find_turn_by_id(persisted.thread.turns, second.id)
|
||||
print(assistant_text_from_turn(persisted_turn))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,25 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import assistant_text_from_turn, ensure_local_sdk_src, find_turn_by_id, runtime_config
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
from codex_app_server import Codex, TextInput
|
||||
|
||||
with Codex(config=runtime_config()) as codex:
|
||||
# Create an initial thread and turn so we have a real thread to resume.
|
||||
original = codex.thread_start(model="gpt-5.4", config={"model_reasoning_effort": "high"})
|
||||
first = original.turn(TextInput("Tell me one fact about Saturn.")).run()
|
||||
print("Created thread:", original.id)
|
||||
|
||||
# Resume the existing thread by ID.
|
||||
resumed = codex.thread_resume(original.id)
|
||||
second = resumed.turn(TextInput("Continue with one more fact.")).run()
|
||||
persisted = resumed.read(include_turns=True)
|
||||
persisted_turn = find_turn_by_id(persisted.thread.turns, second.id)
|
||||
print(assistant_text_from_turn(persisted_turn))
|
||||
@@ -0,0 +1,70 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import ensure_local_sdk_src, runtime_config
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
import asyncio
|
||||
|
||||
from codex_app_server import AsyncCodex, TextInput
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async with AsyncCodex(config=runtime_config()) as codex:
|
||||
thread = await codex.thread_start(model="gpt-5.4", config={"model_reasoning_effort": "high"})
|
||||
first = await (await thread.turn(TextInput("One sentence about structured planning."))).run()
|
||||
second = await (await thread.turn(TextInput("Now restate it for a junior engineer."))).run()
|
||||
|
||||
reopened = await codex.thread_resume(thread.id)
|
||||
listing_active = await codex.thread_list(limit=20, archived=False)
|
||||
reading = await reopened.read(include_turns=True)
|
||||
|
||||
_ = await reopened.set_name("sdk-lifecycle-demo")
|
||||
_ = await codex.thread_archive(reopened.id)
|
||||
listing_archived = await codex.thread_list(limit=20, archived=True)
|
||||
unarchived = await codex.thread_unarchive(reopened.id)
|
||||
|
||||
resumed_info = "n/a"
|
||||
try:
|
||||
resumed = await codex.thread_resume(
|
||||
unarchived.id,
|
||||
model="gpt-5.4",
|
||||
config={"model_reasoning_effort": "high"},
|
||||
)
|
||||
resumed_result = await (await resumed.turn(TextInput("Continue in one short sentence."))).run()
|
||||
resumed_info = f"{resumed_result.id} {resumed_result.status}"
|
||||
except Exception as exc:
|
||||
resumed_info = f"skipped({type(exc).__name__})"
|
||||
|
||||
forked_info = "n/a"
|
||||
try:
|
||||
forked = await codex.thread_fork(unarchived.id, model="gpt-5.4")
|
||||
forked_result = await (await forked.turn(TextInput("Take a different angle in one short sentence."))).run()
|
||||
forked_info = f"{forked_result.id} {forked_result.status}"
|
||||
except Exception as exc:
|
||||
forked_info = f"skipped({type(exc).__name__})"
|
||||
|
||||
compact_info = "sent"
|
||||
try:
|
||||
_ = await unarchived.compact()
|
||||
except Exception as exc:
|
||||
compact_info = f"skipped({type(exc).__name__})"
|
||||
|
||||
print("Lifecycle OK:", thread.id)
|
||||
print("first:", first.id, first.status)
|
||||
print("second:", second.id, second.status)
|
||||
print("read.turns:", len(reading.thread.turns or []))
|
||||
print("list.active:", len(listing_active.data))
|
||||
print("list.archived:", len(listing_archived.data))
|
||||
print("resumed:", resumed_info)
|
||||
print("forked:", forked_info)
|
||||
print("compact:", compact_info)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,63 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import ensure_local_sdk_src, runtime_config
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
from codex_app_server import Codex, TextInput
|
||||
|
||||
|
||||
with Codex(config=runtime_config()) as codex:
|
||||
thread = codex.thread_start(model="gpt-5.4", config={"model_reasoning_effort": "high"})
|
||||
first = thread.turn(TextInput("One sentence about structured planning.")).run()
|
||||
second = thread.turn(TextInput("Now restate it for a junior engineer.")).run()
|
||||
|
||||
reopened = codex.thread_resume(thread.id)
|
||||
listing_active = codex.thread_list(limit=20, archived=False)
|
||||
reading = reopened.read(include_turns=True)
|
||||
|
||||
_ = reopened.set_name("sdk-lifecycle-demo")
|
||||
_ = codex.thread_archive(reopened.id)
|
||||
listing_archived = codex.thread_list(limit=20, archived=True)
|
||||
unarchived = codex.thread_unarchive(reopened.id)
|
||||
|
||||
resumed_info = "n/a"
|
||||
try:
|
||||
resumed = codex.thread_resume(
|
||||
unarchived.id,
|
||||
model="gpt-5.4",
|
||||
config={"model_reasoning_effort": "high"},
|
||||
)
|
||||
resumed_result = resumed.turn(TextInput("Continue in one short sentence.")).run()
|
||||
resumed_info = f"{resumed_result.id} {resumed_result.status}"
|
||||
except Exception as exc:
|
||||
resumed_info = f"skipped({type(exc).__name__})"
|
||||
|
||||
forked_info = "n/a"
|
||||
try:
|
||||
forked = codex.thread_fork(unarchived.id, model="gpt-5.4")
|
||||
forked_result = forked.turn(TextInput("Take a different angle in one short sentence.")).run()
|
||||
forked_info = f"{forked_result.id} {forked_result.status}"
|
||||
except Exception as exc:
|
||||
forked_info = f"skipped({type(exc).__name__})"
|
||||
|
||||
compact_info = "sent"
|
||||
try:
|
||||
_ = unarchived.compact()
|
||||
except Exception as exc:
|
||||
compact_info = f"skipped({type(exc).__name__})"
|
||||
|
||||
print("Lifecycle OK:", thread.id)
|
||||
print("first:", first.id, first.status)
|
||||
print("second:", second.id, second.status)
|
||||
print("read.turns:", len(reading.thread.turns or []))
|
||||
print("list.active:", len(listing_active.data))
|
||||
print("list.archived:", len(listing_archived.data))
|
||||
print("resumed:", resumed_info)
|
||||
print("forked:", forked_info)
|
||||
print("compact:", compact_info)
|
||||
@@ -0,0 +1,42 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import (
|
||||
assistant_text_from_turn,
|
||||
ensure_local_sdk_src,
|
||||
find_turn_by_id,
|
||||
runtime_config,
|
||||
)
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
import asyncio
|
||||
|
||||
from codex_app_server import AsyncCodex, ImageInput, TextInput
|
||||
|
||||
REMOTE_IMAGE_URL = "https://raw.githubusercontent.com/github/explore/main/topics/python/python.png"
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async with AsyncCodex(config=runtime_config()) as codex:
|
||||
thread = await codex.thread_start(model="gpt-5.4", config={"model_reasoning_effort": "high"})
|
||||
turn = await thread.turn(
|
||||
[
|
||||
TextInput("What is in this image? Give 3 bullets."),
|
||||
ImageInput(REMOTE_IMAGE_URL),
|
||||
]
|
||||
)
|
||||
result = await turn.run()
|
||||
persisted = await thread.read(include_turns=True)
|
||||
persisted_turn = find_turn_by_id(persisted.thread.turns, result.id)
|
||||
|
||||
print("Status:", result.status)
|
||||
print(assistant_text_from_turn(persisted_turn))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,33 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import (
|
||||
assistant_text_from_turn,
|
||||
ensure_local_sdk_src,
|
||||
find_turn_by_id,
|
||||
runtime_config,
|
||||
)
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
from codex_app_server import Codex, ImageInput, TextInput
|
||||
|
||||
REMOTE_IMAGE_URL = "https://raw.githubusercontent.com/github/explore/main/topics/python/python.png"
|
||||
|
||||
with Codex(config=runtime_config()) as codex:
|
||||
thread = codex.thread_start(model="gpt-5.4", config={"model_reasoning_effort": "high"})
|
||||
result = thread.turn(
|
||||
[
|
||||
TextInput("What is in this image? Give 3 bullets."),
|
||||
ImageInput(REMOTE_IMAGE_URL),
|
||||
]
|
||||
).run()
|
||||
persisted = thread.read(include_turns=True)
|
||||
persisted_turn = find_turn_by_id(persisted.thread.turns, result.id)
|
||||
|
||||
print("Status:", result.status)
|
||||
print(assistant_text_from_turn(persisted_turn))
|
||||
@@ -0,0 +1,43 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import (
|
||||
assistant_text_from_turn,
|
||||
ensure_local_sdk_src,
|
||||
find_turn_by_id,
|
||||
runtime_config,
|
||||
temporary_sample_image_path,
|
||||
)
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
import asyncio
|
||||
|
||||
from codex_app_server import AsyncCodex, LocalImageInput, TextInput
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
with temporary_sample_image_path() as image_path:
|
||||
async with AsyncCodex(config=runtime_config()) as codex:
|
||||
thread = await codex.thread_start(model="gpt-5.4", config={"model_reasoning_effort": "high"})
|
||||
|
||||
turn = await thread.turn(
|
||||
[
|
||||
TextInput("Read this generated local image and summarize the colors/layout in 2 bullets."),
|
||||
LocalImageInput(str(image_path.resolve())),
|
||||
]
|
||||
)
|
||||
result = await turn.run()
|
||||
persisted = await thread.read(include_turns=True)
|
||||
persisted_turn = find_turn_by_id(persisted.thread.turns, result.id)
|
||||
|
||||
print("Status:", result.status)
|
||||
print(assistant_text_from_turn(persisted_turn))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,34 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import (
|
||||
assistant_text_from_turn,
|
||||
ensure_local_sdk_src,
|
||||
find_turn_by_id,
|
||||
runtime_config,
|
||||
temporary_sample_image_path,
|
||||
)
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
from codex_app_server import Codex, LocalImageInput, TextInput
|
||||
|
||||
with temporary_sample_image_path() as image_path:
|
||||
with Codex(config=runtime_config()) as codex:
|
||||
thread = codex.thread_start(model="gpt-5.4", config={"model_reasoning_effort": "high"})
|
||||
|
||||
result = thread.turn(
|
||||
[
|
||||
TextInput("Read this generated local image and summarize the colors/layout in 2 bullets."),
|
||||
LocalImageInput(str(image_path.resolve())),
|
||||
]
|
||||
).run()
|
||||
persisted = thread.read(include_turns=True)
|
||||
persisted_turn = find_turn_by_id(persisted.thread.turns, result.id)
|
||||
|
||||
print("Status:", result.status)
|
||||
print(assistant_text_from_turn(persisted_turn))
|
||||
@@ -0,0 +1,31 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import (
|
||||
assistant_text_from_turn,
|
||||
ensure_local_sdk_src,
|
||||
find_turn_by_id,
|
||||
runtime_config,
|
||||
server_label,
|
||||
)
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
from codex_app_server import Codex, TextInput
|
||||
|
||||
with Codex(config=runtime_config()) as codex:
|
||||
print("Server:", server_label(codex.metadata))
|
||||
|
||||
thread = codex.thread_start(model="gpt-5.4", config={"model_reasoning_effort": "high"})
|
||||
turn = thread.turn(TextInput("Say hello in one sentence."))
|
||||
result = turn.run()
|
||||
persisted = thread.read(include_turns=True)
|
||||
persisted_turn = find_turn_by_id(persisted.thread.turns, result.id)
|
||||
|
||||
print("Thread:", thread.id)
|
||||
print("Turn:", result.id)
|
||||
print("Text:", assistant_text_from_turn(persisted_turn).strip())
|
||||
@@ -0,0 +1,98 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import (
|
||||
assistant_text_from_turn,
|
||||
ensure_local_sdk_src,
|
||||
find_turn_by_id,
|
||||
runtime_config,
|
||||
)
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
import asyncio
|
||||
import random
|
||||
from collections.abc import Awaitable, Callable
|
||||
from typing import TypeVar
|
||||
|
||||
from codex_app_server import (
|
||||
AsyncCodex,
|
||||
JsonRpcError,
|
||||
ServerBusyError,
|
||||
TextInput,
|
||||
TurnStatus,
|
||||
is_retryable_error,
|
||||
)
|
||||
|
||||
ResultT = TypeVar("ResultT")
|
||||
|
||||
|
||||
async def retry_on_overload_async(
|
||||
op: Callable[[], Awaitable[ResultT]],
|
||||
*,
|
||||
max_attempts: int = 3,
|
||||
initial_delay_s: float = 0.25,
|
||||
max_delay_s: float = 2.0,
|
||||
jitter_ratio: float = 0.2,
|
||||
) -> ResultT:
|
||||
if max_attempts < 1:
|
||||
raise ValueError("max_attempts must be >= 1")
|
||||
|
||||
delay = initial_delay_s
|
||||
attempt = 0
|
||||
while True:
|
||||
attempt += 1
|
||||
try:
|
||||
return await op()
|
||||
except Exception as exc: # noqa: BLE001
|
||||
if attempt >= max_attempts or not is_retryable_error(exc):
|
||||
raise
|
||||
jitter = delay * jitter_ratio
|
||||
sleep_for = min(max_delay_s, delay) + random.uniform(-jitter, jitter)
|
||||
if sleep_for > 0:
|
||||
await asyncio.sleep(sleep_for)
|
||||
delay = min(max_delay_s, delay * 2)
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async with AsyncCodex(config=runtime_config()) as codex:
|
||||
thread = await codex.thread_start(model="gpt-5.4", config={"model_reasoning_effort": "high"})
|
||||
|
||||
try:
|
||||
result = await retry_on_overload_async(
|
||||
_run_turn(thread, "Summarize retry best practices in 3 bullets."),
|
||||
max_attempts=3,
|
||||
initial_delay_s=0.25,
|
||||
max_delay_s=2.0,
|
||||
)
|
||||
except ServerBusyError as exc:
|
||||
print("Server overloaded after retries:", exc.message)
|
||||
print("Text:")
|
||||
return
|
||||
except JsonRpcError as exc:
|
||||
print(f"JSON-RPC error {exc.code}: {exc.message}")
|
||||
print("Text:")
|
||||
return
|
||||
|
||||
persisted = await thread.read(include_turns=True)
|
||||
persisted_turn = find_turn_by_id(persisted.thread.turns, result.id)
|
||||
if result.status == TurnStatus.failed:
|
||||
print("Turn failed:", result.error)
|
||||
|
||||
print("Text:", assistant_text_from_turn(persisted_turn))
|
||||
|
||||
|
||||
def _run_turn(thread, prompt: str):
|
||||
async def _inner():
|
||||
turn = await thread.turn(TextInput(prompt))
|
||||
return await turn.run()
|
||||
|
||||
return _inner
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,47 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import (
|
||||
assistant_text_from_turn,
|
||||
ensure_local_sdk_src,
|
||||
find_turn_by_id,
|
||||
runtime_config,
|
||||
)
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
from codex_app_server import (
|
||||
Codex,
|
||||
JsonRpcError,
|
||||
ServerBusyError,
|
||||
TextInput,
|
||||
TurnStatus,
|
||||
retry_on_overload,
|
||||
)
|
||||
|
||||
with Codex(config=runtime_config()) as codex:
|
||||
thread = codex.thread_start(model="gpt-5.4", config={"model_reasoning_effort": "high"})
|
||||
|
||||
try:
|
||||
result = retry_on_overload(
|
||||
lambda: thread.turn(TextInput("Summarize retry best practices in 3 bullets.")).run(),
|
||||
max_attempts=3,
|
||||
initial_delay_s=0.25,
|
||||
max_delay_s=2.0,
|
||||
)
|
||||
except ServerBusyError as exc:
|
||||
print("Server overloaded after retries:", exc.message)
|
||||
print("Text:")
|
||||
except JsonRpcError as exc:
|
||||
print(f"JSON-RPC error {exc.code}: {exc.message}")
|
||||
print("Text:")
|
||||
else:
|
||||
persisted = thread.read(include_turns=True)
|
||||
persisted_turn = find_turn_by_id(persisted.thread.turns, result.id)
|
||||
if result.status == TurnStatus.failed:
|
||||
print("Turn failed:", result.error)
|
||||
print("Text:", assistant_text_from_turn(persisted_turn))
|
||||
@@ -0,0 +1,96 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import ensure_local_sdk_src, runtime_config
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
import asyncio
|
||||
|
||||
from codex_app_server import (
|
||||
AsyncCodex,
|
||||
TextInput,
|
||||
ThreadTokenUsageUpdatedNotification,
|
||||
TurnCompletedNotification,
|
||||
)
|
||||
|
||||
|
||||
def _status_value(status: object | None) -> str:
|
||||
return str(getattr(status, "value", status))
|
||||
|
||||
|
||||
def _format_usage(usage: object | None) -> str:
|
||||
if usage is None:
|
||||
return "usage> (none)"
|
||||
|
||||
last = getattr(usage, "last", None)
|
||||
total = getattr(usage, "total", None)
|
||||
if last is None or total is None:
|
||||
return f"usage> {usage}"
|
||||
|
||||
return (
|
||||
"usage>\n"
|
||||
f" last: input={last.input_tokens} output={last.output_tokens} reasoning={last.reasoning_output_tokens} total={last.total_tokens} cached={last.cached_input_tokens}\n"
|
||||
f" total: input={total.input_tokens} output={total.output_tokens} reasoning={total.reasoning_output_tokens} total={total.total_tokens} cached={total.cached_input_tokens}"
|
||||
)
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
print("Codex async mini CLI. Type /exit to quit.")
|
||||
|
||||
async with AsyncCodex(config=runtime_config()) as codex:
|
||||
thread = await codex.thread_start(model="gpt-5.4", config={"model_reasoning_effort": "high"})
|
||||
print("Thread:", thread.id)
|
||||
|
||||
while True:
|
||||
try:
|
||||
user_input = (await asyncio.to_thread(input, "you> ")).strip()
|
||||
except EOFError:
|
||||
break
|
||||
|
||||
if not user_input:
|
||||
continue
|
||||
if user_input in {"/exit", "/quit"}:
|
||||
break
|
||||
|
||||
turn = await thread.turn(TextInput(user_input))
|
||||
usage = None
|
||||
status = None
|
||||
error = None
|
||||
printed_delta = False
|
||||
|
||||
print("assistant> ", end="", flush=True)
|
||||
async for event in turn.stream():
|
||||
payload = event.payload
|
||||
if event.method == "item/agentMessage/delta":
|
||||
delta = getattr(payload, "delta", "")
|
||||
if delta:
|
||||
print(delta, end="", flush=True)
|
||||
printed_delta = True
|
||||
continue
|
||||
if isinstance(payload, ThreadTokenUsageUpdatedNotification):
|
||||
usage = payload.token_usage
|
||||
continue
|
||||
if isinstance(payload, TurnCompletedNotification):
|
||||
status = payload.turn.status
|
||||
error = payload.turn.error
|
||||
|
||||
if printed_delta:
|
||||
print()
|
||||
else:
|
||||
print("[no text]")
|
||||
|
||||
status_text = _status_value(status)
|
||||
print(f"assistant.status> {status_text}")
|
||||
if status_text == "failed":
|
||||
print("assistant.error>", error)
|
||||
|
||||
print(_format_usage(usage))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,89 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import ensure_local_sdk_src, runtime_config
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
from codex_app_server import (
|
||||
Codex,
|
||||
TextInput,
|
||||
ThreadTokenUsageUpdatedNotification,
|
||||
TurnCompletedNotification,
|
||||
)
|
||||
|
||||
print("Codex mini CLI. Type /exit to quit.")
|
||||
|
||||
|
||||
def _status_value(status: object | None) -> str:
|
||||
return str(getattr(status, "value", status))
|
||||
|
||||
|
||||
def _format_usage(usage: object | None) -> str:
|
||||
if usage is None:
|
||||
return "usage> (none)"
|
||||
|
||||
last = getattr(usage, "last", None)
|
||||
total = getattr(usage, "total", None)
|
||||
if last is None or total is None:
|
||||
return f"usage> {usage}"
|
||||
|
||||
return (
|
||||
"usage>\n"
|
||||
f" last: input={last.input_tokens} output={last.output_tokens} reasoning={last.reasoning_output_tokens} total={last.total_tokens} cached={last.cached_input_tokens}\n"
|
||||
f" total: input={total.input_tokens} output={total.output_tokens} reasoning={total.reasoning_output_tokens} total={total.total_tokens} cached={total.cached_input_tokens}"
|
||||
)
|
||||
|
||||
|
||||
with Codex(config=runtime_config()) as codex:
|
||||
thread = codex.thread_start(model="gpt-5.4", config={"model_reasoning_effort": "high"})
|
||||
print("Thread:", thread.id)
|
||||
|
||||
while True:
|
||||
try:
|
||||
user_input = input("you> ").strip()
|
||||
except EOFError:
|
||||
break
|
||||
|
||||
if not user_input:
|
||||
continue
|
||||
if user_input in {"/exit", "/quit"}:
|
||||
break
|
||||
|
||||
turn = thread.turn(TextInput(user_input))
|
||||
usage = None
|
||||
status = None
|
||||
error = None
|
||||
printed_delta = False
|
||||
|
||||
print("assistant> ", end="", flush=True)
|
||||
for event in turn.stream():
|
||||
payload = event.payload
|
||||
if event.method == "item/agentMessage/delta":
|
||||
delta = getattr(payload, "delta", "")
|
||||
if delta:
|
||||
print(delta, end="", flush=True)
|
||||
printed_delta = True
|
||||
continue
|
||||
if isinstance(payload, ThreadTokenUsageUpdatedNotification):
|
||||
usage = payload.token_usage
|
||||
continue
|
||||
if isinstance(payload, TurnCompletedNotification):
|
||||
status = payload.turn.status
|
||||
error = payload.turn.error
|
||||
|
||||
if printed_delta:
|
||||
print()
|
||||
else:
|
||||
print("[no text]")
|
||||
|
||||
status_text = _status_value(status)
|
||||
print(f"assistant.status> {status_text}")
|
||||
if status_text == "failed":
|
||||
print("assistant.error>", error)
|
||||
|
||||
print(_format_usage(usage))
|
||||
@@ -0,0 +1,88 @@
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import (
|
||||
assistant_text_from_turn,
|
||||
ensure_local_sdk_src,
|
||||
find_turn_by_id,
|
||||
runtime_config,
|
||||
)
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
import asyncio
|
||||
|
||||
from codex_app_server import (
|
||||
AskForApproval,
|
||||
AsyncCodex,
|
||||
Personality,
|
||||
ReasoningSummary,
|
||||
TextInput,
|
||||
)
|
||||
|
||||
OUTPUT_SCHEMA = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"summary": {"type": "string"},
|
||||
"actions": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
},
|
||||
},
|
||||
"required": ["summary", "actions"],
|
||||
"additionalProperties": False,
|
||||
}
|
||||
|
||||
SUMMARY = ReasoningSummary.model_validate("concise")
|
||||
|
||||
PROMPT = (
|
||||
"Analyze a safe rollout plan for enabling a feature flag in production. "
|
||||
"Return JSON matching the requested schema."
|
||||
)
|
||||
APPROVAL_POLICY = AskForApproval.model_validate("never")
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async with AsyncCodex(config=runtime_config()) as codex:
|
||||
thread = await codex.thread_start(model="gpt-5.4", config={"model_reasoning_effort": "high"})
|
||||
|
||||
turn = await thread.turn(
|
||||
TextInput(PROMPT),
|
||||
approval_policy=APPROVAL_POLICY,
|
||||
output_schema=OUTPUT_SCHEMA,
|
||||
personality=Personality.pragmatic,
|
||||
summary=SUMMARY,
|
||||
)
|
||||
result = await turn.run()
|
||||
persisted = await thread.read(include_turns=True)
|
||||
persisted_turn = find_turn_by_id(persisted.thread.turns, result.id)
|
||||
structured_text = assistant_text_from_turn(persisted_turn).strip()
|
||||
try:
|
||||
structured = json.loads(structured_text)
|
||||
except json.JSONDecodeError as exc:
|
||||
raise RuntimeError(f"Expected JSON matching OUTPUT_SCHEMA, got: {structured_text!r}") from exc
|
||||
|
||||
summary = structured.get("summary")
|
||||
actions = structured.get("actions")
|
||||
if not isinstance(summary, str) or not isinstance(actions, list) or not all(
|
||||
isinstance(action, str) for action in actions
|
||||
):
|
||||
raise RuntimeError(
|
||||
f"Expected structured output with string summary/actions, got: {structured!r}"
|
||||
)
|
||||
|
||||
print("Status:", result.status)
|
||||
print("summary:", summary)
|
||||
print("actions:")
|
||||
for action in actions:
|
||||
print("-", action)
|
||||
print("Items:", 0 if persisted_turn is None else len(persisted_turn.items or []))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,78 @@
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import (
|
||||
assistant_text_from_turn,
|
||||
ensure_local_sdk_src,
|
||||
find_turn_by_id,
|
||||
runtime_config,
|
||||
)
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
from codex_app_server import (
|
||||
AskForApproval,
|
||||
Codex,
|
||||
Personality,
|
||||
ReasoningSummary,
|
||||
TextInput,
|
||||
)
|
||||
|
||||
OUTPUT_SCHEMA = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"summary": {"type": "string"},
|
||||
"actions": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
},
|
||||
},
|
||||
"required": ["summary", "actions"],
|
||||
"additionalProperties": False,
|
||||
}
|
||||
|
||||
SUMMARY = ReasoningSummary.model_validate("concise")
|
||||
|
||||
PROMPT = (
|
||||
"Analyze a safe rollout plan for enabling a feature flag in production. "
|
||||
"Return JSON matching the requested schema."
|
||||
)
|
||||
APPROVAL_POLICY = AskForApproval.model_validate("never")
|
||||
|
||||
with Codex(config=runtime_config()) as codex:
|
||||
thread = codex.thread_start(model="gpt-5.4", config={"model_reasoning_effort": "high"})
|
||||
|
||||
turn = thread.turn(
|
||||
TextInput(PROMPT),
|
||||
approval_policy=APPROVAL_POLICY,
|
||||
output_schema=OUTPUT_SCHEMA,
|
||||
personality=Personality.pragmatic,
|
||||
summary=SUMMARY,
|
||||
)
|
||||
result = turn.run()
|
||||
persisted = thread.read(include_turns=True)
|
||||
persisted_turn = find_turn_by_id(persisted.thread.turns, result.id)
|
||||
structured_text = assistant_text_from_turn(persisted_turn).strip()
|
||||
try:
|
||||
structured = json.loads(structured_text)
|
||||
except json.JSONDecodeError as exc:
|
||||
raise RuntimeError(f"Expected JSON matching OUTPUT_SCHEMA, got: {structured_text!r}") from exc
|
||||
|
||||
summary = structured.get("summary")
|
||||
actions = structured.get("actions")
|
||||
if not isinstance(summary, str) or not isinstance(actions, list) or not all(
|
||||
isinstance(action, str) for action in actions
|
||||
):
|
||||
raise RuntimeError(f"Expected structured output with string summary/actions, got: {structured!r}")
|
||||
|
||||
print("Status:", result.status)
|
||||
print("summary:", summary)
|
||||
print("actions:")
|
||||
for action in actions:
|
||||
print("-", action)
|
||||
print("Items:", 0 if persisted_turn is None else len(persisted_turn.items or []))
|
||||
@@ -0,0 +1,125 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import assistant_text_from_turn, ensure_local_sdk_src, find_turn_by_id, runtime_config
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
import asyncio
|
||||
|
||||
from codex_app_server import (
|
||||
AskForApproval,
|
||||
AsyncCodex,
|
||||
Personality,
|
||||
ReasoningEffort,
|
||||
ReasoningSummary,
|
||||
SandboxPolicy,
|
||||
TextInput,
|
||||
)
|
||||
|
||||
REASONING_RANK = {
|
||||
"none": 0,
|
||||
"minimal": 1,
|
||||
"low": 2,
|
||||
"medium": 3,
|
||||
"high": 4,
|
||||
"xhigh": 5,
|
||||
}
|
||||
PREFERRED_MODEL = "gpt-5.4"
|
||||
|
||||
|
||||
def _pick_highest_model(models):
|
||||
visible = [m for m in models if not m.hidden] or models
|
||||
preferred = next((m for m in visible if m.model == PREFERRED_MODEL or m.id == PREFERRED_MODEL), None)
|
||||
if preferred is not None:
|
||||
return preferred
|
||||
known_names = {m.id for m in visible} | {m.model for m in visible}
|
||||
top_candidates = [m for m in visible if not (m.upgrade and m.upgrade in known_names)]
|
||||
pool = top_candidates or visible
|
||||
return max(pool, key=lambda m: (m.model, m.id))
|
||||
|
||||
|
||||
def _pick_highest_turn_effort(model) -> ReasoningEffort:
|
||||
if not model.supported_reasoning_efforts:
|
||||
return ReasoningEffort.medium
|
||||
|
||||
best = max(
|
||||
model.supported_reasoning_efforts,
|
||||
key=lambda option: REASONING_RANK.get(option.reasoning_effort.value, -1),
|
||||
)
|
||||
return ReasoningEffort(best.reasoning_effort.value)
|
||||
|
||||
|
||||
OUTPUT_SCHEMA = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"summary": {"type": "string"},
|
||||
"actions": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
},
|
||||
},
|
||||
"required": ["summary", "actions"],
|
||||
"additionalProperties": False,
|
||||
}
|
||||
|
||||
SANDBOX_POLICY = SandboxPolicy.model_validate(
|
||||
{
|
||||
"type": "readOnly",
|
||||
"access": {"type": "fullAccess"},
|
||||
}
|
||||
)
|
||||
APPROVAL_POLICY = AskForApproval.model_validate("never")
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async with AsyncCodex(config=runtime_config()) as codex:
|
||||
models = await codex.models(include_hidden=True)
|
||||
selected_model = _pick_highest_model(models.data)
|
||||
selected_effort = _pick_highest_turn_effort(selected_model)
|
||||
|
||||
print("selected.model:", selected_model.model)
|
||||
print("selected.effort:", selected_effort.value)
|
||||
|
||||
thread = await codex.thread_start(
|
||||
model=selected_model.model,
|
||||
config={"model_reasoning_effort": selected_effort.value},
|
||||
)
|
||||
|
||||
first_turn = await thread.turn(
|
||||
TextInput("Give one short sentence about reliable production releases."),
|
||||
model=selected_model.model,
|
||||
effort=selected_effort,
|
||||
)
|
||||
first = await first_turn.run()
|
||||
persisted = await thread.read(include_turns=True)
|
||||
first_persisted_turn = find_turn_by_id(persisted.thread.turns, first.id)
|
||||
|
||||
print("agent.message:", assistant_text_from_turn(first_persisted_turn))
|
||||
print("items:", 0 if first_persisted_turn is None else len(first_persisted_turn.items or []))
|
||||
|
||||
second_turn = await thread.turn(
|
||||
TextInput("Return JSON for a safe feature-flag rollout plan."),
|
||||
approval_policy=APPROVAL_POLICY,
|
||||
cwd=str(Path.cwd()),
|
||||
effort=selected_effort,
|
||||
model=selected_model.model,
|
||||
output_schema=OUTPUT_SCHEMA,
|
||||
personality=Personality.pragmatic,
|
||||
sandbox_policy=SANDBOX_POLICY,
|
||||
summary=ReasoningSummary.model_validate("concise"),
|
||||
)
|
||||
second = await second_turn.run()
|
||||
persisted = await thread.read(include_turns=True)
|
||||
second_persisted_turn = find_turn_by_id(persisted.thread.turns, second.id)
|
||||
|
||||
print("agent.message.params:", assistant_text_from_turn(second_persisted_turn))
|
||||
print("items.params:", 0 if second_persisted_turn is None else len(second_persisted_turn.items or []))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,116 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import assistant_text_from_turn, ensure_local_sdk_src, find_turn_by_id, runtime_config
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
from codex_app_server import (
|
||||
AskForApproval,
|
||||
Codex,
|
||||
Personality,
|
||||
ReasoningEffort,
|
||||
ReasoningSummary,
|
||||
SandboxPolicy,
|
||||
TextInput,
|
||||
)
|
||||
|
||||
REASONING_RANK = {
|
||||
"none": 0,
|
||||
"minimal": 1,
|
||||
"low": 2,
|
||||
"medium": 3,
|
||||
"high": 4,
|
||||
"xhigh": 5,
|
||||
}
|
||||
PREFERRED_MODEL = "gpt-5.4"
|
||||
|
||||
|
||||
def _pick_highest_model(models):
|
||||
visible = [m for m in models if not m.hidden] or models
|
||||
preferred = next((m for m in visible if m.model == PREFERRED_MODEL or m.id == PREFERRED_MODEL), None)
|
||||
if preferred is not None:
|
||||
return preferred
|
||||
known_names = {m.id for m in visible} | {m.model for m in visible}
|
||||
top_candidates = [m for m in visible if not (m.upgrade and m.upgrade in known_names)]
|
||||
pool = top_candidates or visible
|
||||
return max(pool, key=lambda m: (m.model, m.id))
|
||||
|
||||
|
||||
def _pick_highest_turn_effort(model) -> ReasoningEffort:
|
||||
if not model.supported_reasoning_efforts:
|
||||
return ReasoningEffort.medium
|
||||
|
||||
best = max(
|
||||
model.supported_reasoning_efforts,
|
||||
key=lambda option: REASONING_RANK.get(option.reasoning_effort.value, -1),
|
||||
)
|
||||
return ReasoningEffort(best.reasoning_effort.value)
|
||||
|
||||
|
||||
OUTPUT_SCHEMA = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"summary": {"type": "string"},
|
||||
"actions": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
},
|
||||
},
|
||||
"required": ["summary", "actions"],
|
||||
"additionalProperties": False,
|
||||
}
|
||||
|
||||
SANDBOX_POLICY = SandboxPolicy.model_validate(
|
||||
{
|
||||
"type": "readOnly",
|
||||
"access": {"type": "fullAccess"},
|
||||
}
|
||||
)
|
||||
APPROVAL_POLICY = AskForApproval.model_validate("never")
|
||||
|
||||
|
||||
with Codex(config=runtime_config()) as codex:
|
||||
models = codex.models(include_hidden=True)
|
||||
selected_model = _pick_highest_model(models.data)
|
||||
selected_effort = _pick_highest_turn_effort(selected_model)
|
||||
|
||||
print("selected.model:", selected_model.model)
|
||||
print("selected.effort:", selected_effort.value)
|
||||
|
||||
thread = codex.thread_start(
|
||||
model=selected_model.model,
|
||||
config={"model_reasoning_effort": selected_effort.value},
|
||||
)
|
||||
|
||||
first = thread.turn(
|
||||
TextInput("Give one short sentence about reliable production releases."),
|
||||
model=selected_model.model,
|
||||
effort=selected_effort,
|
||||
).run()
|
||||
persisted = thread.read(include_turns=True)
|
||||
first_turn = find_turn_by_id(persisted.thread.turns, first.id)
|
||||
|
||||
print("agent.message:", assistant_text_from_turn(first_turn))
|
||||
print("items:", 0 if first_turn is None else len(first_turn.items or []))
|
||||
|
||||
second = thread.turn(
|
||||
TextInput("Return JSON for a safe feature-flag rollout plan."),
|
||||
approval_policy=APPROVAL_POLICY,
|
||||
cwd=str(Path.cwd()),
|
||||
effort=selected_effort,
|
||||
model=selected_model.model,
|
||||
output_schema=OUTPUT_SCHEMA,
|
||||
personality=Personality.pragmatic,
|
||||
sandbox_policy=SANDBOX_POLICY,
|
||||
summary=ReasoningSummary.model_validate("concise"),
|
||||
).run()
|
||||
persisted = thread.read(include_turns=True)
|
||||
second_turn = find_turn_by_id(persisted.thread.turns, second.id)
|
||||
|
||||
print("agent.message.params:", assistant_text_from_turn(second_turn))
|
||||
print("items.params:", 0 if second_turn is None else len(second_turn.items or []))
|
||||
@@ -0,0 +1,71 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import (
|
||||
assistant_text_from_turn,
|
||||
ensure_local_sdk_src,
|
||||
runtime_config,
|
||||
)
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
import asyncio
|
||||
|
||||
from codex_app_server import AsyncCodex, TextInput
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async with AsyncCodex(config=runtime_config()) as codex:
|
||||
thread = await codex.thread_start(model="gpt-5.4", config={"model_reasoning_effort": "high"})
|
||||
steer_turn = await thread.turn(TextInput("Count from 1 to 40 with commas, then one summary sentence."))
|
||||
steer_result = "sent"
|
||||
try:
|
||||
_ = await steer_turn.steer(TextInput("Keep it brief and stop after 10 numbers."))
|
||||
except Exception as exc:
|
||||
steer_result = f"skipped {type(exc).__name__}"
|
||||
|
||||
steer_event_count = 0
|
||||
steer_completed_status = "unknown"
|
||||
steer_completed_turn = None
|
||||
async for event in steer_turn.stream():
|
||||
steer_event_count += 1
|
||||
if event.method == "turn/completed":
|
||||
steer_completed_turn = event.payload.turn
|
||||
steer_completed_status = getattr(event.payload.turn.status, "value", str(event.payload.turn.status))
|
||||
|
||||
steer_preview = assistant_text_from_turn(steer_completed_turn).strip() or "[no assistant text]"
|
||||
|
||||
interrupt_turn = await thread.turn(TextInput("Count from 1 to 200 with commas, then one summary sentence."))
|
||||
interrupt_result = "sent"
|
||||
try:
|
||||
_ = await interrupt_turn.interrupt()
|
||||
except Exception as exc:
|
||||
interrupt_result = f"skipped {type(exc).__name__}"
|
||||
|
||||
interrupt_event_count = 0
|
||||
interrupt_completed_status = "unknown"
|
||||
interrupt_completed_turn = None
|
||||
async for event in interrupt_turn.stream():
|
||||
interrupt_event_count += 1
|
||||
if event.method == "turn/completed":
|
||||
interrupt_completed_turn = event.payload.turn
|
||||
interrupt_completed_status = getattr(event.payload.turn.status, "value", str(event.payload.turn.status))
|
||||
|
||||
interrupt_preview = assistant_text_from_turn(interrupt_completed_turn).strip() or "[no assistant text]"
|
||||
|
||||
print("steer.result:", steer_result)
|
||||
print("steer.final.status:", steer_completed_status)
|
||||
print("steer.events.count:", steer_event_count)
|
||||
print("steer.assistant.preview:", steer_preview)
|
||||
print("interrupt.result:", interrupt_result)
|
||||
print("interrupt.final.status:", interrupt_completed_status)
|
||||
print("interrupt.events.count:", interrupt_event_count)
|
||||
print("interrupt.assistant.preview:", interrupt_preview)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,63 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_EXAMPLES_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_EXAMPLES_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_EXAMPLES_ROOT))
|
||||
|
||||
from _bootstrap import (
|
||||
assistant_text_from_turn,
|
||||
ensure_local_sdk_src,
|
||||
runtime_config,
|
||||
)
|
||||
|
||||
ensure_local_sdk_src()
|
||||
|
||||
from codex_app_server import Codex, TextInput
|
||||
|
||||
with Codex(config=runtime_config()) as codex:
|
||||
thread = codex.thread_start(model="gpt-5.4", config={"model_reasoning_effort": "high"})
|
||||
steer_turn = thread.turn(TextInput("Count from 1 to 40 with commas, then one summary sentence."))
|
||||
steer_result = "sent"
|
||||
try:
|
||||
_ = steer_turn.steer(TextInput("Keep it brief and stop after 10 numbers."))
|
||||
except Exception as exc:
|
||||
steer_result = f"skipped {type(exc).__name__}"
|
||||
|
||||
steer_event_count = 0
|
||||
steer_completed_status = "unknown"
|
||||
steer_completed_turn = None
|
||||
for event in steer_turn.stream():
|
||||
steer_event_count += 1
|
||||
if event.method == "turn/completed":
|
||||
steer_completed_turn = event.payload.turn
|
||||
steer_completed_status = getattr(event.payload.turn.status, "value", str(event.payload.turn.status))
|
||||
|
||||
steer_preview = assistant_text_from_turn(steer_completed_turn).strip() or "[no assistant text]"
|
||||
|
||||
interrupt_turn = thread.turn(TextInput("Count from 1 to 200 with commas, then one summary sentence."))
|
||||
interrupt_result = "sent"
|
||||
try:
|
||||
_ = interrupt_turn.interrupt()
|
||||
except Exception as exc:
|
||||
interrupt_result = f"skipped {type(exc).__name__}"
|
||||
|
||||
interrupt_event_count = 0
|
||||
interrupt_completed_status = "unknown"
|
||||
interrupt_completed_turn = None
|
||||
for event in interrupt_turn.stream():
|
||||
interrupt_event_count += 1
|
||||
if event.method == "turn/completed":
|
||||
interrupt_completed_turn = event.payload.turn
|
||||
interrupt_completed_status = getattr(event.payload.turn.status, "value", str(event.payload.turn.status))
|
||||
|
||||
interrupt_preview = assistant_text_from_turn(interrupt_completed_turn).strip() or "[no assistant text]"
|
||||
|
||||
print("steer.result:", steer_result)
|
||||
print("steer.final.status:", steer_completed_status)
|
||||
print("steer.events.count:", steer_event_count)
|
||||
print("steer.assistant.preview:", steer_preview)
|
||||
print("interrupt.result:", interrupt_result)
|
||||
print("interrupt.final.status:", interrupt_completed_status)
|
||||
print("interrupt.events.count:", interrupt_event_count)
|
||||
print("interrupt.assistant.preview:", interrupt_preview)
|
||||
@@ -0,0 +1,85 @@
|
||||
# Python SDK Examples
|
||||
|
||||
Each example folder contains runnable versions:
|
||||
|
||||
- `sync.py` (public sync surface: `Codex`)
|
||||
- `async.py` (public async surface: `AsyncCodex`)
|
||||
|
||||
All examples intentionally use only public SDK exports from `codex_app_server`.
|
||||
|
||||
## Prerequisites
|
||||
|
||||
- Python `>=3.10`
|
||||
- Install SDK dependencies for the same Python interpreter you will use to run examples
|
||||
|
||||
Recommended setup (from `sdk/python`):
|
||||
|
||||
```bash
|
||||
python -m venv .venv
|
||||
source .venv/bin/activate
|
||||
python -m pip install -U pip
|
||||
python -m pip install -e .
|
||||
```
|
||||
|
||||
When running examples from this repo checkout, the SDK source uses the local
|
||||
tree and does not bundle a runtime binary. The helper in `examples/_bootstrap.py`
|
||||
uses the installed `codex-cli-bin` runtime package.
|
||||
|
||||
If the pinned `codex-cli-bin` runtime is not already installed, the bootstrap
|
||||
will download the matching GitHub release artifact, stage a temporary local
|
||||
`codex-cli-bin` package, install it into your active interpreter, and clean up
|
||||
the temporary files afterward.
|
||||
|
||||
Current pinned runtime version: `0.116.0-alpha.1`
|
||||
|
||||
## Run examples
|
||||
|
||||
From `sdk/python`:
|
||||
|
||||
```bash
|
||||
python examples/<example-folder>/sync.py
|
||||
python examples/<example-folder>/async.py
|
||||
```
|
||||
|
||||
The examples bootstrap local imports from `sdk/python/src` automatically, so no
|
||||
SDK wheel install is required. You only need the Python dependencies for your
|
||||
active interpreter and an installed `codex-cli-bin` runtime package (either
|
||||
already present or automatically provisioned by the bootstrap).
|
||||
|
||||
## Recommended first run
|
||||
|
||||
```bash
|
||||
python examples/01_quickstart_constructor/sync.py
|
||||
python examples/01_quickstart_constructor/async.py
|
||||
```
|
||||
|
||||
## Index
|
||||
|
||||
- `01_quickstart_constructor/`
|
||||
- first run / sanity check
|
||||
- `02_turn_run/`
|
||||
- inspect full turn output fields
|
||||
- `03_turn_stream_events/`
|
||||
- stream a turn with a small curated event view
|
||||
- `04_models_and_metadata/`
|
||||
- discover visible models for the connected runtime
|
||||
- `05_existing_thread/`
|
||||
- resume a real existing thread (created in-script)
|
||||
- `06_thread_lifecycle_and_controls/`
|
||||
- thread lifecycle + control calls
|
||||
- `07_image_and_text/`
|
||||
- remote image URL + text multimodal turn
|
||||
- `08_local_image_and_text/`
|
||||
- local image + text multimodal turn using a generated temporary sample image
|
||||
- `09_async_parity/`
|
||||
- parity-style sync flow (see async parity in other examples)
|
||||
- `10_error_handling_and_retry/`
|
||||
- overload retry pattern + typed error handling structure
|
||||
- `11_cli_mini_app/`
|
||||
- interactive chat loop
|
||||
- `12_turn_params_kitchen_sink/`
|
||||
- structured output with a curated advanced `turn(...)` configuration
|
||||
- `13_model_select_and_turn_params/`
|
||||
- list models, pick highest model + highest supported reasoning effort, run turns, print message and usage
|
||||
- `14_turn_controls/`
|
||||
- separate best-effort `steer()` and `interrupt()` demos with concise summaries
|
||||
@@ -0,0 +1,152 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import importlib.util
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import zlib
|
||||
from pathlib import Path
|
||||
from typing import Iterable, Iterator
|
||||
|
||||
_SDK_PYTHON_DIR = Path(__file__).resolve().parents[1]
|
||||
_SDK_PYTHON_STR = str(_SDK_PYTHON_DIR)
|
||||
if _SDK_PYTHON_STR not in sys.path:
|
||||
sys.path.insert(0, _SDK_PYTHON_STR)
|
||||
|
||||
from _runtime_setup import ensure_runtime_package_installed
|
||||
|
||||
|
||||
def _ensure_runtime_dependencies(sdk_python_dir: Path) -> None:
|
||||
if importlib.util.find_spec("pydantic") is not None:
|
||||
return
|
||||
|
||||
python = sys.executable
|
||||
raise RuntimeError(
|
||||
"Missing required dependency: pydantic.\n"
|
||||
f"Interpreter: {python}\n"
|
||||
"Install dependencies with the same interpreter used to run this example:\n"
|
||||
f" {python} -m pip install -e {sdk_python_dir}\n"
|
||||
"If you installed with `pip` from another Python, reinstall using the command above."
|
||||
)
|
||||
|
||||
|
||||
def ensure_local_sdk_src() -> Path:
|
||||
"""Add sdk/python/src to sys.path so examples run without installing the package."""
|
||||
sdk_python_dir = _SDK_PYTHON_DIR
|
||||
src_dir = sdk_python_dir / "src"
|
||||
package_dir = src_dir / "codex_app_server"
|
||||
if not package_dir.exists():
|
||||
raise RuntimeError(f"Could not locate local SDK package at {package_dir}")
|
||||
|
||||
_ensure_runtime_dependencies(sdk_python_dir)
|
||||
|
||||
src_str = str(src_dir)
|
||||
if src_str not in sys.path:
|
||||
sys.path.insert(0, src_str)
|
||||
return src_dir
|
||||
|
||||
|
||||
def runtime_config():
|
||||
"""Return an example-friendly AppServerConfig for repo-source SDK usage."""
|
||||
from codex_app_server import AppServerConfig
|
||||
|
||||
ensure_runtime_package_installed(sys.executable, _SDK_PYTHON_DIR)
|
||||
return AppServerConfig()
|
||||
|
||||
|
||||
def _png_chunk(chunk_type: bytes, data: bytes) -> bytes:
|
||||
import struct
|
||||
|
||||
payload = chunk_type + data
|
||||
checksum = zlib.crc32(payload) & 0xFFFFFFFF
|
||||
return struct.pack(">I", len(data)) + payload + struct.pack(">I", checksum)
|
||||
|
||||
|
||||
def _generated_sample_png_bytes() -> bytes:
|
||||
import struct
|
||||
|
||||
width = 96
|
||||
height = 96
|
||||
top_left = (120, 180, 255)
|
||||
top_right = (255, 220, 90)
|
||||
bottom_left = (90, 180, 95)
|
||||
bottom_right = (180, 85, 85)
|
||||
|
||||
rows = bytearray()
|
||||
for y in range(height):
|
||||
rows.append(0)
|
||||
for x in range(width):
|
||||
if y < height // 2 and x < width // 2:
|
||||
color = top_left
|
||||
elif y < height // 2:
|
||||
color = top_right
|
||||
elif x < width // 2:
|
||||
color = bottom_left
|
||||
else:
|
||||
color = bottom_right
|
||||
rows.extend(color)
|
||||
|
||||
header = struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0)
|
||||
return (
|
||||
b"\x89PNG\r\n\x1a\n"
|
||||
+ _png_chunk(b"IHDR", header)
|
||||
+ _png_chunk(b"IDAT", zlib.compress(bytes(rows)))
|
||||
+ _png_chunk(b"IEND", b"")
|
||||
)
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def temporary_sample_image_path() -> Iterator[Path]:
|
||||
with tempfile.TemporaryDirectory(prefix="codex-python-example-image-") as temp_root:
|
||||
image_path = Path(temp_root) / "generated_sample.png"
|
||||
image_path.write_bytes(_generated_sample_png_bytes())
|
||||
yield image_path
|
||||
|
||||
|
||||
def server_label(metadata: object) -> str:
|
||||
server = getattr(metadata, "serverInfo", None)
|
||||
server_name = ((getattr(server, "name", None) or "") if server is not None else "").strip()
|
||||
server_version = ((getattr(server, "version", None) or "") if server is not None else "").strip()
|
||||
if server_name and server_version:
|
||||
return f"{server_name} {server_version}"
|
||||
|
||||
user_agent = ((getattr(metadata, "userAgent", None) or "") if metadata is not None else "").strip()
|
||||
return user_agent or "unknown"
|
||||
|
||||
|
||||
def find_turn_by_id(turns: Iterable[object] | None, turn_id: str) -> object | None:
|
||||
for turn in turns or []:
|
||||
if getattr(turn, "id", None) == turn_id:
|
||||
return turn
|
||||
return None
|
||||
|
||||
|
||||
def assistant_text_from_turn(turn: object | None) -> str:
|
||||
if turn is None:
|
||||
return ""
|
||||
|
||||
chunks: list[str] = []
|
||||
for item in getattr(turn, "items", []) or []:
|
||||
raw_item = item.model_dump(mode="json") if hasattr(item, "model_dump") else item
|
||||
if not isinstance(raw_item, dict):
|
||||
continue
|
||||
|
||||
item_type = raw_item.get("type")
|
||||
if item_type == "agentMessage":
|
||||
text = raw_item.get("text")
|
||||
if isinstance(text, str) and text:
|
||||
chunks.append(text)
|
||||
continue
|
||||
|
||||
if item_type != "message" or raw_item.get("role") != "assistant":
|
||||
continue
|
||||
|
||||
for content in raw_item.get("content") or []:
|
||||
if not isinstance(content, dict) or content.get("type") != "output_text":
|
||||
continue
|
||||
text = content.get("text")
|
||||
if isinstance(text, str) and text:
|
||||
chunks.append(text)
|
||||
|
||||
return "".join(chunks)
|
||||
Reference in New Issue
Block a user