Wrap a legacy app in a REST API

Plenty of business-critical software has no API: a legacy enterprise app behind SSO, reachable only through its GUI. This recipe wraps one workflow of such an app, “create a record”, in a REST endpoint. The pattern comes from a real production deployment and has three parts:

1. A FastAPI service that owns exactly one warm, already-authenticated desktop computer and serializes access to it.
2. A per-workflow state machine. Deterministic steps with measured coordinates, plus one bounded model-recovery step when a step fails.
3. Fail-closed verification. The endpoint returns 2xx only after reading an in-app success indicator off the screen.

This page assumes you already have a warm, logged-in desktop. Warm sessions and self-healing covers how to keep one alive.

The endpoint shape

One endpoint per workflow, one warm desktop behind them all, one request at a time. The status codes are part of the contract:

• 200/201. The app visibly confirmed the outcome. Safe to record as done.
• 503. We never touched the app (no healthy session, or it is busy). Safe for the caller to retry.
• 502. We acted but could not confirm the outcome. Not safe to retry blindly; route to human review.

app.py

import asyncio
import uuid

from fastapi import FastAPI, HTTPException, Response
from pydantic import BaseModel
from tzafon import AsyncLightcone

app = FastAPI()
client = AsyncLightcone()
lock = asyncio.Lock()          # one desktop => one request at a time
computer_id: str | None = None  # set at startup; see warm-sessions recipe


class RecordIn(BaseModel):     # your request schema; shape it to your app
    first_name: str
    last_name: str
    birth_date: str            # YYYY-MM-DD


@app.post("/records", status_code=201)
async def create_record(body: RecordIn, response: Response):
    if computer_id is None:
        # Fail closed BEFORE touching the app: retrying is safe.
        raise HTTPException(503, "no healthy session", headers={"Retry-After": "30"})
    if lock.locked():
        raise HTTPException(503, "worker busy", headers={"Retry-After": "10"})

    async with lock:
        ctx = {"record": body, "request_id": str(uuid.uuid4())}
        outcome = await CREATE_RECORD.run(client, computer_id, ctx)

    if not outcome.ok:
        # We touched the app but never saw the success indicator. The record
        # may or may not exist; surface that honestly instead of guessing.
        raise HTTPException(502, f"outcome unverified: {outcome.reason}")

    # Respond first, clean up after: reset latency is not the caller's problem.
    asyncio.create_task(reset_workspace(client, computer_id))
    return {"request_id": ctx["request_id"], "status": "created"}

Never map “the click sequence completed” to 2xx. The only thing that earns a success response is the application itself confirming the outcome on screen. Everything else is a 502.

A minimal per-workflow state machine

Free-form “agent, please create the record” runs are too variable for a production endpoint. Instead, express the workflow as named states with deterministic actions: clicks at coordinates you measured once from a screenshot, at a fixed display resolution. The model is not in the hot path; it appears only in a single bounded recovery state, entered when a deterministic step fails (a surprise modal, a slightly shifted layout).

workflow.py

from dataclasses import dataclass, field


@dataclass
class Outcome:
    ok: bool
    reason: str = ""
    trace: list[str] = field(default_factory=list)

# Each step: async (client, computer_id, ctx) -> (ok, reason)

async def search_record(client, cid, ctx):
    r = ctx["record"]
    await client.computers.click(cid, x=176, y=98)         # search field
    await client.computers.hotkey(cid, keys=["ctrl", "a"])
    await client.computers.type(cid, text=r.name)
    await client.computers.click(cid, x=176, y=182)        # Search button
    await asyncio.sleep(2)                                  # let results settle
    ctx["exists"] = await screen_says_yes(                  # see extraction recipe
        client, cid, f"Do the search results contain {r.name!r}? Answer YES or NO."
    )
    return True, ""

async def fill_form(client, cid, ctx):
    r = ctx["record"]
    await client.computers.click(cid, x=176, y=64)          # Clear form (F1)
    await client.computers.batch(cid, actions=[
        {"type": "click", "x": 176, "y": 98},
        {"type": "type", "text": r.name},
        {"type": "click", "x": 176, "y": 134},
        {"type": "type", "text": r.reference},
        {"type": "click", "x": 640, "y": 512},              # Save
    ])
    await asyncio.sleep(2)
    return True, ""

CREATE_STEPS = [("fill_form", fill_form),
                ("verify", verify_created)]

The runner attempts each step once; on failure it runs one bounded model-recovery pass and retries the step once. The “does it already exist?” check is explicit control flow, not a step. If the record is already there, we stop rather than create a duplicate (fail-closed, and idempotent):

workflow.py (runner)

MAX_RECOVERY_STEPS = 8  # hard budget for the model loop

