Skip to content

codex_auth_helper API

codex-auth-helper is intentionally small. The public API is documented here so ACP examples can depend on it without requiring readers to inspect the package source first.

Functions

create_codex_responses_model(model_name, *, config=None, http_client=None, auth_http_client=None, instructions, settings=None, transport='responses', connection='http', fallback='error', transport_observer=None)

Source code in packages/helpers/codex-auth-helper/src/codex_auth_helper/factory.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def create_codex_responses_model(
    model_name: str,
    *,
    config: CodexAuthConfig | None = None,
    http_client: Any | None = None,
    auth_http_client: httpx.AsyncClient | None = None,
    instructions: str,
    settings: OpenAIResponsesModelSettings | None = None,
    transport: CodexResponsesTransport = "responses",
    connection: CodexResponsesConnection = "http",
    fallback: CodexResponsesFallback = "error",
    transport_observer: CodexResponsesTransportObserver | None = None,
) -> CodexResponsesModel:
    if instructions is None:
        raise ValueError(
            "`instructions` is required for Codex-backed Pydantic models. "
            "Pass an explicit system instruction string.",
        )
    if transport == "responses_lite" and connection == "websocket":
        raise ValueError("`responses_lite` does not support the WebSocket connection.")
    client = create_codex_async_openai(
        config=config,
        http_client=http_client,
        auth_http_client=auth_http_client,
        transport=transport,
        transport_observer=transport_observer,
    )
    model_settings: OpenAIResponsesModelSettings = {"openai_store": False}
    if settings is not None:
        model_settings.update(settings)
    model_settings["openai_store"] = False
    return CodexResponsesModel(
        model_name,
        default_instructions=instructions,
        provider=OpenAIProvider(openai_client=client),
        settings=model_settings,
        connection=connection,
        fallback=fallback,
    )

create_codex_async_openai(*, config=None, http_client=None, auth_http_client=None, transport='responses', transport_observer=None)

Source code in packages/helpers/codex-auth-helper/src/codex_auth_helper/client.py 
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
def create_codex_async_openai(
    *,
    config: CodexAuthConfig | None = None,
    http_client: Any | None = None,
    auth_http_client: httpx.AsyncClient | None = None,
    transport: CodexResponsesTransport = "responses",
    transport_observer: CodexResponsesTransportObserver | None = None,
) -> CodexAsyncOpenAI:
    resolved_config = config or CodexAuthConfig()
    owns_http_client = http_client is None
    owns_auth_http_client = auth_http_client is None
    resolved_auth_http_client = auth_http_client or httpx.AsyncClient(
        follow_redirects=True,
        timeout=resolved_config.timeout_seconds,
    )
    token_manager = CodexTokenManager(
        config=resolved_config,
        store=CodexAuthStore(resolved_config.auth_path),
        http_client=resolved_auth_http_client,
        owns_http_client=owns_auth_http_client,
    )
    return CodexAsyncOpenAI(
        base_url=resolved_config.api_base_url,
        http_client=http_client,
        token_manager=token_manager,
        owns_http_client=owns_http_client,
        transport=transport,
        transport_observer=transport_observer,
    )

create_codex_openai(*, config=None, http_client=None, auth_http_client=None, transport='responses')

Source code in packages/helpers/codex-auth-helper/src/codex_auth_helper/client.py 
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
def create_codex_openai(
    *,
    config: CodexAuthConfig | None = None,
    http_client: Any | None = None,
    auth_http_client: httpx.AsyncClient | None = None,
    transport: CodexResponsesTransport = "responses",
) -> CodexOpenAI:
    resolved_config = config or CodexAuthConfig()
    owns_http_client = http_client is None
    owns_auth_http_client = auth_http_client is None
    resolved_auth_http_client = auth_http_client or httpx.AsyncClient(
        follow_redirects=True,
        timeout=resolved_config.timeout_seconds,
    )
    token_manager = CodexTokenManager(
        config=resolved_config,
        store=CodexAuthStore(resolved_config.auth_path),
        http_client=resolved_auth_http_client,
        owns_http_client=owns_auth_http_client,
    )
    return CodexOpenAI(
        base_url=resolved_config.api_base_url,
        http_client=http_client,
        token_manager=token_manager,
        owns_http_client=owns_http_client,
        transport=transport,
    )

