codex/sdk/python/notebooks/sdk_walkthrough.ipynb

Codex Python SDK Walkthrough

Public SDK surface only (openai_codex root exports).

In [ ]:

# Cell 1: bootstrap local SDK imports + pinned runtime package
import os
import sys
from pathlib import Path

if sys.version_info < (3, 10):
    raise RuntimeError(
        f'Notebook requires Python 3.10+; current interpreter is {sys.version.split()[0]}.'
    )

def _is_sdk_python_dir(path: Path) -> bool:
    return (path / 'pyproject.toml').exists() and (path / 'src' / 'openai_codex').exists()


def _find_sdk_python_dir(start: Path) -> Path | None:
    checked = set()

    def _consider(candidate: Path) -> Path | None:
        resolved = candidate.resolve()
        if resolved in checked:
            return None
        checked.add(resolved)
        if _is_sdk_python_dir(resolved):
            return resolved
        return None

    for candidate in [start, *start.parents]:
        found = _consider(candidate)
        if found is not None:
            return found

    for candidate in [start / 'sdk' / 'python', *(parent / 'sdk' / 'python' for parent in start.parents)]:
        found = _consider(candidate)
        if found is not None:
            return found

    env_dir = os.environ.get('CODEX_PYTHON_SDK_DIR')
    if env_dir:
        found = _consider(Path(env_dir).expanduser())
        if found is not None:
            return found

    return None


repo_python_dir = _find_sdk_python_dir(Path.cwd())
if repo_python_dir is None:
    raise RuntimeError('Could not locate sdk/python. Set CODEX_PYTHON_SDK_DIR to your sdk/python path.')

repo_python_str = str(repo_python_dir)
if repo_python_str not in sys.path:
    sys.path.insert(0, repo_python_str)

from _runtime_setup import ensure_runtime_package_installed

runtime_version = ensure_runtime_package_installed(
    sys.executable,
    repo_python_dir,
)

src_dir = repo_python_dir / 'src'
examples_dir = repo_python_dir / 'examples'
src_str = str(src_dir)
examples_str = str(examples_dir)
if src_str not in sys.path:
    sys.path.insert(0, src_str)
if examples_str not in sys.path:
    sys.path.insert(0, examples_str)

# Force fresh imports after SDK upgrades in the same notebook kernel.
for module_name in list(sys.modules):
    if module_name == 'openai_codex' or module_name.startswith('openai_codex.'):
        sys.modules.pop(module_name, None)

print('Kernel:', sys.executable)
print('SDK source:', src_dir)
print('Runtime package:', runtime_version)

In [ ]:

# Cell 2: imports (public only)
from _bootstrap import generated_sample_image_data_url, server_label
from openai_codex import (
    AsyncCodex,
    Codex,
    ImageInput,
    LocalImageInput,
    TextInput,
    retry_on_overload,
)

In [ ]:

# Cell 2b: browser login handle lifecycle
with Codex() as codex:
    # Open this URL and call `wait()` without canceling when completing login for real.
    login = codex.login_chatgpt()
    print('Please complete login at:', login.auth_url)
    completed = login.wait()
    account = codex.account()

    print('login.id:', login.login_id)
    print('login.auth_url:', login.auth_url)
    print('login.completed.success:', completed.success)
    print('account:', account.email)

In [ ]:

# Cell 3: simple sync conversation
with Codex() as codex:
    thread = codex.thread_start(model='gpt-5.4', config={'model_reasoning_effort': 'high'})
    result = thread.run('Explain gradient descent in 3 bullets.')
    print(result.final_response)

In [ ]:

# Cell 4: multi-turn continuity in same thread
with Codex() as codex:
    thread = codex.thread_start(model='gpt-5.4', config={'model_reasoning_effort': 'high'})
    first = thread.turn('Give a short summary of transformers.').run()
    second = thread.turn('Now explain that to a high-school student.').run()
    print('first status:', first.status)
    print('second status:', second.status)
    print('second text:', second.final_response)

In [ ]:

