sdk/python: add first-class login support (#23093)

## Why The Python SDK can already create threads and run turns, but authentication still has to be arranged outside the SDK. App-server already exposes account login, account inspection, logout, and `account/login/completed` notifications, so SDK users currently have to work around a missing public client layer for a core setup step. This change makes authentication a normal SDK workflow while preserving the backend flow shape: API-key login completes immediately, and interactive ChatGPT flows return live handles that complete later through app-server notifications. ## What changed - Added public sync and async auth methods on `Codex` / `AsyncCodex`: - `login_api_key(...)` - `login_chatgpt()` - `login_chatgpt_device_code()` - `account(...)` - `logout()` - Added public browser-login and device-code handle types with attempt-local `wait()` and `cancel()` helpers. Cancellation stays on the handle instead of a root-level SDK method. - Extended the Python app-server client and notification router so login completion events are routed by `login_id` without consuming unrelated global notifications. - Kept login request/handle logic in a focused internal `_login.py` module so `api.py` remains the public facade instead of absorbing more auth plumbing. - Exported the new handle types plus curated account/login response types from the SDK surfaces. - Updated SDK docs, added sync/async login walkthrough examples, and added a notebook login walkthrough cell. ## Verification Added SDK coverage for: - API-key login, account readback, and logout through the app-server harness in both sync and async clients. - Browser login cancellation plus `handle.wait()` completion through the real app-server boundary used by the Python SDK harness. - Waiter routing that stays scoped across replaced interactive login attempts, plus async handle cancellation coverage. - Login notification demuxing, replay of early completion events, and async client delegation. - Public export/signature assertions. - Real integration-suite smoke coverage for the new examples and notebook login cell.
2026-07-01 00:31:56 +08:00 · 2026-05-17 05:49:28 +03:00
parent 0445b290fe
commit 4c89772314
19 changed files with 772 additions and 13 deletions
@@ -26,6 +26,8 @@ when you intentionally want to run against a specific local app-server binary.
 from openai_codex import Codex

 with Codex() as codex:
+    # Call login_api_key(...) first when this app-server session is not
+    # already authenticated.
    thread = codex.thread_start(model="gpt-5")
    result = thread.run("Say hello in one sentence.")
    print(result.final_response)
@@ -35,6 +37,34 @@ with Codex() as codex:
 `result.final_response` is `None` when the turn completes without a final-answer
 or phase-less assistant message item.

+## Login
+
+Use the auth helper that matches your app:
+
+```python
+from openai_codex import Codex
+
+with Codex() as codex:
+    codex.login_api_key("sk-...")
+    account = codex.account()
+    print(account.account)
+```
+
+Interactive ChatGPT login returns a handle. Open the provided URL or device-code
+page, then wait for the matching completion event:
+
+```python
+with Codex() as codex:
+    login = codex.login_chatgpt()
+    print(login.auth_url)
+    completed = login.wait()
+    print(completed.success)
+```
+
+Use `login_chatgpt_device_code()` for device-code auth, `handle.cancel()` to
+stop an in-progress interactive login, and `logout()` to clear the active
+app-server account session.
+
 ## Docs map

 - Golden path tutorial: `docs/getting-started.md`
@@ -12,6 +12,10 @@ from openai_codex import (
    Codex,
    AsyncCodex,
    ApprovalMode,
+    ChatgptLoginHandle,
+    DeviceCodeLoginHandle,
+    AsyncChatgptLoginHandle,
+    AsyncDeviceCodeLoginHandle,
    RunResult,
    Thread,
    AsyncThread,
@@ -26,6 +30,11 @@ from openai_codex import (
    MentionInput,
 )
 from openai_codex.types import (
+    Account,
+    AccountLoginCompletedNotification,
+    CancelLoginAccountResponse,
+    CancelLoginAccountStatus,
+    GetAccountResponse,
    InitializeResponse,
    ThreadItem,
    ThreadTokenUsage,
@@ -47,6 +56,11 @@ 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`
@@ -82,6 +96,11 @@ 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]`
@@ -97,6 +116,30 @@ 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.
@@ -176,6 +219,11 @@ The SDK wrappers return and accept public app-server models wherever possible:

 ```python
 from openai_codex.types import (
+    Account,
+    AccountLoginCompletedNotification,
+    CancelLoginAccountResponse,
+    CancelLoginAccountStatus,
+    GetAccountResponse,
    ThreadReadResponse,
    Turn,
    TurnStatus,
@@ -23,6 +23,16 @@ Choose `run()` for most apps. Choose `stream()` for progress UIs, custom timeout

 If your app is not already async, stay with `Codex`.

+## How do I log in?
+
+- `login_api_key(...)` authenticates immediately with an API key.
+- `login_chatgpt()` starts browser login and returns a handle with `auth_url`.
+- `login_chatgpt_device_code()` starts device-code login and returns a handle
+  with `verification_url` and `user_code`.
+- Interactive handles expose `wait()` for the matching
+  `account/login/completed` notification and `cancel()` to stop that attempt.
+- `account()` reads the current account state, and `logout()` clears it.
+
 ## Public kwargs are snake_case

 Public API keyword names are snake_case. The SDK still maps them to wire camelCase under the hood.
@@ -56,7 +66,6 @@ Common causes:

 - published runtime package (`openai-codex-cli-bin`) is not installed
 - local `codex_bin` override points to a missing file
- local auth/session is missing
 - incompatible/old app-server

 ## Why does a turn "hang"?
@@ -19,9 +19,37 @@ Requirements:
 - Python `>=3.10`
 - uv
 - installed `openai-codex-cli-bin` runtime package, or an explicit `codex_bin` override
- local Codex auth/session configured

-## 2) Run your first turn (sync)
+## 2) Authenticate when needed
+
+Existing Codex auth state is reused automatically. To authenticate from the SDK,
+use the flow that fits your app:
+
+```python
+from openai_codex import Codex
+
+with Codex() as codex:
+    codex.login_api_key("sk-...")
+    account = codex.account()
+    print(account.account)
+```
+
+Interactive ChatGPT browser login returns a handle that carries the URL and the
+matching completion event:
+
+```python
+with Codex() as codex:
+    login = codex.login_chatgpt()
+    print(login.auth_url)
+    completed = login.wait()
+    print(completed.success)
+```
+
+Device-code login works the same way with
+`login_chatgpt_device_code()`, which exposes `verification_url`, `user_code`,
+and `wait()`.
+
+## 3) Run your first turn (sync)

 ```python
 from openai_codex import Codex
@@ -47,7 +75,7 @@ What happened:
 - use `thread.turn(...)` when you need a `TurnHandle` for streaming, steering, interrupting, or turn IDs/status
 - one client can consume multiple active turns concurrently; turn streams are routed by turn ID

-## 3) Continue the same thread (multi-turn)
+## 4) Continue the same thread (multi-turn)

 ```python
 from openai_codex import Codex
@@ -62,7 +90,7 @@ with Codex() as codex:
    print("second:", second.final_response)
 ```

-## 4) Async parity
+## 5) Async parity

 Use `async with AsyncCodex()` as the normal async entrypoint. `AsyncCodex`
 initializes lazily, and context entry makes startup/shutdown explicit.
@@ -82,7 +110,7 @@ async def main() -> None:
 asyncio.run(main())
 ```

-## 5) Resume an existing thread
+## 6) Resume an existing thread

 ```python
 from openai_codex import Codex
@@ -95,7 +123,7 @@ with Codex() as codex:
    print(result.final_response)
 ```

-## 6) Public app-server types
+## 7) Public app-server types

 The convenience wrappers live at the package root. Public app-server value and
 event types live under:
@@ -104,7 +132,7 @@ event types live under:
 from openai_codex.types import ThreadReadResponse, Turn, TurnStatus
 ```

-## 7) Next stops
+## 8) Next stops

 - API surface and signatures: `docs/api-reference.md`
 - Common decisions/pitfalls: `docs/faq.md`
@@ -0,0 +1,34 @@
+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 openai_codex import AsyncCodex
+
+
+async def main() -> None:
+    async with AsyncCodex(config=runtime_config()) as codex:
+        # Browser login returns a live handle. Open `auth_url` and await `wait()`
+        # in a real app; this example cancels immediately so it stays non-blocking.
+        login = await codex.login_chatgpt()
+        canceled = await login.cancel()
+        completed = await login.wait()
+        account = await codex.account()
+
+        print("login.id:", login.login_id)
+        print("login.auth_url:", login.auth_url)
+        print("login.cancel.status:", canceled.status)
+        print("login.completed.success:", completed.success)
+        print("account.requires_openai_auth:", account.requires_openai_auth)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
@@ -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 openai_codex import Codex
+
+with Codex(config=runtime_config()) as codex:
+    # Browser login returns a live handle. Open `auth_url` and call `wait()`
+    # in a real app; this example cancels immediately so it stays non-blocking.
+    login = codex.login_chatgpt()
+    canceled = login.cancel()
+    completed = login.wait()
+    account = codex.account()
+
+    print("login.id:", login.login_id)
+    print("login.auth_url:", login.auth_url)
+    print("login.cancel.status:", canceled.status)
+    print("login.completed.success:", completed.success)
+    print("account.requires_openai_auth:", account.requires_openai_auth)
@@ -82,3 +82,5 @@ python examples/01_quickstart_constructor/async.py
  - list models, pick highest model + highest supported reasoning effort, run turns, print message and usage
 - `14_turn_controls/`
  - separate best-effort `steer()` and `interrupt()` demos with concise summaries
+- `15_login_and_account/`
+  - browser-login handle lifecycle, cancellation, and account inspection
@@ -140,6 +140,27 @@
    ")\n"
   ]
  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# Cell 2b: browser login handle lifecycle\n",
+    "with Codex() as codex:\n",
+    "    # Open this URL and call `wait()` without canceling when completing login for real.\n",
+    "    login = codex.login_chatgpt()\n",
+    "    canceled = login.cancel()\n",
+    "    completed = login.wait()\n",
+    "    account = codex.account()\n",
+    "\n",
+    "    print('login.id:', login.login_id)\n",
+    "    print('login.auth_url:', login.auth_url)\n",
+    "    print('login.cancel.status:', canceled.status)\n",
+    "    print('login.completed.success:', completed.success)\n",
+    "    print('account.requires_openai_auth:', account.requires_openai_auth)\n"
+   ]
+  },
  {
   "cell_type": "code",
   "execution_count": null,
@@ -1,10 +1,14 @@
 from ._version import __version__
 from .api import (
    ApprovalMode,
+    AsyncChatgptLoginHandle,
    AsyncCodex,
+    AsyncDeviceCodeLoginHandle,
    AsyncThread,
    AsyncTurnHandle,
+    ChatgptLoginHandle,
    Codex,
+    DeviceCodeLoginHandle,
    ImageInput,
    Input,
    InputItem,
@@ -39,6 +43,10 @@ __all__ = [
    "Codex",
    "AsyncCodex",
    "ApprovalMode",
+    "ChatgptLoginHandle",
+    "DeviceCodeLoginHandle",
+    "AsyncChatgptLoginHandle",
+    "AsyncDeviceCodeLoginHandle",
    "Thread",
    "AsyncThread",
    "TurnHandle",
@@ -0,0 +1,172 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Protocol
+
+from .async_client import AsyncAppServerClient
+from .client import AppServerClient
+from .generated.v2_all import (
+    AccountLoginCompletedNotification,
+    CancelLoginAccountResponse,
+    ChatgptDeviceCodeLoginAccountParams,
+    ChatgptDeviceCodeLoginAccountResponse,
+    ChatgptLoginAccountParams,
+    ChatgptLoginAccountResponse,
+    LoginAccountParams,
+)
+
+
+class _AsyncLoginOwner(Protocol):
+    """Subset of AsyncCodex needed by async login handles."""
+
+    _client: AsyncAppServerClient
+
+    async def _ensure_initialized(self) -> None:
+        """Ensure the owning SDK client has a live app-server connection."""
+        ...
+
+
+def start_chatgpt_login(client: AppServerClient) -> ChatgptLoginHandle:
+    """Start browser ChatGPT login and return the handle for that attempt."""
+    response = client.account_login_start(
+        LoginAccountParams(
+            root=ChatgptLoginAccountParams(type="chatgpt"),
+        )
+    )
+    response_root = response.root
+    if not isinstance(response_root, ChatgptLoginAccountResponse):
+        raise RuntimeError(f"unexpected ChatGPT login response: {response_root!r}")
+    return ChatgptLoginHandle(
+        client,
+        response_root.login_id,
+        response_root.auth_url,
+    )
+
+
+async def async_start_chatgpt_login(owner: _AsyncLoginOwner) -> AsyncChatgptLoginHandle:
+    """Start async browser ChatGPT login and return that attempt's handle."""
+    response = await owner._client.account_login_start(
+        LoginAccountParams(
+            root=ChatgptLoginAccountParams(type="chatgpt"),
+        )
+    )
+    response_root = response.root
+    if not isinstance(response_root, ChatgptLoginAccountResponse):
+        raise RuntimeError(f"unexpected ChatGPT login response: {response_root!r}")
+    return AsyncChatgptLoginHandle(
+        owner,
+        response_root.login_id,
+        response_root.auth_url,
+    )
+
+
+def start_device_code_login(client: AppServerClient) -> DeviceCodeLoginHandle:
+    """Start device-code ChatGPT login and return the handle for that attempt."""
+    response = client.account_login_start(
+        LoginAccountParams(
+            root=ChatgptDeviceCodeLoginAccountParams(type="chatgptDeviceCode"),
+        )
+    )
+    response_root = response.root
+    if not isinstance(response_root, ChatgptDeviceCodeLoginAccountResponse):
+        raise RuntimeError(f"unexpected device-code login response: {response_root!r}")
+    return DeviceCodeLoginHandle(
+        client,
+        response_root.login_id,
+        response_root.verification_url,
+        response_root.user_code,
+    )
+
+
+async def async_start_device_code_login(
+    owner: _AsyncLoginOwner,
+) -> AsyncDeviceCodeLoginHandle:
+    """Start async device-code ChatGPT login and return that attempt's handle."""
+    response = await owner._client.account_login_start(
+        LoginAccountParams(
+            root=ChatgptDeviceCodeLoginAccountParams(type="chatgptDeviceCode"),
+        )
+    )
+    response_root = response.root
+    if not isinstance(response_root, ChatgptDeviceCodeLoginAccountResponse):
+        raise RuntimeError(f"unexpected device-code login response: {response_root!r}")
+    return AsyncDeviceCodeLoginHandle(
+        owner,
+        response_root.login_id,
+        response_root.verification_url,
+        response_root.user_code,
+    )
+
+
+@dataclass(slots=True)
+class ChatgptLoginHandle:
+    """Live browser-login attempt returned by `Codex.login_chatgpt()`."""
+
+    _client: AppServerClient
+    login_id: str
+    auth_url: str
+
+    def wait(self) -> AccountLoginCompletedNotification:
+        """Wait for this browser login attempt's completion notification."""
+        return self._client.wait_for_login_completed(self.login_id)
+
+    def cancel(self) -> CancelLoginAccountResponse:
+        """Cancel this browser login attempt."""
+        return self._client.account_login_cancel(self.login_id)
+
+
+@dataclass(slots=True)
+class DeviceCodeLoginHandle:
+    """Live device-code login attempt returned by `Codex.login_chatgpt_device_code()`."""
+
+    _client: AppServerClient
+    login_id: str
+    verification_url: str
+    user_code: str
+
+    def wait(self) -> AccountLoginCompletedNotification:
+        """Wait for this device-code login attempt's completion notification."""
+        return self._client.wait_for_login_completed(self.login_id)
+
+    def cancel(self) -> CancelLoginAccountResponse:
+        """Cancel this device-code login attempt."""
+        return self._client.account_login_cancel(self.login_id)
+
+
+@dataclass(slots=True)
+class AsyncChatgptLoginHandle:
+    """Live browser-login attempt returned by `AsyncCodex.login_chatgpt()`."""
+
+    _codex: _AsyncLoginOwner
+    login_id: str
+    auth_url: str
+
+    async def wait(self) -> AccountLoginCompletedNotification:
+        """Wait for this browser login attempt's completion notification."""
+        await self._codex._ensure_initialized()
+        return await self._codex._client.wait_for_login_completed(self.login_id)
+
+    async def cancel(self) -> CancelLoginAccountResponse:
+        """Cancel this browser login attempt."""
+        await self._codex._ensure_initialized()
+        return await self._codex._client.account_login_cancel(self.login_id)
+
+
+@dataclass(slots=True)
+class AsyncDeviceCodeLoginHandle:
+    """Live device-code attempt returned by `AsyncCodex.login_chatgpt_device_code()`."""
+
+    _codex: _AsyncLoginOwner
+    login_id: str
+    verification_url: str
+    user_code: str
+
+    async def wait(self) -> AccountLoginCompletedNotification:
+        """Wait for this device-code login attempt's completion notification."""
+        await self._codex._ensure_initialized()
+        return await self._codex._client.wait_for_login_completed(self.login_id)
+
+    async def cancel(self) -> CancelLoginAccountResponse:
+        """Cancel this device-code login attempt."""
+        await self._codex._ensure_initialized()
+        return await self._codex._client.account_login_cancel(self.login_id)
@@ -6,6 +6,7 @@ from collections import deque

 from .errors import AppServerError, map_jsonrpc_error
 from .generated.notification_registry import notification_turn_id
+from .generated.v2_all import AccountLoginCompletedNotification
 from .models import JsonValue, Notification, UnknownNotification

 ResponseQueueItem = JsonValue | BaseException
@@ -25,6 +26,8 @@ class MessageRouter:
        """Create empty response, turn, and global notification queues."""
        self._lock = threading.Lock()
        self._response_waiters: dict[str, queue.Queue[ResponseQueueItem]] = {}
+        self._login_notifications: dict[str, queue.Queue[NotificationQueueItem]] = {}
+        self._pending_login_notifications: dict[str, deque[Notification]] = {}
        self._turn_notifications: dict[str, queue.Queue[NotificationQueueItem]] = {}
        self._pending_turn_notifications: dict[str, deque[Notification]] = {}
        self._global_notifications: queue.Queue[NotificationQueueItem] = queue.Queue()
@@ -51,6 +54,36 @@ class MessageRouter:
            raise item
        return item

+    def register_login(self, login_id: str) -> None:
+        """Register a queue for one interactive login attempt."""
+
+        login_queue: queue.Queue[NotificationQueueItem] = queue.Queue()
+        with self._lock:
+            if login_id in self._login_notifications:
+                return
+            pending = self._pending_login_notifications.pop(login_id, deque())
+            self._login_notifications[login_id] = login_queue
+        for notification in pending:
+            login_queue.put(notification)
+
+    def unregister_login(self, login_id: str) -> None:
+        """Stop routing future notifications for one login attempt."""
+
+        with self._lock:
+            self._login_notifications.pop(login_id, None)
+
+    def next_login_notification(self, login_id: str) -> Notification:
+        """Block until the next notification for a registered login attempt."""
+
+        with self._lock:
+            login_queue = self._login_notifications.get(login_id)
+        if login_queue is None:
+            raise RuntimeError(f"login {login_id!r} is not registered for waiting")
+        item = login_queue.get()
+        if isinstance(item, BaseException):
+            raise item
+        return item
+
    def register_turn(self, turn_id: str) -> None:
        """Register a queue for a turn stream and replay early events."""

@@ -111,6 +144,18 @@ class MessageRouter:
    def route_notification(self, notification: Notification) -> None:
        """Deliver a notification to a turn queue or the global queue."""

+        login_id = self._notification_login_id(notification)
+        if login_id is not None:
+            with self._lock:
+                login_queue = self._login_notifications.get(login_id)
+                if login_queue is None:
+                    self._pending_login_notifications.setdefault(login_id, deque()).append(
+                        notification
+                    )
+                    return
+            login_queue.put(notification)
+            return
+
        turn_id = self._notification_turn_id(notification)
        if turn_id is None:
            self._global_notifications.put(notification)
@@ -132,16 +177,35 @@ class MessageRouter:
        with self._lock:
            response_waiters = list(self._response_waiters.values())
            self._response_waiters.clear()
+            login_queues = list(self._login_notifications.values())
+            self._login_notifications.clear()
+            self._pending_login_notifications.clear()
            turn_queues = list(self._turn_notifications.values())
            self._pending_turn_notifications.clear()
        # Put the same transport failure into every queue so no SDK call blocks
        # forever waiting for a response that cannot arrive.
        for waiter in response_waiters:
            waiter.put(exc)
+        for login_queue in login_queues:
+            login_queue.put(exc)
        for turn_queue in turn_queues:
            turn_queue.put(exc)
        self._global_notifications.put(exc)

+    def _notification_login_id(self, notification: Notification) -> str | None:
+        """Extract the login attempt id from completion notifications."""
+        if notification.method != "account/login/completed":
+            return None
+
+        payload = notification.payload
+        if isinstance(payload, AccountLoginCompletedNotification):
+            return payload.login_id
+        if isinstance(payload, UnknownNotification):
+            raw_login_id = payload.params.get("loginId")
+            if isinstance(raw_login_id, str):
+                return raw_login_id
+        return None
+
    def _notification_turn_id(self, notification: Notification) -> str | None:
        """Extract routing ids from known generated payloads or raw unknown payloads."""
        payload = notification.payload
@@ -22,6 +22,16 @@ from ._inputs import (
    _normalize_run_input,
    _to_wire_input,
 )
+from ._login import (
+    AsyncChatgptLoginHandle,
+    AsyncDeviceCodeLoginHandle,
+    ChatgptLoginHandle,
+    DeviceCodeLoginHandle,
+    async_start_chatgpt_login,
+    async_start_device_code_login,
+    start_chatgpt_login,
+    start_device_code_login,
+)
 from ._run import (
    RunResult,
    _collect_async_run_result,
@@ -30,6 +40,10 @@ from ._run import (
 from .async_client import AsyncAppServerClient
 from .client import AppServerClient, AppServerConfig
 from .generated.v2_all import (
+    ApiKeyLoginAccountParams,
+    GetAccountParams,
+    GetAccountResponse,
+    LoginAccountParams,
    ModelListResponse,
    Personality,
    ReasoningEffort,
@@ -85,6 +99,33 @@ class Codex:
    def close(self) -> None:
        self._client.close()

+    def login_api_key(self, api_key: str) -> None:
+        """Authenticate app-server with an API key."""
+        self._client.account_login_start(
+            LoginAccountParams(
+                root=ApiKeyLoginAccountParams(
+                    api_key=api_key,
+                    type="apiKey",
+                )
+            )
+        )
+
+    def login_chatgpt(self) -> ChatgptLoginHandle:
+        """Start browser-based ChatGPT login and return its live handle."""
+        return start_chatgpt_login(self._client)
+
+    def login_chatgpt_device_code(self) -> DeviceCodeLoginHandle:
+        """Start device-code ChatGPT login and return its live handle."""
+        return start_device_code_login(self._client)
+
+    def account(self, *, refresh_token: bool = False) -> GetAccountResponse:
+        """Read the current app-server account state."""
+        return self._client.account_read(GetAccountParams(refresh_token=refresh_token))
+
+    def logout(self) -> None:
+        """Clear the current app-server account session."""
+        self._client.account_logout()
+
    # BEGIN GENERATED: Codex.flat_methods
    def thread_start(
        self,
@@ -286,6 +327,38 @@ class AsyncCodex:
        self._init = None
        self._initialized = False

+    async def login_api_key(self, api_key: str) -> None:
+        """Authenticate app-server with an API key."""
+        await self._ensure_initialized()
+        await self._client.account_login_start(
+            LoginAccountParams(
+                root=ApiKeyLoginAccountParams(
+                    api_key=api_key,
+                    type="apiKey",
+                )
+            )
+        )
+
+    async def login_chatgpt(self) -> AsyncChatgptLoginHandle:
+        """Start browser-based ChatGPT login and return its live handle."""
+        await self._ensure_initialized()
+        return await async_start_chatgpt_login(self)
+
+    async def login_chatgpt_device_code(self) -> AsyncDeviceCodeLoginHandle:
+        """Start device-code ChatGPT login and return its live handle."""
+        await self._ensure_initialized()
+        return await async_start_device_code_login(self)
+
+    async def account(self, *, refresh_token: bool = False) -> GetAccountResponse:
+        """Read the current app-server account state."""
+        await self._ensure_initialized()
+        return await self._client.account_read(GetAccountParams(refresh_token=refresh_token))
+
+    async def logout(self) -> None:
+        """Clear the current app-server account session."""
+        await self._ensure_initialized()
+        await self._client.account_logout()
+
    # BEGIN GENERATED: AsyncCodex.flat_methods
    async def thread_start(
        self,
@@ -8,7 +8,14 @@ from pydantic import BaseModel

 from .client import AppServerClient, AppServerConfig
 from .generated.v2_all import (
+    AccountLoginCompletedNotification,
    AgentMessageDeltaNotification,
+    CancelLoginAccountResponse,
+    GetAccountParams as V2GetAccountParams,
+    GetAccountResponse,
+    LoginAccountParams as V2LoginAccountParams,
+    LoginAccountResponse,
+    LogoutAccountResponse,
    ModelListResponse,
    ThreadArchiveResponse,
    ThreadCompactStartResponse,
@@ -88,6 +95,14 @@ class AsyncAppServerClient:
        """Register a turn notification queue on the wrapped sync client."""
        self._sync.register_turn_notifications(turn_id)

+    def register_login_notifications(self, login_id: str) -> None:
+        """Register a login notification queue on the wrapped sync client."""
+        self._sync.register_login_notifications(login_id)
+
+    def unregister_login_notifications(self, login_id: str) -> None:
+        """Unregister a login notification queue on the wrapped sync client."""
+        self._sync.unregister_login_notifications(login_id)
+
    def unregister_turn_notifications(self, turn_id: str) -> None:
        """Unregister a turn notification queue on the wrapped sync client."""
        self._sync.unregister_turn_notifications(turn_id)
@@ -107,6 +122,28 @@ class AsyncAppServerClient:
            response_model=response_model,
        )

+    async def account_login_start(
+        self,
+        params: V2LoginAccountParams | JsonObject,
+    ) -> LoginAccountResponse:
+        """Start one account login attempt through the wrapped sync client."""
+        return await self._call_sync(self._sync.account_login_start, params)
+
+    async def account_login_cancel(self, login_id: str) -> CancelLoginAccountResponse:
+        """Cancel one active account login attempt through the wrapped sync client."""
+        return await self._call_sync(self._sync.account_login_cancel, login_id)
+
+    async def account_read(
+        self,
+        params: V2GetAccountParams | JsonObject | None = None,
+    ) -> GetAccountResponse:
+        """Read current account state through the wrapped sync client."""
+        return await self._call_sync(self._sync.account_read, params)
+
+    async def account_logout(self) -> LogoutAccountResponse:
+        """Clear the active account session through the wrapped sync client."""
+        return await self._call_sync(self._sync.account_logout)
+
    async def thread_start(
        self, params: V2ThreadStartParams | JsonObject | None = None
    ) -> ThreadStartResponse:
@@ -211,10 +248,21 @@ class AsyncAppServerClient:
        """Wait for the next global notification without blocking the event loop."""
        return await self._call_sync(self._sync.next_notification)

+    async def next_login_notification(self, login_id: str) -> Notification:
+        """Wait for the next notification routed to one login attempt."""
+        return await self._call_sync(self._sync.next_login_notification, login_id)
+
    async def next_turn_notification(self, turn_id: str) -> Notification:
        """Wait for the next notification routed to one turn."""
        return await self._call_sync(self._sync.next_turn_notification, turn_id)

+    async def wait_for_login_completed(
+        self,
+        login_id: str,
+    ) -> AccountLoginCompletedNotification:
+        """Wait for the completion notification routed to one login attempt."""
+        return await self._call_sync(self._sync.wait_for_login_completed, login_id)
+
    async def wait_for_turn_completed(self, turn_id: str) -> TurnCompletedNotification:
        """Wait for the completion notification routed to one turn."""
        return await self._call_sync(self._sync.wait_for_turn_completed, turn_id)
@@ -17,7 +17,16 @@ from ._version import __version__ as SDK_VERSION
 from .errors import AppServerError, TransportClosedError
 from .generated.notification_registry import NOTIFICATION_MODELS
 from .generated.v2_all import (
+    AccountLoginCompletedNotification,
    AgentMessageDeltaNotification,
+    CancelLoginAccountResponse,
+    ChatgptDeviceCodeLoginAccountResponse,
+    ChatgptLoginAccountResponse,
+    GetAccountParams as V2GetAccountParams,
+    GetAccountResponse,
+    LoginAccountParams as V2LoginAccountParams,
+    LoginAccountResponse,
+    LogoutAccountResponse,
    ModelListResponse,
    ThreadArchiveResponse,
    ThreadCompactStartResponse,
@@ -59,6 +68,8 @@ def _params_dict(
        | V2ThreadListParams
        | V2ThreadForkParams
        | V2TurnStartParams
+        | V2GetAccountParams
+        | V2LoginAccountParams
        | JsonObject
        | None
    ),
@@ -246,7 +257,10 @@ class AppServerClient:
        waiter = self._router.create_response_waiter(request_id)

        try:
-            self._write_message({"id": request_id, "method": method, "params": params or {}})
+            message: JsonObject = {"id": request_id, "method": method}
+            if params is not None:
+                message["params"] = params
+            self._write_message(message)
        except BaseException:
            self._router.discard_response_waiter(request_id)
            raise
@@ -258,12 +272,27 @@ class AppServerClient:

    def notify(self, method: str, params: JsonObject | None = None) -> None:
        """Send a JSON-RPC notification without waiting for a response."""
-        self._write_message({"method": method, "params": params or {}})
+        message: JsonObject = {"method": method}
+        if params is not None:
+            message["params"] = params
+        self._write_message(message)

    def next_notification(self) -> Notification:
        """Return the next notification that is not scoped to an active turn."""
        return self._router.next_global_notification()

+    def register_login_notifications(self, login_id: str) -> None:
+        """Start routing notifications for one interactive login attempt."""
+        self._router.register_login(login_id)
+
+    def unregister_login_notifications(self, login_id: str) -> None:
+        """Stop routing notifications for one interactive login attempt."""
+        self._router.unregister_login(login_id)
+
+    def next_login_notification(self, login_id: str) -> Notification:
+        """Return the next routed notification for the requested login id."""
+        return self._router.next_login_notification(login_id)
+
    def register_turn_notifications(self, turn_id: str) -> None:
        """Start routing notifications for one turn into its dedicated queue."""
        self._router.register_turn(turn_id)
@@ -276,6 +305,43 @@ class AppServerClient:
        """Return the next routed notification for the requested turn id."""
        return self._router.next_turn_notification(turn_id)

+    def account_login_start(
+        self,
+        params: V2LoginAccountParams | JsonObject,
+    ) -> LoginAccountResponse:
+        response = self.request(
+            "account/login/start",
+            _params_dict(params),
+            response_model=LoginAccountResponse,
+        )
+        response_root = response.root
+        if isinstance(
+            response_root,
+            ChatgptLoginAccountResponse | ChatgptDeviceCodeLoginAccountResponse,
+        ):
+            self.register_login_notifications(response_root.login_id)
+        return response
+
+    def account_login_cancel(self, login_id: str) -> CancelLoginAccountResponse:
+        return self.request(
+            "account/login/cancel",
+            {"loginId": login_id},
+            response_model=CancelLoginAccountResponse,
+        )
+
+    def account_read(
+        self,
+        params: V2GetAccountParams | JsonObject | None = None,
+    ) -> GetAccountResponse:
+        return self.request(
+            "account/read",
+            _params_dict(params),
+            response_model=GetAccountResponse,
+        )
+
+    def account_logout(self) -> LogoutAccountResponse:
+        return self.request("account/logout", None, response_model=LogoutAccountResponse)
+
    def thread_start(
        self, params: V2ThreadStartParams | JsonObject | None = None
    ) -> ThreadStartResponse:
@@ -417,6 +483,24 @@ class AppServerClient:
        finally:
            self.unregister_turn_notifications(turn_id)

+    def wait_for_login_completed(
+        self,
+        login_id: str,
+    ) -> AccountLoginCompletedNotification:
+        """Block until the matching interactive login attempt completes."""
+        self.register_login_notifications(login_id)
+        try:
+            while True:
+                notification = self.next_login_notification(login_id)
+                if (
+                    notification.method == "account/login/completed"
+                    and isinstance(notification.payload, AccountLoginCompletedNotification)
+                    and notification.payload.login_id == login_id
+                ):
+                    return notification.payload
+        finally:
+            self.unregister_login_notifications(login_id)
+
    def stream_text(
        self,
        thread_id: str,
@@ -3,8 +3,13 @@
 from __future__ import annotations

 from .generated.v2_all import (
+    Account,
+    AccountLoginCompletedNotification,
    ApprovalsReviewer,
    AskForApproval,
+    CancelLoginAccountResponse,
+    CancelLoginAccountStatus,
+    GetAccountResponse,
    ModelListResponse,
    Personality,
    PlanType,
@@ -35,8 +40,13 @@ from .generated.v2_all import (
 from .models import InitializeResponse, JsonObject, Notification

 __all__ = [
+    "Account",
+    "AccountLoginCompletedNotification",
    "ApprovalsReviewer",
    "AskForApproval",
+    "CancelLoginAccountResponse",
+    "CancelLoginAccountStatus",
+    "GetAccountResponse",
    "InitializeResponse",
    "JsonObject",
    "ModelListResponse",
@@ -206,10 +206,11 @@ class MockResponsesServer:
 class AppServerHarness:
    """Test fixture that points a pinned runtime app-server at MockResponsesServer."""

-    def __init__(self, tmp_path: Path) -> None:
+    def __init__(self, tmp_path: Path, *, requires_openai_auth: bool = False) -> None:
        self.tmp_path = tmp_path
        self.codex_home = tmp_path / "codex-home"
        self.workspace = tmp_path / "workspace"
+        self.requires_openai_auth = requires_openai_auth
        self.responses = MockResponsesServer()

    def __enter__(self) -> AppServerHarness:
@@ -238,6 +239,7 @@ class AppServerHarness:
    def _write_config(self) -> None:
        """Write config.toml that routes model calls to the mock server."""
        config_toml = self.codex_home / "config.toml"
+        requires_openai_auth = "requires_openai_auth = true\n" if self.requires_openai_auth else ""
        config_toml.write_text(
            f"""
 model = "mock-model"
@@ -252,6 +254,7 @@ base_url = "{self.responses.url}/v1"
 wire_api = "responses"
 request_max_retries = 0
 stream_max_retries = 0
+{requires_openai_auth}
 """.lstrip()
        )

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+import base64
+import json
+
+from app_server_harness import AppServerHarness
+
+from openai_codex import AppServerConfig, Codex
+from openai_codex.generated.v2_all import (
+    ChatgptAuthTokensLoginAccountParams,
+    LoginAccountParams,
+)
+
+
+def _app_server_config(harness: AppServerHarness) -> AppServerConfig:
+    """Build an isolated login config without inheriting ambient API-key auth."""
+    config = harness.app_server_config()
+    config.env = {**(config.env or {}), "OPENAI_API_KEY": ""}
+    return config
+
+
+def test_api_key_login_authenticates_follow_up_model_requests(tmp_path) -> None:
+    """API-key login should authorize the next Responses request with that key."""
+    with AppServerHarness(tmp_path, requires_openai_auth=True) as harness:
+        harness.responses.enqueue_assistant_message("api key auth", response_id="api-key-auth")
+
+        with Codex(config=_app_server_config(harness)) as codex:
+            codex.login_api_key("sk-sdk-login-test")
+            result = codex.thread_start().run("prove api key auth")
+            request = harness.responses.single_request()
+
+    assert {
+        "final_response": result.final_response,
+        "authorization": request.header("authorization"),
+    } == {
+        "final_response": "api key auth",
+        "authorization": "Bearer sk-sdk-login-test",
+    }
+
+
+def test_chatgpt_token_login_authenticates_follow_up_model_requests(tmp_path) -> None:
+    """ChatGPT token handoff should authorize later Responses requests with that token."""
+    account_id = "workspace-sdk-chatgpt"
+
+    def _encode(payload: dict[str, object]) -> str:
+        raw = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode("utf-8")
+        return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")
+
+    # App-server parses claims from the access token before persisting external ChatGPT auth.
+    header = _encode({"alg": "none", "typ": "JWT"})
+    claims = _encode(
+        {
+            "email": "sdk-chatgpt@example.com",
+            "https://api.openai.com/auth": {
+                "chatgpt_account_id": account_id,
+                "chatgpt_plan_type": "pro",
+            },
+        }
+    )
+    access_token = f"{header}.{claims}.sig"
+
+    with AppServerHarness(tmp_path, requires_openai_auth=True) as harness:
+        harness.responses.enqueue_assistant_message(
+            "chatgpt token auth",
+            response_id="chatgpt-token-auth",
+        )
+
+        with Codex(config=_app_server_config(harness)) as codex:
+            login = codex._client.account_login_start(
+                LoginAccountParams(
+                    root=ChatgptAuthTokensLoginAccountParams(
+                        access_token=access_token,
+                        chatgpt_account_id=account_id,
+                        chatgpt_plan_type="pro",
+                        type="chatgptAuthTokens",
+                    )
+                )
+            )
+            result = codex.thread_start().run("prove chatgpt token auth")
+            request = harness.responses.single_request()
+
+    assert {
+        "login_type": login.root.type,
+        "final_response": result.final_response,
+        "authorization": request.header("authorization"),
+    } == {
+        "login_type": "chatgptAuthTokens",
+        "final_response": "chatgpt token auth",
+        "authorization": f"Bearer {access_token}",
+    }
@@ -27,6 +27,10 @@ EXPECTED_ROOT_EXPORTS = [
    "Codex",
    "AsyncCodex",
    "ApprovalMode",
+    "ChatgptLoginHandle",
+    "DeviceCodeLoginHandle",
+    "AsyncChatgptLoginHandle",
+    "AsyncDeviceCodeLoginHandle",
    "Thread",
    "AsyncThread",
    "TurnHandle",
@@ -55,8 +59,13 @@ EXPECTED_ROOT_EXPORTS = [
 ]

 EXPECTED_TYPES_EXPORTS = [
+    "Account",
+    "AccountLoginCompletedNotification",
    "ApprovalsReviewer",
    "AskForApproval",
+    "CancelLoginAccountResponse",
+    "CancelLoginAccountStatus",
+    "GetAccountResponse",
    "InitializeResponse",
    "JsonObject",
    "ModelListResponse",
@@ -434,7 +434,7 @@ def test_notebook_sync_cell_smoke(runtime_env: PreparedRuntimeEnv) -> None:
        [
            _notebook_cell_source(1),
            _notebook_cell_source(2),
-            _notebook_cell_source(3),
+            _notebook_cell_source(4),
        ]
    )
    result = _run_python(runtime_env, source, timeout_s=240)
@@ -450,7 +450,7 @@ def test_notebook_advanced_cell_smoke(runtime_env: PreparedRuntimeEnv) -> None:
        [
            _notebook_cell_source(1),
            _notebook_cell_source(2),
-            _notebook_cell_source(7),
+            _notebook_cell_source(8),
        ]
    )
    result = _run_python(runtime_env, source, timeout_s=360)