Skip to content

app_server.md

Latest commit

 

History

History
244 lines (178 loc) · 7.89 KB

app_server.md

File metadata and controls

244 lines (178 loc) · 7.89 KB
Raw

App-Server API (AppServerClient)

AppServerClient connects to codex app-server and exposes a thread and stream API on top of the JSON-RPC protocol.

SDK 1.153.4 bundles Codex app-server 0.153.4; its generated protocol models include that version's experimental schema. Experimental RPCs still require AppServerInitializeOptions(experimental_api=True).

Use it when you need a deeper integration than Codex provides: persistent connections, typed protocol notifications, or server-driven requests.

Import the raw app-server surface from codex.app_server, not from the top-level codex facade.

Happy path

from codex.app_server import AppServerClient, AppServerClientInfo, AppServerInitializeOptions

initialize_options = AppServerInitializeOptions(
    client_info=AppServerClientInfo(
        name="my_integration",
        title="My Integration",
        version="0.1.0",
    )
)

with AppServerClient.connect_stdio(initialize_options=initialize_options) as client:
    thread = client.start_thread()
    summary = thread.run_text("Briefly summarize this repository's purpose.")
    print(summary)

Core objects

  • AppServerClient: manages the app-server connection
  • AppServerThread: represents a server-side thread
  • TurnStream: iterates over typed protocol notifications for a turn

Async equivalents are also available:

  • AsyncAppServerClient
  • AsyncAppServerThread
  • AsyncTurnStream

AsyncAppServerClient.connect_stdio() and connect_websocket() return an already started client.

If you use websocket transport, install the optional extra:

pip install "codex-python[websocket]"

connect_websocket() also accepts AppServerWebSocketOptions for explicit bearer auth, headers, subprotocols, and connection timeouts.

A minimal websocket example is available at examples/app_server_websocket_conversation.py.

Starting and resuming threads

from codex.app_server import AppServerClient

with AppServerClient.connect_stdio() as client:
    new_thread = client.start_thread()
    existing_thread = client.resume_thread("thr_123")

Thread objects expose lifecycle methods such as refresh(), fork(), archive(), revert(), compact(), and set_name().

Paginated thread history

Codex 0.151 defaults durable threads to paginated history when the active store supports it. Page turns or full persisted items with generated protocol response types. Explicitly selecting a history mode remains experimental and requires enabling the experimental API:

from codex.app_server import (
    AppServerClient,
    AppServerInitializeOptions,
    AppServerThreadStartOptions,
)
from codex.protocol import types as protocol

with AppServerClient.connect_stdio(
    initialize_options=AppServerInitializeOptions(experimental_api=True)
) as client:
    thread = client.start_thread(
        AppServerThreadStartOptions(history_mode=protocol.ThreadHistoryMode("paginated"))
    )
    turns = thread.list_turns_page(limit=50)
    items = thread.list_items_page(limit=100)

    for entry in items.data:
        print(entry.turnId, entry.item.root)

A new thread is materialized when its first user message starts; history pagination is unavailable before that point.

Use thread.revert("turn_123") to remove that turn and every later turn. The returned ThreadRevertResponse includes the updated thread and both backwards cursors. rollback() is the deprecated count-based operation and only works with legacy-history threads.

Resume with exclude_turns=True to avoid returning the entire history. The generated resume response retained on thread.resume_response contains turnsBackwardsCursor and itemsBackwardsCursor, which establish the durable-history boundary while newer records arrive as live notifications.

AppServerThreadListOptions supports parent_thread_id and ancestor_thread_id for spawned thread trees, plus project_id and section_id for project organization. Returned protocol.Thread values include parentThreadId, agentNickname, and agentRole when available. Paginated threads also expose search_occurrences_page() for typed message-search results and turn-navigation cursors.

Thread sections

Create, list, and rename sections through client.thread_sections; move or unsection a loaded thread directly:

with AppServerClient.connect_stdio() as client:
    section = client.thread_sections.create(name="Work")
    thread = client.start_thread()
    thread.move_to_section(section.id)
    thread.move_to_section(None)

AppServerThreadListOptions() omits section and project filters. Pass section_id=None or project_id=None explicitly to list unassigned threads.

Running turns

run()

Use run() when you want to consume protocol-native notifications:

from codex.app_server import AppServerClient
from codex.protocol import types as protocol

with AppServerClient.connect_stdio() as client:
    thread = client.start_thread()
    stream = thread.run("Investigate the failing tests")

    for event in stream:
        if isinstance(event, protocol.ItemAgentMessageDeltaNotification):
            print(event.params.delta, end="", flush=True)

run_text()

Use run_text() when you only want the final assistant text:

with AppServerClient.connect_stdio() as client:
    thread = client.start_thread()
    summary = thread.run_text("Summarize the repository")

run_json() and run_model()

Use these helpers when the turn is expected to return structured JSON:

from pydantic import BaseModel

from codex.app_server import AppServerClient, AppServerTurnOptions


class Summary(BaseModel):
    answer: str


schema = {
    "type": "object",
    "properties": {"answer": {"type": "string"}},
    "required": ["answer"],
    "additionalProperties": False,
}

with AppServerClient.connect_stdio() as client:
    thread = client.start_thread()
    payload = thread.run_json(
        "Return JSON matching the schema",
        AppServerTurnOptions(output_schema=schema),
    )
    payload_from_model = thread.run_json(
        "Return JSON matching the schema",
        AppServerTurnOptions(output_schema=Summary),
    )
    result = thread.run_model("Return JSON matching the schema", Summary)

run_model() validates the final assistant message text with pydantic and uses the model class as the output schema by default. If you want JSON back without validation, you can also pass a Pydantic model class directly to AppServerTurnOptions(output_schema=...).

Working with TurnStream

TurnStream keeps the protocol-native event stream, but also aggregates the final state for convenience:

  • final_text
  • final_message
  • final_turn
  • items
  • usage
  • text_deltas
  • final_json()
  • final_model(Model)

Example:

with AppServerClient.connect_stdio() as client:
    thread = client.start_thread()
    stream = thread.run("Summarize the repository")
    stream.wait()
    print(stream.final_text)

Sync and async usage

The sync client mirrors the stable blocking workflow and the current stable typed RPC domains such as client.models, client.account, client.config, and client.command.

The async client remains the canonical surface for lower-level extensibility and future protocol expansion:

from codex.app_server import AsyncAppServerClient


async def main() -> None:
    client = await AsyncAppServerClient.connect_stdio()
    try:
        thread = await client.start_thread()
        summary = await thread.run_text("Summarize the repository")
        print(summary)
    finally:
        await client.close()

For unsupported or experimental methods, use client.rpc on either client instead of expecting full sync/async wrapper parity for every future protocol addition.

For lower-level RPC access, typed request handlers, and protocol-native event iteration patterns, see app_server_advanced.md.