--- title: Wrap a legacy app in a REST API | Lightcone description: Turn a GUI-only enterprise application into a REST endpoint with a FastAPI service, a per-workflow state machine, and one warm authenticated desktop. --- 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](/cookbook/warm-sessions-and-self-healing/index.md) 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](/guides/responses-api/index.md) 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](/cookbook/verified-structured-extraction/index.md). ## 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**](/cookbook/warm-sessions-and-self-healing/index.md): keep the authenticated desktop alive across restarts and repair it automatically - [**Verified structured extraction**](/cookbook/verified-structured-extraction/index.md): read data off the screen as strict JSON and verify before acting - [**Production**](/guides/production/index.md): deployment, timeouts, and capacity patterns - [**Logins and sessions**](/guides/logins-and-sessions/index.md): authenticating through SSO and persisting the result - [**Observability**](/guides/observability/index.md): recording sessions and tracing every request