Everything you need to know about Coasty

pip install requests

{
  "request_id": "req_abc123",
  "actions": [
    {
      "action_type": "click",
      "params": { "x": 512, "y": 340, "button": "left", "clicks": 1 }
    },
    {
      "action_type": "type_text",
      "params": { "text": "hello world" }
    }
  ],
  "reasoning": "I see a search bar at (512, 340)...",
  "status": "continue",
  "usage": {
    "input_tokens": 1523,
    "output_tokens": 245,
    "credits_charged": 5,
    "cost_cents": 5
  }
}

# Automate YOUR screen — no VM needed. pip install requests mss pyautogui pillow
import base64, io, time, uuid, requests, mss, pyautogui
from PIL import Image

API, KEY = "https://coasty.ai/v1", "sk-coasty-test-..."  # test key = free while you build
HDRS = {"X-API-Key": KEY}
pyautogui.FAILSAFE = True          # slam the mouse into a corner to abort instantly

REAL_W, REAL_H = pyautogui.size()  # your actual desktop resolution
SEND_W, SEND_H = 1280, 720         # what we tell the model (SD = 1 credit cheaper)
SX, SY = REAL_W / SEND_W, REAL_H / SEND_H   # scale model coords -> real pixels

def screenshot_b64():
    with mss.mss() as sct:
        shot = sct.grab(sct.monitors[1])                      # primary monitor
        img = Image.frombytes("RGB", shot.size, shot.bgra, "raw", "BGRX")
        img = img.resize((SEND_W, SEND_H))                    # MUST match screen_width/height
        buf = io.BytesIO(); img.save(buf, format="PNG")
        return base64.b64encode(buf.getvalue()).decode()

def execute(a):
    t, p = a["action_type"], a["params"]
    if t == "click":
        pyautogui.click(p["x"] * SX, p["y"] * SY,
                        clicks=p.get("clicks", 1), button=p.get("button", "left"))
    elif t == "type_text":  pyautogui.write(p["text"], interval=0.02)
    elif t == "key_press":  pyautogui.press(p["keys"])
    elif t == "key_combo":  pyautogui.hotkey(*p["keys"])
    elif t == "scroll":     pyautogui.scroll(p["clicks"])     # +up / -down, like pyautogui
    elif t == "drag":
        pyautogui.moveTo(p["x1"] * SX, p["y1"] * SY)
        pyautogui.dragTo(p["x2"] * SX, p["y2"] * SY, duration=0.4)
    elif t == "wait":       time.sleep(p["seconds"])
    return t

# A session keeps screenshot history so the model remembers what it already did.
sess = requests.post(f"{API}/sessions", headers=HDRS, json={
    "cua_version": "v3",
    "screen_width": SEND_W, "screen_height": SEND_H,
    # Optional best-practice steering (Starter+). Pick a preset from the docs:
    "instructions": "Click the visual center of elements. If the target is not visible, scroll toward it, never guess.",
}).json()
sid = sess["session_id"]

task = "Open the calculator and compute 42 * 17"
try:
    for step in range(25):
        r = requests.post(
            f"{API}/sessions/{sid}/predict",
            headers={**HDRS, "Idempotency-Key": f"step-{sid}-{step}-{uuid.uuid4().hex[:8]}"},
            json={"screenshot": screenshot_b64(), "instruction": task},
            timeout=120,
        ).json()
        print(f"step {r['step']}: {r['reasoning'][:80]}")
        for a in r["actions"]:
            if execute(a) in ("done", "fail"):
                raise SystemExit(f"finished: {r['status']} — {r['reasoning']}")
        time.sleep(0.5)                                       # let the UI settle
finally:
    requests.delete(f"{API}/sessions/{sid}", headers=HDRS)    # stop the session clock

Be precise. Before clicking, confirm the target element is actually visible in the CURRENT screenshot — never click from memory of a previous screen. Click the visual center of elements, not their edges. If the element you need is not visible, scroll toward where it should be instead of guessing coordinates. If two elements look similar, prefer the one whose text matches the task exactly. After typing into a field, verify focus landed in the right field before continuing.

requests.post(f"{API}/sessions", headers=HDRS, json={
    "cua_version": "v3",
    "screen_width": 1280, "screen_height": 720,
    "instructions": PRESET,   # the text above — applies to every step in the session
})

import requests

# Provision a fresh Linux desktop VM. Sandbox keys (sk-coasty-test-*)
# return a mock machine instantly with no AWS billing.
r = requests.post(
    "https://coasty.ai/v1/machines",
    headers={
        "X-API-Key": "sk-coasty-live-...",
        "Idempotency-Key": "provision-bot-001",   # safe to retry
    },
    json={
        "display_name": "automation-bot",
        "os_type": "linux",
        "desktop_enabled": True,
    },
)
machine = r.json()["machine"]
print(machine["id"], machine["status"])

import requests

machine_id = "..."  # from provision response
r = requests.post(
    f"https://coasty.ai/v1/machines/{machine_id}/actions",
    headers={"X-API-Key": "sk-coasty-live-..."},
    json={
        "command": "click",
        "parameters": {"x": 512, "y": 340},
    },
)
result = r.json()
print(result["success"], result["duration_ms"], "ms")

import requests

