curl --request POST \
--url "${UNSTRUCTURED_API_URL}/api/v1/workflows/f0a1b2c3-4d5e-6f7a-8b9c-0d1e2f3a4b5c/run" \
--header "unstructured-api-key: ${UNSTRUCTURED_API_KEY}"
import os
from unstructured_client import UnstructuredClient
from unstructured_client.models.operations import RunWorkflowRequest
client = UnstructuredClient(
api_key_auth=os.getenv("UNSTRUCTURED_API_KEY"),
server_url=os.getenv("UNSTRUCTURED_API_URL"),
)
response = client.workflows.run_workflow(
request=RunWorkflowRequest(
workflow_id="f0a1b2c3-4d5e-6f7a-8b9c-0d1e2f3a4b5c"
)
)
print(response)
import asyncio
import os
from unstructured_client import UnstructuredClient
from unstructured_client.models.operations import RunWorkflowRequest
async def run_workflow():
client = UnstructuredClient(
api_key_auth=os.getenv("UNSTRUCTURED_API_KEY"),
server_url=os.getenv("UNSTRUCTURED_API_URL"),
)
response = await client.workflows.run_workflow_async(
request=RunWorkflowRequest(
workflow_id="f0a1b2c3-4d5e-6f7a-8b9c-0d1e2f3a4b5c"
)
)
print(response)
asyncio.run(run_workflow())
{
"id": "a1b2c3d4-5e6f-7a8b-9c0d-1e2f3a4b5c6d",
"workflow_id": "f0a1b2c3-4d5e-6f7a-8b9c-0d1e2f3a4b5c",
"workflow_name": "my-workflow",
"status": "SCHEDULED",
"created_at": "2026-04-29T10:00:00Z",
"runtime": null,
"input_file_ids": null,
"output_node_files": null,
"job_type": "ephemeral"
}
Workflow
Run workflow
Run a workflow manually, by triggering a new job if none is currently active.
POST
/
api
/
v1
/
workflows
/
{workflow_id}
/
run
curl --request POST \
--url "${UNSTRUCTURED_API_URL}/api/v1/workflows/f0a1b2c3-4d5e-6f7a-8b9c-0d1e2f3a4b5c/run" \
--header "unstructured-api-key: ${UNSTRUCTURED_API_KEY}"
import os
from unstructured_client import UnstructuredClient
from unstructured_client.models.operations import RunWorkflowRequest
client = UnstructuredClient(
api_key_auth=os.getenv("UNSTRUCTURED_API_KEY"),
server_url=os.getenv("UNSTRUCTURED_API_URL"),
)
response = client.workflows.run_workflow(
request=RunWorkflowRequest(
workflow_id="f0a1b2c3-4d5e-6f7a-8b9c-0d1e2f3a4b5c"
)
)
print(response)
import asyncio
import os
from unstructured_client import UnstructuredClient
from unstructured_client.models.operations import RunWorkflowRequest
async def run_workflow():
client = UnstructuredClient(
api_key_auth=os.getenv("UNSTRUCTURED_API_KEY"),
server_url=os.getenv("UNSTRUCTURED_API_URL"),
)
response = await client.workflows.run_workflow_async(
request=RunWorkflowRequest(
workflow_id="f0a1b2c3-4d5e-6f7a-8b9c-0d1e2f3a4b5c"
)
)
print(response)
asyncio.run(run_workflow())
{
"id": "a1b2c3d4-5e6f-7a8b-9c0d-1e2f3a4b5c6d",
"workflow_id": "f0a1b2c3-4d5e-6f7a-8b9c-0d1e2f3a4b5c",
"workflow_name": "my-workflow",
"status": "SCHEDULED",
"created_at": "2026-04-29T10:00:00Z",
"runtime": null,
"input_file_ids": null,
"output_node_files": null,
"job_type": "ephemeral"
}
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.
Path parameters
string
required
The unique identifier of the workflow to run.
Body
array
One or more files to process at runtime. Provide as a multipart file upload.
Response
string
required
Unique identifier for the job.
string
required
ID of the workflow that triggered this job.
string
required
Name of the workflow that triggered this job.
string
required
Job status:
SCHEDULED, IN_PROGRESS, COMPLETED, STOPPED, or FAILED.string
required
ISO 8601 timestamp when the job was created.
string
Job duration.
array
IDs of input files submitted with the job.
array
Metadata about output files produced by the job.
string
Job type. Defaults to
ephemeral.curl --request POST \
--url "${UNSTRUCTURED_API_URL}/api/v1/workflows/f0a1b2c3-4d5e-6f7a-8b9c-0d1e2f3a4b5c/run" \
--header "unstructured-api-key: ${UNSTRUCTURED_API_KEY}"
import os
from unstructured_client import UnstructuredClient
from unstructured_client.models.operations import RunWorkflowRequest
client = UnstructuredClient(
api_key_auth=os.getenv("UNSTRUCTURED_API_KEY"),
server_url=os.getenv("UNSTRUCTURED_API_URL"),
)
response = client.workflows.run_workflow(
request=RunWorkflowRequest(
workflow_id="f0a1b2c3-4d5e-6f7a-8b9c-0d1e2f3a4b5c"
)
)
print(response)
import asyncio
import os
from unstructured_client import UnstructuredClient
from unstructured_client.models.operations import RunWorkflowRequest
async def run_workflow():
client = UnstructuredClient(
api_key_auth=os.getenv("UNSTRUCTURED_API_KEY"),
server_url=os.getenv("UNSTRUCTURED_API_URL"),
)
response = await client.workflows.run_workflow_async(
request=RunWorkflowRequest(
workflow_id="f0a1b2c3-4d5e-6f7a-8b9c-0d1e2f3a4b5c"
)
)
print(response)
asyncio.run(run_workflow())
{
"id": "a1b2c3d4-5e6f-7a8b-9c0d-1e2f3a4b5c6d",
"workflow_id": "f0a1b2c3-4d5e-6f7a-8b9c-0d1e2f3a4b5c",
"workflow_name": "my-workflow",
"status": "SCHEDULED",
"created_at": "2026-04-29T10:00:00Z",
"runtime": null,
"input_file_ids": null,
"output_node_files": null,
"job_type": "ephemeral"
}
Was this page helpful?