create_codex_chat_openai(model_name, *, config=None, http_client=None, auth_http_client=None, instructions, sync_http_client=None, sync_auth_http_client=None, include_response_headers=False, streaming=None, model_kwargs=None, output_version='responses/v1', reasoning=None, temperature=None, transport='responses', use_previous_response_id=False, connection='http', fallback='error', transport_observer=None)

Build a Codex-backed LangChain chat model via langchain-openai.

The returned model is pinned to the OpenAI Responses API and reuses the same Codex auth flow as the Pydantic helper path.

Source code in packages/helpers/codex-auth-helper/src/codex_auth_helper/factory.py 
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def create_codex_chat_openai(
    model_name: str,
    *,
    config: CodexAuthConfig | None = None,
    http_client: Any | None = None,
    auth_http_client: httpx.AsyncClient | None = None,
    instructions: str,
    sync_http_client: Any | None = None,
    sync_auth_http_client: httpx.AsyncClient | None = None,
    include_response_headers: bool = False,
    streaming: bool | None = None,
    model_kwargs: dict[str, Any] | None = None,
    output_version: Literal["v0", "responses/v1"] = "responses/v1",
    reasoning: dict[str, Any] | None = None,
    temperature: float | None = None,
    transport: CodexResponsesTransport = "responses",
    use_previous_response_id: bool = False,
    connection: CodexResponsesConnection = "http",
    fallback: CodexResponsesFallback = "error",
    transport_observer: CodexResponsesTransportObserver | None = None,
) -> CodexChatOpenAI:
    """Build a Codex-backed LangChain chat model via `langchain-openai`.

    The returned model is pinned to the OpenAI Responses API and reuses the
    same Codex auth flow as the Pydantic helper path.
    """
    try:
        from .langchain import CodexChatOpenAI
    except ModuleNotFoundError as exc:
        raise ModuleNotFoundError(
            "Install the optional LangChain dependency first: "
            'uv add "codex-auth-helper[langchain]" or '
            'pip install "codex-auth-helper[langchain]".',
        ) from exc

    validate_responses_connection(connection, fallback)
    if transport == "responses_lite" and connection == "websocket":
        raise ValueError("`responses_lite` does not support the WebSocket connection.")
    if use_previous_response_id:
        raise ValueError("Codex continuation is owned by responses_session(), not response IDs.")
    if connection == "websocket" and include_response_headers:
        raise ValueError("Response HTTP headers are unavailable over Codex WebSocket.")
    async_root_client = create_codex_async_openai(
        config=config,
        http_client=http_client,
        auth_http_client=auth_http_client,
        transport=transport,
        transport_observer=transport_observer,
    )
    sync_root_client = create_codex_openai(
        config=config,
        http_client=sync_http_client,
        auth_http_client=sync_auth_http_client,
        transport=transport,
    )
    chat_model_kwargs = dict(model_kwargs or {})
    if instructions is None:
        raise ValueError(
            "`instructions` is required for Codex-backed LangChain models. "
            "Pass an explicit system instruction string.",
        )
    if "store" in chat_model_kwargs:
        raise ValueError(
            "Do not pass `model_kwargs['store']`; Codex-backed ChatOpenAI always forces "
            "`store=False`.",
        )
    if "instructions" in chat_model_kwargs:
        raise ValueError(
            "Pass `instructions` either through the dedicated parameter or "
            "`model_kwargs['instructions']`, not both.",
        )
    chat_model_kwargs["instructions"] = instructions
    chat_openai_kwargs: dict[str, Any] = {
        "model": model_name,
        "async_client": async_root_client.chat.completions,
        "client": sync_root_client.chat.completions,
        "include_response_headers": include_response_headers,
        "model_kwargs": chat_model_kwargs,
        "output_version": output_version,
        "reasoning": reasoning,
        "root_async_client": async_root_client,
        "root_client": sync_root_client,
        "store": False,
        "temperature": temperature,
        "use_previous_response_id": use_previous_response_id,
        "use_responses_api": True,
        "responses_connection": connection,
        "responses_fallback": fallback,
    }
    # Leave native stream selection automatic over WS: ainvoke must use the
    # atomic response path, while explicit astream still exposes live deltas.
    if streaming is not None or connection == "http":
        chat_openai_kwargs["streaming"] = True if streaming is None else streaming
    return CodexChatOpenAI(
        **chat_openai_kwargs,
    )

