Add Python SDK public API and examples

Co-authored-by: Codex <noreply@openai.com>

sdk/python/examples/01_quickstart_constructor/async.py

@@ -0,0 +1,30 @@
+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:
+        print("Server:", codex.metadata.server_name, codex.metadata.server_version)
+
+        thread = await codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"})
+        turn = await thread.turn(TextInput("Say hello in one sentence."))
+        result = await turn.run()
+
+        print("Status:", result.status)
+        print("Text:", result.text)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

sdk/python/examples/01_quickstart_constructor/sync.py

@@ -0,0 +1,20 @@
+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:
+    print("Server:", codex.metadata.server_name, codex.metadata.server_version)
+
+    thread = codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"})
+    result = thread.turn(TextInput("Say hello in one sentence.")).run()
+    print("Status:", result.status)
+    print("Text:", result.text)

sdk/python/examples/02_turn_run/async.py

@@ -0,0 +1,37 @@
+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", config={"model_reasoning_effort": "high"})
+        turn = await thread.turn(TextInput("Give 3 bullets about SIMD."))
+        result = await turn.run()
+
+        print("thread_id:", result.thread_id)
+        print("turn_id:", result.turn_id)
+        print("status:", result.status)
+        if result.error is not None:
+            print("error:", result.error)
+        print("text:", result.text)
+        print("items.count:", len(result.items))
+        if result.usage is None:
+            raise RuntimeError("missing usage for completed turn")
+        print("usage.thread_id:", result.usage.thread_id)
+        print("usage.turn_id:", result.usage.turn_id)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

sdk/python/examples/02_turn_run/sync.py

@@ -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 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", config={"model_reasoning_effort": "high"})
+    result = thread.turn(TextInput("Give 3 bullets about SIMD.")).run()
+
+    print("thread_id:", result.thread_id)
+    print("turn_id:", result.turn_id)
+    print("status:", result.status)
+    if result.error is not None:
+        print("error:", result.error)
+    print("text:", result.text)
+    print("items.count:", len(result.items))
+    if result.usage is None:
+        raise RuntimeError("missing usage for completed turn")
+    print("usage.thread_id:", result.usage.thread_id)
+    print("usage.turn_id:", result.usage.turn_id)

sdk/python/examples/03_turn_stream_events/async.py

@@ -0,0 +1,44 @@
+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", config={"model_reasoning_effort": "high"})
+        turn = await thread.turn(TextInput("Count from 1 to 200 with commas, then one summary sentence."))
+
+        # Best effort controls: models can finish quickly, so races are expected.
+        try:
+            _ = await turn.steer(TextInput("Keep it brief and stop after 20 numbers."))
+            print("steer: sent")
+        except Exception as exc:
+            print("steer: skipped", type(exc).__name__)
+
+        try:
+            _ = await turn.interrupt()
+            print("interrupt: sent")
+        except Exception as exc:
+            print("interrupt: skipped", type(exc).__name__)
+
+        event_count = 0
+        async for event in turn.stream():
+            event_count += 1
+            print(event.method, event.payload)
+
+        print("events.count:", event_count)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

sdk/python/examples/03_turn_stream_events/sync.py

@@ -0,0 +1,36 @@
+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", config={"model_reasoning_effort": "high"})
+    turn = thread.turn(TextInput("Count from 1 to 200 with commas, then one summary sentence."))
+
+    # Best effort controls: models can finish quickly, so races are expected.
+    try:
+        _ = turn.steer(TextInput("Keep it brief and stop after 20 numbers."))
+        print("steer: sent")
+    except Exception as exc:
+        print("steer: skipped", type(exc).__name__)
+
+    try:
+        _ = turn.interrupt()
+        print("interrupt: sent")
+    except Exception as exc:
+        print("interrupt: skipped", type(exc).__name__)
+
+    event_count = 0
+    for event in turn.stream():
+        event_count += 1
+        print(event.method, event.payload)
+
+    print("events.count:", event_count)

sdk/python/examples/04_models_and_metadata/async.py

