Back to Blog
Tutorial

Michael Rodriguez7 min
Home

Most computer use setups block until a task finishes, which hides what your agent does in between. With Server-Sent Events (SSE) and Last-Event-ID you can stream events as they happen, let users watch in real time, and reconnect cleanly after a network hiccup. Coasty's GET /v1/runs/{id}/events endpoint gives you exactly this, emitting an event per step or state change.

How it works

Submit a Task Run with POST /v1/runs, then open an SSE stream with GET /v1/runs/{id}/events. Each event contains a type (queued, running, awaiting_human, succeeded, failed, cancelled, timed_out), a timestamp, and a payload (like fields.step or fields.message). If the connection drops, send the Last-Event-ID header on the next request to resume from that event without missing anything. You can cancel or resume a run later with POST /v1/runs/{id}/cancel and POST /v1/runs/{id}/resume.

python
import os, json, time, requests, threading

def run_task():
    api_key = os.getenv("COASTY_API_KEY")
    resp = requests.post(
        "https://coasty.ai/v1/runs",
        json={
            "machine_id": os.getenv("COASTY_MACHINE_ID"),
            "task": "Open Chrome and navigate to https://example.com",
            "cua_version": "v4",
            "max_steps": 40,
            "deadline_seconds": 300
        },
        headers={"X-API-Key": api_key}
    )
    resp.raise_for_status()
    run_id = resp.json()["id"]
    return run_id

def stream_events(run_id):
    api_key = os.getenv("COASTY_API_KEY")
    url = f"https://coasty.ai/v1/runs/{run_id}/events"
    headers = {"X-API-Key": api_key}
    last_id = None
    while True:
        params = {}
        if last_id:
            params["Last-Event-ID"] = last_id
        resp = requests.get(url, headers=headers, params=params, stream=True)
        resp.raise_for_status()
        for line in resp.iter_lines():
            if not line:
                continue
            line = line.decode("utf-8")
            if line.startswith("data:"):
                data = json.loads(line[5:])
                print(f"[{data.get('type')}] {data.get('fields', {}).get('message', '')}")
                event_id = data.get("id")
                if event_id:
                    last_id = event_id

def main():
    run_id = run_task()
    thread = threading.Thread(target=stream_events, args=(run_id,), daemon=True)
    thread.start()
    thread.join()

if __name__ == "__main__":
    main()

Event types and payloads

  • queued: run is queued on the server.
  • running: agent is executing steps, billed $0.05 per step.
  • awaiting_human: human approval was requested; set on_awaiting_human (pause, fail, cancel) when submitting the run.
  • succeeded, failed, cancelled, timed_out: final terminal states.
  • payload fields vary by type, often include message and step details.

Set Last-Event-ID on reconnect to resume from the last event without gaps.

Where this beats brittle automation

Instead of guessing IDs or waiting for a full HTTP response, you see each step in real time. You can pause and resume runs, handle human approval inline, and detect failures early. This is why computer use agents that see the screen and act like humans are more robust than pure API-only tools.

Start streaming live agent progress with SSE and Last-Event-ID. Build dashboards, dashboards for debugging, and observability tools that show runs in real time. Get a key at https://coasty.ai/developers to begin.

© 2026 Coasty

Backed byYCombinator