The Workflow DSL Explained: task, assert, if, loop, parallel
Most automation struggles with flaky selectors and fragile workflows. You write a script, a UI changes, your tool breaks. The Coasty Workflow DSL solves this by encoding a versioned JSON workflow that the server executes as a series of Task Runs. Each step can describe a task, assert a state, branch with if conditions, repeat with loop guards, or run in parallel. The server drives a real desktop, browser, or terminal at $0.05 per agent step, so your DSL becomes a production-ready automation蓝图.
How the Workflow DSL works
You POST a workflow JSON to POST /v1/workflows. The body is a versioned DSL with step types like task, assert, if, loop, parallel, human_approval, retry, succeed, and fail. Each step can reference variables like {{inputs.x}} or stepId.field. You can guard steps with budget_cents, max_iterations, or deadline_seconds. After creating a workflow, you POST /v1/workflows/{id}/runs to execute it. The server creates a Task Run for the workflow, which internally spawns runs for each task step. Each run costs $0.05 per agent step. You can monitor progress via GET /v1/runs/{id}/events, which streams Server-Sent Events. The run states include queued, running, awaiting_human, succeeded, failed, cancelled, and timed_out.
import os
import json
import requests
COASTY_API_KEY = os.getenv("COASTY_API_KEY")
BASE_URL = "https://coasty.ai/v1"
workflow = {
"version": "1.0",
"steps": [
{
"stepId": "login",
"type": "task",
"task": "Log in to the app with username {{inputs.username}} and password {{inputs.password}}",
"assert": {
"stepId": "login_success",
"type": "assert",
"description": "Assert that user is logged in by checking dashboard appears"
}
},
{
"stepId": "loop_tasks",
"type": "loop",
"condition": {
"field": "count",
"operator": "<",
"value": "{{inputs.iterations}}"
},
"steps": [
{
"stepId": "task_{{count}}",
"type": "task",
"task": "Click menu, select Reports, open CSV download"
},
{
"stepId": "assert_{{count}}",
"type": "assert",
"description": "Assert that CSV download started"
}
]
},
{
"stepId": "parallel_checkout",
"type": "parallel",
"steps": [
{
"stepId": "checkout_sm",
"type": "task",
"task": "Checkout with credit card ending in 1234"
},
{
"stepId": "checkout_lg",
"type": "task",
"task": "Checkout with PayPal"
}
]
},
{
"stepId": "final_assert",
"type": "assert",
"description": "Assert all orders placed successfully"
},
{
"stepId": "succeed",
"type": "succeed"
}
]
}
headers = {
"X-API-Key": COASTY_API_KEY,
"Content-Type": "application/json"
}
response = requests.post(f"{BASE_URL}/workflows", headers=headers, json=workflow)
response.raise_for_status()
workflow_id = response.json()["id"]
print("Created workflow", workflow_id)
run_response = requests.post(
f"{BASE_URL}/workflows/{workflow_id}/runs",
headers=headers,
json={"max_steps": 1000, "deadline_seconds": 600}
)
run_response.raise_for_status()
run_id = run_response.json()["id"]
print("Started run", run_id)
# Stream events to check status
events_url = f"{BASE_URL}/runs/{run_id}/events"
while True:
r = requests.get(events_url, headers=headers, stream=True)
for line in r.iter_lines():
if line:
print(line.decode("utf-8"))
if b"data: succeeded" in line:
print("Workflow succeeded")
break
if "data: succeeded" in line.decode("utf-8"):
break
Step types and their purpose
- ●task: Describes a user action like clicking, typing, or navigating. Each task run costs $0.05 per agent step. The server executes tasks by driving a computer with vision.
- ●assert: Validates that a condition holds after a previous step. Use assert to confirm that a dashboard loaded, a file downloaded, or a UI element appears.
- ●if: Conditionally executes a block of steps when a condition evaluates to true. Conditions reference fields like stepId.field or input variables using double braces.
- ●loop: Repeats a block of steps while a condition is true. Use max_iterations and budget_cents as guards to prevent runaway loops.
- ●parallel: Runs multiple steps concurrently. Useful for parallel checkout paths, parallel API calls, or parallel tool invocations.
- ●human_approval: Pauses the workflow and waits for human confirmation. You can configure on_awaiting_human to pause, fail, or cancel the run.
- ●retry: Retries a previous step on failure. Use it to make workflows resilient to transient UI glitches.
- ●succeed: Marks the workflow as successful. If a workflow reaches succeed, the overall run state becomes succeeded.
- ●fail: Marks the workflow as failed. Use fail to enforce hard guards or to halt execution on critical errors.
Every task step is billed $0.05 per agent step, and you can guard loops and workflows with budget_cents, max_iterations, and deadline_seconds.
Where this beats brittle automation
Traditional automation relies on precise selectors, XPath, or API endpoints. When UIs change, selectors break, and you must rewrite scripts. The Coasty Workflow DSL describes what to do in natural language and lets the server see the screen. The server uses computer vision to locate elements, click, and type, so your workflow adapts to layout changes. You can assert states, branch on conditions, and run parallel tasks without hard-coding element IDs. This makes workflows durable across releases and easier to maintain for teams that do not want to manage brittle selectors.
Now you can build multi-step, stateful agents with conditionals, loops, and parallelism using the Coasty Workflow DSL. Define your workflow once, launch runs via POST /v1/workflows/{id}/runs, and monitor progress with Server-Sent Events. Get your API key at https://coasty.ai/developers and start building reliable computer use agents today.