codex-app-server-client 0.1.0

Python client SDK for the Codex app-server JSON-RPC protocol

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 for installation)

Install from PyPI

pip install codex-app-server-client

Or with uv:

uv pip install codex-app-server-client

Install from source

For contributors or to install the latest development version:

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

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

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:

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

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

# 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:

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

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)

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

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:

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:

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