# Run a shell command (PowerShell on Windows, bash on Linux).
# Output is truncated VM-side to 5000 chars.
r = requests.post(
    f"https://coasty.ai/v1/machines/{machine_id}/terminal",
    headers={"X-API-Key": "sk-coasty-live-..."},
    json={
        "command": "uname -a && uptime",
        "timeout_ms": 10_000,
    },
)
print(r.json()["result"]["output"])

import requests

API_KEY = "sk-coasty-live-..."

# Start a task run. Returns status "queued" plus a ONE-TIME webhook_secret
# (only present when you pass webhook_url). Idempotency-Key makes retries safe.
r = requests.post(
    "https://coasty.ai/v1/runs",
    headers={
        "X-API-Key": API_KEY,
        "Idempotency-Key": "invoice-run-4821",
    },
    json={
        "machine_id": "550e8400-e29b-41d4-a716-446655440000",
        "task": "Open the invoice in the browser and read the total.",
        "cua_version": "v3",          # "v4" needs professional tier or above
        "max_steps": 40,
        "on_awaiting_human": "pause", # pause | fail | cancel
        "webhook_url": "https://your.app/hooks/coasty",
    },
)
run = r.json()
print(run["id"], run["status"])               # ... queued
webhook_secret = run.get("webhook_secret")    # shown once — store it now

import requests

# Stream lifecycle as Server-Sent Events. Reconnect with Last-Event-ID to
# replay everything after the last seq you saw — no gaps on a dropped socket.
with requests.get(
    f"https://coasty.ai/v1/runs/{run_id}/events",
    headers={"X-API-Key": API_KEY, "Last-Event-ID": "42"},
    stream=True,
) as r:
    for line in r.iter_lines(decode_unicode=True):
        if line.startswith("data:"):
            print(line[5:].strip())   # status / step / awaiting_human / terminal

import requests

# A workflow is a versioned JSON DSL composing many runs with branching,
# loops, parallelism, asserts, retries, and human approvals. {{var}} pulls
# from earlier steps' results (bound via save_as / step id).
r = requests.post(
    "https://coasty.ai/v1/workflows",
    headers={
        "X-API-Key": "sk-coasty-live-...",
        "Idempotency-Key": "ar-collections-v1",
    },
    json={
        "name": "AR collections",
        "definition": {
            "steps": [
                {"id": "invoice", "type": "task", "save_as": "invoice",
                 "machine_id": "550e8400-e29b-41d4-a716-446655440000",
                 "task": "Open the invoice and read its status + total."},
                {"id": "check", "type": "assert",
                 "condition": {"op": "truthy", "value": "{{invoice.passed}}"},
                 "message": "Agent failed to read the invoice"},
                {"id": "branch", "type": "if",
                 "condition": {"op": "contains",
                               "left": "{{invoice.result}}", "right": "PAID"},
                 "then": [{"id": "ok", "type": "succeed",
                           "output": {"state": "paid"}}],
                 "else": [{"id": "ask", "type": "human_approval",
                           "message": "Invoice unpaid — send reminder?"},
                          {"id": "fin", "type": "succeed",
                           "output": {"state": "reminded"}}]},
            ]
        },
    },
)
workflow = r.json()
print(workflow["id"], workflow["version"])

import requests

# Daily 9:00 AM ET email summary, fired by the Coasty scheduler.
# Per fire: needs >= 20 cr ($0.20) in your API wallet to dispatch (gate only);
# agent runtime then bills your Coasty credit balance at 10 credits/min.
r = requests.post(
    "https://coasty.ai/v1/schedules",
    headers={
        "X-API-Key": "sk-coasty-live-...",
        "Idempotency-Key": "morning-briefing-001",
    },
    json={
        "name": "morning briefing",
        "machine_id": "550e8400-e29b-41d4-a716-446655440000",
        "task_prompt": "Summarize unread Gmail and post the top 5 to Slack.",
        "frequency": "daily",
        "time": "09:00",
        "timezone": "America/New_York",
    },
)
schedule = r.json()
print(schedule["id"], schedule["next_run_at"])

# Add a webhook trigger — returns the signing secret ONCE.
r = requests.post(
    f"https://coasty.ai/v1/schedules/{schedule_id}/triggers",
    headers={"X-API-Key": "sk-coasty-live-..."},
    json={"kind": "webhook", "rate_limit_per_minute": 60},
)
trigger = r.json()
webhook_url    = trigger["webhook_url"]      # https://coasty.ai/v1/triggers/webhook/whk_...
webhook_secret = trigger["webhook_secret"]   # whsec_<64 hex>  — STORE THIS

# Save webhook_secret in your secrets manager. We hash + persist it; we
# cannot show it again. Lose it = generate a new trigger.

# Customer-side webhook signing — produces a Coasty-Signature header
# the public /v1/triggers/webhook/{id} endpoint accepts.
import hmac, hashlib, time

def sign_coasty_webhook(secret: str, body: bytes) -> dict:
    ts = int(time.time())
    signed_payload = f"{ts}.".encode("utf-8") + body
    sig = hmac.new(secret.encode("utf-8"), signed_payload, hashlib.sha256).hexdigest()
    return {"Coasty-Signature": f"t={ts},v1={sig}"}

# Fire the webhook from your own app:
import requests
body = b'{"event":"order.placed","order_id":"123"}'
headers = {**sign_coasty_webhook(webhook_secret, body), "Content-Type": "application/json"}
requests.post(webhook_url, data=body, headers=headers)