app_server_advanced.md

Advanced App-Server Usage

This guide covers the lower-level app-server surface that sits under the thread convenience helpers.

Use it when you need typed stable RPC domains, protocol-native events, raw JSON-RPC fallbacks, or typed handlers for server-driven requests.

Import this surface from codex.app_server; the top-level codex package is reserved for the higher-level Codex facade.

Protocol-native event iteration

thread.run() returns a TurnStream that yields generated protocol models from codex.protocol.types.

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)
        elif isinstance(event, protocol.TurnCompletedNotificationModel):
            print("\nturn finished:", event.params.turn.status.root)

This is the right level when you want exact protocol semantics instead of a wrapper event layer.

Typed stable RPC domains

The advanced client exposes typed domain clients for stable non-thread RPC methods:

from codex.app_server import AppServerClient

with AppServerClient.connect_stdio() as client:
    models = client.models.list(limit=20, include_hidden=False)
    print(models.data[0].displayName)

    account = client.account.read()
    print(account.requiresOpenaiAuth)

    result = client.command.exec(command=["git", "status"], cwd="/repo")
    print(result.exitCode)

These domain clients cover the stable RPC surface that does not naturally belong on a thread object, including models, apps, skills, account, config, mcp_servers, feedback, command, external_agent_config, and windows_sandbox.

client.memory covers the experimental memory RPCs and requires experimental_api=True. client.memory.status() reports v2 readiness. client.memory.reset() irreversibly deletes all stored memories for the app-server's CODEX_HOME, not just the current thread. Per-thread controls live on the thread: thread.read_memory_summary() and thread.set_memory_mode("disabled").

set_memory_mode() only controls whether a thread is used to generate memories. A "disabled" thread still receives existing memories in its prompts; set the memories.use_memories config to false to stop that. With memories.disable_on_external_context on, the server can mark a thread polluted after outside content enters it, which also excludes it from generation. That state cannot be requested through the API, but calling set_memory_mode("enabled") overwrites it, so only do that if the thread's content is safe to learn from.

The concrete sync wrapper classes behind these attributes are internal implementation details. Rely on client.models and the other domain attributes instead of importing wrapper types directly.

Raw JSON-RPC

The rpc client remains the low-level escape hatch for experimental methods, version-skewed endpoints, or anything that does not have a typed wrapper yet.

from codex.app_server import AppServerClient

with AppServerClient.connect_stdio() as client:
    result = client.rpc.request("experimentalFeature/list", {"limit": 20})
    print(result)

If you want typed results, use request_typed() with one of the Pydantic models from codex.app_server.models or codex.protocol.types.

Connection-wide notifications

Use client.events.subscribe() when you want notifications outside a single turn stream.

from codex.app_server import AppServerClient, GenericNotification

with AppServerClient.connect_stdio() as client:
    subscription = client.events.subscribe({"codex/event/mcp_startup_update"})
    try:
        event = subscription.next()
        if isinstance(event, GenericNotification):
            print(event.method, event.params)
    finally:
        subscription.close()

Typed request handlers

Use on_request() when the server sends a JSON-RPC request that expects a client response.

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


def handle_tool_call(request: protocol.ItemToolCallRequest) -> dict[str, object]:
    return {
        "contentItems": [
            {
                "type": "text",
                "text": f"Handled {request.params.tool}",
            }
        ]
    }


with AppServerClient.connect_stdio() as client:
    client.on_request(
        "item/tool/call",
        handle_tool_call,
        request_model=protocol.ItemToolCallRequest,
    )
    thread = client.start_thread()
    thread.run_text("Use the configured dynamic tool if needed.")

The async client exposes the same pattern with async handlers. If you prefer the annotation-driven flow, use @dynamic_tool plus tools= on start_thread().

Handlers run concurrently and do not block other messages, so they can call back into the client and a slow handler no longer delays notifications. They may overlap, so protect any shared state. close() cancels handlers that are still running.

On the sync client, handlers run on a worker thread. Sync @dynamic_tool functions run on a worker thread on both clients. On the async client, a plain function passed to on_request() still runs on the event loop and must not block it; use an async handler for slow work.

For a complete decorator-driven dynamic-tool example that activates tools on start_thread() and handles the resulting item/tool/call requests, see examples/app_server_dynamic_tool.py.

When to use the advanced surface

Prefer the advanced app-server APIs when you need:

• stable typed access to lower-level RPC domains
• exact access to generated protocol notifications
• lower-level JSON-RPC methods not wrapped by a thread helper
• typed responses to server-driven requests
• explicit control over protocol details in a host integration

If you only want the final answer or a validated JSON payload, prefer the simpler thread helpers in app_server.md.