# Cell 5: full thread lifecycle and branching (sync)
with Codex() as codex:
    thread = codex.thread_start(model='gpt-5.4', config={'model_reasoning_effort': 'high'})
    first = thread.turn('One sentence about structured planning.').run()
    second = thread.turn('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 = codex.thread_resume(
        unarchived.id,
        model='gpt-5.4',
        config={'model_reasoning_effort': 'high'},
    )
    resumed_result = resumed.turn('Continue in one short sentence.').run()

    forked = codex.thread_fork(unarchived.id, model='gpt-5.4')
    forked_result = forked.turn('Take a different angle in one short sentence.').run()

    compact_result = unarchived.compact()

    print('Lifecycle OK:', thread.id)
    print('first:', first.id, first.status)
    print('second:', second.id, second.status)
    print('read.turns:', len(reading.thread.turns))
    print('list.active:', len(listing_active.data))
    print('list.archived:', len(listing_archived.data))
    print('resumed:', resumed_result.id, resumed_result.status)
    print('forked:', forked_result.id, forked_result.status)
    print('compact:', compact_result.model_dump(mode='json', by_alias=True))

In [ ]:

# Cell 5b: one turn with most optional turn params
from pathlib import Path
from openai_codex import Sandbox
from openai_codex.types import (
    ReasoningEffort,
    ReasoningSummary,
)

output_schema = {
    'type': 'object',
    'properties': {
        'summary': {'type': 'string'},
        'actions': {'type': 'array', 'items': {'type': 'string'}},
    },
    'required': ['summary', 'actions'],
    'additionalProperties': False,
}

summary = ReasoningSummary.model_validate('concise')

with Codex() as codex:
    thread = codex.thread_start(model='gpt-5.4', config={'model_reasoning_effort': 'high'})
    turn = thread.turn(
        'Propose a safe production feature-flag rollout. Return JSON matching the schema.',
        cwd=str(Path.cwd()),
        effort=ReasoningEffort.medium,
        model='gpt-5.4',
        output_schema=output_schema,
        sandbox=Sandbox.read_only,
        summary=summary,
    )
    result = turn.run()
    print('status:', result.status)
    print(result.final_response)

In [ ]:

# Cell 5c: choose highest model + highest supported reasoning, then run turns
from pathlib import Path
from openai_codex import Sandbox
from openai_codex.types import (
    ReasoningEffort,
    ReasoningSummary,
)

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]
    if not visible:
        raise RuntimeError('models response did not include visible 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)]
    if not top_candidates:
        raise RuntimeError('models response did not include top-level visible models')
    return max(top_candidates, key=lambda m: (m.model, m.id))


def pick_highest_turn_effort(model) -> ReasoningEffort:
    if not model.supported_reasoning_efforts:
        raise RuntimeError(f'{model.model} did not advertise supported reasoning efforts')
    best = max(model.supported_reasoning_efforts, key=lambda opt: reasoning_rank[opt.reasoning_effort.value])
    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,
}

with Codex() 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(
        'Give one short sentence about reliable production releases.',
        model=selected_model.model,
        effort=selected_effort,
    ).run()
    print('agent.message:', first.final_response)
    print('items:', len(first.items))

    second = thread.turn(
        'Return JSON for a safe feature-flag rollout plan.',
        cwd=str(Path.cwd()),
        effort=selected_effort,
        model=selected_model.model,
        output_schema=output_schema,
        sandbox=Sandbox.read_only,
        summary=ReasoningSummary.model_validate('concise'),
    ).run()
    print('agent.message.params:', second.final_response)
    print('items.params:', len(second.items))

In [ ]:

# Cell 6: multimodal with an image data URL
image_data_url = generated_sample_image_data_url()

with Codex() as codex:
    thread = codex.thread_start(model='gpt-5.4', config={'model_reasoning_effort': 'high'})
    result = thread.turn([
        TextInput('What do you see in this image? 3 bullets.'),
        ImageInput(image_data_url),
    ]).run()
    print('status:', result.status)
    print(result.final_response)

In [ ]:

