List workflows

Retrieve a list of workflows, optionally filtered by source, destination, state, name, date range, and supports pagination and sorting.

GET /api/v1/workflows/

cURL

curl --request GET \
  --url "${UNSTRUCTURED_API_URL}/api/v1/workflows/?status=active&page=1&page_size=20" \
  --header "unstructured-api-key: ${UNSTRUCTURED_API_KEY}"

import os
from unstructured_client import UnstructuredClient
from unstructured_client.models.operations import ListWorkflowsRequest

client = UnstructuredClient(
    api_key_auth=os.getenv("UNSTRUCTURED_API_KEY"),
    server_url=os.getenv("UNSTRUCTURED_API_URL"),
)

response = client.workflows.list_workflows(
    request=ListWorkflowsRequest(
        status="active",
        page=1,
        page_size=20,
    )
)
print(response)

import asyncio
import os
from unstructured_client import UnstructuredClient
from unstructured_client.models.operations import ListWorkflowsRequest

async def list_workflows():
    client = UnstructuredClient(
        api_key_auth=os.getenv("UNSTRUCTURED_API_KEY"),
        server_url=os.getenv("UNSTRUCTURED_API_URL"),
    )
    response = await client.workflows.list_workflows_async(
        request=ListWorkflowsRequest(
            status="active",
            page=1,
            page_size=20,
        )
    )
    print(response)

asyncio.run(list_workflows())

Response

[
  {
    "id": "f0a1b2c3-4d5e-6f7a-8b9c-0d1e2f3a4b5c",
    "name": "my-workflow",
    "workflow_type": "auto",
    "status": "active",
    "source_id": "7f3e2a1b-4c5d-6e7f-8a9b-0c1d2e3f4a5b",
    "destination_id": "1a2b3c4d-5e6f-7a8b-9c0d-1e2f3a4b5c6d",
    "schedule": "daily",
    "dag_nodes": null,
    "created_at": "2026-04-29T10:00:00Z",
    "updated_at": null
  }
]

You can specify multiple query parameters, for example ?source_id=<connector-id>&status=<status>.

Query parameters

source_id · string

Filter by source connector ID.

destination_id · string

Filter by destination connector ID.

status · string

Filter by workflow state.

Value	Description
active	Workflow is enabled and will run on its schedule.
inactive	Workflow is disabled and will not run.
paused	Workflow is temporarily paused.

name · string

Filter by workflow name.

page · integer

Page number for pagination. Default: 1.

page_size · integer

Results per page. Default: 20.

created_since · string

Return workflows created after this ISO 8601 timestamp.

created_before · string

Return workflows created before this ISO 8601 timestamp.

sort_by · string

Field to sort results by. Default: id.

sort_direction · string

Sort order. Default: asc.

Value	Description
asc	Ascending order.
desc	Descending order.

dag_node_configuration_id · string

Filter by DAG node configuration ID.

show_only_soft_deleted · boolean

If true, return only soft-deleted workflows. Default: false.

show_recommender_workflows · boolean

If true, include recommender system workflows. Default: false.

Response

id · string · required

Unique identifier for the workflow.

name · string · required

Workflow name.

workflow_type · string · required

Workflow type: custom or auto.

status · string · required

Workflow state: active, inactive, or paused.

created_at · string · required

ISO 8601 timestamp when the workflow was created.

source_id · string

Source connector ID.

destination_id · string

Destination connector ID.

schedule · string

Repeating run schedule.

dag_nodes · array

Workflow processing pipeline nodes.

For more information on workflow nodes, see Workflow nodes.

updated_at · string

ISO 8601 timestamp when the workflow was last updated.

cURL

curl --request GET \
  --url "${UNSTRUCTURED_API_URL}/api/v1/workflows/?status=active&page=1&page_size=20" \
  --header "unstructured-api-key: ${UNSTRUCTURED_API_KEY}"

import os
from unstructured_client import UnstructuredClient
from unstructured_client.models.operations import ListWorkflowsRequest

client = UnstructuredClient(
    api_key_auth=os.getenv("UNSTRUCTURED_API_KEY"),
    server_url=os.getenv("UNSTRUCTURED_API_URL"),
)

response = client.workflows.list_workflows(
    request=ListWorkflowsRequest(
        status="active",
        page=1,
        page_size=20,
    )
)
print(response)

import asyncio
import os
from unstructured_client import UnstructuredClient
from unstructured_client.models.operations import ListWorkflowsRequest

async def list_workflows():
    client = UnstructuredClient(
        api_key_auth=os.getenv("UNSTRUCTURED_API_KEY"),
        server_url=os.getenv("UNSTRUCTURED_API_URL"),
    )
    response = await client.workflows.list_workflows_async(
        request=ListWorkflowsRequest(
            status="active",
            page=1,
            page_size=20,
        )
    )
    print(response)

asyncio.run(list_workflows())

Response

[
  {
    "id": "f0a1b2c3-4d5e-6f7a-8b9c-0d1e2f3a4b5c",
    "name": "my-workflow",
    "workflow_type": "auto",
    "status": "active",
    "source_id": "7f3e2a1b-4c5d-6e7f-8a9b-0c1d2e3f4a5b",
    "destination_id": "1a2b3c4d-5e6f-7a8b-9c0d-1e2f3a4b5c6d",
    "schedule": "daily",
    "dag_nodes": null,
    "created_at": "2026-04-29T10:00:00Z",
    "updated_at": null
  }
]