### Run Quickstart Example

Source: https://github.com/cheolwanpark/codex-client/blob/main/typescript/examples/README.md

Execute the quickstart example using pnpm. This command builds the examples and runs the compiled TypeScript code.

```bash
pnpm example:quickstart
```

--------------------------------

### Quick Start Session with Auto-Accept

Source: https://github.com/cheolwanpark/codex-client/blob/main/python/README.md

This example demonstrates the quickest way to get started by creating a session, starting an ephemeral thread, and asking a question with automatic approval of any necessary actions. It's suitable for simple, non-interactive use cases.

```python
from __future__ import annotations

import asyncio

from codex_client import ApprovalPolicy, Session, client_info


async def main() -> None:
    async with await Session.create(
        client_info=client_info("my-app", "0.1.0"),
        approval_policy=ApprovalPolicy.auto_accept(),
    ) as session:
        thread = await session.start_ephemeral_thread()
        answer = await thread.ask("Reply with exactly OK.")
        print(answer.strip())


if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Run Python Examples

Source: https://github.com/cheolwanpark/codex-client/blob/main/python/examples/README.md

Execute the provided Python example scripts using `uv run`. Ensure the `codex` CLI is installed and authenticated.

```bash
uv run python examples/01_quickstart.py
```

```bash
uv run python examples/02_streaming_turn.py
```

```bash
uv run python examples/03_resume_thread.py
```

```bash
uv run python examples/04_multi_session.py
```

```bash
uv run python examples/05_approval_policy.py
```

```bash
uv run python examples/06_low_level_client.py
```

--------------------------------

### Run Resume Example

Source: https://github.com/cheolwanpark/codex-client/blob/main/typescript/examples/README.md

Execute the resume example using pnpm. This command builds the examples and runs the compiled TypeScript code.

```bash
pnpm example:resume
```

--------------------------------

### Run Approval Example

Source: https://github.com/cheolwanpark/codex-client/blob/main/typescript/examples/README.md

Execute the approval example using pnpm. This command builds the examples and runs the compiled TypeScript code.

```bash
pnpm example:approval
```

--------------------------------

### Run Weather Assistant Example (Interactive)

Source: https://github.com/cheolwanpark/codex-client/blob/main/src/examples/README.md

Launches the weather assistant demo in interactive mode, allowing conversational weather queries. Requires Node.js for MCP demos.

```bash
uv run weather/main.py --interactive
```

--------------------------------

### Run Low-Level Client Example

Source: https://github.com/cheolwanpark/codex-client/blob/main/typescript/examples/README.md

Execute the low-level client example using pnpm. This command builds the examples and runs the compiled TypeScript code.

```bash
pnpm example:low-level
```

--------------------------------

### Run Streaming Example

Source: https://github.com/cheolwanpark/codex-client/blob/main/typescript/examples/README.md

Execute the streaming example using pnpm. This command builds the examples and runs the compiled TypeScript code.

```bash
pnpm example:streaming
```

--------------------------------

### Install Dependencies and Authenticate

Source: https://github.com/cheolwanpark/codex-client/blob/main/src/examples/README.md

Installs project dependencies using uv sync and authenticates the Codex CLI. Ensure Python 3.11+ and uv are installed.

```bash
cd src/examples
uv sync

uv run codex-client login
```

--------------------------------

### Install Dependencies

Source: https://github.com/cheolwanpark/codex-client/blob/main/typescript/README.md

Installs project dependencies using pnpm. Ensure Node.js 20+ and pnpm are installed.

```bash
cd typescript
pnpm install
```

--------------------------------

### Start a turn and get a one-shot answer with thread.ask()

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

Use `thread.ask()` for simple prompts where you need a single, final text response. It automatically handles session setup and thread management for ephemeral threads. Supports both string prompts and lists of user inputs.

```python
import asyncio
from codex_client import ApprovalPolicy, Session, client_info

async def main() -> None:
    async with await Session.create(
        client_info=client_info("my-app", "1.0.0"),
        approval_policy=ApprovalPolicy.auto_accept(),
    ) as session:
        thread = await session.start_ephemeral_thread()

        # String shorthand — auto-converted to TextUserInput
        answer = await thread.ask("What is 2 + 2? Reply with just the number.")
        print(answer.strip())  # 4

        # Multi-part input using explicit UserInput list
        from codex_client import text_input
        answer2 = await thread.ask(
            [text_input("What did I just ask you?")],
        )
        print(answer2.strip())

asyncio.run(main())
```

--------------------------------

### Quick Start Session Usage

Source: https://github.com/cheolwanpark/codex-client/blob/main/typescript/README.md

Demonstrates basic usage of the Session runtime to create a session, start an ephemeral thread, and ask a question. Ensure a local Codex CLI session is authenticated.

```typescript
import { Session, clientInfo } from "@codex-client/typescript";

const session = await Session.create({
  clientInfo: clientInfo("my-app", "0.1.0"),
});

