Files
codex/sdk/python/docs/api-reference.md
Ahmed Ibrahim e7bffc5a20 [codex] Accept string input for Python turns (#23162)
## Summary
- Allow thread.turn and turn.steer, including async variants, to accept
RunInput so plain strings work alongside typed input objects.
- Export RunInput and update the SDK artifact generator so regenerated
turn methods keep the same signature and normalization.
- Update docs, examples, notebook cells, and tests to use string
shorthand for text-only turns while keeping typed inputs for multimodal
input.

## Validation
- uv run --extra dev ruff format .
- uv run --extra dev ruff check --output-format=github .
- python3 -m py_compile sdk/python/src/openai_codex/__init__.py
sdk/python/src/openai_codex/api.py
sdk/python/src/openai_codex/_inputs.py
sdk/python/scripts/update_sdk_artifacts.py
sdk/python/tests/test_public_api_signatures.py
sdk/python/tests/test_app_server_streaming.py
sdk/python/tests/test_app_server_turn_controls.py
sdk/python/tests/test_real_app_server_integration.py
- python3 -c "import json;
json.load(open('sdk/python/notebooks/sdk_walkthrough.ipynb'))"
- sdk/python/.venv/bin/python -c "import inspect, openai_codex; from
openai_codex import Thread, AsyncThread, TurnHandle, AsyncTurnHandle,
RunInput; funcs=[Thread.run, Thread.turn, AsyncThread.run,
AsyncThread.turn, TurnHandle.steer, AsyncTurnHandle.steer]; assert
all(inspect.signature(fn).parameters['input'].annotation == 'RunInput'
for fn in funcs); assert RunInput is openai_codex.RunInput"
2026-05-17 09:05:44 -07:00

9.0 KiB

OpenAI Codex SDK — API Reference

Public surface of openai_codex for app-server v2.

This SDK surface is experimental. Turn streams are routed by turn ID so one client can consume multiple active turns concurrently. Thread starts default to ApprovalMode.auto_review; turn starts accept an optional approval_mode override.

Package Entry

from openai_codex import (
    Codex,
    AsyncCodex,
    ApprovalMode,
    ChatgptLoginHandle,
    DeviceCodeLoginHandle,
    AsyncChatgptLoginHandle,
    AsyncDeviceCodeLoginHandle,
    Thread,
    AsyncThread,
    TurnHandle,
    AsyncTurnHandle,
    TurnResult,
    Input,
    InputItem,
    RunInput,
    TextInput,
    ImageInput,
    LocalImageInput,
    SkillInput,
    MentionInput,
)
from openai_codex.types import (
    Account,
    AccountLoginCompletedNotification,
    CancelLoginAccountResponse,
    CancelLoginAccountStatus,
    GetAccountResponse,
    InitializeResponse,
    ThreadItem,
    ThreadTokenUsage,
    TurnError,
    TurnStatus,
)
  • Version: openai_codex.__version__
  • Requires Python >= 3.10
  • Public app-server value and event types live in openai_codex.types

Codex (sync)

Codex(config: AppServerConfig | None = None)

Properties/methods:

  • metadata -> InitializeResponse
  • close() -> None
  • login_api_key(api_key: str) -> None
  • login_chatgpt() -> ChatgptLoginHandle
  • login_chatgpt_device_code() -> DeviceCodeLoginHandle
  • account(*, refresh_token: bool = False) -> GetAccountResponse
  • logout() -> None
  • thread_start(*, approval_mode=ApprovalMode.auto_review, base_instructions=None, config=None, cwd=None, developer_instructions=None, ephemeral=None, model=None, model_provider=None, personality=None, sandbox=None) -> Thread
  • thread_list(*, archived=None, cursor=None, cwd=None, limit=None, model_providers=None, sort_key=None, source_kinds=None) -> ThreadListResponse
  • thread_resume(thread_id: str, *, approval_mode=ApprovalMode.auto_review, base_instructions=None, config=None, cwd=None, developer_instructions=None, model=None, model_provider=None, personality=None, sandbox=None) -> Thread
  • thread_fork(thread_id: str, *, approval_mode=ApprovalMode.auto_review, base_instructions=None, config=None, cwd=None, developer_instructions=None, model=None, model_provider=None, sandbox=None) -> Thread
  • thread_archive(thread_id: str) -> ThreadArchiveResponse
  • thread_unarchive(thread_id: str) -> Thread
  • models(*, include_hidden: bool = False) -> ModelListResponse

Context manager:

with Codex() as codex:
    ...

AsyncCodex (async parity)

AsyncCodex(config: AppServerConfig | None = None)

Preferred usage:

async with AsyncCodex() as codex:
    ...

AsyncCodex initializes lazily. Context entry is the standard path because it ensures startup and shutdown are paired explicitly.

Properties/methods:

  • metadata -> InitializeResponse
  • close() -> Awaitable[None]
  • login_api_key(api_key: str) -> Awaitable[None]
  • login_chatgpt() -> Awaitable[AsyncChatgptLoginHandle]
  • login_chatgpt_device_code() -> Awaitable[AsyncDeviceCodeLoginHandle]
  • account(*, refresh_token: bool = False) -> Awaitable[GetAccountResponse]
  • logout() -> Awaitable[None]
  • thread_start(*, approval_mode=ApprovalMode.auto_review, base_instructions=None, config=None, cwd=None, developer_instructions=None, ephemeral=None, model=None, model_provider=None, personality=None, sandbox=None) -> Awaitable[AsyncThread]
  • thread_list(*, archived=None, cursor=None, cwd=None, limit=None, model_providers=None, sort_key=None, source_kinds=None) -> Awaitable[ThreadListResponse]
  • thread_resume(thread_id: str, *, approval_mode=ApprovalMode.auto_review, base_instructions=None, config=None, cwd=None, developer_instructions=None, model=None, model_provider=None, personality=None, sandbox=None) -> Awaitable[AsyncThread]
  • thread_fork(thread_id: str, *, approval_mode=ApprovalMode.auto_review, base_instructions=None, config=None, cwd=None, developer_instructions=None, ephemeral=None, model=None, model_provider=None, sandbox=None) -> Awaitable[AsyncThread]
  • thread_archive(thread_id: str) -> Awaitable[ThreadArchiveResponse]
  • thread_unarchive(thread_id: str) -> Awaitable[AsyncThread]
  • models(*, include_hidden: bool = False) -> Awaitable[ModelListResponse]

Async context manager:

async with AsyncCodex() as codex:
    ...

Login handles

ChatgptLoginHandle / AsyncChatgptLoginHandle

  • login_id: str
  • auth_url: str
  • wait() -> AccountLoginCompletedNotification
  • cancel() -> CancelLoginAccountResponse

Async handle methods return awaitables.

DeviceCodeLoginHandle / AsyncDeviceCodeLoginHandle

  • login_id: str
  • verification_url: str
  • user_code: str
  • wait() -> AccountLoginCompletedNotification
  • cancel() -> CancelLoginAccountResponse

Async handle methods return awaitables.

wait() consumes only the completion notification for its matching login attempt. API-key login completes synchronously and does not return a handle.

Thread / AsyncThread

Thread and AsyncThread share the same shape and intent.

Thread

  • run(input: str | Input, *, approval_mode=None, cwd=None, effort=None, model=None, output_schema=None, personality=None, sandbox_policy=None, service_tier=None, summary=None) -> TurnResult
  • turn(input: str | Input, *, approval_mode=None, cwd=None, effort=None, model=None, output_schema=None, personality=None, sandbox_policy=None, service_tier=None, summary=None) -> TurnHandle
  • read(*, include_turns: bool = False) -> ThreadReadResponse
  • set_name(name: str) -> ThreadSetNameResponse
  • compact() -> ThreadCompactStartResponse

AsyncThread

  • run(input: str | Input, *, approval_mode=None, cwd=None, effort=None, model=None, output_schema=None, personality=None, sandbox_policy=None, service_tier=None, summary=None) -> Awaitable[TurnResult]
  • turn(input: str | Input, *, approval_mode=None, cwd=None, effort=None, model=None, output_schema=None, personality=None, sandbox_policy=None, service_tier=None, summary=None) -> Awaitable[AsyncTurnHandle]
  • read(*, include_turns: bool = False) -> Awaitable[ThreadReadResponse]
  • set_name(name: str) -> Awaitable[ThreadSetNameResponse]
  • compact() -> Awaitable[ThreadCompactStartResponse]

run(...) is the common-case convenience path. It accepts plain strings, starts the turn, consumes notifications until completion, and returns a small result object with:

  • id: str
  • status: TurnStatus
  • error: TurnError | None
  • started_at: int | None
  • completed_at: int | None
  • duration_ms: int | None
  • final_response: str | None
  • items: list[ThreadItem]
  • usage: ThreadTokenUsage | None

final_response is None when the turn finishes without a final-answer or phase-less assistant message item.

Use turn(...) when you need low-level turn control (stream(), steer(), interrupt()) before collecting the turn result.

TurnHandle / AsyncTurnHandle

TurnHandle

  • steer(input: str | Input) -> TurnSteerResponse
  • interrupt() -> TurnInterruptResponse
  • stream() -> Iterator[Notification]
  • run() -> TurnResult

Behavior notes:

  • stream() and run() consume only notifications for their own turn ID
  • one Codex instance can stream multiple active turns concurrently

AsyncTurnHandle

  • steer(input: str | Input) -> Awaitable[TurnSteerResponse]
  • interrupt() -> Awaitable[TurnInterruptResponse]
  • stream() -> AsyncIterator[Notification]
  • run() -> Awaitable[TurnResult]

Behavior notes:

  • stream() and run() consume only notifications for their own turn ID
  • one AsyncCodex instance can stream multiple active turns concurrently

Inputs

@dataclass class TextInput: text: str
@dataclass class ImageInput: url: str
@dataclass class LocalImageInput: path: str
@dataclass class SkillInput: name: str; path: str
@dataclass class MentionInput: name: str; path: str

InputItem = TextInput | ImageInput | LocalImageInput | SkillInput | MentionInput
Input = list[InputItem] | InputItem
RunInput = Input | str

Use a plain str as shorthand for TextInput(...) anywhere a turn input is accepted: thread.run("..."), thread.turn("..."), and turn.steer("...").

Public Types

The SDK wrappers return and accept public app-server models wherever possible:

from openai_codex.types import (
    Account,
    AccountLoginCompletedNotification,
    CancelLoginAccountResponse,
    CancelLoginAccountStatus,
    GetAccountResponse,
    ThreadReadResponse,
    Turn,
    TurnStatus,
)

Retry + errors

from openai_codex import (
    retry_on_overload,
    JsonRpcError,
    MethodNotFoundError,
    InvalidParamsError,
    ServerBusyError,
    is_retryable_error,
)
  • retry_on_overload(...) retries transient overload errors with exponential backoff + jitter.
  • is_retryable_error(exc) checks if an exception is transient/overload-like.

Example

from openai_codex import Codex

with Codex() as codex:
    thread = codex.thread_start(model="gpt-5.4", config={"model_reasoning_effort": "high"})
    result = thread.run("Say hello in one sentence.")
    print(result.final_response)