@@ -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 ensure_local_sdk_src, runtime_config
+
+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("metadata:", codex.metadata)
+
+        models = await codex.models(include_hidden=True)
+        print("models.count:", len(models.data))
+        if models.data:
+            print("first model id:", models.data[0].id)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

sdk/python/examples/04_models_and_metadata/sync.py

@@ -0,0 +1,20 @@
+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
+
+with Codex(config=runtime_config()) as codex:
+    print("metadata:", codex.metadata)
+
+    models = codex.models()
+    print("models.count:", len(models.data))
+    if models.data:
+        print("first model id:", models.data[0].id)

sdk/python/examples/05_existing_thread/async.py

@@ -0,0 +1,32 @@
+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:
+        original = await codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"})
+
+        first_turn = await original.turn(TextInput("Tell me one fact about Saturn."))
+        first = await first_turn.run()
+        print("Created thread:", first.thread_id)
+
+        resumed = await codex.thread_resume(first.thread_id)
+        second_turn = await resumed.turn(TextInput("Continue with one more fact."))
+        second = await second_turn.run()
+        print(second.text)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

sdk/python/examples/05_existing_thread/sync.py

@@ -0,0 +1,23 @@
+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:
+    # Create an initial thread and turn so we have a real thread to resume.
+    original = codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"})
+    first = original.turn(TextInput("Tell me one fact about Saturn.")).run()
+    print("Created thread:", first.thread_id)
+
+    # Resume the existing thread by ID.
+    resumed = codex.thread_resume(first.thread_id)
+    second = resumed.turn(TextInput("Continue with one more fact.")).run()
+    print(second.text)

sdk/python/examples/06_thread_lifecycle_and_controls/async.py

@@ -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", 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",
+                config={"model_reasoning_effort": "high"},
+            )
+            resumed_result = await (await resumed.turn(TextInput("Continue in one short sentence."))).run()
+            resumed_info = f"{resumed_result.turn_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")
+            forked_result = await (await forked.turn(TextInput("Take a different angle in one short sentence."))).run()
+            forked_info = f"{forked_result.turn_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.turn_id, first.status)
+        print("second:", second.turn_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())

sdk/python/examples/06_thread_lifecycle_and_controls/sync.py

@@ -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", 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",
+            config={"model_reasoning_effort": "high"},
+        )
+        resumed_result = resumed.turn(TextInput("Continue in one short sentence.")).run()
+        resumed_info = f"{resumed_result.turn_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")
+        forked_result = forked.turn(TextInput("Take a different angle in one short sentence.")).run()
+        forked_info = f"{forked_result.turn_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.turn_id, first.status)
+    print("second:", second.turn_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)

sdk/python/examples/07_image_and_text/async.py

@@ -0,0 +1,35 @@
+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, 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", 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()
+
+        print("Status:", result.status)
+        print(result.text)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

sdk/python/examples/07_image_and_text/sync.py

@@ -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
+
+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", config={"model_reasoning_effort": "high"})
+    result = thread.turn(
+        [
+            TextInput("What is in this image? Give 3 bullets."),
+            ImageInput(REMOTE_IMAGE_URL),
+        ]
+    ).run()
+
+    print("Status:", result.status)
+    print(result.text)

sdk/python/examples/08_local_image_and_text/async.py

@@ -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 ensure_local_sdk_src, runtime_config
+
+ensure_local_sdk_src()
+
+import asyncio
+
+from codex_app_server import AsyncCodex, LocalImageInput, TextInput
+
+IMAGE_PATH = Path(__file__).resolve().parents[1] / "assets" / "sample_scene.png"
+if not IMAGE_PATH.exists():
+    raise FileNotFoundError(f"Missing bundled image: {IMAGE_PATH}")
+
+
+async def main() -> None:
+    async with AsyncCodex(config=runtime_config()) as codex:
+        thread = await codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"})
+
+        turn = await thread.turn(
+            [
+                TextInput("Read this local image and summarize what you see in 2 bullets."),
+                LocalImageInput(str(IMAGE_PATH.resolve())),
+            ]
+        )
+        result = await turn.run()
+
+        print("Status:", result.status)
+        print(result.text)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