try {
  const thread = await session.startEphemeralThread();
  const answer = await thread.ask("Reply with exactly OK.");
  console.log(answer.trim());
} finally {
  await session.close();
}
```

--------------------------------

### Run Weather Assistant Example (Scripted)

Source: https://github.com/cheolwanpark/codex-client/blob/main/src/examples/README.md

Executes the weather assistant demo with scripted scenarios, including current weather, forecasts, comparisons, travel planning, and state management. Requires Node.js for MCP demos.

```bash
uv run weather/main.py
```

--------------------------------

### Run Interactive Chat Example

Source: https://github.com/cheolwanpark/codex-client/blob/main/src/examples/README.md

Launches the interactive chat demo, which supports multi-turn conversations with streaming responses, reasoning traces, and command execution. Try prompts like 'Introduce yourself' or 'Create a fibonacci function'.

```bash
uv run interactive/main.py
```

--------------------------------

### Development Commands

Source: https://github.com/cheolwanpark/codex-client/blob/main/python/README.md

Commands for setting up the development environment, generating the protocol client, and running tests. Ensure you have a local 'codex' CLI installation for integration tests.

```bash
uv sync --dev
```

```bash
uv run python scripts/generate_protocol_client.py
```

```bash
uv run pytest -m "not integration"
```

```bash
uv run pytest -m integration
```

--------------------------------

### Run MCP Transport Example

Source: https://github.com/cheolwanpark/codex-client/blob/main/src/examples/README.md

Tests MCP server connectivity using stdio and HTTP transports. This example automatically runs connectivity tests and basic tool operations with the Everything MCP Server.

```bash
uv run mcp/main.py
```

--------------------------------

### Development Workflow

Source: https://github.com/cheolwanpark/codex-client/blob/main/typescript/README.md

Commands for local development, including code generation, example building, testing, and type checking. Ensure the 'codex' CLI is available for integration tests.

```bash
pnpm generate
pnpm examples:build
pnpm test
pnpm typecheck
pnpm build
```

--------------------------------

### Start a Managed Session with Session.create()

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

Use Session.create() to spawn the codex app-server, perform the initialize/initialized handshake, and obtain a Session object. Provide an ApprovalPolicy; auto_decline is the default if none is specified. Custom launch options like command, args, cwd, and env can also be configured.

```python
import asyncio
from codex_client import ApprovalPolicy, Session, client_info, thread_params

async def main() -> None:
    async with await Session.create(
        client_info=client_info("my-app", "1.0.0"),
        approval_policy=ApprovalPolicy.auto_accept(),
        # Optional: custom launch options
        command="codex",
        args=("app-server",),
        cwd="/path/to/workdir",
        env={"CODEX_LOG": "debug"},
    ) as session:
        # session.client exposes the underlying TypedCodexClient
        models = await session.list_models()
        print([m["id"] for m in models.get("models", [])])

asyncio.run(main())
```

--------------------------------

### Stream Turn Events for Incremental Output

Source: https://github.com/cheolwanpark/codex-client/blob/main/python/README.md

This example demonstrates how to stream turn events for incremental output, plan updates, and final turn status. Use `start_turn` when you need to process agent messages as they arrive or monitor the turn's progress.

```python
from __future__ import annotations

import asyncio

from codex_client import ApprovalPolicy, Session, TurnEventType, client_info


async def main() -> None:
    async with await Session.create(
        client_info=client_info("my-app", "0.1.0"),
        approval_policy=ApprovalPolicy.auto_accept(),
    ) as session:
        thread = await session.start_ephemeral_thread()
        turn = await thread.start_turn("Explain this SDK in one paragraph.")

        async for event in turn:
            if event["type"] == TurnEventType.AGENT_MESSAGE_DELTA:
                print(event["delta"], end="", flush=True)
            elif event["type"] == TurnEventType.PLAN_UPDATED:
                print("\n[plan updated]")
                for step in event["plan"]:
                    print(step["status"], step["step"])
            elif event["type"] == TurnEventType.COMPLETED:
                print(f"\nstatus={event['turn']['status']}")


if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Basic Ask Flow with Session

Source: https://github.com/cheolwanpark/codex-client/blob/main/python/README.md

This snippet shows a basic interaction flow using the runtime layer. It creates a session, starts a thread, and asks a question, printing the stripped answer. Use this when you need a straightforward prompt-to-answer mechanism.

```python
from __future__ import annotations

import asyncio

from codex_client import ApprovalPolicy, Session, client_info


async def main() -> None:
    async with await Session.create(
        client_info=client_info("my-app", "0.1.0"),
        approval_policy=ApprovalPolicy.auto_accept(),
    ) as session:
        thread = await session.start_ephemeral_thread()
        answer = await thread.ask("Summarize this repository in one sentence.")
        print(answer.strip())


if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### thread.ask()

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

Initiates a one-shot prompt to the agent, waits for the completion, and returns the final concatenated agent message text. This is the most straightforward method for getting a single answer from a prompt.

```APIDOC
## thread.ask()

### Description
Starts a turn, waits for completion, and returns the concatenated `agentMessage` text. It is the simplest path from prompt to answer.

### Method Signature
`await thread.ask(prompt: Union[str, List[UserInput]]) -> str`

### Parameters
* **prompt** (Union[str, List[UserInput]]) - The user's input, which can be a string or a list of `UserInput` objects.

### Request Example
```python
# String shorthand
answer = await thread.ask("What is 2 + 2? Reply with just the number.")

# Multi-part input
from codex_client import text_input
answer2 = await thread.ask([
    text_input("What did I just ask you?")
])
```

### Response
* **answer** (str) - The agent's response text.
```

--------------------------------

### Create Ephemeral and Persistent Threads

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

Start an ephemeral thread that is discarded when the session ends, or a persistent thread that survives across sessions. Use thread_params() to construct the ThreadStartParams dictionary with options like ephemeral, model, cwd, and approval_policy.

```python
import asyncio
from codex_client import ApprovalPolicy, Session, client_info, thread_params

