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())
[
{
"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
}
]
Workflow
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 --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())
[
{
"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
string
Filter by source connector ID.
string
Filter by destination connector ID.
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. |
string
Filter by workflow name.
integer
Page number for pagination. Default:
1.integer
Results per page. Default:
20.string
Return workflows created after this ISO 8601 timestamp.
string
Return workflows created before this ISO 8601 timestamp.
string
Field to sort results by. Default:
id.string
Sort order. Default:
asc.| Value | Description |
|---|---|
asc | Ascending order. |
desc | Descending order. |
string
Filter by DAG node configuration ID.
boolean
If
true, return only soft-deleted workflows. Default: false.boolean
If
true, include recommender system workflows. Default: false.Response
string
required
Unique identifier for the workflow.
string
required
Workflow name.
string
required
Workflow type:
custom or auto.string
required
Workflow state:
active, inactive, or paused.string
required
ISO 8601 timestamp when the workflow was created.
string
Source connector ID.
string
Destination connector ID.
string
Repeating run schedule.
array
Workflow processing pipeline nodes.For more information on workflow nodes, see Workflow nodes.
string
ISO 8601 timestamp when the workflow was last updated.
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())
[
{
"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
}
]
Was this page helpful?