第10章 App Server Protocol

The Terminal UI provides an excellent interactive experience, but modern IDE integration requires a programmatic interface. When VS Code, Cursor, or Windsurf needs to communicate with Codex, they don't spawn a TUI—they speak the app-server protocol.

This protocol exposes Codex's full capabilities through JSON-RPC 2.0, enabling everything from simple command execution to multi-turn conversations with streaming responses. It's the bridge between your IDE and the agent, handling authentication, file operations, tool execution, and real-time event streaming—all while maintaining the security boundaries that keep Codex safe.

🤿 深入底层：The app-server isn't just a thin wrapper around the CLI. It uses the same core agent logic (CodexThread, ToolOrchestrator) but provides a different access pattern. While the TUI owns the event loop and renders directly, the app-server exposes that event loop through JSON-RPC, letting clients drive the conversation programmatically.

Protocol Overview

Codex's app-server uses JSON-RPC 2.0—a lightweight, stateless protocol that maps cleanly to request/response patterns while supporting server-initiated notifications. Unlike REST APIs that require you to poll for status, JSON-RPC lets the server push updates to your client as they happen.

JSON-RPC 2.0 Basics

A JSON-RPC message is either a request, a response, a notification, or an error. Here's the anatomy:

// Request (expects response)
{
  "method": "thread/start",
  "id": 1,
  "params": {
    "cwd": "/Users/me/project"
  }
}

// Response
{
  "id": 1,
  "result": {
    "thread": { "id": "thr_123", ... }
  }
}

// Notification (fire-and-forget, no id)
{
  "method": "item/started",
  "params": {
    "item": { "type": "userMessage", ... }
  }
}

// Error
{
  "id": 1,
  "error": {
    "code": -32601,
    "message": "Method not found",
    "data": null
  }
}

Notice that "jsonrpc": "2.0" is omitted on the wire—Codex's implementation skips this header for brevity. The id field correlates requests with responses; notifications omit id entirely because they don't expect a response.

Requests vs Notifications

Understanding this distinction is crucial:

Type	Has id	Expects Response	Use Case
Request	Yes	Yes	Method calls like thread/start, fs/readFile
Notification	No	No	Events like item/started, turn/completed

When you send a request, you block until a response arrives (or times out). Notifications are one-way—the server emits them when something happens, and you process them asynchronously. This enables streaming: a single turn/start request triggers dozens of item/started, item/agentMessage/delta, and item/completed notifications as the agent generates its response.

⚠️ 避坑指南：Don't assume notification ordering across different threads. Notifications for the same (threadId, turnId) arrive in order, but interleaved notifications from multiple concurrent threads may arrive out of order. Always check the threadId and turnId fields to scope events correctly.

Transport Options

Codex supports two transports with different tradeoffs:

stdio (default): Newline-delimited JSON (JSONL) over stdin/stdout. One complete JSON-RPC message per line. Perfect for subprocess spawning—your IDE launches codex app-server --listen stdio:// and communicates via the subprocess's standard streams. No network overhead, no port conflicts, automatic cleanup when the subprocess dies.

