Codex App Server in Production, Part 2: The FastAPI Bridge and the Event Vocabulary

One HTTP POST. One sentence about a coffee shop. 419 server-sent events fall out: a session_start, 379 fragments of prose, nineteen items opening and closing (seven stretches of reasoning, four shell commands, two file edits, six messages), and a receipt reading 159,680 tokens, 54.9 seconds. By the end of this page you'll watch that stream live in your terminal with curl -N, coming from your own FastAPI server. But the stream is the demo, not the product. The product is two designs that the next eleven parts will stand on without ever rewriting: a client class worth studying line by line, and a vocabulary with exactly six words in it. This is the most load-bearing page in the series, so it moves slowly where it matters.

Why a server at all

Part 1's script speaks fluent JSON-RPC, and it has three fatal properties: it runs on your machine, as your Codex login, with your filesystem. Nobody else can use it, and a browser must never be able to. A web page can't spawn subprocesses, shouldn't hold credentials, and has no business anywhere near the folder an agent writes to. Every AI product solves this the same way: the engine lives behind an HTTP endpoint on a machine you control, and clients talk to the endpoint. FastAPI is our front door. If FastAPI itself is new to you, LangGraph Part 2 teaches it from zero, and LangGraph Part 5 does the same for SSE; this part assumes both and spends its pages on what's genuinely new here: putting a bidirectional agent protocol behind a one-way wire.

From script to client class

Part 1 ended with a confession: the toy request() method reads lines until its reply appears and throws away every notification that arrives in between. For two requests in a straight line, fine. For a server, fatal, three times over. Requests and notifications interleave freely once turns run long. Two browser tabs mean two threads speaking over one stdio pipe, and their notifications must not get mixed. And somewhere in Act II, the engine starts sending us requests (approvals, in Part 7), which a read-until-my-reply loop would swallow whole.

So Part 2 opens with the real thing. Restructure the backend into a package and add the web dependencies:

cd backend
mkdir app && touch app/__init__.py
uv add fastapi==0.136.3 "uvicorn[standard]==0.49.0"

Then the first of three files, app/codex_client.py. Its state is six moving parts, and the whole class is those six parts cooperating:

import asyncio
import json
from collections import deque

MAX_LINE_BYTES = 8 * 1024 * 1024


class CodexError(Exception):
    """A JSON-RPC error reply, or a dead engine."""


class CodexClient:
    def __init__(self) -> None:
        self.proc = None
        self._next_id = 0
        self._pending: dict[int, asyncio.Future] = {}
        self._queues: dict[str, asyncio.Queue] = {}
        self._server_handlers: dict[str, callable] = {}
        self._stderr_tail: deque[str] = deque(maxlen=40)
        self._write_lock = asyncio.Lock()

Read the six as a mailroom. _next_id is the numbering stamp for outgoing notes. _pending is a wall of mailboxes, one per note we've sent and not yet heard back about. _queues is a row of pigeonholes, one per thread, where unnumbered notes get sorted. _server_handlers is a desk that's currently unstaffed; we're building it anyway, and there's a section below on why. _stderr_tail keeps the engine's last forty lines of stderr, which sounds paranoid until the day it's the only witness. And _write_lock guarantees one writer at a time, because two coroutines interleaving bytes into one stdin would corrupt the line framing that the whole protocol sits on. MAX_LINE_BYTES raises asyncio's default 64 KB line limit: a single notification can carry a whole file diff, and 64 KB of HTML is not a big page.

Starting the client is spawning the engine plus two background readers, then the handshake you already know:

    async def start(self) -> None:
        self.proc = await asyncio.create_subprocess_exec(
            "codex", "app-server",
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            limit=MAX_LINE_BYTES,
        )
        asyncio.create_task(self._read_stdout())
        asyncio.create_task(self._read_stderr())
        await self.request("initialize", {
            "clientInfo": {"name": "pagewright", "title": "Pagewright", "version": "0.2"},
            "capabilities": {},
        })
        await self._send({"jsonrpc": "2.0", "method": "initialized", "params": {}})

Same spawn as Part 1 with two upgrades: stderr is now piped instead of ignored, and reading is no longer something request() does inline. Two long-lived tasks own the pipes, full stop. Everything else in the class communicates with them through the six data structures.