async def main() -> None:
    async with await Session.create(
        client_info=client_info("my-app", "1.0.0"),
        approval_policy=ApprovalPolicy.auto_accept(),
    ) as session:
        # Ephemeral thread (not stored after session close)
        ephemeral = await session.start_ephemeral_thread()
        print(f"ephemeral id: {ephemeral.id}, status: {ephemeral.status}")

        # Persistent thread with a specific model and working directory
        persistent = await session.start_thread(
            thread_params(
                ephemeral=False,
                model="o4-mini",
                cwd="/tmp/project",
                approval_policy="unless-allow-listed",
            )
        )
        print(f"persistent id: {persistent.id}")

asyncio.run(main())
```

--------------------------------

### Iterate over turn events with thread.start_turn()

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

Use `thread.start_turn()` to get a `Turn` object, which is an async iterable of `TurnEvent` dictionaries. This allows for processing streaming responses, plan updates, command outputs, and errors as they occur. Requires explicit session and thread creation.

```python
import asyncio
from codex_client import ApprovalPolicy, Session, TurnEventType, client_info, turn_options

async def main() -> None:
    async with await Session.create(
        client_info=client_info("my-app", "1.0.0"),
        approval_policy=ApprovalPolicy.auto_accept(),
    ) as session:
        thread = await session.start_ephemeral_thread()
        turn = await thread.start_turn(
            "List three prime numbers.",
            options=turn_options(model="o4-mini", effort="low"),
        )

        async for event in turn:
            match event["type"]:
                case TurnEventType.AGENT_MESSAGE_DELTA:
                    print(event["delta"], end="", flush=True)
                case TurnEventType.PLAN_UPDATED:
                    for step in event["plan"]:
                        print(f"\n[plan] {step['status']} {step['step']}")
                case TurnEventType.COMMAND_OUTPUT_DELTA:
                    print(f"\n[cmd] {event['delta']}", end="")
                case TurnEventType.COMPLETED:
                    print(f"\n[done] status={event['turn']['status']}")
                case TurnEventType.ERROR:
                    print(f"\n[error] {event['error']}")

asyncio.run(main())
```

--------------------------------

### ThreadStartParams and TurnOptions

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

Demonstrates how to configure thread parameters and turn options for the codex client.

```python
params = thread_params(
    ephemeral=False,
    model="o4-mini",
    cwd="/tmp/repo",
    approval_policy="unless-allow-listed",
    base_instructions="You are a code review assistant.",
    sandbox="read-only",
)

opts = turn_options(
    model="o4-mini",
    effort="high",
    cwd="/tmp/repo",
    summary="detailed",
)
```

--------------------------------

### Session.create()

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

Spawns `codex app-server`, performs the `initialize`/`initialized` handshake, and returns a context-manager-compatible `Session`. An `ApprovalPolicy` must be provided.

```APIDOC
## Session.create()

### Description
Spawns `codex app-server` over stdio (or accepts a custom `Transport`), runs the `initialize`/`initialized` handshake, and returns a context-manager-compatible `Session`. An `ApprovalPolicy` must be provided; the default when none is given is `auto_decline`.

### Method
`Session.create()`

### Parameters
- `client_info` (dict) - Information about the client application.
- `approval_policy` (ApprovalPolicy) - The policy for handling approval requests.
- `command` (str, optional) - The command to launch the app-server.
- `args` (tuple, optional) - Arguments for the app-server command.
- `cwd` (str, optional) - The working directory for the app-server.
- `env` (dict, optional) - Environment variables for the app-server.

### Request Example
```python
import asyncio
from codex_client import ApprovalPolicy, Session, client_info

async def main() -> None:
    async with await Session.create(
        client_info=client_info("my-app", "1.0.0"),
        approval_policy=ApprovalPolicy.auto_accept(),
        command="codex",
        args=("app-server",),
        cwd="/path/to/workdir",
        env={"CODEX_LOG": "debug"},
    ) as session:
        # Use the session
        pass

asyncio.run(main())
```

### Response
- `Session` object - A context manager for the session.
```

--------------------------------

### Use Low-Level Typed Client Directly

Source: https://github.com/cheolwanpark/codex-client/blob/main/python/README.md

Demonstrates direct control over requests and notifications using `TypedCodexClient`. This approach bypasses the runtime wrapper for granular control.

```python
from __future__ import annotations

from codex_client import (
    NotificationMethod,
    StdioTransport,
    TypedCodexClient,
    client_info,
    text_input,
)


async def main() -> None:
    client = TypedCodexClient.from_transport(StdioTransport())
    chunks: list[str] = []

    client.on_notification(
        NotificationMethod.ITEM_AGENT_MESSAGE_DELTA,
        lambda params: chunks.append(params["delta"]),
    )

    try:
        await client.initialize({"clientInfo": client_info("my-app", "0.1.0")})
        await client.send_initialized()

        thread = await client.thread_start({"ephemeral": True})
        thread_id = thread["thread"]["id"]

        await client.turn_start(
            {
                "threadId": thread_id,
                "input": [text_input("Reply with exactly OK.")],
            }
        )

        result = await client.wait_for_notification(
            NotificationMethod.TURN_COMPLETED,
            timeout=60.0,
        )
        print(result["turn"]["status"])
        print("".join(chunks).strip())
    finally:
        await client.close()


if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Customize Approval Policy

Source: https://github.com/cheolwanpark/codex-client/blob/main/python/README.md

Illustrates how to implement a custom approval policy for command execution. The `on_command_execution` function is called to approve or decline commands.