Classes

CodexResponsesModel(model_name, *, default_instructions, provider='openai', profile=None, settings=None, connection='http', fallback='error')

Bases: OpenAIResponsesModel

Source code in packages/helpers/codex-auth-helper/src/codex_auth_helper/model.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def __init__(
    self,
    model_name: str,
    *,
    default_instructions: str,
    provider: Any = "openai",
    profile: Any = None,
    settings: ModelSettings | None = None,
    connection: CodexResponsesConnection = "http",
    fallback: CodexResponsesFallback = "error",
) -> None:
    validate_responses_connection(connection, fallback)
    self._default_instructions = default_instructions
    self.responses_connection: CodexResponsesConnection = connection
    self.responses_fallback: CodexResponsesFallback = fallback
    super().__init__(
        model_name,
        provider=provider,
        profile=profile,
        settings=settings,
    )

responses_turn() async

Keep routing state for one native agent run, including tool continuations.

Source code in packages/helpers/codex-auth-helper/src/codex_auth_helper/model.py 
74
75
76
77
78
79
80
81
@asynccontextmanager
async def responses_turn(self) -> AsyncIterator[None]:
    """Keep routing state for one native agent run, including tool continuations."""
    client = self.client
    if not isinstance(client, CodexAsyncOpenAI):
        raise TypeError("CodexResponsesModel requires a CodexAsyncOpenAI client.")
    async with self.responses_session(), client.responses_turn():
        yield

CodexAsyncOpenAI(*, base_url, http_client, token_manager, owns_http_client, transport='responses', transport_observer=None)

Bases: AsyncOpenAI

Source code in packages/helpers/codex-auth-helper/src/codex_auth_helper/client.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def __init__(
    self,
    *,
    base_url: str,
    http_client: Any | None,
    token_manager: CodexTokenManager,
    owns_http_client: bool,
    transport: CodexResponsesTransport = "responses",
    transport_observer: CodexResponsesTransportObserver | None = None,
) -> None:
    _validate_transport(transport)
    self.token_manager = token_manager
    self._owns_http_client = owns_http_client
    self.responses_transport: CodexResponsesTransport = transport
    self.responses_transport_observer = transport_observer
    self._turn_context = ResponsesTurnContext()
    super().__init__(
        api_key=token_manager.get_access_token,
        base_url=base_url,
        http_client=http_client,
    )
    self._http_responses = self.responses
    self._responses_session_var: ContextVar[_ResponsesSessionState | None] = ContextVar(
        f"codex_responses_session_{id(self)}",
        default=None,
    )
    self.__dict__["responses"] = _CodexAsyncResponsesProxy(self)

recover_response(exc, *, retries) async

Recover an uncommitted, non-streaming framework model response only.

The caller must rebuild the complete native request, not replay a delta or rerun its agent/tools. Never call this after exposing partial output.

Source code in packages/helpers/codex-auth-helper/src/codex_auth_helper/client.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
async def recover_response(self, exc: Exception, *, retries: int) -> bool:
    """Recover an uncommitted, non-streaming framework model response only.

    The caller must rebuild the complete native request, not replay a delta
    or rerun its agent/tools. Never call this after exposing partial output.
    """
    state = self.current_responses_session()
    if (
        state is None
        or state.info.requested_connection != "websocket"
        or state.info.effective_connection == "http"
        or not state.replay_safe
        or not _transient_response_failure(exc)
    ):
        return False
    fallback = retries == self.max_retries and state.fallback == "http"
    if retries >= self.max_retries and not fallback:
        return False
    with suppress(Exception):
        await state.close_connection()
    state.continuation_reason = "transport_recovery"
    state.acknowledged_message_count = 0
    if fallback:
        state.info.effective_connection = "http"
        state.info.fallback_used = True
    self._observe(
        state,
        {},
        phase="fallback" if fallback else "retry",
        failure_category="transport",
    )
    await asyncio.sleep(min(0.25 * 2**retries, 2.0))
    return True

responses_turn() async

Scope server routing state to one user turn, including its tool calls.