Here's the sending half, and the pattern at its heart carries Parts 7, 10, and 11:

    async def request(self, method: str, params: dict, timeout: float = 60) -> dict:
        self._next_id += 1
        rid = self._next_id
        fut = asyncio.get_running_loop().create_future()
        self._pending[rid] = fut
        await self._send({"jsonrpc": "2.0", "id": rid, "method": method, "params": params})
        try:
            msg = await asyncio.wait_for(fut, timeout)
        finally:
            self._pending.pop(rid, None)
        if "error" in msg:
            raise CodexError(msg["error"].get("message", str(msg["error"])))
        return msg["result"]

    async def _send(self, obj: dict) -> None:
        async with self._write_lock:
            self.proc.stdin.write((json.dumps(obj) + "\n").encode())
            await self.proc.stdin.drain()

    async def _respond(self, rid, result: dict) -> None:
        await self._send({"jsonrpc": "2.0", "id": rid, "result": result})

request() no longer reads anything. It stamps a number, hangs an empty Future on the mailbox wall under that number, drops the note through the door, and goes to sleep on the future. When the reply eventually arrives, whoever finds it (the reader task, next fence) puts it in the right mailbox, and request() wakes up holding the answer. The finally guarantees the mailbox comes off the wall even on a timeout, and an "error" reply becomes a typed CodexError, keeping Part 1's discipline: errors are values, not mysteries. Note _respond too, three quiet lines for sending a reply rather than a request. Nothing calls it from our code yet. It exists because JSON-RPC is symmetrical, and the engine is allowed to ask us questions.

The reader task: one loop, three destinations

Everything the engine will ever say arrives through this one loop. It's the busiest twenty lines in Pagewright:

    async def _read_stdout(self) -> None:
        while True:
            line = await self.proc.stdout.readline()
            if not line:
                self._fail_pending("codex app-server exited: " + " | ".join(self._stderr_tail))
                return
            try:
                msg = json.loads(line)
            except json.JSONDecodeError:
                continue
            if "id" in msg and ("result" in msg or "error" in msg):
                fut = self._pending.get(msg["id"])
                if fut and not fut.done():
                    fut.set_result(msg)
            elif "id" in msg and "method" in msg:
                asyncio.create_task(self._dispatch_server_request(msg))
            elif "method" in msg:
                thread_id = (msg.get("params") or {}).get("threadId")
                if thread_id:
                    await self.queue_for(thread_id).put(msg)

The three-way switch is Part 1's Rosetta stone compiled into code, with one new case. An id plus a result or error is a response: find the mailbox, fill it. A method with no id is a notification: read its threadId and drop it in that thread's pigeonhole, so two turns running at once can never contaminate each other's stream. And the case Part 1 never met, an id and a method together: that's the engine sending us a request, expecting a response with that id back through the door. The grammar allowed it all along; we've only ever seen the polite half of the conversation.

The pigeonholes and the unstaffed desk are four lines each:

    def on_server_request(self, method: str, handler) -> None:
        """Register an async handler for server-initiated requests (Part 7)."""
        self._server_handlers[method] = handler

    def queue_for(self, thread_id: str) -> asyncio.Queue:
        """The notification mailbox for one thread."""
        return self._queues.setdefault(thread_id, asyncio.Queue())

    async def _dispatch_server_request(self, msg: dict) -> None:
        handler = self._server_handlers.get(msg["method"])
        if handler is None:
            await self._respond(msg["id"], {})
            return
        await self._respond(msg["id"], await handler(msg.get("params") or {}))

_server_handlers is empty and will stay empty for four more parts. So why build the seam now? Because it costs six lines today and a rewrite later. When Part 7 flips the approval policy and the engine starts asking "may I run this command?", the reader task already routes the question, the dispatcher already sends the answer, and the only work left is deciding. The alternative, retrofitting a request handler into a reader loop that never expected one, is the kind of surgery that breaks streams in production. This seam is the single most deliberate design decision in the class; the series will collect the interest in Part 7 and keep collecting it in Parts 10 and 11, when questions and review rides arrive through the same slot. Note the graceful default, too: an unhandled server request gets an empty response instead of silence, because the engine is blocked waiting on that id, and a contract you didn't answer is still a contract.

Last, the two smallest methods, which exist for the worst day:

    async def _read_stderr(self) -> None:
        while True:
            line = await self.proc.stderr.readline()
            if not line:
                return
            self._stderr_tail.append(line.decode(errors="replace").strip())

    def _fail_pending(self, reason: str) -> None:
        for fut in self._pending.values():
            if not fut.done():
                fut.set_exception(CodexError(reason))

