The Workflow DSL Explained: Task, Assert, If, Loop, Parallel
Most automation tries to click a button by ID or selector. If the UI changes, the script breaks. The workflow DSL lets you orchestrate a computer use agent to drive the screen, validate results, and adapt to changes. You define a versioned JSON workflow, POST it to /v1/workflows, and Coasty executes it step by step, handling retries, cancellations, and pausing for human approval when needed.
Workflow DSL basics
- ●A workflow is a versioned JSON object describing a sequence of steps.
- ●Step types: task, assert, if, loop, parallel, human_approval, retry, succeed, fail.
- ●Variables follow double-brace syntax like {{inputs.x}} or stepId.field.
- ●Hard guards include budget_cents, max_iterations, and deadline_seconds.
- ●Task steps are billed $0.05 per agent step.
- ●A workflow run is created via POST /v1/workflows or POST /v1/workflows/{id}/runs.
Task steps
Task steps run a computer use agent to perform actions on the screen. Each task step can include instructions that append to the base prompt. The DSL passes machine_id and task to the underlying /v1/runs endpoint, which drives the agent on a real desktop, browser, or terminal. You can specify max_steps and deadline_seconds for a task.
Assert steps
Assert steps validate the current state after a task. They map a screenshot and an element description to coordinates using the /v1/ground endpoint, then check for expected conditions. If an assertion fails, the workflow can retry, pause for human approval, or mark the run as failed depending on your configuration.
If and loop conditionals
If steps evaluate structured conditions. You can check variable values, loop counters, or outcomes of previous steps. Loop steps iterate a block until a condition is met or max_iterations is reached. These conditionals let your agent adapt to dynamic state without external orchestration.
Parallel steps
Parallel steps run multiple sub-workflows or tasks concurrently. This reduces total runtime for independent operations, such as launching multiple browser tabs or checking several API endpoints from a VM. Each parallel branch gets its own state and can resolve into variables that subsequent steps can reference.
import os, json, requests, time
def create_workflow():
key = os.getenv("COASTY_API_KEY")
if not key:
raise RuntimeError("COASTY_API_KEY env var required")
url = "https://coasty.ai/v1/workflows"
headers = {
"Authorization": f"Bearer {key}",
"Content-Type": "application/json"
}
workflow = {
"name": "checkout_and_check",
"version": "1.0",
"steps": [
{
"type": "task",
"id": "place_order",
"task": "Go to shop.example.com, add a product to cart, and proceed to checkout.",
"instructions": "Use the computer use API to navigate the website and complete the purchase."
},
{
"type": "assert",
"id": "order_confirmed",
"description": "order confirmation page",
"expected": "order completed"
},
{
"type": "if",
"id": "is_prime",
"condition": {"type": "eval", "expression": "{{place_order.prime}}"},
"then": {
"type": "task",
"id": "get_prime_bonus",
"task": "Claim the Prime member bonus offer if visible."
},
"else": {
"type": "task",
"id": "check_coupons",
"task": "Look for any available coupon codes and apply one."
}
},
{
"type": "loop",
"id": "retry_until_paid",
"max_iterations": 3,
"condition": {"type": "eval", "expression": "{{retry_until_paid.step1.paid}} == false"},
"steps": [
{
"type": "task",
"id": "step1",
"task": "Retry payment by selecting the saved card and submitting."
},
{
"type": "assert",
"id": "payment_ok",
"description": "payment success message",
"expected": "payment successful"
}
]
},
{
"type": "parallel",
"id": "verify",
"branches": [
{
"type": "task",
"id": "verify_payment",
"task": "Confirm payment status in account history."
},
{
"type": "task",
"id": "verify_shipping",
"task": "Check shipping estimate on order details."
}
]
},
{
"type": "succeed",
"id": "finish",
"message": "Checkout completed successfully."
}
],
"hard_guards": {
"budget_cents": 1000,
"max_iterations": 100,
"deadline_seconds": 600
}
}
resp = requests.post(url, headers=headers, json=workflow)
resp.raise_for_status()
result = resp.json()
print("Workflow created", result.get("id"))
return result.get("id")
def run_workflow(workflow_id):
key = os.getenv("COASTY_API_KEY")
if not key:
raise RuntimeError("COASTY_API_KEY env var required")
url = f"https://coasty.ai/v1/workflows/{workflow_id}/runs"
headers = {
"Authorization": f"Bearer {key}",
"Content-Type": "application/json"
}
run = {
"machine_id": "your_machine_id",
"task": "Execute the workflow",
"cua_version": "v3",
"max_steps": 50,
"deadline_seconds": 600
}
resp = requests.post(url, headers=headers, json=run)
resp.raise_for_status()
run_result = resp.json()
print("Run started", run_result.get("id"))
return run_result.get("id")
def get_run_events(run_id):
key = os.getenv("COASTY_API_KEY")
if not key:
raise RuntimeError("COASTY_API_KEY env var required")
url = f"https://coasty.ai/v1/runs/{run_id}/events"
headers = {
"Authorization": f"Bearer {key}"
}
resp = requests.get(url, headers=headers, stream=True)
resp.raise_for_status()
for line in resp.iter_lines(decode_unicode=True):
if line:
print(line)
if __name__ == "__main__":
wid = create_workflow()
rid = run_workflow(wid)
get_run_events(rid)Use the workflow DSL to orchestrate task, assert, if, loop, and parallel steps with a single POST /v1/workflows.
Where this beats brittle automation
Traditional automation relies on fixed IDs, class names, or XPath selectors. A single UI change breaks the script. The workflow DSL lets your computer use agent see the screen, drive real desktops and browsers, and adapt to changing layouts. By combining tasks with asserts, conditionals, loops, and parallel branches, you can build agents that are resilient, reusable, and easier to maintain.
Start building resilient computer use agents with the workflow DSL. Define your steps, POST the workflow, and let Coasty execute the full trajectory. Get your API key at https://coasty.ai/developers today.