```python
from __future__ import annotations

import asyncio

from codex_client import ApprovalPolicy, Session, approve_command, client_info
from codex_client.protocol_types import (
    CommandExecutionRequestApprovalParams,
    CommandExecutionRequestApprovalResponse,
)


async def on_command_execution(
    params: CommandExecutionRequestApprovalParams,
) -> CommandExecutionRequestApprovalResponse:
    print(f"command approval requested for {params['itemId']}")
    return approve_command()


async def main() -> None:
    policy = ApprovalPolicy.custom(on_command_execution=on_command_execution)

    async with await Session.create(
        client_info=client_info("my-app", "0.1.0"),
        approval_policy=policy,
    ) as session:
        thread = await session.start_ephemeral_thread()
        print((await thread.ask("Reply with exactly OK.")).strip())


if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Sync Dependencies

Source: https://github.com/cheolwanpark/codex-client/blob/main/src/examples/README.md

Synchronizes project dependencies using uv. Run this if you encounter 'ModuleNotFoundError: No module named 'codex_client''.

```bash
uv sync
```

--------------------------------

### Utility Helpers for Protocol Payloads

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

Helper functions like `client_info()`, `text_input()`, `thread_params()`, and `turn_options()` reduce boilerplate and improve readability when constructing protocol payloads.

```python
from codex_client import (
    approve_command, approve_file_change, approve_file_change_for_session,
    client_info, decline_command, decline_file_change,
    text_input, thread_params, tool_answers, tool_call_failure, tool_call_success, turn_options,
)

# ClientInfo dict
info = client_info("my-app", "2.0.0", title="My Application")
# {"name": "my-app", "version": "2.0.0", "title": "My Application"}

# TextUserInput dict
inp = text_input("Summarize the current directory in one sentence.")
# {"type": "text", "text": "Summarize..."}

```

--------------------------------

### Codex Client Constants

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

Shows how to use string enums for notification methods, server request methods, and turn event types.

```python
from codex_client import NotificationMethod, ServerRequestMethod, TurnEventType

# NotificationMethod — server-to-client notification method strings
print(NotificationMethod.TURN_COMPLETED)            # "turn/completed"
print(NotificationMethod.ITEM_AGENT_MESSAGE_DELTA)  # "item/agentMessage/delta"
print(NotificationMethod.TURN_PLAN_UPDATED)         # "turn/plan/updated"
print(NotificationMethod.ITEM_COMMAND_EXECUTION_OUTPUT_DELTA)  # "item/commandExecution/outputDelta"

# ServerRequestMethod — server requests that expect a host response
print(ServerRequestMethod.ITEM_COMMAND_EXECUTION_REQUEST_APPROVAL)  # "item/commandExecution/requestApproval"
print(ServerRequestMethod.ITEM_FILE_CHANGE_REQUEST_APPROVAL)        # "item/fileChange/requestApproval"
print(ServerRequestMethod.ITEM_TOOL_REQUEST_USER_INPUT)             # "item/tool/requestUserInput"
print(ServerRequestMethod.ITEM_TOOL_CALL)                           # "item/tool/call"

# TurnEventType — event["type"] values in Turn async iteration
print(TurnEventType.AGENT_MESSAGE_DELTA)  # "agent_message_delta"
print(TurnEventType.PLAN_UPDATED)         # "plan_updated"
print(TurnEventType.COMPLETED)            # "completed"
print(TurnEventType.ERROR)                # "error"
```

--------------------------------

### Execute shell commands with session.exec()

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

Use `session.exec()` to run shell commands within the app-server's sandboxed environment. Specify the command, working directory, and a timeout. The result includes output and exit code.

```python
import asyncio
from codex_client import ApprovalPolicy, Session, client_info

async def main() -> None:
    async with await Session.create(
        client_info=client_info("my-app", "1.0.0"),
        approval_policy=ApprovalPolicy.auto_accept(),
    ) as session:
        result = await session.exec(
            ["git", "log", "--oneline", "-5"],
            cwd="/path/to/repo",
            timeout_ms=10_000,
        )
        print(result.get("output", ""))
        print(f"exit code: {result.get('exitCode')}")

asyncio.run(main())
```

--------------------------------

### thread.start_turn()

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

Initiates a turn in a thread and returns a `Turn` object, which is an asynchronous iterable of `TurnEvent` dictionaries. This method allows for streaming responses and observing intermediate states like plan updates and command outputs.

```APIDOC
## thread.start_turn()

### Description
Starts a turn, returning a `Turn` that is an async iterable of `TurnEvent` dicts. Iteration is single-use. Use `TurnEventType` constants to identify events.

### Method Signature
`async def start_turn(prompt: Union[str, List[UserInput]], *, options: TurnOptions = None) -> Turn`

### Parameters
* **prompt** (Union[str, List[UserInput]]) - The user's input for the turn.
* **options** (TurnOptions, optional) - Options for the turn, such as specifying the model or effort level.

### Request Example
```python
from codex_client import TurnEventType, turn_options

turn = await thread.start_turn(
    "List three prime numbers.",
    options=turn_options(model="o4-mini", effort="low")
)

async for event in turn:
    match event["type"]:
        case TurnEventType.AGENT_MESSAGE_DELTA:
            print(event["delta"], end="", flush=True)
        case TurnEventType.PLAN_UPDATED:
            # ... handle plan updates
        case TurnEventType.COMMAND_OUTPUT_DELTA:
            # ... handle command output
        case TurnEventType.COMPLETED:
            print(f"\n[done] status={event['turn']['status']}")
        case TurnEventType.ERROR:
            print(f"\n[error] {event['error']}")
```