The stderr reader files the engine's complaints into the ring buffer, forty lines deep, oldest evicted first. It's not idle chatter in there: botch one line of config.toml and the engine logs a precise ERROR codex_app_server: Invalid configuration ... string values must be quoted to stderr while your stdout stream stays innocent. And _fail_pending is the reader task's dying act: when readline() returns empty, the engine is gone, so every request still waiting on the mailbox wall gets an exception carrying those last words, immediately. Without it, a dead engine looks exactly like a slow engine for sixty seconds per pending request. With it, the crash report quotes the corpse. You'll see precisely when this fires, and when it can't, before this page ends.

Checkpoint. Right now you have: a client that can hold one engine process and any number of conversations with it, with nothing to talk to it yet. Two files to go.

One process, many threads

Notice what CodexClient is not: per-request. Pagewright will run exactly one codex app-server process per backend, spawned at startup, shared by every thread and every user. That's not thrift; it's how every official Codex surface works. The engine is built to multiplex: threads are cheap entries in its state, notifications carry threadId precisely so clients can demultiplex, and the rollout archive from Part 1 lives per-engine-home, not per-process. Spawning a fresh engine per HTTP request would pay process startup on every message, scatter your sessions across dying processes, and still not isolate anything the sandbox doesn't already isolate better.

Designing the envelope

Now the design section this part is named for. The browser needs to see the agent work. The agent's work arrives as app-server notifications. And we are not forwarding those notifications to the browser, even though it would cost zero code today. Method names like item/agentMessage/delta are the engine's interface, versioned by OpenAI on OpenAI's schedule, marked experimental besides. Every pixel of frontend coupled to them is a pixel that breaks when the protocol moves. So we mint our own vocabulary: every event is one JSON object with a type field, framed as a server-sent event. Labeled parcels, one conveyor. Six labels today:

Two properties make this table worth a section of prose. First, nothing in it says Codex. No method names, no threadId versus session_id leakage, no JSON-RPC. A client written against these six words is coupled to Pagewright, not to the engine inside it. That's not hypothetical portability: the production app this series is distilled from streams Claude and Codex through one contract shaped like this table, and its frontend genuinely cannot tell which engine is answering. Two engines, two protocols, one belt. Second, the parser contract is one sentence: switch on type, ignore types you don't recognize. The second clause is where the compounding happens. file_change joins the belt in Part 4, approval_request in Part 7, plan_update in Part 10, and the client you write in Part 3 keeps working through all of it, blind to labels it predates, crashing on none of them. We're designing today for parcels we haven't invented yet.

If you've ever integrated a vendor's API and had a minor version rename one field out from under you, you know the failure this table prevents. Everybody has one of these scars: the webhook payload that quietly changed shape, the SDK upgrade that turned a string into an object, the Friday deploy that was fine until a dependency's dependency wasn't. The translator we're about to write is the only code in Pagewright that will ever need to care when codex app-server changes its notifications, and it's one file. That's the whole argument for the envelope, and it's also why this section came before the code.

The translator

New file, app/events.py: the one file allowed to speak both languages. First the wire framing and a helper that decides what a parcel carries:

def sse(event: dict) -> str:
    """One envelope on the wire: data: {...}\n\n"""
    return f"data: {json.dumps(event)}\n\n"


def item_detail(item: dict) -> dict:
    """The few fields of an item worth putting on the wire, by kind."""
    kind = item.get("type", "unknown")
    if kind == "commandExecution":
        return {"command": item.get("command", "")}
    if kind == "fileChange":
        return {"files": [
            {"path": c.get("path", ""), "kind": c.get("kind", {}).get("type", "")}
            for c in item.get("changes", [])
        ]}
    if kind == "agentMessage":
        return {}
    return {}

sse() is the entire SSE format: a data: prefix and a blank-line delimiter (LangGraph Part 5 has the from-zero story, including the client bug that blank line causes). item_detail() encodes a quieter judgment: the wire narrates work, it doesn't haul cargo. A commandExecution item ships its command string, because a UI will print it on a badge. A fileChange ships paths and change kinds, not the diffs riding inside the notification, because nothing in Part 3 renders diffs. When Part 4 builds the diff drawer, the detail gets richer; today, restraint.

Then the mapping itself, Part 1's stream anatomy cashed in as a translate() function:

def translate(note: dict) -> dict | None:
    """Map one app-server notification onto the envelope, or drop it."""
    method, p = note["method"], note.get("params", {})
    if method == "item/agentMessage/delta":
        return {"type": "text_delta", "text": p.get("delta", "")}
    if method == "item/started":
        item = p.get("item", {})
        if item.get("type") in ("userMessage",):
            return None
        return {"type": "item_start", "item_id": item.get("id", ""),
                "kind": item.get("type", ""), "detail": item_detail(item)}
    if method == "item/completed":
        item = p.get("item", {})
        if item.get("type") in ("userMessage",):
            return None
        return {"type": "item_done", "item_id": item.get("id", ""),
                "kind": item.get("type", ""), "detail": item_detail(item)}
    if method == "turn/completed":
        turn = p.get("turn", {})
        return {"type": "complete", "status": turn.get("status", ""),
                "duration_ms": turn.get("durationMs")}
    if method == "error":
        return {"type": "error", "message": p.get("message", "unknown error")}
    return None

Four notification methods in, five parcel types out, and the two Nones teach as much as the mappings. The engine emits a userMessage item echoing what you sent, because rollout files replay whole conversations and the archive needs both sides. The wire doesn't: the client typed the message, it already rendered it, and echoing it back would paint every message twice. Dropped. And the final return None is the same forward-compatibility contract we imposed on the frontend, pointed the other way: when a future CLI version adds notification methods this translator never heard of, they fall through harmlessly instead of crashing the stream. Both sides of the translator ignore the unknown; that's what lets both sides evolve.

Look also at what translate() deliberately isn't: stateful. One notification in, at most one envelope out. Part 1's biggest gotcha, usage living in thread/tokenUsage/updated rather than on turn/completed, means someone has to remember the running totals and staple them to the receipt. That someone is not the translator; a pure function shouldn't hoard state. It's the endpoint, next.

The agent behind a URL

Third file, app/main.py. Setup first, all of it familiar shapes:

MODEL = "gpt-5.4-mini"
SITE = Path("site")

client = CodexClient()


@asynccontextmanager
async def lifespan(app: FastAPI):
    SITE.mkdir(exist_ok=True)
    await client.start()
    yield


app = FastAPI(lifespan=lifespan)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_methods=["*"],
    allow_headers=["*"],
)

The one-process rule becomes two lines of code: client is module-level, and lifespan starts the engine once, when uvicorn boots, not per request. The CORS block is forward wiring for Part 3's Next.js dev server; no browser calls this today, but the permission slip is signed. Then the turn runner, in two halves. First half, the setup and the opening parcel:

class ChatRequest(BaseModel):
    message: str


async def run_turn(message: str):
    thread = await client.request("thread/start", {
        "cwd": str(SITE.resolve()),
        "sandbox": "workspace-write",
        "approvalPolicy": "never",
        "model": MODEL,
    })
    thread_id = thread["thread"]["id"]
    queue = client.queue_for(thread_id)
    await client.request("turn/start", {
        "threadId": thread_id,
        "input": [{"type": "text", "text": message}],
    })

Part 1's two most consequential lines survive unchanged: workspace-write and approvalPolicy: "never", sane together because the sandbox walls are real (the scissors callout holds; Part 6 opens the dials). One subtle ordering matters here: queue_for(thread_id) runs before turn/start. Grab the pigeonhole before any mail can arrive, and no notification can slip through in the gap. Each /chat call starts a fresh thread for now; carrying a conversation across requests is Part 5's whole subject, and the engine is already archiving everything we'll need for it. Second half, the pump:

    yield sse({"type": "session_start", "session_id": thread_id})
    usage: dict = {}
    try:
        while True:
            note = await queue.get()
            if note["method"] == "thread/tokenUsage/updated":
                usage = note["params"].get("tokenUsage", {}).get("total", {})
                continue
            event = translate(note)
            if event is None:
                continue
            if event["type"] == "complete":
                event["usage"] = usage
            yield sse(event)
            if event["type"] in ("complete", "error"):
                return
    except CodexError as exc:
        yield sse({"type": "error", "message": str(exc)})

An async generator: every yield is one parcel leaving the building. The loop drains this thread's queue, keeps the latest usage totals to itself, translates everything else, and staples usage onto the complete parcel as it goes out the door, converting Part 1's "the receipt has no bill on it" gotcha into a wire format that pretends it never happened. No client of ours will ever know usage arrived separately. The except CodexError turns a client-layer failure into the sixth parcel type, because past the first byte of a stream, errors have to be data. And the endpoint itself is almost an anticlimax:

@app.post("/chat")
async def chat(req: ChatRequest):
    return StreamingResponse(run_turn(req.message), media_type="text/event-stream")

StreamingResponse pushes each yielded string out the socket the moment it exists. No buffering, no waiting for the turn to finish. That's the entire web layer of this part: three lines, because the two files above did their jobs.

