## 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"
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 -> InitializeResponseclose() -> Nonelogin_api_key(api_key: str) -> Nonelogin_chatgpt() -> ChatgptLoginHandlelogin_chatgpt_device_code() -> DeviceCodeLoginHandleaccount(*, refresh_token: bool = False) -> GetAccountResponselogout() -> Nonethread_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) -> Threadthread_list(*, archived=None, cursor=None, cwd=None, limit=None, model_providers=None, sort_key=None, source_kinds=None) -> ThreadListResponsethread_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) -> Threadthread_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) -> Threadthread_archive(thread_id: str) -> ThreadArchiveResponsethread_unarchive(thread_id: str) -> Threadmodels(*, 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 -> InitializeResponseclose() -> 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: strauth_url: strwait() -> AccountLoginCompletedNotificationcancel() -> CancelLoginAccountResponse
Async handle methods return awaitables.
DeviceCodeLoginHandle / AsyncDeviceCodeLoginHandle
login_id: strverification_url: struser_code: strwait() -> AccountLoginCompletedNotificationcancel() -> 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) -> TurnResultturn(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) -> TurnHandleread(*, include_turns: bool = False) -> ThreadReadResponseset_name(name: str) -> ThreadSetNameResponsecompact() -> 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: strstatus: TurnStatuserror: TurnError | Nonestarted_at: int | Nonecompleted_at: int | Noneduration_ms: int | Nonefinal_response: str | Noneitems: 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) -> TurnSteerResponseinterrupt() -> TurnInterruptResponsestream() -> Iterator[Notification]run() -> TurnResult
Behavior notes:
stream()andrun()consume only notifications for their own turn ID- one
Codexinstance 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()andrun()consume only notifications for their own turn ID- one
AsyncCodexinstance 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)