# Cell 7: multimodal with local image (generated temporary file)
with temporary_sample_image_path() as local_image_path:
    with Codex() as codex:
        thread = codex.thread_start(model='gpt-5.4', config={'model_reasoning_effort': 'high'})
        result = thread.turn([
            TextInput('Describe the colors and layout in this generated local image in 2 bullets.'),
            LocalImageInput(str(local_image_path.resolve())),
        ]).run()
        print('status:', result.status)
        print(result.final_response)

In [ ]:

# Cell 8: retry-on-overload pattern
with Codex() as codex:
    thread = codex.thread_start(model='gpt-5.4', config={'model_reasoning_effort': 'high'})

    result = retry_on_overload(
        lambda: thread.turn('List 5 failure modes in distributed systems.').run(),
        max_attempts=3,
        initial_delay_s=0.25,
        max_delay_s=2.0,
    )
    print('status:', result.status)
    print(result.final_response)

In [ ]:

# Cell 9: full thread lifecycle and branching (async)
import asyncio


async def async_lifecycle_demo():
    async with AsyncCodex() as codex:
        thread = await codex.thread_start(model='gpt-5.4', config={'model_reasoning_effort': 'high'})
        first = await (await thread.turn('One sentence about structured planning.')).run()
        second = await (await thread.turn('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 = await codex.thread_resume(
            unarchived.id,
            model='gpt-5.4',
            config={'model_reasoning_effort': 'high'},
        )
        resumed_result = await (await resumed.turn('Continue in one short sentence.')).run()

        forked = await codex.thread_fork(unarchived.id, model='gpt-5.4')
        forked_result = await (await forked.turn('Take a different angle in one short sentence.')).run()

        compact_result = await unarchived.compact()

        print('Lifecycle OK:', thread.id)
        print('first:', first.id, first.status)
        print('second:', second.id, second.status)
        print('read.turns:', len(reading.thread.turns))
        print('list.active:', len(listing_active.data))
        print('list.archived:', len(listing_archived.data))
        print('resumed:', resumed_result.id, resumed_result.status)
        print('forked:', forked_result.id, forked_result.status)
        print('compact:', compact_result.model_dump(mode='json', by_alias=True))


await async_lifecycle_demo()

In [ ]:

# Cell 10: async turn controls (steer + interrupt)
import asyncio


async def async_stream_demo():
    async with AsyncCodex() as codex:
        thread = await codex.thread_start(model='gpt-5.4', config={'model_reasoning_effort': 'high'})
        steer_turn = await thread.turn('Count from 1 to 40 with commas, then one summary sentence.')

        steer_result = await steer_turn.steer('Keep it brief and stop after 10 numbers.')

        steer_event_count = 0
        steer_completed_status = None
        steer_deltas = []
        async for event in steer_turn.stream():
            steer_event_count += 1
            if event.method == 'item/agentMessage/delta':
                steer_deltas.append(event.payload.delta)
                continue
            if event.method == 'turn/completed':
                steer_completed_status = event.payload.turn.status.value

        if steer_completed_status is None:
            raise RuntimeError('stream ended without turn/completed')
        steer_preview = ''.join(steer_deltas).strip()

        interrupt_turn = await thread.turn('Count from 1 to 200 with commas, then one summary sentence.')
        interrupt_result = await interrupt_turn.interrupt()

        interrupt_event_count = 0
        interrupt_completed_status = None
        interrupt_deltas = []
        async for event in interrupt_turn.stream():
            interrupt_event_count += 1
            if event.method == 'item/agentMessage/delta':
                interrupt_deltas.append(event.payload.delta)
                continue
            if event.method == 'turn/completed':
                interrupt_completed_status = event.payload.turn.status.value

        if interrupt_completed_status is None:
            raise RuntimeError('stream ended without turn/completed')
        interrupt_preview = ''.join(interrupt_deltas).strip()

        print('steer.result:', steer_result.model_dump(mode='json', by_alias=True))
        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.model_dump(mode='json', by_alias=True))
        print('interrupt.final.status:', interrupt_completed_status)
        print('interrupt.events.count:', interrupt_event_count)
        print('interrupt.assistant.preview:', interrupt_preview)


await async_stream_demo()