Source code in packages/helpers/codex-auth-helper/src/codex_auth_helper/client.py 
233
234
235
236
237
@asynccontextmanager
async def responses_turn(self) -> AsyncIterator[None]:
    """Scope server routing state to one user turn, including its tool calls."""
    with self._turn_context.scope():
        yield

CodexOpenAI(*, base_url, http_client, token_manager, owns_http_client, transport='responses')

Bases: OpenAI

Source code in packages/helpers/codex-auth-helper/src/codex_auth_helper/client.py 
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
def __init__(
    self,
    *,
    base_url: str,
    http_client: Any | None,
    token_manager: CodexTokenManager,
    owns_http_client: bool,
    transport: CodexResponsesTransport = "responses",
) -> None:
    _validate_transport(transport)
    self.token_manager = token_manager
    self._owns_http_client = owns_http_client
    self.responses_transport: CodexResponsesTransport = transport
    self._turn_context = ResponsesTurnContext()
    super().__init__(
        api_key=token_manager.get_access_token_sync,
        base_url=base_url,
        http_client=http_client,
    )

responses_turn()

Scope server routing state to one synchronous user turn.

Source code in packages/helpers/codex-auth-helper/src/codex_auth_helper/client.py 
577
578
579
def responses_turn(self) -> AbstractContextManager[None]:
    """Scope server routing state to one synchronous user turn."""
    return self._turn_context.scope()

CodexResponsesSessionInfo(requested_connection, effective_connection=None, fallback_used=False, observer_failures=0) dataclass

Sanitized connection outcome for one logical model run.

CodexResponsesTransportEvent(session_id, attempt, phase, requested_connection, effective_connection, fallback_used, continuation, continuation_reason, acknowledged_message_count, input_item_count, input_bytes, request_bytes, failure_category=None) dataclass

Content-free evidence for one Responses transport attempt.

CodexResponsesConnectionError

Bases: RuntimeError

Raised when an explicitly requested Responses WebSocket cannot be opened.

CodexResponsesProtocolError

Bases: RuntimeError

Raised when a WebSocket response cannot be handled without unsafe replay.

CodexAuthConfig(auth_path=default_auth_path(), api_base_url='https://chatgpt.com/backend-api/codex', client_id='app_EMoamEEZ73f0CkXaXp7hrann', default_token_ttl=timedelta(hours=1), issuer='https://auth.openai.com', refresh_margin=timedelta(seconds=30), timeout_seconds=30.0) dataclass

CodexAuthState(access_token, refresh_token, account_id=None, auth_mode=None, expires_at=None, id_token=None, last_refresh=None, openai_api_key=None) dataclass

CodexAuthStore(path) dataclass

CodexTokenManager(config, store, http_client, owns_http_client=False) dataclass

recover_after_unauthorized(rejected_token, *, refresh) async

Reload first; optionally refresh once after an explicit 401 rejection.

Source code in packages/helpers/codex-auth-helper/src/codex_auth_helper/auth/manager.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
async def recover_after_unauthorized(self, rejected_token: str, *, refresh: bool) -> str | None:
    """Reload first; optionally refresh once after an explicit 401 rejection."""
    while not self._refresh_lock.acquire(blocking=False):
        await asyncio.sleep(0.01)
    try:
        self._reload_guarded()
        if self._state.access_token != rejected_token and not self._should_refresh(self._state):
            return self._state.access_token
        if refresh:
            self._state = await self._refresh_locked()
            return self._state.access_token
        return None
    finally:
        self._refresh_lock.release()

recover_after_unauthorized_sync(rejected_token, *, refresh)

Synchronous equivalent of the bounded unauthorized recovery step.

Source code in packages/helpers/codex-auth-helper/src/codex_auth_helper/auth/manager.py 
110
111
112
113
114
115
116
117
118
119
def recover_after_unauthorized_sync(self, rejected_token: str, *, refresh: bool) -> str | None:
    """Synchronous equivalent of the bounded unauthorized recovery step."""
    with self._refresh_lock:
        self._reload_guarded()
        if self._state.access_token != rejected_token and not self._should_refresh(self._state):
            return self._state.access_token
        if refresh:
            self._state = self._refresh_locked_sync()
            return self._state.access_token
        return None

CodexAuthAccountMismatchError

Bases: OpenAIError