sdk/python/examples/08_local_image_and_text/sync.py

@@ -0,0 +1,29 @@
+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, LocalImageInput, TextInput
+
+IMAGE_PATH = Path(__file__).resolve().parents[1] / "assets" / "sample_scene.png"
+if not IMAGE_PATH.exists():
+    raise FileNotFoundError(f"Missing bundled image: {IMAGE_PATH}")
+
+with Codex(config=runtime_config()) as codex:
+    thread = codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"})
+
+    result = thread.turn(
+        [
+            TextInput("Read this local image and summarize what you see in 2 bullets."),
+            LocalImageInput(str(IMAGE_PATH.resolve())),
+        ]
+    ).run()
+
+    print("Status:", result.status)
+    print(result.text)

sdk/python/examples/09_async_parity/sync.py

@@ -0,0 +1,23 @@
+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:
+    print("Server:", codex.metadata.server_name, codex.metadata.server_version)
+
+    thread = codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"})
+    turn = thread.turn(TextInput("Say hello in one sentence."))
+    result = turn.run()
+
+    print("Thread:", result.thread_id)
+    print("Turn:", result.turn_id)
+    print("Text:", result.text.strip())

sdk/python/examples/10_error_handling_and_retry/async.py

@@ -0,0 +1,91 @@
+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
+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", 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
+
+        if result.status == TurnStatus.failed:
+            print("Turn failed:", result.error)
+
+        print("Text:", result.text)
+
+
+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())

sdk/python/examples/10_error_handling_and_retry/sync.py

@@ -0,0 +1,40 @@
+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,
+    JsonRpcError,
+    ServerBusyError,
+    TextInput,
+    TurnStatus,
+    retry_on_overload,
+)
+
+with Codex(config=runtime_config()) as codex:
+    thread = codex.thread_start(model="gpt-5", 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:
+        if result.status == TurnStatus.failed:
+            print("Turn failed:", result.error)
+        print("Text:", result.text)

sdk/python/examples/11_cli_mini_app/async.py

@@ -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,
+    TurnCompletedNotificationPayload,
+)
+
+
+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.inputTokens} output={last.outputTokens} reasoning={last.reasoningOutputTokens} total={last.totalTokens} cached={last.cachedInputTokens}\n"
+        f"  total: input={total.inputTokens} output={total.outputTokens} reasoning={total.reasoningOutputTokens} total={total.totalTokens} cached={total.cachedInputTokens}"
+    )
+
+
+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", 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, TurnCompletedNotificationPayload):
+                    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())

sdk/python/examples/11_cli_mini_app/sync.py

@@ -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,
+    TurnCompletedNotificationPayload,
+)
+
+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.inputTokens} output={last.outputTokens} reasoning={last.reasoningOutputTokens} total={last.totalTokens} cached={last.cachedInputTokens}\n"
+        f"  total: input={total.inputTokens} output={total.outputTokens} reasoning={total.reasoningOutputTokens} total={total.totalTokens} cached={total.cachedInputTokens}"
+    )
+
+
+with Codex(config=runtime_config()) as codex:
+    thread = codex.thread_start(model="gpt-5", 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, TurnCompletedNotificationPayload):
+                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))

sdk/python/examples/12_turn_params_kitchen_sink/async.py

@@ -0,0 +1,75 @@
+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 (
+    AskForApproval,
+    AsyncCodex,
+    Personality,
+    ReasoningEffort,
+    ReasoningSummary,
+    SandboxPolicy,
+    TextInput,
+)
+
+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"},
+    }
+)
+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", config={"model_reasoning_effort": "high"})
+
+        turn = await thread.turn(
+            TextInput(PROMPT),
+            approval_policy=APPROVAL_POLICY,
+            cwd=str(Path.cwd()),
+            effort=ReasoningEffort.medium,
+            model="gpt-5",
+            output_schema=OUTPUT_SCHEMA,
+            personality=Personality.pragmatic,
+            sandbox_policy=SANDBOX_POLICY,
+            summary=SUMMARY,
+        )
+        result = await turn.run()
+
+        print("Status:", result.status)
+        print("Text:", result.text)
+        print("Usage:", result.usage)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