websocket: One JSON-RPC message per WebSocket text frame. Supports concurrent clients and remote connections (with authentication). Experimental and currently unsupported for production—use at your own risk. Loopback listeners (ws://127.0.0.1:PORT) are appropriate for local development and SSH port-forwarding. Non-loopback listeners require explicit auth configuration:

# Capability token auth
codex app-server --listen ws://0.0.0.0:8080 \
  --ws-auth capability-token \
  --ws-token-file /absolute/path/to/token.txt

# Signed bearer token (HMAC-JWT/JWS)
codex app-server --listen ws://0.0.0.0:8080 \
  --ws-auth signed-bearer-token \
  --ws-shared-secret-file /absolute/path/to/secret.key \
  --ws-issuer "https://my-corp.com" \
  --ws-audience "codex-app-server"

Clients present credentials via the Authorization: Bearer <token> header during the WebSocket handshake.

Codex CLI Adaptations

Codex extends standard JSON-RPC in a few ways:

• Backpressure handling: When request ingress saturates, new requests receive error code -32001 with message "Server overloaded; retry later." Clients should retry with exponential backoff and jitter.
• Connection-scoped notifications: Unlike traditional JSON-RPC, notifications aren't broadcast to all clients. Each connection only receives notifications for threads it has subscribed to (via thread/start, thread/resume, or thread/fork).
• Server-initiated requests: For approvals, the server sends JSON-RPC requests to the client. The client must respond with a decision, and the server acknowledges with serverRequest/resolved. This bidirectional pattern enables approval dialogs without polling.

Core Methods: Thread and Turn Management

The app-server's core API revolves around three primitives: threads (conversations), turns (user-message/response pairs), and items (individual events within a turn). Understanding this hierarchy is key to driving Codex programmatically.

thread/start Method

Creating a new conversation begins with thread/start. This method initializes a CodexThread, loads configuration, and returns a thread object you can use for subsequent operations:

{
  "method": "thread/start",
  "id": 10,
  "params": {
    "model": "gpt-5.1-codex",
    "cwd": "/Users/me/project",
    "approvalPolicy": "never",
    "sandbox": "workspaceWrite",
    "personality": "friendly",
    "serviceName": "my_ide_extension"
  }
}

The response contains the newly created thread:

{
  "id": 10,
  "result": {
    "thread": {
      "id": "thr_123",
      "preview": "",
      "modelProvider": "openai",
      "createdAt": 1730910000
    }
  }
}

You'll also receive a thread/started notification with the full thread object, including the current status. This notification is your signal that the thread is ready for turns.

☕ 茶水间时间：Why the separate notification? Because thread/start triggers async work—loading skills, initializing the thread, setting up watchers. The response returns immediately with the thread ID, but the thread/started notification fires only when initialization completes. This separation lets you start displaying UI while the thread finishes loading in the background.

turn/submit Method

Once you have a thread, you send user input via turn/submit (or turn/start in the newer API). This creates a new turn and streams the agent's response:

{
  "method": "turn/submit",
  "id": 30,
  "params": {
    "threadId": "thr_123",
    "input": [
      { "type": "text", "text": "Run tests" }
    ]
  }
}

The response acknowledges the turn creation:

{
  "id": 30,
  "result": {
    "turn": {
      "id": "turn_456",
      "status": "inProgress",
      "items": [],
      "error": null
    }
  }
}

Then the notification stream begins:

{"method":"turn/started","params":{"turn":{"id":"turn_456","status":"inProgress",...}}}
{"method":"item/started","params":{"item":{"type":"userMessage","id":"turn_456",...}}}
{"method":"item/completed","params":{"item":{"type":"userMessage","id":"turn_456",...}}}
{"method":"item/started","params":{"item":{"type":"commandExecution","id":"item_1",...}}}
{"method":"item/commandExecution/outputDelta","params":{"itemId":"item_1","outputDelta":"running cargo test...\n"}}
{"method":"item/completed","params":{"item":{"type":"commandExecution","id":"item_1","status":"completed",...}}}
{"method":"turn/completed","params":{"turn":{"id":"turn_456","status":"completed",...}}}

The turn/started notification marks when the turn actually begins processing. From there, you'll see a series of item/started and item/completed pairs as the agent works through the request. Each item represents a discrete unit of work—user messages, agent responses, tool calls, file edits.

turn/create Method

For ad-hoc, single-shot interactions, turn/create combines thread creation and turn submission into one call:

{
  "method": "turn/create",
  "id": 40,
  "params": {
    "input": [
      { "type": "text", "text": "What files are in this directory?" }
    ],
    "cwd": "/Users/me/project"
  }
}

This creates an ephemeral thread (not persisted to disk), starts a turn, and streams notifications as usual. When the turn completes, the thread is automatically unloaded. Useful for quick operations where you don't need conversation history.

Item/Notification Pattern

The streaming model is built around the item lifecycle:

1. item/started: Announces a new unit of work with full item payload
2. Item-specific deltas: Stream incremental updates (item/agentMessage/delta, item/commandExecution/outputDelta)
3. item/completed: Marks the item as finished with final state

For agent messages, you receive the accumulated text via item/agentMessage/delta notifications:

{"method":"item/started","params":{"item":{"type":"agentMessage","id":"msg_1","text":""}}}
{"method":"item/agentMessage/delta","params":{"itemId":"msg_1","delta":"I'll"}}
{"method":"item/agentMessage/delta","params":{"itemId":"msg_1","delta":" run the tests for you."}}
{"method":"item/completed","params":{"item":{"type":"agentMessage","id":"msg_1","text":"I'll run the tests for you."}}}

Concatenate delta values in order to reconstruct the full message. This enables real-time streaming UI—each delta updates the display incrementally as tokens arrive.

Streaming Updates

The turn-level notifications track overall progress:

• turn/started: Turn begins processing
• turn/diff/updated: Aggregated diff after each file change
• turn/plan/updated: Agent's working plan (if in plan mode)
• turn/completed: Turn finishes with final status and token usage

The turn/completed notification includes everything you need for final state:

{
  "method": "turn/completed",
  "params": {
    "turn": {
      "id": "turn_456",
      "status": "completed",
      "items": [...],
      "error": null
    }
  }
}

Status can be completed, interrupted, or failed. Failed turns include an error object with message, error code, and optional details.

File System and Command Methods

Beyond conversation management, the app-server exposes direct file system and command execution APIs. These enable IDE features like file browsing, quick edits, and command running without starting a full turn.

fs/* Methods

File system operations work on absolute paths and return base64-encoded data:

{
  "method": "fs/readFile",
  "id": 40,
  "params": {
    "path": "/Users/me/project/src/main.rs"
  }
}
{
  "id": 40,
  "result": {
    "dataBase64": "ZnVuY3Rpb24gbWFpbigpIHsKICAgcHJpbnRsbiEoIkhlbGxvLCBXb3JsZCEiKTsKfQ=="
  }
}

The available fs/* methods:

Method	Purpose	Key Parameters
fs/readFile	Read file contents	path (absolute)
fs/writeFile	Write base64-encoded data	path, dataBase64
fs/createDirectory	Create directory (recursive by default)	path, recursive
fs/getMetadata	Get file/directory metadata	path
fs/readDirectory	List direct children	path
fs/remove	Remove file/directory tree	path, recursive, force
fs/copy	Copy files or directories	sourcePath, destPath, recursive
fs/watch	Subscribe to change notifications	path
fs/unwatch	Cancel watch subscription	watchId

The fs/watch method is particularly powerful for real-time features:

{
  "method": "fs/watch",
  "id": 44,
  "params": {
    "path": "/Users/me/project/.git/HEAD"
  }
}
{
  "id": 44,
  "result": {
    "watchId": "0195ec6b-1d6f-7c2e-8c7a-56f2c4a8b9d1",
    "path": "/Users/me/project/.git/HEAD"
  }
}

When the watched path changes, you receive notifications:

{
  "method": "fs/changed",
  "params": {
    "watchId": "0195ec6b-1d6f-7c2e-8c7a-56f2c4a8b9d1",
    "changedPaths": ["/Users/me/project/.git/HEAD"]
  }
}

command/* Methods

Command execution runs shell commands under the server's sandbox:

{
  "method": "command/exec",
  "id": 32,
  "params": {
    "command": ["ls", "-la"],
    "processId": "ls-1",
    "cwd": "/Users/me/project",
    "env": { "FOO": "override" },
    "size": { "rows": 40, "cols": 120 },
    "sandboxPolicy": { "type": "workspaceWrite" },
    "outputBytesCap": 1048576,
    "timeoutMs": 10000
  }
}

The response includes exit code and captured output:

{
  "id": 32,
  "result": {
    "exitCode": 0,
    "stdout": "total 16\ndrwxr-xr-x  ...",
    "stderr": ""
  }
}

For interactive commands, enable streaming with streamStdin and streamStdoutStderr:

{
  "method": "command/exec",
  "id": 33,
  "params": {
    "command": ["bash", "-i"],
    "processId": "bash-1",
    "tty": true,
    "streamStdin": true,
    "streamStdoutStderr": true
  }
}

Then handle streaming output notifications:

{
  "method": "command/exec/outputDelta",
  "params": {
    "processId": "bash-1",
    "stream": "stdout",
    "deltaBase64": "YmFzaC00LjQkIA==",
    "capReached": false
  }
}

Write to stdin via command/exec/write:

{
  "method": "command/exec/write",
  "id": 34,
  "params": {
    "processId": "bash-1",
    "deltaBase64": "cHdkCg=="
  }
}

Terminate with command/exec/terminate:

{
  "method": "command/exec/terminate",
  "id": 37,
  "params": {
    "processId": "bash-1"
  }
}

Approval Transmission

When a command or file edit requires user approval, the server sends a server-initiated request to your client. This reverses the usual client/server pattern—now the server is the requester.

For command approvals:

{
  "method": "item/commandExecution/requestApproval",
  "id": 50,
  "params": {
    "threadId": "thr_123",
    "turnId": "turn_456",
    "itemId": "item_1",
    "command": ["rm", "-rf", "/tmp/test"],
    "cwd": "/Users/me/project",
    "reason": "Untrusted command"
  }
}

Your client responds with a decision:

{
  "id": 50,
  "result": {
    "decision": "accept"
  }
}

Supported decisions: accept, acceptForSession, decline, cancel. Some approvals support acceptWithExecpolicyAmendment for modifying sandbox rules mid-session.

Sandbox Integration

All command/* and fs/* methods respect the configured sandbox policy. The policy determines:

• Filesystem access: Which directories can be read/written
• Network access: Whether external connections are allowed
• Execution scope: Whether commands run in isolation

Specify policy per-request:

{
  "method": "command/exec",
  "id": 33,
  "params": {
    "command": ["curl", "https://api.example.com/data"],
    "sandboxPolicy": {
      "type": "workspaceWrite",
      "writableRoots": ["/Users/me/project"],
      "networkAccess": true
    }
  }
}

Or inherit from thread-level config set during thread/start.

⚠️ 避坑指南：Never trust client-provided paths for sandbox decisions. Always resolve paths relative to the configured cwd and validate against the sandbox policy before execution. The app-server handles this for you, but if you're building custom tooling, remember that path traversal attacks are real.

Security Considerations

The app-server's security model relies on:

1. Transport-level auth: WebSocket connections require valid bearer tokens
2. Sandbox enforcement: All file/command operations pass through the policy layer
3. Approval gating: Dangerous operations require explicit user consent
4. Process isolation: Commands execute in sandboxed subprocesses

When using external sandbox mode (sandboxPolicy: { "type": "externalSandbox" }), Codex skips its own enforcement and trusts the environment to provide isolation. This is appropriate for containerized deployments where the container boundary provides the security guarantee.

Transport Implementation

The stdio and websocket transports share a common architecture but differ in framing and connection management. Understanding these differences helps when choosing between transports or debugging connection issues.

stdio Transport

The stdio transport is the simplest—JSONL (newline-delimited JSON) over subprocess streams:

// codex-rs/app-server/src/transport/stdio.rs

pub(crate) async fn start_stdio_connection(
    transport_event_tx: mpsc::Sender<TransportEvent>,
    stdio_handles: &mut Vec<JoinHandle<()>>,
) -> IoResult<()> {
    let connection_id = ConnectionId(0);
    let (writer_tx, mut writer_rx) = mpsc::channel(CHANNEL_CAPACITY);
    
    // Signal connection opened
    transport_event_tx.send(TransportEvent::ConnectionOpened {
        connection_id,
        writer: writer_tx,
        disconnect_sender: None,
    }).await?;
    
    // Spawn stdin reader
    stdio_handles.push(tokio::spawn(async move {
        let stdin = io::stdin();
        let reader = BufReader::new(stdin);
        let mut lines = reader.lines();
        
        loop {
            match lines.next_line().await {
                Ok(Some(line)) => {
                    // Parse JSON-RPC from line
                    if !forward_incoming_message(...).await {
                        break;
                    }
                }
                Ok(None) => break, // EOF
                Err(err) => {
                    error!("Failed reading stdin: {err}");
                    break;
                }
            }
        }
    }));
    
    // Spawn stdout writer
    stdio_handles.push(tokio::spawn(async move {
        let mut stdout = io::stdout();
        while let Some(queued_message) = writer_rx.recv().await {
            let json = serialize_outgoing_message(queued_message.message)?;
            json.push('\n');
            stdout.write_all(json.as_bytes()).await?;
        }
    }));
    
    Ok(())
}

The framing is trivial: each line is one JSON-RPC message. No length prefixes, no chunking—just read until newline, parse JSON, process, repeat.

Connection lifecycle: The stdio transport is inherently single-client. Connection opens when the subprocess spawns, closes when stdin closes or the subprocess exits. There's no reconnection—when the connection dies, restart the subprocess.

Message framing: Each JSON-RPC message becomes exactly one line. Newlines within JSON strings are escaped (\n). The protocol is line-oriented, so malformed JSON on one line doesn't corrupt subsequent messages.

Error handling: Parse errors log a warning and continue. EOF terminates the connection. The bounded queue (128 messages by default) provides backpressure—if the client can't keep up, the server queues messages; when full, new requests receive the -32001 overload error.

websocket Transport

The websocket transport is more complex, supporting multiple concurrent connections with authentication:

// codex-rs/app-server/src/transport/websocket.rs

pub(crate) async fn start_websocket_acceptor(
    bind_address: SocketAddr,
    transport_event_tx: mpsc::Sender<TransportEvent>,
    shutdown_token: CancellationToken,
    auth_policy: AuthPolicy,
) -> IoResult<JoinHandle<()>> {
    let listener = TcpListener::bind(bind_address).await?;
    
    let handle = tokio::spawn(async move {
        loop {
            select! {
                _ = shutdown_token.cancelled() => break,
                
                result = listener.accept() => {
                    match result {
                        Ok((stream, addr)) => {
                            // Spawn connection handler
                            tokio::spawn(handle_connection(
                                stream,
                                addr,
                                transport_event_tx.clone(),
                                auth_policy.clone(),
                            ));
                        }
                        Err(err) => error!("Accept failed: {err}"),
                    }
                }
            }
        }
    });
    
    Ok(handle)
}

Each connection gets its own reader/writer pair with a unique ConnectionId. The transport tracks active connections and routes messages to the appropriate recipient.

Connection lifecycle: Connections open on WebSocket handshake, close on frame close or network error. Multiple clients can connect simultaneously, each with its own connection ID and isolated state. Graceful shutdown signals all clients to disconnect before the server exits.

Message framing: One JSON-RPC message per WebSocket text frame. No newlines required—each frame contains exactly one JSON object. Binary frames are not used. The framing is provided by the WebSocket protocol itself.

Error handling: Connection-level errors close that specific connection without affecting others. Parse errors log and drop the malformed message. The bounded queue applies per-connection, so one slow client doesn't affect others.

Connection Lifecycle

Both transports follow the same state machine:

The initialize → initialized handshake is required before any other methods. This negotiation lets the client declare capabilities (experimentalApi, optOutNotificationMethods) and receive server metadata (codexHome, platformFamily).

Message Framing Comparison

Aspect	stdio	websocket
Framing	Newline-delimited JSON (JSONL)	One message per WS text frame
Concurrency	Single client only	Multiple concurrent clients
Authentication	None (subprocess boundary)	Bearer token in handshake
Reconnection	Restart subprocess	Reconnect with same auth
Backpressure	Per-connection queue	Per-connection queue
Transport overhead	None (local pipes)	WebSocket framing overhead

Error Handling

Both transports handle errors at multiple levels:

1. Transport level: Connection failures, malformed frames
2. JSON-RPC level: Invalid requests, method not found, parse errors
3. Application level: Invalid params, execution failures

Transport-level errors close the connection. JSON-RPC errors return error responses. Application-level errors may send error notifications or include error info in responses.

Transport Selection

Choose based on your use case:

• Use stdio when: Single client, local execution, subprocess spawning is acceptable, simplicity is valued
• Use websocket when: Multiple concurrent clients, remote execution, reconnection support needed, browser-based clients

Most IDE integrations use stdio—it's simpler, more secure (no network exposure), and the subprocess model fits IDE extension architectures. Websocket shines for web-based clients or scenarios where a long-running server serves multiple clients.

Building a Client

Now let's build a complete app-server client from scratch. We'll implement connection handling, method invocation, notification processing, and error recovery.

Connection Handling

First, establish the connection and handle the initialization handshake:

// codex-rs/app-server/src/bin/notify_capture.rs

interface JSONRPCRequest {
  method: string;
  id: number;
  params?: any;
}

interface JSONRPCResponse {
  id: number;
  result?: any;
  error?: {
    code: number;
    message: string;
    data?: any;
  };
}

interface JSONRPCNotification {
  method: string;
  params: any;
}

type JSONRPCMessage = JSONRPCResponse | JSONRPCNotification;

class CodexClient {
  private process: ChildProcess;
  private nextId = 1;
  private pendingRequests = new Map<number, {
    resolve: (value: any) => void;
    reject: (error: any) => void;
  }>();
  private notificationHandlers = new Map<string, (params: any) => void>();

  async connect(command: string = "codex app-server --listen stdio://"): Promise<void> {
    this.process = spawn(command, { shell: true });
    
    // Handle stdout (responses and notifications)
    this.process.stdout?.on('line', (line: string) => {
      try {
        const message = JSON.parse(line) as JSONRPCMessage;
        this.handleMessage(message);
      } catch (err) {
        console.error('Failed to parse message:', line, err);
      }
    });
    
    // Wait for process to be ready
    await new Promise<void>((resolve) => {
      this.process.stdout?.once('readable', resolve);
    });
  }

  private handleMessage(message: JSONRPCMessage): void {
    if ('id' in message) {
      // Response to a request
      const pending = this.pendingRequests.get(message.id);
      if (pending) {
        this.pendingRequests.delete(message.id);
        if (message.error) {
          pending.reject(new Error(message.error.message));
        } else {
          pending.resolve(message.result);
        }
      }
    } else {
      // Notification
      const handler = this.notificationHandlers.get(message.method);
      if (handler) {
        handler(message.params);
      }
    }
  }

  async initialize(clientInfo: { name: string; title: string; version: string }): Promise<void> {
    const response = await this.request('initialize', { clientInfo });
    
    // Send initialized notification
    this.sendNotification('initialized', {});
    
    return response;
  }

  private request(method: string, params?: any): Promise<any> {
    return new Promise((resolve, reject) => {
      const id = this.nextId++;
      const request: JSONRPCRequest = { method, id, params };
      
      this.pendingRequests.set(id, { resolve, reject });
      
      const json = JSON.stringify(request);
      this.process.stdin?.write(json + '\n');
      
      // Timeout after 30 seconds
      setTimeout(() => {
        if (this.pendingRequests.has(id)) {
          this.pendingRequests.delete(id);
          reject(new Error('Request timeout'));
        }
      }, 30000);
    });
  }

  private sendNotification(method: string, params: any): void {
    const notification = { method, params };
    const json = JSON.stringify(notification);
    this.process.stdin?.write(json + '\n');
  }

  on(method: string, handler: (params: any) => void): void {
    this.notificationHandlers.set(method, handler);
  }

  async disconnect(): Promise<void> {
    this.process.stdin?.end();
    await new Promise<void>((resolve) => {
      this.process.on('exit', resolve);
    });
  }
}

Method Invocation

With the connection established, invoke core methods:

async function startThread(client: Codex, cwd: string): Promise<string> {
  const response = await client.request('thread/start', {
    cwd,
    model: 'gpt-5.1-codex',
    sandbox: 'workspaceWrite'
  });
  return response.thread.id;
}

async function submitTurn(client: Codex, threadId: string, input: string): Promise<void> {
  await client.request('turn/submit', {
    threadId,
    input: [{ type: 'text', text: input }]
  });
}

async function interruptTurn(client: Codex, threadId: string, turnId: string): Promise<void> {
  await client.request('turn/interrupt', {
    threadId,
    turnId
  });
}

Notification Processing

Register handlers for streaming events:

function setupNotificationHandlers(client: Codex): void {
  // Thread lifecycle
  client.on('thread/started', (params) => {
    console.log('Thread started:', params.thread.id);
  });
  
  client.on('thread/status/changed', (params) => {
    console.log('Thread status changed:', params.threadId, params.status);
  });
  
  // Turn lifecycle
  client.on('turn/started', (params) => {
    console.log('Turn started:', params.turn.id);
  });
  
  client.on('turn/completed', (params) => {
    console.log('Turn completed:', params.turn.status);
    if (params.turn.error) {
      console.error('Turn error:', params.turn.error);
    }
  });
  
  // Item lifecycle with streaming
  const activeMessages = new Map<string, string>();
  
  client.on('item/started', (params) => {
    const { item } = params;
    console.log('Item started:', item.type, item.id);
    
    if (item.type === 'agentMessage') {
      activeMessages.set(item.id, '');
    }
  });
  
  client.on('item/agentMessage/delta', (params) => {
    const { itemId, delta } = params;
    const current = activeMessages.get(itemId) || '';
    activeMessages.set(itemId, current + delta);
    
    // Update UI with streaming text
    process.stdout.write(delta);
  });
  
  client.on('item/completed', (params) => {
    const { item } = params;
    console.log('Item completed:', item.type, item.id);
    
    if (item.type === 'agentMessage') {
      activeMessages.delete(item.id);
    }
  });
}

Error Recovery

Build resilience into your client:

class ResilientCodexClient extends CodexClient {
  private maxRetries = 3;
  private retryDelay = 1000; // ms
  
  async requestWithRetry(method: string, params?: any): Promise<any> {
    let lastError: any;
    
    for (let attempt = 0; attempt < this.maxRetries; attempt++) {
      try {
        return await this.request(method, params);
      } catch (error) {
        lastError = error;
        
        // Check if retryable
        if (error.message.includes('Server overloaded')) {
          const delay = this.retryDelay * Math.pow(2, attempt);
          console.log(`Retrying after ${delay}ms...`);
          await this.sleep(delay);
        } else {
          // Not retryable
          throw error;
        }
      }
    }
    
    throw lastError;
  }
  
  private async reconnect(): Promise<void> {
    console.log('Connection lost, reconnecting...');
    
    try {
      await this.disconnect();
    } catch (err) {
      // Ignore disconnect errors
    }
    
    await this.connect();
    await this.initialize({
      name: 'my_client',
      title: 'My Client',
      version: '1.0.0'
    });
  }
  
  private sleep(ms: number): Promise<void> {
    return new Promise(resolve => setTimeout(resolve, ms));
  }
}

Complete TypeScript Example

Putting it all together:

async function main(): Promise<void> {
  const client = new ResilientCodexClient();
  
  try {
    // Connect and initialize
    await client.connect();
    await client.initialize({
      name: 'my_ide_extension',
      title: 'My IDE Extension',
      version: '0.1.0'
    });
    
    // Setup notification handlers
    setupNotificationHandlers(client);
    
    // Start a thread
    const threadId = await client.requestWithRetry('thread/start', {
      cwd: process.cwd(),
      sandbox: 'workspaceWrite'
    });
    console.log('Thread created:', threadId);
    
    // Submit a turn
    await client.requestWithRetry('turn/submit', {
      threadId: threadId.thread.id,
      input: [{ type: 'text', text: 'List all TypeScript files in src/' }]
    });
    
    // Wait for completion (via notifications)
    await new Promise<void>((resolve) => {
      client.on('turn/completed', () => resolve());
    });
    
  } finally {
    await client.disconnect();
  }
}

main().catch(console.error);

Python Client Example

For Python environments:

import json
import subprocess
from typing import Any, Callable, Dict
from dataclasses import dataclass

@dataclass
class JSONRPCRequest:
    method: str
    id: int
    params: Dict[str, Any] | None = None

@dataclass
class JSONRPCResponse:
    id: int
    result: Any | None = None
    error: Dict[str, Any] | None = None

class CodexClient:
    def __init__(self):
        self.process = subprocess.Popen(
            ["codex", "app-server", "--listen", "stdio://"],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            bufsize=1  # Line buffered
        )
        self.next_id = 1
        self.pending_requests = {}
        self.notification_handlers = {}
        
        # Start stdout reader
        import threading
        self.reader_thread = threading.Thread(
            target=self._read_stdout,
            daemon=True
        )
        self.reader_thread.start()
    
    def _read_stdout(self):
        """Read and process messages from stdout."""
        for line in iter(self.process.stdout.readline, ''):
            if not line:
                break
            try:
                message = json.loads(line.strip())
                self._handle_message(message)
            except json.JSONDecodeError as e:
                print(f"Failed to parse: {line}", e)
    
    def _handle_message(self, message):
        """Route message to appropriate handler."""
        if 'id' in message:
            # Response
            pending = self.pending_requests.pop(message['id'], None)
            if pending:
                if message.get('error'):
                    pending['reject'](Exception(message['error']['message']))
                else:
                    pending['resolve'](message.get('result'))
        else:
            # Notification
            handler = self.notification_handlers.get(message['method'])
            if handler:
                handler(message.get('params', {}))
    
    def initialize(self, client_info: Dict[str, str]):
        """Initialize the connection."""
        response = self._request('initialize', {'clientInfo': client_info})
        self._send_notification('initialized', {})
        return response
    
    def _request(self, method: str, params: Dict[str, Any] | None = None):
        """Send a request and wait for response."""
        import concurrent.futures
        
        future = concurrent.futures.Future()
        request_id = self.next_id
        self.next_id += 1
        
        self.pending_requests[request_id] = {
            'resolve': future.set_result,
            'reject': future.set_exception
        }
        
        request = JSONRPCRequest(method, request_id, params)
        self._write_message(json.dumps(request.__dict__))
        
        return future.result(timeout=30)
    
    def _send_notification(self, method: str, params: Dict[str, Any]):
        """Send a notification (no response expected)."""
        notification = {
            'method': method,
            'params': params
        }
        self._write_message(json.dumps(notification))
    
    def _write_message(self, message: str):
        """Write a message to stdin."""
        self.process.stdin.write(message + '\n')
        self.process.stdin.flush()
    
    def on(self, method: str, handler: Callable[[Dict], None]):
        """Register a notification handler."""
        self.notification_handlers[method] = handler
    
    def start_thread(self, cwd: str) -> str:
        """Start a new thread."""
        response = self._request('thread/start', {
            'cwd': cwd,
            'sandbox': 'workspaceWrite'
        })
        return response['thread']['id']
    
    def submit_turn(self, thread_id: str, input_text: str):
        """Submit a turn to the thread."""
        self._request('turn/submit', {
            'threadId': thread_id,
            'input': [{'type': 'text', 'text': input_text}]
        })
    
    def disconnect(self):
        """Clean up the connection."""
        self.process.stdin.close()
        self.process.wait()

The app-server exposes powerful capabilities—from streaming conversations to direct file system access—all through a simple JSON-RPC interface. But those capabilities must be executed safely, which brings us to the sandboxing system.