Create workflow

curl --request POST \
  --url "${UNSTRUCTURED_API_URL}/api/v1/workflows/" \
  --header "unstructured-api-key: ${UNSTRUCTURED_API_KEY}" \
  --header "Content-Type: application/json" \
  --data '{
    "name": "my-workflow",
    "workflow_type": "custom"
  }'

import os
from unstructured_client import UnstructuredClient
from unstructured_client.models.operations import CreateWorkflowRequest
from unstructured_client.models.shared import CreateWorkflow, WorkflowType

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

response = client.workflows.create_workflow(
    request=CreateWorkflowRequest(
        create_workflow=CreateWorkflow(
            name="my-workflow",
            workflow_type=WorkflowType.CUSTOM,
        )
    )
)
print(response)

import asyncio
import os
from unstructured_client import UnstructuredClient
from unstructured_client.models.operations import CreateWorkflowRequest
from unstructured_client.models.shared import CreateWorkflow, WorkflowType

async def create_workflow():
    client = UnstructuredClient(
        api_key_auth=os.getenv("UNSTRUCTURED_API_KEY"),
        server_url=os.getenv("UNSTRUCTURED_API_URL"),
    )
    response = await client.workflows.create_workflow_async(
        request=CreateWorkflowRequest(
            create_workflow=CreateWorkflow(
                name="my-workflow",
                workflow_type=WorkflowType.CUSTOM,
            )
        )
    )
    print(response)

asyncio.run(create_workflow())

{
  "id": "9b8c7d6e-5f4a-3b2c-1d0e-9f8a7b6c5d4e",
  "name": "my-workflow",
  "key": "my-workflow-a1b2c3",
  "workflow_type": "custom",
  "status": "active",
  "sources": [],
  "destinations": [],
  "workflow_nodes": [],
  "schedule": null,
  "reprocess_all": false,
  "skip_preflight": false,
  "edit_lock": null,
  "created_at": "2026-04-29T10:00:00Z",
  "updated_at": null
}

POST

api

workflows

curl --request POST \
  --url "${UNSTRUCTURED_API_URL}/api/v1/workflows/" \
  --header "unstructured-api-key: ${UNSTRUCTURED_API_KEY}" \
  --header "Content-Type: application/json" \
  --data '{
    "name": "my-workflow",
    "workflow_type": "custom"
  }'

import os
from unstructured_client import UnstructuredClient
from unstructured_client.models.operations import CreateWorkflowRequest
from unstructured_client.models.shared import CreateWorkflow, WorkflowType

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

response = client.workflows.create_workflow(
    request=CreateWorkflowRequest(
        create_workflow=CreateWorkflow(
            name="my-workflow",
            workflow_type=WorkflowType.CUSTOM,
        )
    )
)
print(response)

import asyncio
import os
from unstructured_client import UnstructuredClient
from unstructured_client.models.operations import CreateWorkflowRequest
from unstructured_client.models.shared import CreateWorkflow, WorkflowType

async def create_workflow():
    client = UnstructuredClient(
        api_key_auth=os.getenv("UNSTRUCTURED_API_KEY"),
        server_url=os.getenv("UNSTRUCTURED_API_URL"),
    )
    response = await client.workflows.create_workflow_async(
        request=CreateWorkflowRequest(
            create_workflow=CreateWorkflow(
                name="my-workflow",
                workflow_type=WorkflowType.CUSTOM,
            )
        )
    )
    print(response)

asyncio.run(create_workflow())

{
  "id": "9b8c7d6e-5f4a-3b2c-1d0e-9f8a7b6c5d4e",
  "name": "my-workflow",
  "key": "my-workflow-a1b2c3",
  "workflow_type": "custom",
  "status": "active",
  "sources": [],
  "destinations": [],
  "workflow_nodes": [],
  "schedule": null,
  "reprocess_all": false,
  "skip_preflight": false,
  "edit_lock": null,
  "created_at": "2026-04-29T10:00:00Z",
  "updated_at": null
}