Credentials changed to a different account during an active client lifetime.

CodexAuthRefreshError(reason)

Bases: OpenAIError

Permanent refresh rejection without exposing the credential response body.

Source code in packages/helpers/codex-auth-helper/src/codex_auth_helper/auth/errors.py 
17
18
19
def __init__(self, reason: Literal["expired", "reused", "invalidated", "rejected"]) -> None:
    self.reason = reason
    super().__init__(f"Codex credentials could not be refreshed ({reason}); sign in again.")

Turn Lifecycle

async with model.responses_turn(): scopes one user request, including its tool loop, for both Pydantic AI and LangChain. It opens a Responses session if needed. Wrap several separate turns in model.responses_session() to reuse a connection; each turn still starts with fresh routing state. Kedi owns this run boundary automatically. The lower-level async and sync clients expose equivalent async and sync responses_turn() contexts.

The first server x-codex-turn-state is retained only until turn exit and echoed unchanged. HTTP reads response headers. WebSocket reads handshake headers and codex.response.metadata, then uses response.create.client_metadata and any replacement handshake. The managed slot overrides caller-supplied turn state, without changing other metadata. It never becomes prompt history or a metric value; cancellation and nested scopes cannot carry it into a new turn. This is routing affinity, not a provider cache-hit guarantee.

Credential Recovery

Expired credentials are reloaded before refresh. Same-file managers serialize refreshes inside the process and reject account changes. An explicit Responses HTTP 401 or handshake 401 may reload once, then refresh once; custom HTTP Authorization and unrelated endpoints are not intercepted. Permanent auth errors and exhausted 401s do not switch WebSocket sessions to HTTP. Recovery does not replay an accepted response or an ambiguous send/stream failure.

Refresh failures expose a safe CodexAuthRefreshError.reason (expired, reused, invalidated, or rejected), not raw server error bodies. Account changes raise CodexAuthAccountMismatchError. Create a new client after changing login. Access-token expiry is authoritative when available. Atomic file writes and a pre-write rotation check remain, but there is no cross-process refresh lock.

Request Headers

The native Pydantic AI and LangChain factories and both lower-level clients send originator: codex-auth-helper and User-Agent: codex-auth-helper/<installed version>. Effective Responses model and tier settings determine x-codex-routing-hint; extra_body overrides are included when deriving that hint. An explicitly supplied service tier adds ;tier=<effective tier>; an omitted tier does not add a tier hint.

WebSocket handshakes additionally send OpenAI-Beta: responses_websockets=2026-02-06. The HTTP path does not add it. Caller-provided values take precedence regardless of header-name casing, and the caller's header mapping is not mutated. Changed handshake options replace the connection before dispatch; native model integrations resume with full history instead of reusing the old connection's response ID.

These defaults identify this package, not Codex CLI or another client. They are not a guarantee of cache affinity, cache hits, or future backend acceptance.

Types

CodexResponsesTransport is the public literal transport selector:

  • "responses" keeps the standard Responses request shape and is the default.
  • "responses_lite" emits the Codex Responses Lite request shape and disables native parallel tool-call batches as required by that backend contract.

CodexResponsesConnection selects the Pydantic model's outbound Responses connection:

  • "http" preserves the existing default request path.
  • "websocket" keeps one connection for an explicit logical run and enables append-only continuation with previous_response_id.

WebSocket shutdown allows 250 ms for the closing handshake. This deadline does not limit response generation or tool execution.

History comparisons use independent snapshots of request-relevant values, including completed model output. Editing messages in place resets continuation; uncopyable values use the full-request path. Usage, run IDs, and other local run bookkeeping do not invalidate an otherwise unchanged prefix. Concurrent requests to one session are rejected before handshake or send can overlap. Separate sessions remain independent.

Transport metrics are computed only when transport_observer is configured; an absent observer does not trigger payload-size serialization.

CodexResponsesFallback controls initial WebSocket handshake failure:

  • "error" is strict and is the default.
  • "http" permits a recorded fallback before any request is submitted. It does not replay ambiguous send or stream failures.

CodexResponsesTransportObserver is a synchronous callback receiving content-free CodexResponsesTransportEvent values. Keep callbacks non-blocking; exceptions are counted on the active CodexResponsesSessionInfo and isolated from request execution.