Discover multiple flows

List with the CLI

Connect to your workspace, then run:

dagy flows list
dagy deployments list
dagy -e staging flows list
dagy deployments list --all-envs --format json

Listings use the current environment unless --all-envs is set. Supported formats are table, json, yaml, and csv.

dagy flows list --limit 50 uses pages of 50 and continues until the API has no next page. dagy deployments list --limit 50 requests one listing of at most 50. An empty CLI listing prints a human-readable message rather than an empty JSON array, even with --format json; use the API when you need a strict machine-readable response.

Page through flows in Python

DagyClient.list_flows() returns one page, not every matching flow. Continue with the response's opaque next_token until it is absent:

import os
from dagy import DagyClient

client = DagyClient(os.environ["DAGY_API_URL"])
next_token = None

while True:
    page = client.list_flows(limit=50, next_token=next_token)
    for item in page.get("items", []):
        print(item["flow_name"], item["flow_version"])
    next_token = page.get("next_token")
    if not next_token:
        break

This example uses saved CLI bearer credentials. For API-key integration, pass the key explicitly as DagyClient(base_url, token=key) and grant the required scopes. Do not treat pagination tokens as stable identifiers or construct them yourself.

To inspect a known flow or environment:

latest = client.get_flow_latest("order_pipeline")
if latest is None:
    print("The flow has not been registered in this organization")

page = client.list_deployments(environment="develop", limit=100)
for deployment in page.get("items", []):
    print(deployment["deployment_name"], deployment["flow_version"])

The named SDK list_flows method accepts only limit and next_token; use the CLI or public HTTP query parameters when you need additional filters. get_flow_latest returns None only for a 404 and raises APIError for authorization, connectivity, or other errors.

Trigger the selected deployment

response = client.trigger_run({
    "deployment": "orders-develop",
    "environment": "develop",
    "parameters": {"source": "sample"},
})
print(response["run_id"])

A trigger response means the request was accepted; it does not mean the run finished successfully. Use the runs API or workspace run detail to inspect final status, task outputs, and logs.

If a deployment is missing, check the active organization, environment, spelling, and permission to list flows. See API errors for the difference between authorization failure and not found.