### Response
* **Turn** - An asynchronous iterable yielding `TurnEvent` dictionaries.
```

--------------------------------

### TypedCodexClient for Direct Protocol Access

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

Use `TypedCodexClient` for granular control over protocol interactions. Manually manage client lifecycle and register handlers for specific notifications and server requests.

```python
import asyncio
from codex_client import (
    NotificationMethod, ServerRequestMethod,
    StdioTransport, TypedCodexClient,
    client_info, text_input,
)

async def main() -> None:
    client = TypedCodexClient.from_transport(StdioTransport())
    deltas: list[str] = []

    # Register notification handlers (returns unsubscribe callable)
    unsubscribe = client.on_notification(
        NotificationMethod.ITEM_AGENT_MESSAGE_DELTA,
        lambda params: deltas.append(params["delta"]),
    )

    # Register server request handler
    client.on_server_request(
        ServerRequestMethod.ITEM_COMMAND_EXECUTION_REQUEST_APPROVAL,
        lambda _params: {"decision": "accept"},
    )

    try:
        await client.initialize({"clientInfo": client_info("my-app", "1.0.0")})
        await client.send_initialized()

        thread = await client.thread_start({"ephemeral": True})
        thread_id = thread["thread"]["id"]

        await client.turn_start({
            "threadId": thread_id,
            "input": [text_input("Reply with exactly OK.")],
        })

        # Wait for a specific notification with optional predicate and timeout
        completed = await client.wait_for_notification(
            NotificationMethod.TURN_COMPLETED,
            predicate=lambda p: p["turn"]["status"] == "completed",
            timeout=60.0,
        )
        print(f"status: {completed['turn']['status']}")
        print(f"text: {''.join(deltas).strip()}")
    finally:
        unsubscribe()
        await client.close()

asyncio.run(main())

```

--------------------------------

### Manage threads with fork(), rollback(), and archive()

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

Provides methods for thread lifecycle management. `fork()` creates a copy of the thread, `rollback()` reverts the thread to a previous state, and `archive()` marks the thread as inactive. Requires explicit session and thread creation.

```python
import asyncio
from codex_client import ApprovalPolicy, Session, client_info, thread_params

async def main() -> None:
    async with await Session.create(
        client_info=client_info("my-app", "1.0.0"),
        approval_policy=ApprovalPolicy.auto_accept(),
    ) as session:
        main_thread = await session.start_thread(thread_params(ephemeral=False))
        await main_thread.ask("Reply with exactly MAIN.")

        # Fork the thread at current state
        forked = await main_thread.fork()
        print(f"forked thread id: {forked.id}")
        await forked.ask("Reply with exactly FORK.")

        # Rollback main thread by 1 turn
        await main_thread.rollback(1)
        print(f"main thread turns after rollback: {len(main_thread.data.get('turns', []))}")

        # Set a human-readable name
        await main_thread.set_name("My Experiment Thread")
        print(f"thread name: {main_thread.name}")

        # Archive when done
        await main_thread.archive()

asyncio.run(main())
```

--------------------------------

### TypedCodexClient.use() - Middleware Pipeline

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

Attach middleware for logging, tracing, replay, or approval instrumentation. Each middleware receives a MiddlewareContext and a next callable.

```APIDOC
## `TypedCodexClient.use()` — middleware pipeline

### Description
Attach middleware for logging, tracing, replay, or approval instrumentation. Each middleware receives a `MiddlewareContext` and a `next` callable.

### Example Usage
```python
import asyncio
from codex_client import (
    ApprovalPolicy, Middleware, MiddlewareContext, Session, client_info,
)

async def logging_middleware(ctx: MiddlewareContext, next) -> None:
    direction = ctx.direction  # "incoming" | "outgoing"
    method = ctx.method or "(response)"
    if ctx.direction == "outgoing":
        print(f"→ {method}")
    await next()
    if ctx.direction == "incoming":
        print(f"← {method}")

async def main() -> None:
    async with await Session.create(
        client_info=client_info("my-app", "1.0.0"),
        approval_policy=ApprovalPolicy.auto_accept(),
        middleware=[logging_middleware],
    ) as session:
        thread = await session.start_ephemeral_thread()
        print((await thread.ask("Reply with exactly OK.")).strip())

asyncio.run(main())
```
```

--------------------------------

### Parallel Sessions with Codex Client

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

Demonstrates running multiple independent sessions concurrently using the `Session` object.

```python
import asyncio
from codex_client import ApprovalPolicy, Session, client_info

async def ask_once(prompt: str, label: str) -> str:
    async with await Session.create(
        client_info=client_info("my-app", "1.0.0"),
        approval_policy=ApprovalPolicy.auto_accept(),
    ) as session:
        thread = await session.start_ephemeral_thread()
        return (await thread.ask(prompt)).strip()

async def main() -> None:
    results = await asyncio.gather(
        ask_once("Reply with exactly A.", "session-a"),
        ask_once("Reply with exactly B.", "session-b"),
        ask_once("Reply with exactly C.", "session-c"),
    )
    print(results)  # ['A', 'B', 'C']

asyncio.run(main())
```

--------------------------------

### Run Multiple Independent Sessions

Source: https://github.com/cheolwanpark/codex-client/blob/main/python/README.md

Shows how to run multiple independent sessions concurrently using asyncio.gather. Each session manages its own connection and thread lifecycle.

```python
import asyncio