This endpoint creates a workflow that persists until it is explicitly deleted (a long-lived workflow). To create a workflow that exists only for the duration of a single job run using local files as input, use the create job endpoint instead.

Body

string

required

Workflow name.

string

required

Workflow type. Use custom to configure processing nodes. The other accepted values, basic, advanced, and platinum, are non-operational and will be removed in a future release.

Workflows with neither source_id nor source_ids use a local file source. A local-source workflow must use custom and cannot have a repeating schedule. Run these workflows through the API or Python SDK because Unstructured Pipelines cannot run them.

string

ID of the source connector.

string

ID of the destination connector.

array

Processing nodes. Each node requires name, type, and subtype, and can include an id and settings object. For a custom workflow with source or destination connectors, this field is required.For more information on workflow nodes, see Workflow nodes.

string

ID of a pre-built workflow template to use as the basis for the workflow.

string

Repeating run schedule. Valid values and their cron equivalents:

Value	cron	Description
`every 15 minutes`	`/15 * * *`	Every 15 minutes.
`every hour`	`0 * * * *`	At the first minute of every hour.
`every 2 hours`	`0 /2 * *`	At the first minute of every second hour.
`every 4 hours`	`0 /4 * *`	At the first minute of every fourth hour.
`every 6 hours`	`0 /6 * *`	At the first minute of every sixth hour.
`every 8 hours`	`0 /8 * *`	At the first minute of every eighth hour.
`every 10 hours`	`0 /10 * *`	At the first minute of every tenth hour.
`every 12 hours`	`0 /12 * *`	At the first minute of every twelfth hour.
`daily`	`0 0 * * *`	At the first minute of every day.
`weekly`	`0 0 * * 0`	At the first minute of every Sunday.
`monthly`	`0 0 1 * *`	At the first minute of the first day of every month.

If omitted, the workflow does not automatically run on a repeating schedule.Workflows with a local source cannot be set to run on a repeating schedule.

boolean

Default: false. If true, reprocesses all documents in the source location on every run. If false, the workflow excludes from future processing any files Unstructured determines are unchanged since the last time the workflow ran.Unstructured determines if a document has changed based on the document version. For each workflow, Unstructured maintains a record of documents (and their versions, if present) processed by that workflow. Each document record consists of:

A record_id derived from the document name and path.
A record_version derived from either the document Etag (if the source provider generates one) or the source provider’s native version identifier.

When you set reprocess_all to false for a source connector that supports reprocess_all, Unstructured uses this list of records to determine whether or not to process each document:

If the record_id does not exist in the workflow records, Unstructured processes the document.
If the record_id exists, but the record_version has changed, or there is no record_version, Unstructured processes the document.

The following table lists out the possible record-id and record_version combinations, and the action Unstructured takes in each case:

`record_id`	`record_version`	Action
Exists	Unchanged	Do not process file
Exists	Changed	Process file
Exists	(none)	Process file
New	(Does not apply)	Process file

Renaming a document results in a new record_id; Unstructured will then reprocess the renamed document when the workflow runs.

The following table lists the source connectors that support the reprocess_all setting. The Record version base column specifies the versioning information Unstructured uses to generated the corresponding record version for each processed document.Source connectors that do not support reprocess_all reprocess every document in the source location each time the workflow runs.

Connector	`record_version` base
Amazon S3	ETag
Azure Blob Storage	ETag
Box	Provider version ID
Dropbox	Provider version ID
Elastisearch	Provider version ID
Google Cloud Storage	ETag
Google Drive	Provider version ID
Microsoft OneDrive	Provider version ID
Microsoft SharePoint	Provider version ID

Additional considerations to take into account when setting reprocess_all to false:

Unstructured only adds document records for documents that it successfully processes. Documents that failed to process will be reprocessed the next time the workflow is run.
Because S3 ETags are content-based, changing the metadata on an S3 object will not result in it being reprocessed.
For source providers that support the S3 protocol, be aware that deleting an object and then reuploading it to the source location will maintain the same record_id, but may result in a different record_version being generated. This is especially true of multipart uploads. This results in Unstructured reprocessing the document.
For source providers that offer Key Management Services (KMS), be aware that server-side encryption can change document ETags. This results in the the record_version of a document changing, and Unstructured reprocessing the document.
If you clone or recreate a source connector, the resulting connector does not include the document processing history of the previous connector.
Changing a workflow’s configuration does not automatically result in Unstructured reprocessing all documents. For example, changing chunker, embedder, enrichment, or partitioner settings may not result in reprocessing all document. To reprocess all documents using new workflow settings, set reprocess_all to true for at least the next workflow run.

array

Source connector IDs for a workflow with multiple sources. Use either source_ids or source_id.

array

Destination connector IDs for a workflow with multiple destinations. Use either destination_ids or destination_id.

boolean

Skip the job preflight check for this workflow. Preflight runs by default.

string

Workspace-scoped workflow identifier. If omitted, the API derives it from name with a six-character suffix. You cannot change the key after creating the workflow.

Response

string

required

Unique identifier for the workflow.

string

required

Workflow name.

string

Workflow type, when available: custom, basic, advanced, or platinum.

string

required

Workflow state: active or inactive.

string

required

ISO 8601 timestamp when the workflow was created.

object

Repeating run schedule, with crontab_entries containing cron_expression values.

string

ISO 8601 timestamp when the workflow was last updated.

array

required

Source connector IDs bound to the workflow.

array

required

Destination connector IDs bound to the workflow.

array

required

Workflow processing pipeline nodes.

string

Workspace-scoped workflow identifier.

boolean

true reprocesses all source documents on every run. With false, supported source connectors can skip unchanged documents. Other connectors still reprocess all documents.

boolean

true skips job preflight checks. false uses the default preflight behavior.

object

Edit lock details while someone edits the workflow in the console. Updates and deletes by other users return HTTP 423 until the editor saves or the lock expires.

curl --request POST \
  --url "${UNSTRUCTURED_API_URL}/api/v1/workflows/" \
  --header "unstructured-api-key: ${UNSTRUCTURED_API_KEY}" \
  --header "Content-Type: application/json" \
  --data '{
    "name": "my-workflow",
    "workflow_type": "custom"
  }'

import os
from unstructured_client import UnstructuredClient
from unstructured_client.models.operations import CreateWorkflowRequest
from unstructured_client.models.shared import CreateWorkflow, WorkflowType

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

response = client.workflows.create_workflow(
    request=CreateWorkflowRequest(
        create_workflow=CreateWorkflow(
            name="my-workflow",
            workflow_type=WorkflowType.CUSTOM,
        )
    )
)
print(response)

import asyncio
import os
from unstructured_client import UnstructuredClient
from unstructured_client.models.operations import CreateWorkflowRequest
from unstructured_client.models.shared import CreateWorkflow, WorkflowType

async def create_workflow():
    client = UnstructuredClient(
        api_key_auth=os.getenv("UNSTRUCTURED_API_KEY"),
        server_url=os.getenv("UNSTRUCTURED_API_URL"),
    )
    response = await client.workflows.create_workflow_async(
        request=CreateWorkflowRequest(
            create_workflow=CreateWorkflow(
                name="my-workflow",
                workflow_type=WorkflowType.CUSTOM,
            )
        )
    )
    print(response)

asyncio.run(create_workflow())

{
  "id": "9b8c7d6e-5f4a-3b2c-1d0e-9f8a7b6c5d4e",
  "name": "my-workflow",
  "key": "my-workflow-a1b2c3",
  "workflow_type": "custom",
  "status": "active",
  "sources": [],
  "destinations": [],
  "workflow_nodes": [],
  "schedule": null,
  "reprocess_all": false,
  "skip_preflight": false,
  "edit_lock": null,
  "created_at": "2026-04-29T10:00:00Z",
  "updated_at": null
}

Developer guide

API reference

Troubleshooting

Body

Response

​Body

​Response

Body

Response