Metadata-Version: 2.4
Name: codex-app-server-client
Version: 0.1.0
Summary: Python client SDK for the Codex app-server JSON-RPC protocol
Project-URL: Repository, https://github.com/paras-anekantvad/codex-app-server-client-sdk
Project-URL: Homepage, https://github.com/paras-anekantvad/codex-app-server-client-sdk
Author: Paras Doshi
License: MIT
License-File: LICENSE
Classifier: Development Status :: 3 - Alpha
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Python: >=3.11
Requires-Dist: pydantic>=2.0
Provides-Extra: dev
Requires-Dist: mypy>=1.16.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=1.0.0; extra == 'dev'
Requires-Dist: pytest-cov>=5.0.0; extra == 'dev'
Requires-Dist: pytest>=8.0.0; extra == 'dev'
Requires-Dist: ruff>=0.12.0; extra == 'dev'
Description-Content-Type: text/markdown

# Codex App Server Client SDK

Python client for the Codex app-server JSON-RPC protocol. Build AI-powered applications with full type safety, streaming support, and bidirectional communication.

## Installation

### Prerequisites

- **Python**: ≥3.11
- **Codex CLI**: A `codex` binary on your PATH (see [OpenAI Codex](https://github.com/openai/codex) for installation)

### Install from PyPI

```bash
pip install codex-app-server-client
```

Or with [uv](https://docs.astral.sh/uv/):

```bash
uv pip install codex-app-server-client
```

### Install from source

For contributors or to install the latest development version:

```bash
git clone https://github.com/paras-anekantvad/codex-app-server-client-sdk.git
cd codex-app-server-client-sdk
uv sync --all-extras
```

## Quick Examples

### Simple: Ask a question

```python
import asyncio
from codex_app_server_client import CodexAppServer

async def main():
    async with CodexAppServer() as codex:
        thread = await codex.start_thread()
        result = await thread.run("Explain Python decorators in 3 sentences")
        print(result.final_response)

asyncio.run(main())
```

### Streaming: Print deltas as they arrive

```python
from codex_app_server_client import CodexAppServer
from codex_app_server_client.types.events import AgentMessageDeltaEvent

async with CodexAppServer() as codex:
    thread = await codex.start_thread()
    async for event in thread.run_streamed("Write a haiku about recursion"):
        if isinstance(event, AgentMessageDeltaEvent):
            print(event.delta, end="", flush=True)
    print()  # newline
```

### Advanced: Policy control with event callbacks

Block disallowed actions in real-time:

```python
from codex_app_server_client import CodexAppServer, EventAction
from codex_app_server_client.types.events import ItemCompletedEvent, ThreadEvent

ALLOWED_TYPES = {"agentMessage", "plan", "reasoning", "webSearch"}

def policy_filter(method: str, event: ThreadEvent) -> EventAction | None:
    if isinstance(event, ItemCompletedEvent):
        item_type = event.item.get("type") if isinstance(event.item, dict) else None
        if item_type and item_type not in ALLOWED_TYPES:
            return EventAction.INTERRUPT  # block unsafe actions
    return None

async with CodexAppServer() as codex:
    thread = await codex.start_thread()
    result = await thread.run(
        "Write and run a shell script",
        on_event=policy_filter,
    )
    if result.status == "interrupted":
        print("Turn blocked due to policy violation")
```

## Key Features

- **Type-safe**: Full protocol coverage with Pydantic models
- **Async & sync**: Both `asyncio` and synchronous APIs
- **Streaming**: Real-time events via `run_streamed()`
- **Policy control**: `on_event` callbacks for approval/interrupt logic
- **Bidirectional**: Handle server-initiated approval requests
- **Thread management**: Resume conversations, fork threads, rollback turns
- **Error handling**: Typed exceptions and structured error payloads

## Common Patterns

### Resume an existing thread

```python
async with CodexAppServer() as codex:
    # List recent threads
    threads = await codex.list_threads(limit=10)
    thread_id = threads[0].id
    
    # Resume and continue conversation
    thread = await codex.resume_thread(thread_id)
    result = await thread.run("Summarize our conversation so far")
```

### Select a specific model

```python
# Set model at thread creation
thread = await codex.start_thread(
    model="gpt-5.2-codex",
    cwd="/path/to/project",
)

# Or override per-turn
result = await thread.run(
    "Quick question",
    model="gpt-5.3-codex",
)
```

### Handle approval requests

For bidirectional scenarios where the server requests client approval:

```python
from codex_app_server_client import AsyncCodexClient

client = AsyncCodexClient()

async def approve_safe_commands(method: str, params: dict) -> dict:
    if method == "commandExecution/approve":
        command = params.get("command", "")
        if command.startswith(("ls", "cat", "grep")):
            return {"decision": "approve"}
    return {"decision": "decline"}

client.on_server_request("commandExecution/approve", approve_safe_commands)
await client.start()
# ... use client ...
```

### Set timeouts and handle errors

```python
try:
    result = await thread.run(
        "Long-running task",
        timeout_s=60,  # raise TimeoutError after 60s
    )
    if result.status == "failed":
        print(f"Turn failed: {result.error.message}")
except TimeoutError:
    print("Turn timed out")
```

### Sync API (for non-async codebases)

```python
from codex_app_server_client import SyncCodexAppServer

with SyncCodexAppServer() as codex:
    thread = codex.start_thread()
    result = thread.run("Hello!")
    print(result.final_response)
```

## API Reference

### High-level API

- **`CodexAppServer`** / **`SyncCodexAppServer`**: Context managers for managing the app-server lifecycle. Exposes `start_thread()`, `resume_thread()`, `list_threads()`.
- **`AsyncThread`** / **`SyncThread`**: Bound to a thread ID. Main methods:
  - `run(input, **kwargs) -> TurnResult`: Execute a turn, block until complete
  - `run_streamed(input, **kwargs) -> AsyncIterator[ThreadEvent]`: Stream events in real-time
- **`TurnResult`**: Dataclass with:
  - `final_response: str` — agent's text response
  - `status: str` — `"completed"`, `"interrupted"`, or `"failed"`
  - `items: list[dict]` — all completed items
  - `usage: ThreadTokenUsage | None` — token counts
  - `error: TurnError | None` — error details if failed

### Low-level API

- **`AsyncCodexClient`** / **`SyncCodexClient`**: Direct protocol access. Every app-server method has a typed wrapper:
  - `thread_start()`, `thread_resume()`, `thread_list()`, `thread_fork()`, `thread_rollback()`
  - `turn_start()`, `turn_interrupt()`, `turn_steer()`
  - `account_read()`, `account_login_start()`
  - `on_notification(method, callback)` — register event handlers
  - `on_server_request(method, callback)` — handle server-initiated requests

### Types

All in `codex_app_server_client.types.*`:

- `ThreadInfo`, `TurnInfo`, `ThreadItem` (15 variants)
- `ThreadEvent` (25+ event types): `TurnStartedEvent`, `ItemCompletedEvent`, `AgentMessageDeltaEvent`, etc.
- `TurnStartParams`, `ThreadStartParams`, `ApprovalPolicy`, `SandboxPolicy`

See `src/codex_app_server_client/__init__.py` for all exports.

## Configuration

```python
CodexAppServer(
    codex_bin="/path/to/codex",         # Path to codex binary (default: auto-detect)
    client_name="my_app",               # Client identifier for logging
    client_version="1.0.0",             # Your app version
    experimental_api=True,              # Enable experimental protocol features
    extra_args=["--model", "gpt-5.2-codex"],  # Extra CLI args for codex subprocess
    env={"OPENAI_API_KEY": "sk-..."},   # Environment variables
)
```

## Development

Use the Makefile for common tasks:

```bash
make install      # install dependencies (uv sync --all-extras)
make test         # run tests with coverage
make lint         # run ruff linter
make format       # format code with ruff
make type-check   # run mypy type checker
make fix          # auto-fix linting issues
make pre-commit   # run all checks (CI-like)
```

Or run `uv` commands directly:

```bash
uv sync --all-extras                 # install dependencies
uv run pytest                        # run tests
uv run ruff check src/ tests/        # lint
uv run ruff format src/ tests/       # format
uv run mypy src/                     # typecheck (strict)
```

## License

MIT