from codex_client import ApprovalPolicy, Session, client_info


async def run_once(prompt: str) -> str:
    async with await Session.create(
        client_info=client_info("my-app", "0.1.0"),
        approval_policy=ApprovalPolicy.auto_accept(),
    ) as session:
        thread = await session.start_ephemeral_thread()
        return (await thread.ask(prompt)).strip()


async def main() -> None:
    answers = await asyncio.gather(
        run_once("Reply with exactly SESSION_A."),
        run_once("Reply with exactly SESSION_B."),
    )
    print(answers)


if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Custom ApprovalPolicy with Event Handlers

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

Implement custom logic for approving or declining server requests like commands, file changes, and tool inputs. Requires defining async handler functions for each event type.

```python
import asyncio
from codex_client import (
    ApprovalPolicy, Session, approve_command, approve_file_change_for_session,
    client_info, tool_answers, tool_call_success,
)
from codex_client.protocol_types import (
    CommandExecutionRequestApprovalParams, CommandExecutionRequestApprovalResponse,
    DynamicToolCallParams, DynamicToolCallResponse,
    FileChangeRequestApprovalParams, FileChangeRequestApprovalResponse,
    ToolRequestUserInputParams, ToolRequestUserInputResponse,
)

async def on_command(params: CommandExecutionRequestApprovalParams) -> CommandExecutionRequestApprovalResponse:
    print(f"[approval] command item={params['itemId']}")
    return approve_command()  # {"decision": "accept"}

async def on_file(params: FileChangeRequestApprovalParams) -> FileChangeRequestApprovalResponse:
    print(f"[approval] file item={params['itemId']}")
    return approve_file_change_for_session()  # {"decision": "acceptForSession"}

async def on_tool_input(params: ToolRequestUserInputParams) -> ToolRequestUserInputResponse:
    return tool_answers({q["id"]: ["yes"] for q in params["questions"]})

async def on_dynamic_tool(params: DynamicToolCallParams) -> DynamicToolCallResponse:
    return tool_call_success()  # {"success": True, "contentItems": []}

async def main() -> None:
    policy = ApprovalPolicy.custom(
        on_command_execution=on_command,
        on_file_change=on_file,
        on_tool_request_user_input=on_tool_input,
        on_dynamic_tool_call=on_dynamic_tool,
    )
    # Built-in alternatives:
    # policy = ApprovalPolicy.auto_accept()   — accept everything
    # policy = ApprovalPolicy.auto_decline()  — decline everything (default)
    # policy = ApprovalPolicy.commands_only() — accept commands, decline files/tools

    async with await Session.create(
        client_info=client_info("my-app", "1.0.0"),
        approval_policy=policy,
    ) as session:
        thread = await session.start_ephemeral_thread()
        print((await thread.ask("Reply with exactly OK.")).strip())

asyncio.run(main())

```

--------------------------------

### Tool User-Input Response

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

Shows how to format a response for a tool that requires user input.

```python
tool_answers({"q1": ["answer A", "answer B"], "q2": ["yes"]})
# {"answers": {"q1": {"answers": ["answer A", "answer B"]}, "q2": {"answers": ["yes"]}}}
```

--------------------------------

### Error Handling with Codex Client

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

Demonstrates how to handle various exceptions that can occur when using the codex client, such as timeouts and transport errors.

```python
import asyncio
from codex_client import (
    ApprovalPolicy, Session, StdioTransport, TypedCodexClient,
    client_info,
)
from codex_client.errors import (
    ClientClosedError,
    JsonRpcCodecError,
    MiddlewareAbortedError,
    ProtocolClientError,
    ProtocolStreamError,
    RequestTimeoutError,
    TransportClosedError,
    UnknownResponseIdError,
)

async def main() -> None:
    client = TypedCodexClient.from_transport(StdioTransport())
    try:
        await client.initialize({"clientInfo": client_info("my-app", "1.0.0")})
        await client.send_initialized()
        result = await client.request("model/list", {}, timeout=3.0)
        print(result)
    except RequestTimeoutError:
        print("request timed out — retry or increase timeout")
    except TransportClosedError:
        print("the codex app-server process closed unexpectedly")
    except ProtocolStreamError as e:
        print(f"malformed frame or unexpected exit: {e}")
    except ProtocolClientError as e:
        # Base for ClientClosedError, UnknownResponseIdError, MiddlewareAbortedError
        print(f"protocol client error: {e}")
    finally:
        await client.close()

asyncio.run(main())
```

--------------------------------

### StdioTransport - Spawn and Connect to `codex app-server`

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

`StdioTransport` manages the subprocess lifecycle: spawning, pumping stdout as JSON-RPC frames, buffering stderr for diagnostics, and clean shutdown. Allows custom spawn options.

```APIDOC
## `StdioTransport` — spawn and connect to `codex app-server`

### Description
`StdioTransport` manages the subprocess lifecycle: spawning, pumping stdout lines as JSON-RPC frames, buffering stderr for diagnostics, and clean shutdown.

### Example Usage
```python
import asyncio
from codex_client import StdioTransport
from codex_client.connection import ProtocolConnection
from codex_client.codec import JsonRpcCodec