async def attempt(client, cid, ctx, name, step, trace) -> tuple[bool, str]:
    ok, reason = await step(client, cid, ctx)
    if not ok:
        trace.append(f"{name} failed: {reason}")
        # One bounded recovery: let the model dismiss the surprise
        # (unexpected dialog, stale panel), then retry the step once.
        healed = await run_recovery(
            client, cid,
            "Close any dialog or popup and return the app to its main "
            "search screen. Do not save, delete, or submit anything.",
            max_steps=MAX_RECOVERY_STEPS,
        )
        ok, reason = (await step(client, cid, ctx)) if healed else (False, reason)
    trace.append(f"{name} ok" if ok else f"{name} failed: {reason}")
    return ok, reason

async def run_machine(client, cid, ctx) -> Outcome:
    trace: list[str] = []

    ok, reason = await attempt(client, cid, ctx, "search", search_record, trace)
    if not ok:
        return Outcome(ok=False, reason=f"search: {reason}", trace=trace)

    # Idempotency: if it already exists, stop instead of creating a duplicate.
    if ctx["exists"]:
        return Outcome(ok=True, reason="record already exists; nothing to create", trace=trace)

    for name, step in CREATE_STEPS:
        ok, reason = await attempt(client, cid, ctx, name, step, trace)
        if not ok:
            return Outcome(ok=False, reason=f"{name}: {reason}", trace=trace)
    return Outcome(ok=True, trace=trace)

run_recovery is a standard computer-use loop against the Responses API with the computer_use tool. The crucial detail is the max_steps budget and a prompt that forbids side effects. Recovery may only restore known state, never complete the business action; if it could, you would lose the ability to reason about what the endpoint did.

Fail-closed verification

The verify step is what makes the 2xx trustworthy. Do not infer success from “no error was thrown”. Read a positive indicator the app itself renders (a confirmation banner, a green checkmark, the new record’s ID on screen). Cheap pixel checks on a fixed region are fast and deterministic; poll a few times because legacy UIs paint slowly:

workflow.py (verify)

import io, httpx
from PIL import Image

SUCCESS_REGION = (24, 640, 220, 668)   # measured banner region
SUCCESS_RGB = (46, 160, 67)            # the app's confirmation green

async def verify_created(client, cid, ctx):
    for _ in range(6):
        shot = await client.computers.screenshot(cid)
        img = Image.open(io.BytesIO(httpx.get(shot.result.screenshot_url).content))
        crop = img.crop(SUCCESS_REGION).convert("RGB")
        hits = sum(
            1 for px in crop.getdata()
            if all(abs(a - b) < 40 for a, b in zip(px, SUCCESS_RGB))
        )
        if hits > 200:                 # banner is visibly painted
            return True, ""
        await asyncio.sleep(0.7)
    return False, "confirmation banner not visible after save"

If the banner never appears, the runner returns ok=False and the endpoint answers 502, the honest answer when a save may or may not have landed. A human (or a follow-up read-only lookup workflow) resolves it. For reading richer indicators like extracted IDs or matched names, use the model as described in Verified structured extraction.

Reset the workspace in the background

Every request must start from the same known screen, or your measured coordinates rot. But the caller should not wait for cleanup. Respond first, then reset:

app.py (reset)

async def reset_workspace(client, cid):
    async with lock:                    # same lock as requests: no interleaving
        await client.computers.hotkey(cid, keys=["esc"])
        await client.computers.hotkey(cid, keys=["esc"])
        await client.computers.click(cid, x=64, y=32)   # nav: main screen
        await client.computers.click(cid, x=176, y=64)  # clear all fields
        # Verify the reset the same fail-closed way; if it fails, mark the
        # session unhealthy so the next request 503s instead of typing into
        # a half-open form.

The reset holds the same lock as requests, so no caller can observe the desktop halfway through cleanup. If the reset itself fails, do not limp on. Mark the session unhealthy and let the self-healing loop repair it.

What you learned

1. One warm desktop, one lock, one request at a time. A serialized worker is far easier to reason about than a pool of half-authenticated ones.
2. Deterministic steps with measured coordinates, with the model confined to a bounded recovery state that may restore state but never complete the action.
3. Fail-closed status codes: 2xx only after reading an in-app success indicator; 503 means “never touched the app, retry”; 502 means “outcome uncertain, human review”.
4. Background workspace reset after every request, under the same lock, so each request starts from a known screen.

Next steps

• Warm sessions and self-healing: keep the authenticated desktop alive across restarts and repair it automatically
• Verified structured extraction: read data off the screen as strict JSON and verify before acting
• Production: deployment, timeouts, and capacity patterns
• Logins and sessions: authenticating through SSO and persisting the result
• Observability: recording sessions and tracing every request