Watch it with curl

Boot it from backend/:

uv run uvicorn app.main:app

And in a second terminal, the payoff. -N tells curl not to buffer:

Sit and watch a full run once; it's thirty to sixty seconds of genuine theater. The reasoning items open and close, the prose types itself out a word at a time, rg --files and ls -la scroll past as the agent cases the empty folder, and then a fileChange parcel lands carrying "kind": "add" for index.html. In my run the agent then did something I didn't ask for and want you to notice: it re-read its own HTML with sed, found a real problem, and a second fileChange came down the belt, "kind": "update", with the agent explaining it had caught a brand-name wrapper marked up as a span that should have been a block element. The proofreading habit you watched in Part 1's terminal survived the trip through two files and an HTTP socket, which is the point: nothing about the agent changed today. What changed is that anything, anywhere, can now watch it work.

The meter ritual, from that same complete parcel: 151,319 tokens in, 8,361 out, 54.9 seconds, and 129,280 of those input tokens, 85 percent, were cache hits. Same shape as Part 1: an agent turn is a conversation with itself, every readback rides back through the model, and caching is why the six-figure input number costs cents rather than dollars. The engine still quotes tokens, never currency, so the arithmetic against OpenAI's current price list remains yours to do. What's new is that the numbers now ship to every client on the receipt, which makes cost a UI feature from Part 3 onward instead of a server-side secret.

Break it on purpose: kill the engine mid-turn

The client walk promised you'd see the crash machinery in action, and an untested failure path is a rumor. So: start a build, and while the parcels are flowing, walk to a second terminal and murder the engine.

Read that capture closely, because what it doesn't show is the lesson. No error parcel. No dropped connection. The stream ends on an opened reasoning item and then produces nothing, forever, and curl sits there looking precisely the way it looks when the model is taking its time. Here's why, traced through the code you wrote twenty minutes ago. The reader task saw stdout close and ran _fail_pending, exactly as designed. But _fail_pending fails pending requests, and mid-turn there usually are none: thread/start and turn/start were answered within milliseconds, long ago, and run_turn is parked on queue.get(), waiting for notifications from a process that no longer exists. Nothing is pending. Nothing fails. The queue never fills again.

So the machinery has a precise jurisdiction, and honesty about it matters more than the demo. _fail_pending and the stderr tail protect every moment when a request is in flight: the handshake at boot (an engine that dies on startup fails initialize instantly, last words attached, instead of hanging your server for sixty seconds), and, from Part 7 on, every approval future that spends minutes on the mailbox wall while a human decides. What nothing in this part can protect is a turn that's already streaming, because a one-way wire has no heartbeat: silence and thinking are indistinguishable from the receiving end. That's not a bug in our client; it's a property of the wire, and it has a proper fix with a name, liveness plus a durable event log, which is Part 9's entire reason to exist. Until then the honest recovery is the blunt one: restart uvicorn (Ctrl-C, uv run uvicorn app.main:app), which respawns the engine through lifespan, and note what survived: site/index.html is still on disk, and the thread's rollout file is still in the archive. The workspace is truth; the stream is narration.

The pipeline, drawn

Everything this part built, one picture, and the two-worlds split is the thing to internalize:

Upstairs is the protocol world: bidirectional, engine-flavored, private. Downstairs is the web world: one-way, six-worded, public. Exactly one component stands on the staircase, translate(), and that's the whole architecture of this series in one sentence: when the experimental protocol upstairs changes shape, one file notices; when the product downstairs grows features, the belt grows labels; and neither world can reach around the translator to touch the other. The diagram also shows where every future part plugs in. Part 3 replaces the curl box. Part 4 enriches the translator. Part 7 staffs the dashed desk. Part 9 slides a database between the queue and the wire. The boxes never move again.

Commit it, from the pagewright/ project root, in a terminal that isn't hosting the server:

git add .
git commit -m "part 2: a real client, a FastAPI bridge, and six parcels on one belt"

Your agent now streams its work to anything that can open an HTTP connection, and the only thing that ever connects is curl. In Part 3 the stream gets a face: a chat UI where the prose types itself, commands become live badges with scrolling output, and the reasoning gets a drawer you can peek into.

Every part of this series has a companion folder in the codex-app-server-in-production repo: the complete, tested project exactly as it exists at the end of that part. This part's folder is part-02-fastapi-bridge. Code blocks with a GitHub icon in the header link straight to the exact file, and "View full file" shows the whole file in place with this section's lines highlighted.