async def main() -> None:
    # Custom spawn options
    transport = StdioTransport(
        command="codex",
        args=("app-server",),
        cwd="/tmp/workspace",
        env={"CODEX_LOG_LEVEL": "debug"},
    )

    # Use as async context manager for automatic cleanup
    async with transport:
        # Send a raw newline-delimited JSON frame
        frame = JsonRpcCodec.encode({"id": 0, "method": "initialize", "params": {
            "clientInfo": {"name": "test", "version": "0.0.1"}
        }})
        await transport.send(frame)

        # Read one response frame
        async for raw_line in transport:
            msg = JsonRpcCodec.decode(raw_line)
            print(msg)
            break  # just read the first message

asyncio.run(main())
```
```

--------------------------------

### Dynamic Tool Call Responses

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

Illustrates how to represent the success or failure of a dynamic tool call.

```python
tool_call_success([{"type": "text", "text": "result"}])
# {"success": True, "contentItems": [{"type": "text", "text": "result"}]}

tool_call_failure()
# {"success": False, "contentItems": []}
```

--------------------------------

### Thread Management: fork(), rollback(), archive()

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

Provides methods for managing threads: `fork()` creates a branched copy of the thread, `rollback()` reverts the thread to a previous state by a specified number of turns, and `archive()` marks the thread as archived.

```APIDOC
## Thread Management

### `thread.fork()`

#### Description
Creates a branched copy of the current thread at its current state.

#### Method Signature
`async def fork() -> Thread`

#### Response
* **Thread** - A new `Thread` object representing the forked branch.

### `thread.rollback(n: int)`

#### Description
Reverts the thread to a state `n` turns prior to the current state.

#### Method Signature
`async def rollback(n: int)`

#### Parameters
* **n** (int) - The number of turns to roll back.

### `thread.archive()`

#### Description
Flags the thread as archived, typically for cleanup or historical purposes.

#### Method Signature
`async def archive()`

### Request Example
```python
# Fork the thread
forked = await main_thread.fork()

# Rollback main thread by 1 turn
await main_thread.rollback(1)

# Archive the thread
await main_thread.archive()
```
```

--------------------------------

### Attach Middleware to Session

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

Use `TypedCodexClient.use()` to attach middleware for logging, tracing, or other instrumentation. Each middleware receives a `MiddlewareContext` and a `next` callable.

```python
import asyncio
import json
from codex_client import (
    ApprovalPolicy, Middleware, MiddlewareContext, Session, client_info,
)

async def logging_middleware(ctx: MiddlewareContext, next) -> None:
    direction = ctx.direction  # "incoming" | "outgoing"
    method = ctx.method or "(response)"
    if ctx.direction == "outgoing":
        print(f"→ {method}")
    await next()
    if ctx.direction == "incoming":
        print(f"← {method}")

async def main() -> None:
    async with await Session.create(
        client_info=client_info("my-app", "1.0.0"),
        approval_policy=ApprovalPolicy.auto_accept(),
        middleware=[logging_middleware],
    ) as session:
        thread = await session.start_ephemeral_thread()
        print((await thread.ask("Reply with exactly OK.")).strip())

asyncio.run(main())
```

--------------------------------

### Authenticate Codex CLI

Source: https://github.com/cheolwanpark/codex-client/blob/main/src/examples/README.md

Command to authenticate the Codex CLI. Use this if you encounter 'Authentication required' errors.

```bash
uv run codex-client login
```

--------------------------------

### Approval Response Helpers

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

Provides helper functions for generating approval responses.

```python
approve_command()                     # {"decision": "accept"}
decline_command()                     # {"decision": "decline"}
approve_file_change()                 # {"decision": "accept"}
approve_file_change_for_session()     # {"decision": "acceptForSession"}
decline_file_change()                 # {"decision": "decline"}
```

--------------------------------

### Utility Helpers for Protocol Payloads

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

Helper functions to reduce boilerplate and improve readability when building protocol payloads, such as client information, text inputs, and thread parameters.

```APIDOC
## Utility helpers for building protocol payloads

### Description
`client_info()`, `text_input()`, `thread_params()`, `turn_options()`, and the approval response helpers reduce boilerplate and keep protocol shapes readable.

### Example Usage
```python
from codex_client import (
    approve_command, approve_file_change, approve_file_change_for_session,
    client_info, decline_command, decline_file_change,
    text_input, thread_params, tool_answers, tool_call_failure, tool_call_success, turn_options,
)

# ClientInfo dict
info = client_info("my-app", "2.0.0", title="My Application")
# {"name": "my-app", "version": "2.0.0", "title": "My Application"}

# TextUserInput dict
inp = text_input("Summarize the current directory in one sentence.")
# {"type": "text", "text": "Summarize..."}
```
```

--------------------------------

### Resume Persistent Thread Across Sessions

Source: https://github.com/cheolwanpark/codex-client/blob/main/python/README.md

Demonstrates how to resume a persistent thread in a new session. The thread ID is saved after the first session and used to reconnect in the second session.

```python
from __future__ import annotations

import asyncio

from codex_client import ApprovalPolicy, Session, client_info, thread_params


async def main() -> None:
    async with await Session.create(
        client_info=client_info("my-app", "0.1.0"),
        approval_policy=ApprovalPolicy.auto_accept(),
    ) as first_session:
        thread = await first_session.start_thread(thread_params(ephemeral=False))
        await thread.ask("Reply with exactly FIRST.")
        thread_id = thread.id

    async with await Session.create(
        client_info=client_info("my-app", "0.1.0"),
        approval_policy=ApprovalPolicy.auto_accept(),
    ) as second_session:
        thread = await second_session.resume_thread(thread_id)
        snapshot = await second_session.read_thread(thread.id, include_turns=True)
        print(len(snapshot.get("turns", [])))
        print((await thread.ask("Reply with exactly SECOND.")).strip())


if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### session.start_ephemeral_thread() / session.start_thread()

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

Creates a new thread. An ephemeral thread is discarded when the session ends, while a persistent thread survives and can be resumed across sessions.

```APIDOC
## session.start_ephemeral_thread() / session.start_thread()

