Coasty/

Developer API

Name: Coasty AI Employee
Brand: Coasty
Price: 19 USD
Availability: InStock
Rating: 4.8 (1250 reviews)

OpenAPI llms.txt

Workflows

Compose many runs with a versioned JSON DSL

Workflows

Workflows, at a glance

Compose many runs with a versioned JSON DSL

POST /v1/workflows

Request map4 boundaries

01
Define
JSON steps + inputs
02
Version
immutable run snapshot
03
Execute
tasks and control flow
04
Return
structured output

A workflow combines task runs and deterministic control steps into a versioned program with one durable output.

A workflow composes many runs into one versioned program, with branching, loops, and guards expressed as a JSON DSL. Each task step is itself an agent run, so a workflow is the way to chain tasks, gate them on conditions, and pass results between them. Workflows are versioned: re-creating the same slug bumps the version, and a PUT does too.

Create one with POST /v1/workflows. The slug must match [a-z0-9_-]. The response is a Workflow carrying an id, a version, and the current dsl_version (2026-06-01).

import os, requests

BASE = "https://coasty.ai/v1"
HEADERS = {"X-API-Key": os.environ["COASTY_API_KEY"]}

definition = {
    "steps": [
        {
            "id": "fetch",
            "type": "task",
            "task": "Open order {{inputs.order_id}} and read the invoice total",
            "save_as": "invoice",
        },
        {
            "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": "no", "type": "fail", "message": "Invoice not marked paid"}],
        },
    ],
}

# 1. Create the workflow. Re-using the same slug bumps its version.
wf = requests.post(
    f"{BASE}/workflows",
    headers=HEADERS,
    json={
        "name": "Invoice reconciliation",
        "slug": "invoice-reconcile",
        "inputs_schema": {"type": "object", "properties": {"order_id": {"type": "string"}}},
        "definition": definition,
    },
    timeout=30,
).json()
print(wf["id"], "v", wf["version"], wf["dsl_version"])

# 2. Start a run of the saved workflow.
run = requests.post(
    f"{BASE}/workflows/{wf['id']}/runs",
    headers=HEADERS,
    json={"inputs": {"order_id": "ord_4821"}, "machine_id": "mch_test_0123456789abcdef", "budget_cents": 500},
    timeout=30,
).json()
print(run["id"], run["status"])

Endpoint	Purpose
`POST /v1/workflows`	Create a workflow (or bump its version when the slug already exists).
`GET /v1/workflows`	List workflows. Filter with `?limit=`.
`GET /v1/workflows/{id}`	Fetch a workflow and its definition.
`PUT /v1/workflows/{id}`	Update any mutable field; every successful PUT bumps the version exactly once.
`DELETE /v1/workflows/{id}`	Archive a workflow.

A concurrent update based on a stale workflow version fails with 409 CONFLICT. Fetch the workflow again, reapply the intended change to the latest state, and retry.

Workflows need the workflows:read and workflows:write scopes, granted to new keys by default. See the Workflow DSL for the full step and condition catalogue.