sdk/python/examples/12_turn_params_kitchen_sink/sync.py

@@ -0,0 +1,67 @@
+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 (
+    AskForApproval,
+    Codex,
+    Personality,
+    ReasoningEffort,
+    ReasoningSummary,
+    SandboxPolicy,
+    TextInput,
+)
+
+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"},
+    }
+)
+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", config={"model_reasoning_effort": "high"})
+
+    turn = thread.turn(
+        TextInput(PROMPT),
+        approval_policy=APPROVAL_POLICY,
+        cwd=str(Path.cwd()),
+        effort=ReasoningEffort.medium,
+        model="gpt-5",
+        output_schema=OUTPUT_SCHEMA,
+        personality=Personality.pragmatic,
+        sandbox_policy=SANDBOX_POLICY,
+        summary=SUMMARY,
+    )
+    result = turn.run()
+
+    print("Status:", result.status)
+    print("Text:", result.text)
+    print("Usage:", result.usage)

sdk/python/examples/13_model_select_and_turn_params/async.py

@@ -0,0 +1,117 @@
+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 (
+    AskForApproval,
+    AsyncCodex,
+    Personality,
+    ReasoningEffort,
+    ReasoningSummary,
+    SandboxPolicy,
+    TextInput,
+)
+
+REASONING_RANK = {
+    "none": 0,
+    "minimal": 1,
+    "low": 2,
+    "medium": 3,
+    "high": 4,
+    "xhigh": 5,
+}
+
+
+def _pick_highest_model(models):
+    visible = [m for m in models if not m.hidden] or models
+    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()
+
+        print("agent.message:", first.text)
+        print("usage:", first.usage)
+
+        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()
+
+        print("agent.message.params:", second.text)
+        print("usage.params:", second.usage)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

sdk/python/examples/13_model_select_and_turn_params/sync.py

@@ -0,0 +1,108 @@
+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 (
+    AskForApproval,
+    Codex,
+    Personality,
+    ReasoningEffort,
+    ReasoningSummary,
+    SandboxPolicy,
+    TextInput,
+)
+
+REASONING_RANK = {
+    "none": 0,
+    "minimal": 1,
+    "low": 2,
+    "medium": 3,
+    "high": 4,
+    "xhigh": 5,
+}
+
+
+def _pick_highest_model(models):
+    visible = [m for m in models if not m.hidden] or models
+    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()
+
+    print("agent.message:", first.text)
+    print("usage:", first.usage)
+
+    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()
+
+    print("agent.message.params:", second.text)
+    print("usage.params:", second.usage)

sdk/python/examples/README.md

@@ -0,0 +1,78 @@
+# 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`
+therefore resolves `codex` from:
+
+- `CODEX_PYTHON_SDK_CODEX_BIN`, if set
+- otherwise `codex` on `PATH`
+
+## 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
+extra package build step is required beyond installing dependencies for your
+active interpreter and making a `codex` binary available.
+
+## 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 and print raw notifications
+- `04_models_and_metadata/`
+  - read server metadata and model list
+- `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 bundled 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/`
+  - one turn using most optional `turn(...)` params (sync + async)
+- `13_model_select_and_turn_params/`
+  - list models, pick highest model + highest supported reasoning effort, run turns, print message and usage

sdk/python/examples/_bootstrap.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+import importlib.util
+import os
+import shutil
+import sys
+from pathlib import Path
+
+
+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 = Path(__file__).resolve().parents[1]
+    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 local repo usage."""
+    from codex_app_server import AppServerConfig
+
+    codex_bin = os.environ.get("CODEX_PYTHON_SDK_CODEX_BIN") or shutil.which("codex")
+    if codex_bin is None:
+        raise RuntimeError(
+            "Examples require a Codex CLI binary when run from this repo checkout.\n"
+            "Set CODEX_PYTHON_SDK_CODEX_BIN=/absolute/path/to/codex, or ensure `codex` is on PATH."
+        )
+    return AppServerConfig(codex_bin=codex_bin)