### Description
Creates a new thread. An ephemeral thread is discarded when the session ends. A persistent thread survives and can be resumed across sessions. `thread_params()` builds the `ThreadStartParams` dict with keyword arguments.

### Method
`session.start_ephemeral_thread()` or `session.start_thread(thread_params)`

### Parameters
- `thread_params` (dict, optional) - Parameters for starting the thread. If not provided, default parameters are used. Can be constructed using `thread_params()`.
  - `ephemeral` (bool) - Whether the thread is ephemeral.
  - `model` (str) - The model to use for the thread.
  - `cwd` (str) - The working directory for the thread.
  - `approval_policy` (str) - The approval policy for the thread.

### Request Example
```python
import asyncio
from codex_client import Session, client_info, thread_params

async def main() -> None:
    async with await Session.create(
        client_info=client_info("my-app", "1.0.0"),
        approval_policy=ApprovalPolicy.auto_accept(),
    ) as session:
        # Ephemeral thread
        ephemeral = await session.start_ephemeral_thread()
        print(f"ephemeral id: {ephemeral.id}, status: {ephemeral.status}")

        # Persistent thread
        persistent = await session.start_thread(
            thread_params(
                ephemeral=False,
                model="o4-mini",
                cwd="/tmp/project",
                approval_policy="unless-allow-listed",
            )
        )
        print(f"persistent id: {persistent.id}")

asyncio.run(main())
```

### Response
- `Thread` object - The newly created thread.
```

--------------------------------

### Typed Protocol Client Usage

Source: https://github.com/cheolwanpark/codex-client/blob/main/typescript/README.md

Illustrates using the TypedCodexClient for direct typed requests and notifications. This approach bypasses the Session runtime wrapper. Requires an authenticated local Codex CLI session.

```typescript
import {
  StdioTransport,
  TypedCodexClient,
  clientInfo,
  textInput,
} from "@codex-client/typescript";

const client = TypedCodexClient.fromTransport(new StdioTransport());
const chunks: string[] = [];

client.onNotification("item/agentMessage/delta", (params) => {
  chunks.push(params.delta);
});

try {
  await client.initialize({ clientInfo: clientInfo("my-app", "0.1.0") });
  await client.sendInitialized();

  const thread = await client.threadStart({ ephemeral: true });
  await client.turnStart({
    input: [textInput("Reply with exactly OK.")],
    threadId: thread.thread.id,
  });

  const completed = await client.waitForNotification("turn/completed", {
    timeoutMs: 60_000,
  });
  console.log(completed.turn.status);
  console.log(chunks.join("").trim());
} finally {
  await client.close();
}
```

--------------------------------

### session.exec()

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

Executes a shell command within the sandboxed environment of the application server associated with the current session. It returns the output and exit code of the command.

```APIDOC
## session.exec()

### Description
Executes a shell command in the context of the current session using the app-server's sandboxed command runner.

### Method Signature
`async def exec(cmd: List[str], *, cwd: str = None, timeout_ms: int = None) -> Dict[str, Any]`

### Parameters
* **cmd** (List[str]) - A list of strings representing the command and its arguments.
* **cwd** (str, optional) - The current working directory for the command.
* **timeout_ms** (int, optional) - The timeout for the command execution in milliseconds.

### Request Example
```python
result = await session.exec(
    ["git", "log", "--oneline", "-5"],
    cwd="/path/to/repo",
    timeout_ms=10_000,
)
print(result.get("output", ""))
print(f"exit code: {result.get('exitCode')}")
```

### Response
* **output** (str) - The standard output of the command.
* **exitCode** (int) - The exit code of the command.
```

--------------------------------

### Manage StdioTransport Subprocess Lifecycle

Source: https://context7.com/cheolwanpark/codex-client/llms.txt

`StdioTransport` manages spawning, pumping stdout lines as JSON-RPC frames, buffering stderr, and clean shutdown. Custom spawn options like command, arguments, working directory, and environment variables can be provided.

```python
import asyncio
from codex_client import StdioTransport
from codex_client.connection import ProtocolConnection
from codex_client.codec import JsonRpcCodec

async def main() -> None:
    # Custom spawn options
    transport = StdioTransport(
        command="codex",
        args=("app-server",),
        cwd="/tmp/workspace",
        env={"CODEX_LOG_LEVEL": "debug"},
    )

    # Use as async context manager for automatic cleanup
    async with transport:
        # Send a raw newline-delimited JSON frame
        frame = JsonRpcCodec.encode({"id": 0, "method": "initialize", "params": {
            "clientInfo": {"name": "test", "version": "0.0.1"}
        }})
        await transport.send(frame)

        # Read one response frame
        async for raw_line in transport:
            msg = JsonRpcCodec.decode(raw_line)
            print(msg)
            break  # just read the first message

asyncio.run(main())
```

--------------------------------

### Generate Protocol Types

Source: https://github.com/cheolwanpark/codex-client/blob/main/typescript/README.md

Regenerates TypeScript definitions for the Codex protocol from schema files. Run this command after updating schema definitions.

```bash
cd typescript
pnpm generate
```