DEV Community: Orkes Developers

Automating Serialization/Deserialization Tests with Orkes Conductor and LLMs

Orkes Developers — Thu, 29 May 2025 13:33:11 +0000

Writing unit tests for serialization and deserialization can be time-consuming and repetitive work. It’s monotonous and prone to human error, but still essential for ensuring that data is correctly sent across systems (API to server, application to application, SDK clients to server).

Unlike humans, software systems don’t get tired, bored, or distracted, even if faced with such mind-numbing work of writing serialization and deserialization (SerDe) tests.

Enter Orkes Conductor, a workflow orchestration platform. Combined with large language models (LLMs), Conductor transformed a week-long grind into a streamlined automation pipeline, cutting the testing time by over half.

In this post, we’ll walk through what SerDe tests are, why they matter, how we automated their creation using orchestration, and what made Orkes Conductor the ideal orchestration tool for the job.

Quick recap: serialization/deserialization

Serialization refers to the conversion of a data object from its current state into a format that can be stored in a database or transmitted through a network. For example, a Python object into JSON, or a Java object into a stream of bytes.

Deserialization is the exact opposite of serialization: it refers to the process of reconstructing the data object from its serialized format so that it can be used again. For example, from JSON back into a Python object so that it can be used in a function.

Serialization and deserialization are key processes for transmitting data across distributed systems in a standardized format while maintaining usability.

Why serialization/deserialization tests matter

Serialization and deserialization tests are critical for data integrity. They ensure that your data objects can:

Be accurately converted to/from different formats (such as from POJOs to JSON).
Retain structure and meaning across services.
Interoperate reliably between client-side and server-side components.

Neglecting these tests can introduce subtle, high-impact bugs, especially in systems where APIs evolve or where SDKs are maintained separately from backend logic.

Yet writing these tests is typically:

Manual: You have to craft JSON input/output cases for each class.
Boilerplate-heavy: The structure of each test is nearly identical, with minor differences in class names or references.
Error-prone: A wrong class reference or mismatched field can easily slip through.

That’s exactly what our engineering team was facing at Orkes: dozens of nearly identical tests with minor variations. Instead of slogging through them manually, we built an automated solution using Orkes Conductor.

Automating SerDe tests using Conductor + LLMs

In our case at Orkes, we needed to create tests to validate the mappings between SDK POJOs and server-side POJOs to ensure that data is transmitted correctly.

Accurate serialization and deserialization are crucial for the SDK's functionality, as they ensure that workflows and tasks are correctly interpreted by the Conductor server. By including comprehensive tests, developers can confidently make changes to the SDK without introducing regressions in data handling.

Using an orchestration tool like Orkes Conductor, we can automate repetitive actions across multiple services while ensuring that intermittent failures are remediated through retries and state persistence.

Here’s the high-level breakdown of the automated process:

Retrieve test resources like mapping files and JSON templates.
Generate a test branch in the repository.
Generate the SerDe tests using LLMs and prompt engineering.
Polish and publish the test code to the branch.

Let’s explore how Conductor’s enterprise functionalities enable seamless automation at each step.

1. Initial setup and data retrieval

Fetch a mapping file that links each SDK POJO to its corresponding server-side POJO.
Pull JSON templates generated from server POJOs. These templates will serve as the base for test generation.

{
  "templates" : {
    "IndexedDoc" : {
      "content" : {
        "score" : 123.456,
        "metadata" : {
          "sample_key" : "sample_value"
        },
        "docId" : "sample_docId",
        "text" : "sample_text",
        "parentDocId" : "sample_parentDocId"
      },
      "dependencies" : [ ]
    },
// rest of the JSON template

Parse the mapping file to each extract server-to-SDK relationship. Tests will be created for each server-SDK POJO mapping later.

Using built-in system tasks like HTTP Task and JSON JQ Transform Task, these steps can be quickly assembled without needing to code everything from scratch.

2. Branch management

Retrieve the SHA of the SDK repo’s target branch.
Use the SHA to create a new feature branch specifically for test generation.

Using dynamic references like “${variable}” in the HTTP Task’s URI https://api.github.com/repos/${previousTask.output.owner}/${previousTask.output.repo}/git/refs/heads/${workflow.input.sdkbranch}, the workflow is repeatable across multiple execution runs. Going beyond, the workflow becomes reusable and can be embedded in other workflows as well.

3. Iterative test generation using LLMs

Now we arrive at the heart of the workflow. For each server-SDK POJO mapping, the workflow will:

Fetch the SDK POJO source code and file SHA for version control.
Use the class path to generate a JSON representation.
Clean the Java class (remove annotations/comments) using GPT-4o.
Generate the actual test code via prompt engineering and LLM assistance.
Encode the test, create the proper file path, and commit the file to the new branch.

The suite of pre-built LLM tasks like LLM Text Complete facilitates rapid assembly of AI-powered tasks using any LLM provider. Conductor also supports prompt management, so you can create, test, and refine your prompts. Here is a snippet of an initial prompt to generate the test code:

4. Polish and publish

With all the test code generated, it’s time to wrap up:

Run Spotless to auto-format the code for style consistency.
Create a Pull Request for all the new test files.

Conductor tracks the data flow of every single task for each workflow run, making it easy to audit what has happened. Here is the execution trace of an example execution:

{
 "ownerApp": "user@example.com",
 "createTime": 1744977613276,
 "updateTime": 1744978814626,
 "createdBy": "user@example.com",
 "status": "COMPLETED",
 "endTime": 1744978814621,
 "workflowId": "af4f073b-1c4c-11f0-8645-4e34a87470b0",
 "tasks": [
   {
     "taskType": "HTTP",
     "status": "COMPLETED",
     "inputData": {
       "asyncComplete": false,
       "http_request": {
         "headers": {
           "Authorization": "token \${workflow.secrets.githubToken}",
           "Accept": "application/json"
         },
         "method": "GET",
         "uri": "https://raw.githubusercontent.com/orkes-io/sdk-codegen/base-code/conductor-java-sdk/SdkServerpojoMappings.json"
       },
     },
     "referenceTaskName": "getMappingFile",
     "retryCount": 0,
     "seq": 1,
     "pollCount": 1,
     "taskDefName": "get_mapping_file",
     "scheduledTime": 1744977613280,
     "startTime": 1744977613299,
     "endTime": 1744977613321,
     "updateTime": 1744977613299,
     "startDelayInSeconds": 0,
     "retried": false,
     "executed": true,
     "workflowInstanceId": "af4f073b-1c4c-11f0-8645-4e34a87470b0",
     "workflowType": "SDK_POJO_ser_deser_test",
     "taskId": "af4fa37c-1c4c-11f0-8645-4e34a87470b0",
     "callbackAfterSeconds": 0,
     "workerId": "orkes-workers-deployment-6c5b9f55dc-5xknz",
     "outputData": {
       "response": {
...

Why use Orkes Conductor?

Orchestration at scale

Conductor is built for chaining together complex, fault-tolerant workflows. This automation pipeline included over a dozen steps, with multiple dependencies across GitHub, OpenAI, and other services — exactly what Conductor was designed to handle.

LLM-ready

Not every task is rigid. For creative, generative work like cleaning up Java classes or writing test code, LLMs excel even when traditional automation breaks down. Conductor made it easy to drop in LLM-powered steps wherever flexible automation was required.

Reusable components

Once we built workers for custom tasks like applying Spotless auto-formatting and generating JSON templates, they became modular, reusable components. Now, creating a new workflow for a different use is just a matter of assembling these pieces together.

Developer-friendly

With REST APIs, SDKs, and a powerful UI, Conductor meets you where you are. Whether you need full developer tooling to build integrated workflows with your systems or visual dashboards to get complete visibility into the pipeline’s execution, Conductor’s product suite makes it seamless.

Scaling work from days to minutes

The original estimate for manually writing SerDe tests for this task was more than 5 days. Using Conductor and LLMs? Just 2.5 days to build and test the workflow itself. Now that the automated system exists, regenerating or extending the tests takes less than an hour.

More than just saving time, Orkes Conductor reframed how to think about developer productivity. Using Orkes Conductor with LLMs lets you:

Automate repetitive tasks, even those with variations.
Focus your energy on high-value work, like reviewing code and refining scalable systems.
Maximize your impact by reusing components across projects.

The next time you need to ask yourself: “Can I orchestrate this instead?”

Because chances are, with Orkes Conductor, the answer is yes. Take a stab at automating repetitive work using our free Developer Playground.

Workflow-Level Resilience in Orkes Conductor: Timeouts and Failure Workflows

Orkes Developers — Mon, 12 May 2025 09:07:15 +0000

Building resilient, production-grade workflows means preparing for the unexpected—from task stalls to external service outages. While task-level timeouts catch issues in isolated steps, workflow-level resilience settings act as a safety net for your entire orchestration. They ensure your system behaves predictably under stress and provides a graceful fallback when things go wrong.

In this post, we’ll explore two key features in Orkes Conductor that help you build robust workflows:

Workflow Timeouts
Failure Workflows (a.k.a. Compensation flows)

Workflow timeouts: Don’t let things hang

A workflow timeout defines how long a workflow is allowed to run before it's forcibly marked as timed out. This is crucial when your business logic needs to meet service-level agreements (SLAs) or avoid workflows stalling indefinitely.

Workflow timeout parameters

Parameter	Description
`timeoutSeconds`	Maximum duration (in seconds) for which the workflow is allowed to run. If the workflow hasn’t reached a terminal state within this time, it is marked as TIMED_OUT. Set to 0 to disable.
`timeoutPolicy`	Action to take when a timeout occurs. Supports: TIME_OUT_WF–Terminates the workflow as TIMED_OUT. ALERT_ONLY–Logs an alert but lets the workflow continue.

Use case: E-commerce checkout with 30-minute SLA

Imagine a checkout flow involving payment, inventory locking, and order confirmation. You don’t want stale carts holding inventory hostage for hours. A 30-minute timeout ensures the workflow either completes or fails cleanly.

Here’s a simplified implementation in Python using the Conductor SDK:

def register_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) HTTP task to fetch product price (simulated with dummy URL)
    fetch_random_number_task = HttpTask(
        task_ref_name="fetch_random_number",
        http_input={
            "uri": "https://www.random.org/integers/?num=1&min=1&max=100&col=1&base=10&format=plain&rnd=new",
            "method": "GET",
            "headers": {
                "Content-Type": "application/json"
            }
        }
    )

    # 2) Set variable for base price
    set_base_price = SetVariableTask(task_ref_name='set_base_price')
    set_base_price.input_parameters.update({
        'base_price': '${fetch_random_number.output.response.body}'
    })

    # 3) Inline task to calculate final price
    calculate_price_task = InlineTask(
        task_ref_name='calculate_final_price',
        script='''
            (function() {
                let basePrice = $.base_price;
                let loyaltyDiscount = $.loyalty_discount === "gold" ? 0.2 : 0;
                let promotionDiscount = $.promotion_discount ? 0.1 : 0;
                return basePrice * (1 - loyaltyDiscount - promotionDiscount);
            })();
        ''',
        bindings={
            'base_price': '${workflow.variables.base_price}',
            'loyalty_discount': '${workflow.input.loyalty_status}',
            'promotion_discount': '${workflow.input.is_promotion_active}',
        }
    )

    # 4) Set final calculated price
    set_price_variable = SetVariableTask(task_ref_name='set_final_price_variable')
    set_price_variable.input_parameters.update({
        'final_price': '${calculate_final_price.output.result}'
    })

    # Define the workflow with a 30-minute timeout
    workflow = ConductorWorkflow(
        name='checkout_workflow',
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.description = "E-commerce checkout workflow with 30-min timeout"
    workflow.timeout_seconds(1800)  # 30 minutes
    workflow.timeout_policy(TimeoutPolicy.TIME_OUT_WORKFLOW)

    workflow.add(fetch_random_number_task)
    workflow.add(set_base_price)
    workflow.add(calculate_price_task)
    workflow.add(set_price_variable)

    # Register the workflow definition
    workflow.register(overwrite=True)
    return workflow

Check out the full sample code for the E-commerce workflow.

If the workflow exceeds 30 minutes, it is marked as TIMED_OUT automatically, allowing you to alert a team, start a cleanup flow, or retry.

E-commerce workflow with a 30-minute timeout

Failure workflows: Your fallback plan

What happens when a workflow fails unexpectedly, due to a timeout, an API error, or an unhandled edge case? That’s where failure workflows come in.

These are separate workflows that are triggered when the main workflow fails. They allow you to compensate, clean up, and notify downstream systems or users.

Failure workflow parameters

Parameter	Description
`failureWorkflow`	The name of the fallback workflow to be triggered if this one fails. The default is empty.

Use case: Hotel booking with compensation flow

Let’s say your travel booking app orchestrates a hotel reservation workflow. If the booking fails (maybe the payment went through, but the room wasn’t confirmed), you’d want to:

Trigger a refund flow, and
Notify the customer that the booking failed

Main workflow code

def register_hotel_booking_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) HTTP task to reserve a hotel (simulated with dummy URL)
    reserve_hotel_task = HttpTask(
        task_ref_name="reserve_hotel",
        http_input={
            "uri": "https://httpbin.org/post",
            "method": "POST",
            "headers": {"Content-Type": "application/json"},
            "body": {
                "hotel_id": "${workflow.input.hotel_id}",
                "checkin": "${workflow.input.checkin_date}",
                "checkout": "${workflow.input.checkout_date}",
                "customer_id": "${workflow.input.customer_id}"
            }
        }
    )

    # 2) Set variable to confirm reservation status (simulate from body)
    set_status = SetVariableTask(task_ref_name='set_reservation_status')
    set_status.input_parameters.update({
        'reservation_status': '${reserve_hotel.output.response.body.json.status}'
    })

    # 3) Inline task to check booking status
    evaluate_reservation = InlineTask(
        task_ref_name='check_booking_status',
        script='''
            (function() {
                if ($.reservation_status !== 'confirmed') {
                    throw new Error("Booking failed");
                }
                return "confirmed";
            })();
        ''',
        bindings={
            'reservation_status': '${workflow.variables.reservation_status}'
        }
    )

    workflow = ConductorWorkflow(
        name='hotel_booking_workflow',
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.description = "Hotel reservation flow with SLA and failure handling"
    workflow.timeout_seconds(900)  # 15 minutes
    workflow.timeout_policy(TimeoutPolicy.TIME_OUT_WORKFLOW)
    workflow.failure_workflow("hotel_booking_failure_handler")

    workflow.add(reserve_hotel_task)
    workflow.add(set_status)
    workflow.add(evaluate_reservation)

    workflow.register(overwrite=True)
    return workflow

Failure workflow code

def register_failure_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Notify customer (simulated with dummy URL)
    notify_customer_task = HttpTask(
        task_ref_name="notify_customer",
        http_input={
            "uri": "https://httpbin.org/post",
            "method": "POST",
            "headers": {"Content-Type": "application/json"},
            "body": {
                "customer_id": "${workflow.input.customer_id}",
                "message": "Your hotel booking could not be completed. We apologize for the inconvenience."
            }
        }
    )

    # 2) Trigger refund (simulated with dummy URL)
    refund_payment_task = HttpTask(
        task_ref_name="trigger_refund",
        http_input={
            "uri": "https://httpbin.org/post",
            "method": "POST",
            "headers": {"Content-Type": "application/json"},
            "body": {
                "payment_id": "${workflow.input.payment_id}",
                "reason": "Hotel booking failed"
            }
        }
    )

    failure_workflow = ConductorWorkflow(
        name="hotel_booking_failure_handler",
        executor=workflow_executor
    )
    failure_workflow.version = 1
    failure_workflow.description = "Handles failed hotel bookings with customer notification and refund"

    failure_workflow.add(notify_customer_task)
    failure_workflow.add(refund_payment_task)

    failure_workflow.register(overwrite=True)
    return failure_workflow

Check out the full sample code for the hotel booking workflow.

Hotel booking workflow with a failure handler workflow

Best practices

Always define timeoutSeconds at both workflow and critical task levels to prevent resource overuse.
Use failureWorkflow for any workflow that produces side effects or artifacts that need cleanup in the event of failure.

Wrap up

Building production-ready workflows in Orkes Conductor means planning for both success and failure. Timeout policies and failure workflows aren’t just safeguards—they’re essential tools for maintaining system health, meeting SLAs, and ensuring a reliable user experience. When combined thoughtfully, they allow your workflows to self-regulate, recover from disruptions, and maintain a clean system state, even when things don’t go as planned.

—

Orkes Conductor is an enterprise-grade Unified Application Platform for process automation, API and microservices orchestration, agentic workflows, and more. Check out the full set of features, or try it yourself using our free Developer Playground.

Task-Level Resilience in Orkes Conductor: Timeouts and Retries in Action

Orkes Developers — Mon, 12 May 2025 08:52:28 +0000

In distributed systems, individual task failure is not a matter of if, but when. APIs go down, services stall, and workers disappear. What matters is how your system responds. With Orkes Conductor, you don’t just handle these failures—you design for them.

Conductor provides fine-grained control over how each task behaves under failure. With customizable timeouts and retries, you can recover from transient issues without human intervention, ensure critical steps don’t hang indefinitely, and build workflows that fail gracefully instead of catastrophically.

In this blog, we’ll explore three core capabilities that enable resilient task execution in Orkes Conductor:

Task Retries
Task Timeouts
System Task Timeouts

Task retries: Recovering from flaky failures

One of the most common failure scenarios is a transient error—momentary service unavailability, network hiccups, or throttling by an external API. Conductor lets you retry failed tasks automatically, using configurable backoff strategies to avoid overwhelming downstream services.

Retry parameters

Parameter	Description
`retryCount`	The maximum number of retry attempts. Default is 3.
`retryLogic`	Retry strategy for the tasks. Supports: FIXED–Retries after a fixed interval defined by `retryDelaySeconds`. LINEAR_BACKOFF–Retries occur with a delay that increases linearly based on `retryDelaySeconds x backoffScaleFactor x attempt_number`. EXPONENTIAL_BACKOFF–Retries occur with a delay that increases exponentially based on `retryDelaySeconds x (backoffScaleFactor ^ attempt_number`.
`retryDelaySeconds`	The delay between retries. This combines with the backoff logic to calculate the actual wait time. Note: The actual duration depends on the retry policy set in `retryLogic`.
`backoffScaleFactor`	Multiplier applied to `retryDelaySeconds` to adjust how fast delays increase. Default is 1.

Use case: Flaky email provider

Imagine your email provider fails intermittently. The first request sends a 500 error, but the second or third might succeed. This is a perfect scenario for retries.

from conductor.client.configuration.configuration import Configuration
from conductor.client.http.models import TaskDef
from conductor.client.orkes_clients import OrkesClients


def main():
    api_config = Configuration()
    clients = OrkesClients(configuration=api_config)
    metadata_client = clients.get_metadata_client()

    task_def = TaskDef()
    task_def.name = 'send_email_task'
    task_def.description = 'Send an email with retry on intermittent failures'
    task_def.retry_count = 3
    task_def.retry_logic = 'EXPONENTIAL_BACKOFF'
    task_def.retry_delay_seconds = 2
    task_def.backoff_scale_factor = 2

    metadata_client.register_task_def(task_def=task_def)

    print(f'Registered the task -- view at {api_config.ui_host}/taskDef/{task_def.name}')


if __name__ == '__main__':
    main()

Check out the full sample code for the send email task.

Here, if the email task fails, it will automatically retry up to 3 times with increasing delays—2s, 4s, and 8s—allowing time for the service to recover between attempts.

Email validation workflow using a Send Email task with exponential-backoff retries

Task timeouts: Preventing workflow stalls

Retries help you recover, but timeouts prevent you from getting stuck in the first place. Whether a worker goes offline or an external service hangs, task-level timeouts ensure your workflow doesn’t wait forever.

Timeout parameters

Parameter	Description
`pollTimeoutSeconds`	The time to wait for a worker to poll this task before marking it as TIMED_OUT.
`responseTimeoutSeconds`	The time to wait for a worker to send a status update (like IN_PROGRESS) after polling.
`timeoutSeconds`	Total time allowed for the task to reach a terminal state.
`timeoutPolicy`	Action to take when a timeout occurs: RETRY–Retries the task using retry settings. TIME_OUT_WF–Marks the whole workflow as TIMED_OUT. ALERT_ONLY–Logs an alert but lets the task continue.

Use case: Slow inventory API

Say you're calling a third-party inventory API that sometimes takes too long to respond. You don't want to wait forever, but you also don’t want to fail immediately. Here's how you'd configure a balanced timeout with retries:

from conductor.client.configuration.configuration import Configuration
from conductor.client.http.models import TaskDef
from conductor.client.orkes_clients import OrkesClients


def main():
    api_config = Configuration()
    clients = OrkesClients(configuration=api_config)
    metadata_client = clients.get_metadata_client()

    task_def = TaskDef()
    task_def.name = 'inventory_check_task'
    task_def.description = 'Check inventory status with timeout and retry settings'

    # Retry settings
    task_def.retry_count = 2
    task_def.retry_logic = 'FIXED'
    task_def.retry_delay_seconds = 5

    # Timeout settings
    task_def.timeout_seconds = 30
    task_def.poll_timeout_seconds = 10
    task_def.response_timeout_seconds = 15
    task_def.timeout_policy = 'RETRY'

    metadata_client.register_task_def(task_def=task_def)

    print(f'Registered the task -- view at {api_config.ui_host}/taskDef/{task_def.name}')


if __name__ == '__main__':
    main()

Check out the full sample code for the check inventory task.

This setup gives your worker 30 seconds to complete the task. If it doesn’t respond or fails, Conductor will retry it twice, waiting 5 seconds between each attempt.

Inventory workflow with a Check Inventory task with a 30-second timeout

Together, these retry and timeout configurations help you build workflows that are not just reactive, but resilient by design.

Next, we’ll look at how the same principles apply at the workflow level, giving you end-to-end control over your system’s behavior.

System task resilience

A key part of building resilient workflows is defining how long system tasks should wait on external services or heavy computations. In Orkes Conductor, each system task has default or configurable timeout settings to control this behavior.

HTTP task resilience

In Orkes Conductor, HTTP timeouts are defined by two parameters:

Parameter	Description	Default Values
`connectionTimeout`	The maximum time (in milliseconds) to establish a TCP connection to the remote server.	30 sec
`readTimeout`	The maximum time (in milliseconds) to wait for a response after the connection is established and the request is sent.	60 sec

In Orkes Conductor, these defaults are enforced to ensure platform stability.

Orkes Conductor making an HTTP request to an external server, illustrating timeout applications.

Internal timeouts resilience

Some system tasks have implicit timeout behaviors based on internal implementation. If you're designing workflows around system tasks, it's critical to understand and respect these limits.

System Task	Connection Timeout	Read Timeout
HTTP	30 sec	60 sec
LLM	60 sec	60 sec
Opsgenie	30 sec	60 sec
Inline	n/a	4 sec (max execution time)
Business Rule	10 sec	120 sec

Wrap up

Task failures are unavoidable, but with proper retry and timeout configurations, they don’t have to break your workflows. Conductor’s task-level resilience features help you avoid cascading failures, handle transient issues gracefully, and prevent workflows from hanging indefinitely.

In the next article, we’ll scale this approach up and explore workflow-level failure handling strategies like timeout policies and compensation flows that give you end-to-end resilience.

Next up:

Workflow-Level Resilience

—

Fail Fast, Recover Smart: Timeouts, Retries, and Recovery in Orkes Conductor

Orkes Developers — Mon, 12 May 2025 08:31:39 +0000

In distributed systems, failure isn’t a possibility—it’s a certainty. APIs hang, workers crash, and networks drop. What matters isn’t avoiding failure, but recovering from it quickly and cleanly.

Orkes Conductor gives you the tools to do exactly that. Timeouts and retries aren’t just configurable—they’re core to how Conductor ensures reliability at scale. In this post, you’ll learn how to use them effectively at both the task and workflow levels, how they interact with failure workflows, and how to design resilient systems that heal themselves without manual intervention.

Let’s dive in.

Why failure handling is critical in workflow applications

Imagine you're orchestrating a payment flow. After the user places an order, your workflow triggers tasks to charge the card, update inventory, and send a confirmation email.

All good—until the inventory API goes silent.

Now you're in a mess: the card is charged, but inventory isn’t updated, and the user doesn’t get a confirmation. This is exactly the kind of situation timeouts and retries are designed to avoid.

Unresponsive HTTP task in an inventory workflow

In Orkes Conductor, timeouts and retries work together to provide resilience and control:

Timeouts prevent your workflow from hanging indefinitely.
Retries recover from transient errors without manual intervention.

Together, they help you gracefully handle:

Unresponsive services: If a downstream API is taking too long, timeout the task and retry or move on.
Crashed or unresponsive workers: If a worker crashes mid-task, timeouts ensure the task doesn’t hang forever.
Transient errors: If a task fails quickly (like a network blip or 500 error), retries can automatically reattempt the operation.
Misconfigured or under-provisioned systems: If no worker is available to pick up a task, timeouts ensure the system doesn’t just wait forever.

This is more than resilience—it’s control. Instead of writing custom error-handling logic for every edge case, you get:

At-least-once delivery
Automatic retry logic
Configurable timeouts
Self-healing workflows

Task timeouts vs task retries

As you design resilient workflows, it’s important to understand the difference between timeouts and retries for individual tasks—they solve different problems:

Aspect	Retry	Timeout
What it is	A second (or third, etc.) attempt to run the same task after it fails.	A time limit for how long a task is allowed to run (or how long you wait for it to start/progress).
Trigger	Happens after a task fails (e.g., returns FAILED).	Happens if the task takes too long to respond to or complete.
Example	API call fails with 500 Internal Server Error → Retry it after a delay.	API call hangs with no response after 30 seconds → Timeout triggers a recovery action.
Configured via	`retryCount`, `retryLogic`, `retryDelaySeconds`, `backoffScaleFactor`	`timeoutSeconds`, `pollTimeoutSeconds`, `responseTimeoutSeconds`, `timeoutPolicy`

In short:

Retries deal with tasks that fail fast.
Timeouts deal with tasks that hang slow.

Both are essential for robust, fault-tolerant tasks.

Failed worker resulting in task-timeout and retries

Workflow timeouts and failure workflows

Just like tasks, entire workflows can time out or fail, and Conductor gives you powerful tools to handle those cases gracefully.

Aspect	Workflow Timeout	Failure Workflow
What it is	A maximum time limit for the whole workflow to complete.	A backup workflow that runs automatically when the original workflow fails.
Trigger	Happens when the workflow exceeds its configured timeout.	Happens when the workflow ends in FAILED or TERMINATED status.
Example	A data pipeline takes too long (e.g., > 2 hours) → Workflow is marked TIMED_OUT.	A workflow fails due to repeated task failures → A failure workflow sends alerts and logs diagnostics.
Configured via	`timeoutSeconds`, `timeoutPolicy`	`failureWorkflow`

In short:

Workflow timeouts protect against indefinitely running flows.
Failure workflows let you define recovery logic and postmortems for failed flows.

Together, they make workflows self-aware and recoverable.

Long-running worker exceeds workflow timeout and triggers compensation workflow

Combining retries, timeouts, and failure Workflows

Retries give tasks more chances. Timeouts limit how long each attempt can run. Workflow timeouts cap the total duration of the workflow. If all else fails, failure workflows take over.

Here’s how it all fits together:

Task fails → Task Retry: it’s retried (up to retryCount).
Each retry has a timeout → Task Timeout: Limits how long you wait per attempt.
Too many failures or long delay → Workflow Timeout: Workflow fails or times out.
Workflow fails → Failure Workflow: A failure workflow runs to clean up or alert.

This layered approach gives you fault-tolerance at every level:

Default vs custom configuration: When to override

Orkes Conductor ships with sane defaults.

Task retry settings

{
  "retryCount": 3,
  "retryDelaySeconds": 60,
  "retryLogic": "FIXED",
  "backoffScaleFactor": 1
}

Task timeout settings

{
  "responseTimeoutSeconds": 600,
  "timeoutSeconds": 3600,
  "pollTimeoutSeconds": 3600,
  "timeoutPolicy": "TIME_OUT_WF"
}

Workflow settings

{
  "timeoutSeconds": 0,
  "timeoutPolicy": "ALERT_ONLY",
  "failureWorkflow": ""
}

These are a solid starting point, but not always ideal for your SLAs.

Override the defaults when:

You have strict latency goals, such as completing checkout in under 5 minutes.
You interact with unreliable third-party APIs, like a flaky partner API.
You need predictable system behavior under load or failure.

On the flip side, don’t override just for the sake of it. Defaults help reduce configuration sprawl and are good enough for many internal service workflows.

Wrap up

In distributed systems, failure is inevitable, but downtime doesn’t have to be. With Orkes Conductor, resilience is built in, not bolted on.

Timeouts help prevent indefinite hangs, and retries offer automatic recovery from transient issues. Workflow-level timeouts and failure workflows ensure your entire system can fail gracefully and bounce back, without manual intervention.

By combining these tools thoughtfully, you get more than reliability—you get confidence—confidence that your workflows will keep running, even when parts of your system don’t.

So, whether you're orchestrating high-stakes payment flows or background data pipelines, remember: design for failure, recover automatically, and keep moving forward.

Go more in-depth with task and workflow resilience with these follow-up articles:

—

Orkes Operators: Parallelism and Reusability

Orkes Developers — Mon, 28 Apr 2025 12:57:39 +0000

Workflows don’t just move forward—they branch, scale, and reuse logic. Whether you’re processing tasks in parallel, delegating work to subflows, or triggering asynchronous side-jobs, these patterns let you build smarter, more modular orchestration.

In this post, we’ll explore operators that boost concurrency and composability in Orkes Conductor:

Fork/Join for running multiple branches in parallel
Dynamic Fork for scalable, input-driven concurrency
Sub Workflow for reusing logic across workflows
Start Workflow for async, fire-and-forget execution

From concurrent notifications to reusable payment flows, these operators help you scale orchestration cleanly—without duplicating logic or overcomplicating control flow. Let’s dive in.

Fork/Join

The Fork/Join task in Orkes Conductor enables parallel execution of multiple task branches within a workflow. When a Fork task is reached, it splits the workflow into separate paths that run concurrently, allowing different sequences of tasks to execute in parallel. Once all branches are complete, a Join task is used to synchronize them before continuing with the rest of the workflow. This pattern is ideal for optimizing performance and handling tasks that can be executed independently, such as running multiple validation checks or data fetch operations simultaneously.

Example implementation

A common use case for the Fork/Join task is in a multi-channel notification workflow. Suppose a Fork task dispatches three parallel notifications—email, SMS, and HTTP. Each branch sends its respective message independently to maximize concurrency. Since HTTP responses can be retried and don’t need to block the overall flow, the Join task is configured with joinOn and is set to only the email_notification and sms_notification tasks. This allows the workflow to proceed as soon as the critical email and SMS tasks are complete, while the HTTP notification continues in the background. This setup enables partial synchronization—balancing reliability with improved efficiency.

Here’s the workflow visualized:

Workflow using a Fork/Join operator

Here’s the code snippet for creating the workflow in code:


def register_notification_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Define each notification path
    email_branch = [
        SimpleTask(
            task_def_name="process_notification_payload",
            task_reference_name="process_notification_payload_email"
        ),
        SimpleTask(
            task_def_name="email_notification",
            task_reference_name="email_notification_ref"
        )
    ]

    sms_branch = [
        SimpleTask(
            task_def_name="process_notification_payload",
            task_reference_name="process_notification_payload_sms"
        ),
        SimpleTask(
            task_def_name="sms_notification",
            task_reference_name="sms_notification_ref"
        )
    ]

    http_branch = [
        SimpleTask(
            task_def_name="process_notification_payload",
            task_reference_name="process_notification_payload_http"
        ),
        HttpTask(
            task_ref_name="http_notification_ref",
            http_input={
                "uri": "${workflow.input.http_target_url}",
                "method": "POST",
                "headers": {
                    "Content-Type": "application/json"
                },
                "body": {
                    "message": "Notification triggered"
                }
            }
        )
    ]

    # 2) Fork-Join setup (only join on email + sms)
    fork_join = ForkTask(
        task_ref_name="my_fork_join_ref",
        forked_tasks=[email_branch, sms_branch, http_branch],
        join_on=["email_notification_ref", "sms_notification_ref"]
    )

    workflow = ConductorWorkflow(
        name="notification_workflow_with_fork_join",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(fork_join)
    workflow.register(overwrite=True)

    return workflow


@worker_task(task_definition_name="process_notification_payload")
def process_notification_payload() -> dict:
    print("🛠️ Processing notification payload...")
    return {"payload_processed": True}


@worker_task(task_definition_name="email_notification")
def email_notification() -> dict:
    print("📧 Email sent to test@example.com")
    return {
        "email_sent_at": "2021-11-06T07:37:17+0000",
        "email_sent_to": "test@example.com"
    }


@worker_task(task_definition_name="sms_notification")
def sms_notification() -> dict:
    print("📱 SMS sent to +1-xxx-xxx-xxxx")
    return {
        "sms_sent_at": "2021-11-06T07:37:17+0129",
        "sms_sent_to": "+1-xxx-xxx-xxxx"
    }

Check out the full sample code for the Fork/Join Workflow.

Dynamic Fork

The Dynamic Fork task allows you to create and execute multiple parallel task branches based on dynamic input at runtime. Unlike a standard fork, which defines branches statically, Dynamic Forks generates parallel tasks on the fly using input parameters. This is particularly useful when the number of tasks isn’t known ahead of time—such as sending notifications to a dynamic list of users or processing a batch of files. Each dynamically generated task runs independently, and a corresponding Join task is used to synchronize them once all have been completed.

Example implementation

Imagine a content moderation workflow that needs to process different types of user-generated content—text, images, and videos. Each content type requires a different moderation approach, like scanning text for profanity, running image recognition for flagged visuals, or transcribing and analyzing video audio. Since the type and number of content items vary per request, a Dynamic Fork is ideal. At runtime, it dynamically generates tasks like moderate_text, moderate_image, and moderate_video, each configured with input specific to the content item. Once all moderation tasks are complete, a Join task consolidates the results for final review. This setup ensures flexibility and scalability for diverse, real-time moderation needs.

Here’s the workflow visualized:

Workflow using a Dynamic Fork operator

Here’s the code snippet for creating the workflow in code:


def register_moderation_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Define Join task
    join_task = JoinTask(
        task_ref_name="join_ref",
        join_on=[]
    )

    # 2) Define Dynamic Fork Task with inputParameters dict
    dynamic_fork = DynamicForkTask(
        task_ref_name="fork_join_dynamic_ref",
        join_task=join_task
    )
    dynamic_fork.input_parameters.update({
        "dynamicTasks": "${workflow.input.dynamicTasks}",
        "dynamicTasksInputs": "${workflow.input.dynamicTasksInputs}"
    })

    workflow = ConductorWorkflow(
        name="DynamicForkExample",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(dynamic_fork)
    workflow.register(overwrite=True)

    return workflow


@worker_task(task_definition_name="moderate_text")
def moderate_text() -> dict:
    print("📝 Moderating text...")
    return {"status": "Text moderation complete"}


@worker_task(task_definition_name="moderate_image")
def moderate_image() -> dict:
    print("🖼️ Moderating image...")
    return {"status": "Image moderation complete"}


@worker_task(task_definition_name="moderate_video")
def moderate_video() -> dict:
    print("🎥 Moderating video...")
    return {"status": "Video moderation complete"}

Check out the full sample code for the Dynamic Fork Workflow.

Sub Workflow

The Sub Workflow task in Orkes Conductor allows you to encapsulate and execute another workflow synchronously within the context of a parent workflow. This task enables modularity and reusability, letting you delegate specific parts of a workflow to be handled by a separate workflow definition. The sub-workflow can run independently and return results to the parent workflow, helping to simplify complex workflows and manage different logic in isolation. It also provides better scalability and maintainability by breaking down large workflows into smaller, reusable components.

Example implementation

In a subscription management workflow, the Sub Workflow task is used to modularize payment processing. Instead of duplicating the entire payment logic within the subscription workflow, the payment_for_subscription workflow is called as a sub-workflow whenever payment needs to be processed. This ensures that any updates made to the payment workflow are automatically reflected across all parent workflows, improving maintainability and consistency. The parent workflow can invoke the sub-workflow multiple times, allowing for seamless integration of payment logic into the larger subscription process.

Here’s the workflow visualized:

Workflow using a Sub Workflow operator

Here’s the code snippet for creating the workflow in code:


# --- Payment Workflow (Sub-Workflow for Subscription) ---

def register_payment_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Process payment (for subscription)
    process_payment = SimpleTask(
        task_def_name="process_payment",
        task_reference_name="process_payment"
    )
    process_payment.input_parameters.update({
        "amount": "${workflow.input.amount}"
    })

    workflow = ConductorWorkflow(
        name="payment_for_subscription",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(process_payment)
    workflow.register(overwrite=True)

    return workflow


# --- Subscription Workflow Definition (using Sub-Workflow) ---

def register_subscription_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Check if the user is subscribed
    check_status = SimpleTask(
        task_def_name="check_subscription_status",
        task_reference_name="check_subscription_status"
    )

    # 2) Trigger payment using sub-workflow
    trigger_payment = SubWorkflowTask(
        task_ref_name="trigger_payment",
        workflow_name="payment_for_subscription",  # The sub-workflow to invoke
        version=1
    )
    trigger_payment.input_parameters.update({
        "amount": "${workflow.input.amount}",  # Mapping input to sub-workflow
        "subscription_id": "${workflow.input.subscription_id}",
        "user_id": "${workflow.input.user_id}"
    })

    # 3) Send welcome email
    send_email = SimpleTask(
        task_def_name="send_welcome_email",
        task_reference_name="send_welcome_email"
    )

    workflow = ConductorWorkflow(
        name="subscription_management",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(check_status)
    workflow.add(trigger_payment)
    workflow.add(send_email)
    workflow.register(overwrite=True)

    return workflow


@worker_task(task_definition_name="check_subscription_status")
def check_subscription_status() -> dict:
    print("📡 Checking subscription status...")
    return {"status": "active"}


@worker_task(task_definition_name="send_welcome_email")
def send_welcome_email() -> dict:
    print("📨 Sending welcome email to subscriber...")
    return {"status": "email_sent"}


@worker_task(task_definition_name="process_payment")
def process_payment(amount) -> dict:
    print(f"💰 Charging customer: ${amount.input_data['amount']}")
    return {"status": "Payment successful"}

Check out the full sample code for the Sub Workflow Workflow.

Start Workflow

The Start Workflow task is an operator in Orkes Conductor that launches another workflow asynchronously from within the current one. Unlike the Sub Workflow task, which pauses the parent workflow until the child workflow completes, the Start Workflow task triggers the target workflow and immediately continues execution without waiting for its result. This is ideal for use cases like fire-and-forget processing, background jobs, or decoupled side operations where real-time feedback isn't needed, allowing for greater concurrency and throughput in orchestration.

Example implementation

A common use case for the Start Workflow task is submitting background jobs that don’t need to block the main workflow. For example, in an e-commerce checkout workflow, once the payment is processed, a Start Workflow task can be used to asynchronously trigger an email_receipt_workflow that sends a receipt to the customer. This ensures the checkout flow completes quickly without waiting for email processing, which can be handled independently.

Here’s the workflow visualized:

Workflow using a Start Workflow operator

Here’s the code snippet for creating the workflow in code:


# --- Email Receipt Workflow (Sub-Workflow for emails) ---

def register_email_receipt_workflow(executor: WorkflowExecutor) -> ConductorWorkflow:
    email_task = SimpleTask(
        task_def_name="send_receipt_email",
        task_reference_name="send_receipt_email_ref"
    )

    workflow = ConductorWorkflow(
        name="email_receipt_workflow",
        executor=executor
    )
    workflow.version = 1
    workflow.add(email_task)
    workflow.register(overwrite=True)
    return workflow


# --- Main Checkout Workflow Definition (using Sub-Workflow) ---

def register_checkout_workflow(executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Task to process payment
    payment_task = SimpleTask(
        task_def_name="process_payment",
        task_reference_name="process_payment_ref"
    )

    # 2) Task to asynchronously start email_receipt_workflow
    start_email_workflow_input = StartWorkflowRequest(
        name="email_receipt_workflow",
        version=1,
        input={
            "customer_email": "${workflow.input.customer_email}",
            "order_id": "${workflow.input.order_id}"
        }
    )

    start_email_workflow_task = StartWorkflowTask(
        task_ref_name="start_email_workflow_ref",
        workflow_name="email_receipt_workflow",
        start_workflow_request=start_email_workflow_input,
        version=1
    )

    workflow = ConductorWorkflow(
        name="checkout_workflow",
        executor=executor
    )
    workflow.version = 1
    workflow.add(payment_task)
    workflow.add(start_email_workflow_task)
    workflow.register(overwrite=True)
    return workflow

@worker_task(task_definition_name="process_payment")
def process_payment() -> dict:
    print("💳 Payment processed successfully.")
    return {"payment_status": "success"}


@worker_task(task_definition_name="send_receipt_email")
def send_receipt_email() -> dict:
    print("📧 Receipt email sent to customer.")
    return {"email_status": "sent"}

Check out the full sample code for the Start Workflow Workflow.

Wrap up

Parallelism and reusability aren’t just performance optimizations—they’re architectural superpowers. With operators like Fork/Join, Dynamic Fork, Sub Workflow, and Start Workflow, Orkes Conductor makes it easy to run tasks concurrently, compose modular logic, and trigger workflows asynchronously.

No threads to manage, no glue code to write—just clean, declarative orchestration that scales.

—

Orkes Operators: Loops, Waits, and Human-in-the-Loop

Orkes Developers — Mon, 28 Apr 2025 12:44:18 +0000

Not all workflows are linear. Some wait, some repeat, and some need a human to step in before moving forward.

In this post, we explore three essential operators that add patience, persistence, and people to your workflows:

Do While for flexible, condition-based iteration
Wait for clean, pause-and-resume control
Human for structured human input and approvals

Whether you're looping through review stages, waiting for external input, or routing manual approvals across teams, these operators help you build dynamic, resilient, and human-aware workflows without writing custom orchestration logic.

Do While

The Do While task is a loop operator that repeatedly executes a set of tasks as long as a specified condition remains true. Similar to a do-while loop in programming, it ensures the loop body runs at least once before evaluating the condition. This is useful for polling, retries, approval loops, or any scenario requiring repeated execution until a dynamic condition is met. The loop condition is evaluated after each iteration, and loop variables can be updated within the loop to control its execution.

Example implementation

This workflow handles a ticket that must go through multiple internal review stages: support agent, QA team, and team lead. Using a Do While Task, the workflow loops through these stages automatically, sending the ticket to each reviewer in sequence. The loop runs once for each reviewer using the itemslist and exits cleanly once all reviews are complete.

Here’s the workflow visualized:

Workflow using a Do While operator

Here’s the code snippet for creating the workflow in code:


def register_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Task to simulate review step with HTTP POST
    review_task = HttpTask(
        task_ref_name="log_review_comment",
        http_input={
            "uri": "https://jsonplaceholder.typicode.com/posts",
            "method": "POST",
            "headers": {
                "Content-Type": "application/json"
            },
            "body": {
                "reviewer": "${review_ticket_loop.output.item}",
                "comment": "Reviewed and updated by ${review_ticket_loop.output.item}"
            }
        }
    )

    # 2) DoWhileTask to iterate over reviewer stages
    review_loop = DoWhileTask(
        task_ref_name="review_ticket_loop",
        termination_condition="$.review_ticket_loop.iteration < $.review_ticket_loop.input.items.length",
        tasks=[review_task]
    )
    review_loop.input_parameters.update({
        "items": ["support_agent", "qa_team", "team_lead"]
    })

    workflow = ConductorWorkflow(
        name="ticket_review_workflow",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(review_loop)
    workflow.register(overwrite=True)

    return workflow

Check out the full sample code for the Do While Workflow.

Wait

The Wait task is a control flow operator used to pause workflow execution until an external signal is received. It’s ideal for scenarios that require human input, external approvals, or asynchronous events from another system. When the workflow reaches a Wait task, it halts and remains paused until it's explicitly resumed using the Conductor API or UI. This makes it easy to build workflows that involve manual steps or depend on external triggers without resorting to constant polling or custom state management.

Example implementation

In an employee onboarding workflow, after provisioning accounts and sending welcome materials, the process may require the new hire to complete a digital form or provide verification documents. A Wait task can pause the workflow at this point until the user submits the required information. Once the input is received—via a webhook or manual trigger—the workflow resumes automatically, continuing with the next steps like assigning a manager or scheduling training.

Here’s the workflow visualized:

Workflow using a Wait operator

Here’s the code snippet for creating the workflow in code:


def register_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Simulate account provisioning
    provision_account = HttpTask(
        task_ref_name="provision_account",
        http_input={
            "uri": "https://jsonplaceholder.typicode.com/users",
            "method": "POST",
            "headers": {
                "Content-Type": "application/json"
            },
            "body": {
                "employeeId": "${workflow.input.employeeId}",
                "email": "${workflow.input.email}"
            }
        }
    )

    # 2) Send welcome materials
    send_welcome = HttpTask(
        task_ref_name="send_welcome_materials",
        http_input={
            "uri": "https://jsonplaceholder.typicode.com/posts",
            "method": "POST",
            "headers": {
                "Content-Type": "application/json"
            },
            "body": {
                "employeeId": "${workflow.input.employeeId}",
                "message": "Welcome to the company!"
            }
        }
    )

    # 3) Wait for the new employee to submit verification (simulate with 10s wait)
    wait_for_input = WaitTask(
        task_ref_name="wait_for_employee_response",
        wait_for_seconds=10
    )

    # 4) Assign a manager
    assign_manager = SetVariableTask(task_ref_name="assign_manager")
    assign_manager.input_parameters.update({
        "manager": "alice.smith@company.com"
    })

    # 5) Schedule training
    schedule_training = SetVariableTask(task_ref_name="schedule_training")
    schedule_training.input_parameters.update({
        "training_date": "2025-04-20"
    })

    workflow = ConductorWorkflow(
        name="employee_onboarding_workflow",
        executor=workflow_executor
    )
    workflow.version = 1

    workflow.add(provision_account)
    workflow.add(send_welcome)
    workflow.add(wait_for_input)
    workflow.add(assign_manager)
    workflow.add(schedule_training)

    workflow.register(overwrite=True)
    return workflow

Check out the full sample code for the Wait Workflow.

Human

The Human task in Orkes Conductor is used to pause a workflow and wait for manual input or approval from a human before continuing. It's ideal for workflows that involve human-in-the-loop decisions, such as document reviews, access approvals, or data verification. When a Human task is reached, the workflow enters a WAITING state until a user provides the required input through the API or UI. This enables seamless integration of manual steps into automated workflows while maintaining visibility and control over the process.

Example implementation

In this example, an expense approval workflow uses two Human tasks to gather approvals from different departments. First, the HR department is prompted with a form showing the expense details in read-only fields. If HR approves, the workflow continues to the Finance department for a second round of approval using the same form template. Upon Finance’s approval, a notification workflow is triggered. If either department rejects the request, the workflow terminates early. This use case shows how Human tasks enable structured, multi-step approvals with manual input inside an otherwise automated process.

Here’s the workflow visualized:

Workflow using a Human operator

To create this workflow in Conductor, you must first define the expense_approval_form using the appropriate fields in the Orkes UI. Here’s the code snippet for creating the expense_approval_form in JSON:


{
  "createTime": 1744849075315,
  "updateTime": 1744849570917,
  "createdBy": "USER:john.doe@acme.com",
  "updatedBy": "USER:john.doe@acme.com",
  "name": "expense_approval_form",
  "version": 1,
  "jsonSchema": {
    "$schema": "http://json-schema.org/draft-07/schema",
    "properties": {
      "approvalStatus": {
        "type": "string",
        "enum": [
          "APPROVED",
          "REJECTED"
        ]
      }
    },
    "required": [
      "approvalStatus"
    ]
  },
  "templateUI": {
    "type": "VerticalLayout",
    "elements": [
      {
        "type": "Control",
        "scope": "#/properties/approvalStatus",
        "label": "Department Approval",
        "options": {}
      }
    ]
  }
}

Here’s the code snippet for creating the workflow in code:


# --- Notification Workflow ---

def register_notification_workflow(executor: WorkflowExecutor) -> ConductorWorkflow:
    notify_task = SimpleTask(
        task_def_name="notify_expense_approval",
        task_reference_name="notify_expense_approval_ref"
    )

    workflow = ConductorWorkflow(
        name="notification_workflow",
        executor=executor
    )
    workflow.version = 1
    workflow.add(notify_task)
    workflow.register(overwrite=True)
    return workflow


# --- Expense Approval Workflow ---

def register_expense_approval_workflow(executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) HR Approval Step
    hr_approval = HumanTask(
        task_ref_name="hr_approval_task",
        display_name="HR Approval",
        form_template="expense_approval_form",
        form_version=1,
        assignment_completion_strategy=AssignmentCompletionStrategy.TERMINATE
    )

    check_hr = SwitchTask(
        task_ref_name="check_hr_approval",
        case_expression="${hr_approval_task.output.approvalStatus}"
    ).switch_case("APPROVED", [])

    # 2) Finance Approval Step
    finance_approval = HumanTask(
        task_ref_name="finance_approval_task",
        display_name="Finance Approval",
        form_template="expense_approval_form",
        form_version=1,
        assignment_completion_strategy=AssignmentCompletionStrategy.TERMINATE
    )

    check_finance = SwitchTask(
        task_ref_name="check_finance_approval",
        case_expression="${finance_approval_task.output.approvalStatus}"
    ).switch_case("APPROVED", [])

    # 3) Start Notification Workflow
    start_notification = StartWorkflowTask(
        task_ref_name="start_notification_workflow",
        workflow_name="notification_workflow",
        start_workflow_request=StartWorkflowRequest(
            name="notification_workflow",
            version=1,
            input={
                "expense_id": "${workflow.input.expense_id}",
                "submitted_by": "${workflow.input.submitted_by}"
            }
        ),
        version=1
    )

    # Add the notification task inside the finance approval success branch
    check_finance.switch_case("APPROVED", [start_notification])

    # Add finance approval after HR approves
    check_hr.switch_case("APPROVED", [finance_approval, check_finance])

    workflow = ConductorWorkflow(
        name="expense_approval_workflow",
        executor=executor
    )
    workflow.version = 1
    workflow.add(hr_approval)
    workflow.add(check_hr)
    workflow.register(overwrite=True)
    return workflow


# --- Worker for notification task ---

@worker_task(task_definition_name="notify_expense_approval")
def notify_expense_approval() -> dict:
    print("🔔 Expense approval notification sent.")
    return {"notified": True}

Check out the full sample code for the Human Workflow.

Wrap up

Loops, waits, and human steps bring real-world rhythm to your workflows, allowing them to persist, pause, or involve people when needed. With tasks like Do While, Wait, and Human, Conductor makes asynchronous logic declarative, structured, and resilient. No brittle timers, no custom coordination—just clean, event-driven flow.

Next up:

Parallelism and Reusability

Orkes Operators: Branching and Conditionals

Orkes Developers — Mon, 28 Apr 2025 12:10:02 +0000

Smart workflows don’t just follow a straight line—they make decisions.

Orkes Conductor provides a set of declarative operators that let your workflows respond to input, handle conditions, and dynamically route execution without hardcoding logic into your services.

In this article, we’ll walk through four core decision-making operators:

Switch for branching logic based on runtime conditions
Dynamic for generating tasks on the fly based on data
Set Variable for setting and updating internal state during execution
Terminate for gracefully ending workflows early when needed

With these operators, you can model flexible, maintainable decision paths that react to real-world complexity—all without writing imperative glue code.

Switch

The Switch task is a control flow operator that enables conditional branching within a workflow, similar to a switch-case or if-else statement in code. It evaluates an expression at runtime and routes execution to a matching case, with support for a default path if no match is found. This makes it ideal for directing workflow logic based on dynamic values such as user input, AI-powered decisions, or status flags.

Example implementation

Let’s say you're building an order processing workflow that needs to handle different shipping methods. Based on the shippingType selected by the user (standard, express, or overnight), the workflow should branch into a different fulfillment path. A Switch task can evaluate the shippingType and route the workflow accordingly.

Here’s the workflow visualized:

Workflow using a Switch operator.

Here’s the code snippet for creating the workflow in code:

def register_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Task to fetch order details
    fetch_order_details = HttpTask(
        task_ref_name="fetch_order_details",
        http_input={
            "uri": "https://jsonplaceholder.typicode.com/posts/${workflow.input.orderId}",
            "method": "GET",
            "headers": {
                "Content-Type": "application/json"
            }
        }
    )

    # 2) SetVariable tasks for each shipping method
    standard_shipping = SetVariableTask(task_ref_name="standard_shipping_var")
    standard_shipping.input_parameters.update({
        "selected_shipping_method": "standard"
    })

    express_shipping = SetVariableTask(task_ref_name="express_shipping_var")
    express_shipping.input_parameters.update({
        "selected_shipping_method": "express"
    })

    overnight_shipping = SetVariableTask(task_ref_name="overnight_shipping_var")
    overnight_shipping.input_parameters.update({
        "selected_shipping_method": "overnight"
    })

    # 3) Define the Switch task to route based on shippingType
    switch_shipping = SwitchTask(
        task_ref_name="switch_shipping_method",
        case_expression="${workflow.input.shippingType}"
    )
    switch_shipping \
        .switch_case("standard", [standard_shipping]) \
        .switch_case("express", [express_shipping]) \
        .switch_case("overnight", [overnight_shipping]) \
        .default_case([])

    workflow = ConductorWorkflow(
        name="order_processing_workflow",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(fetch_order_details)
    workflow.add(switch_shipping)

    workflow.register(True)

    return workflow

Check out the full sample code for the Switch Workflow.

Dynamic

The Dynamic task in Orkes Conductor enables workflows to determine and execute tasks at runtime based on input data, rather than defining them statically in the workflow definition. This is especially useful for cases where the number or type of tasks isn’t known in advance—such as processing a variable-length list of items or executing a dynamic sequence of operations. The task determines a llist of task(s) to be executed, allowing workflows to adapt their structure on the fly while remaining fully managed and observable within Conductor.

Example implementation

In a shipping workflow, the courier used to fulfill an order—such as FedEx or UPS—depends on the destination address, which isn’t known until runtime. A Dynamic task allows the workflow to decide which shipping task to execute on the fly, based on the result of a previous task that determines the best courier. For instance, if the shipping_info task outputs "ship_via_fedex", the dynamic task will execute the ship_via_fedex task; if it returns "ship_via_ups", the workflow dynamically routes to ship_via_ups. This eliminates hardcoded branching and makes the workflow adaptable to real-time decision-making.

Here’s the workflow visualized:

Workflow using a Dynamic operator.

Here’s the code snippet for creating the workflow in code:

def register_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Task to simulate courier selection
    shipping_info = HttpTask(
        task_ref_name="shipping_info_ref",
        http_input={
            "uri": "https://jsonplaceholder.typicode.com/posts",
            "method": "POST",
            "headers": {
                "Content-Type": "application/json"
            },
            "body": {
                "order_id": "${workflow.input.order_id}",
                "destination": "${workflow.input.address}"
            }
        }
    )

    # 2) Dynamic task that picks the shipping method based on workflow input
    dynamic_shipping = DynamicTask(
        dynamic_task="${workflow.input.shipping_service}",
        task_reference_name="dynamic_shipping"
    )

    workflow = ConductorWorkflow(
        name="Shipping_Flow",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(shipping_info)
    workflow.add(dynamic_shipping)
    workflow.register(overwrite=True)

    return workflow


# Worker for FedEx
@worker_task(task_definition_name="ship_via_fedex")
def ship_via_fedex() -> dict:
    print("📦 Shipping via FedEx")
    return {"status": "FedEx shipment created"}


# Worker for UPS
@worker_task(task_definition_name="ship_via_ups")
def ship_via_ups() -> dict:
    print("📦 Shipping via UPS")
    return {"status": "UPS shipment created"}

Check out the full sample code for the Dynamic Workflow.

Set Variable

The Set Variable task in Orkes Conductor is used to define or update variables within the scope of a workflow execution. It allows you to store values—such as computed results, flags, or counters—that can be referenced later by other tasks. This task is especially useful for tracking workflow state, managing intermediate data, or preparing inputs for conditional logic or dynamic tasks. Variables set using this task are stored in the workflow’s context and persist throughout its execution.

Example implementation

In an e-commerce order processing workflow, the Set Variable task is used to store intermediate values like the subtotal, discount, and tax during price calculation. For instance, after the calculate_subtotal task, a Set Variable task updates the subtotal variable, which is then modified by the apply_discount task and used in the final price calculation. These variables are passed to subsequent tasks like process_payment, ensuring the correct amount is charged.

Here’s the workflow visualized:

Workflow using a Set Variable operator.

Here’s the code snippet for creating the workflow in code:

def register_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Calculate subtotal
    calc_subtotal = SimpleTask(
        task_def_name="calculate_subtotal",
        task_reference_name="calculate_subtotal"
    )

    # 2) Store subtotal
    set_subtotal = SetVariableTask(task_ref_name="set_subtotal")
    set_subtotal.input_parameters.update({
        "subtotal": "${calculate_subtotal.output.subtotal}"
    })

    # 3) Apply discount
    apply_discount = SimpleTask(
        task_def_name="apply_discount",
        task_reference_name="apply_discount"
    )

    # 4) Store discounted price
    set_discounted_price = SetVariableTask(task_ref_name="set_discounted_price")
    set_discounted_price.input_parameters.update({
        "discounted_price": "${apply_discount.output.discounted_total}"
    })

    # 5) Process payment
    process_payment = SimpleTask(
        task_def_name="process_payment",
        task_reference_name="process_payment",
        input_parameters={
            "amount": "${workflow.variables.discounted_price}"
        }
    )

    workflow = ConductorWorkflow(
        name="ecommerce_order_processing",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(calc_subtotal)
    workflow.add(set_subtotal)
    workflow.add(apply_discount)
    workflow.add(set_discounted_price)
    workflow.add(process_payment)
    workflow.register(overwrite=True)

    return workflow


@worker_task(task_definition_name="calculate_subtotal")
def calculate_subtotal() -> dict:
    print("🧮 Calculating subtotal...")
    return {"subtotal": 200.0}


@worker_task(task_definition_name="apply_discount")
def apply_discount() -> dict:
    print("🏷️ Applying discount...")
    return {"discounted_total": 180.0}


@worker_task(task_definition_name="process_payment")
def process_payment(amount) -> dict:
    print(f"💰 Charging customer: {amount.input_data['amount']}")
    return {"status": "Payment successful"}

Check out the full sample code for the Set Variable Workflow.

Terminate

The Terminate task in Orkes Conductor allows you to immediately end a workflow, either successfully or with a failure, before all tasks are completed. This is useful for scenarios where certain conditions are met, such as an error or a manual intervention, that require halting the workflow early. The task can be configured with an optional message, like a termination reason, providing context. This helps ensure that workflows end gracefully, preventing unnecessary tasks from being executed when they are no longer needed.

Example implementation

In a shipping workflow, the Terminate task ends the workflow if an invalid shipping provider is provided. During the workflow, a decision is made to ship with a specific provider based on user input. If the input does not match the available options (e.g., FedEx or UPS), the workflow enters the default case of a Switch task and triggers a Terminate task, halting the workflow with a "FAILED" status and a relevant reason.

Here’s the workflow visualized:

Workflow using a Terminate operator.

Here’s the code snippet for creating the workflow in code:

def register_shipping_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Decide based on shipping provider
    switch_provider = SwitchTask(
        task_ref_name="switch_provider",
        case_expression="${workflow.input.shipping_provider}"
    )

    # 2a) FedEx path
    fedex_task = SimpleTask(
        task_def_name="ship_with_fedex",
        task_reference_name="ship_with_fedex"
    )

    # 2b) UPS path
    ups_task = SimpleTask(
        task_def_name="ship_with_ups",
        task_reference_name="ship_with_ups"
    )

    # 3) Default path — Terminate the workflow if the provider is invalid
    terminate_task = TerminateTask(
        task_ref_name="terminate_invalid_provider",
        status=WorkflowStatus.FAILED,
        termination_reason="${workflow.input.termination_reason}"
    )

    # 4) Configure Switch task cases
    switch_provider.switch_case("FEDEX", [fedex_task])
    switch_provider.switch_case("UPS", [ups_task])
    switch_provider.default_case([terminate_task])

    workflow = ConductorWorkflow(
        name="shipping_workflow_with_validation",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(switch_provider)
    workflow.register(overwrite=True)

    return workflow


@worker_task(task_definition_name="ship_with_fedex")
def ship_with_fedex() -> dict:
    print("📦 Shipping with FedEx...")
    return {"status": "Shipped via FedEx"}


@worker_task(task_definition_name="ship_with_ups")
def ship_with_ups() -> dict:
    print("📦 Shipping with UPS...")
    return {"status": "Shipped via UPS"}

Check out the full sample code for the Terminate Workflow.

Wrap up

Branching and conditionals bring workflows to life, allowing them to adapt in real time based on input, state, or external data. With tasks like Switch, Dynamic, SetVariable, and Terminate, Conductor makes decision logic clear, declarative, and maintainable. No tangled logic, no hidden conditionals—just clean, visible control flow.

Next up:

—

Control the Flow: Building Dynamic Workflows with Orkes Operators

Orkes Developers — Mon, 28 Apr 2025 12:09:24 +0000

Orchestration is a powerful design framework for coordinating distributed services and systems into automated processes. Using a workflow engine, such as Orkes Conductor, provides a myriad of benefits like durable executions, state tracking, and developer productivity.

A key part of this orchestration model is operators, which are control flow primitives that resemble programming constructs. By using operators, developers can model dynamic, intelligent workflows without writing imperative control logic, making it easier to scale, maintain, and evolve complex systems.

What are operators?

In Conductor, operators define the control flow, structure, and decision-making within a workflow. Operators are one of the three core task types used in workflows, alongside system tasks and worker tasks. While system and worker tasks focus on performing actions (like sending HTTP requests or processing data), operator tasks define the control flow.

Operators act as declarative control structures, similar to constructs in traditional programming languages. For example, a Switch operator mimics a switch-case statement, a Fork allows for parallel task execution, and a Do While enables looping logic. These constructs make it possible to express complex workflow logic (branching, looping, and conditional execution) all without writing procedural code.

Operators are essential for orchestrating agentic workflows, where conditional or parallel logic provides a mapped space for the agent to act autonomously within bounds. Similarly, using operators, human-in-the-loop flows can be dynamically controlled based on the manual input received mid-flow.

Like system tasks, operator tasks are fully managed by Conductor and do not require any external workers to run. They are evaluated and executed internally by the engine, making them lightweight and efficient building blocks for composing complex workflows.

Why use operators?

Using operators in your workflows offers a range of benefits, especially when compared to managing control flow directly in code:

Declarative logic: Operators allow you to express branching, looping, and parallelism in a clean, declarative way. No need to write imperative control logic or orchestration code.
Separation of concerns: Business logic stays inside your services, while orchestration logic lives inside your workflow. This separation makes it easier to test, reason about, and maintain your applications.
Dynamic execution paths: You can define workflows that respond to real-time data and conditions, dynamically choosing tasks or branches to execute.
Fully managed by Conductor: Operators require no external workers or custom handlers. They’re executed entirely within the Conductor engine, reducing complexity and operational overhead.

By leveraging operators, you can design robust, readable, and adaptable workflows without getting bogged down by control-flow code scattered across services.

The power of declarative logic

Declarative logic is all about describing what should happen, not how to make it happen. In traditional code, developers must spell out the exact sequence of steps and control flow — managing state, writing loops, handling conditionals, and coordinating service calls. This imperative approach can quickly become brittle and hard to maintain as workflows grow in complexity.

By contrast, declarative logic lets you define the intent of the workflow and offload the execution details to the orchestration engine. This not only simplifies development but also makes workflows easier to read, reason about, and debug.

Operators in Orkes Conductor are a perfect example of this approach. They serve as control flow declarations: you define that a workflow should loop, branch, or fork — and Conductor takes care of executing that logic under the hood. Want to repeat a task until a condition is met? Just add a Do While with the right condition. Need to split work across multiple paths and rejoin later? Use Fork and Join. There’s no need to write a single line of plumbing code.

This declarative style brings huge benefits:

Maintainability: Simply update the workflow definition to change the workflow logic. No need to rewrite service code.
Readability: Declarative workflows are self-documenting. You can inspect them in the UI or JSON and instantly understand the flow.
Modularity: You can compose and reuse smaller workflows, using operators like Sub Workflow or Start Workflow as high-level building blocks.

Declarative vs imperative logic

In this example, we demonstrate how both imperative and declarative workflows handle iterating over a list of tasks and making HTTP requests for each. The imperative workflow explicitly defines the loop, including custom error handling for each HTTP request, while the declarative workflow uses Orkes Conductor's built-in Do While operator to manage iteration, with automatic error handling provided by the engine. This showcases the difference in complexity and control between the two approaches.

Imperative logic vs declarative logic.

Declarative orchestration is especially valuable when building agentic workflows, human-in-the-loop workflows, event-driven systems, or other dynamic execution paths. Operators let you adapt to changing inputs or user actions without embedding decision logic inside services.

Overview of Conductor’s operators

Build workflows declaratively with operators.

With over a dozen operators and extensive SDK support, you can quickly build out and deploy powerful workflows. Let’s explore example workflows that make use of these most commonly used tasks:

Operator	Description
Switch	Branches the workflow based on conditions, similar to a switch-case construct. Learn more: Branching and Conditionals
Dynamic	Dynamically evaluates conditions at runtime to determine what to do. Learn more: Branching and Conditionals
Set Variable	Sets variables within the workflow, allowing for dynamic control during execution. Learn more: Branching and Conditionals
Terminate	Ends the workflow execution early, often used for terminating workflows under conditions. Learn more: Branching and Conditionals
Do While	Loops through tasks until a condition is no longer true, used for repeated actions. Learn more: Loops, Waits, and Human-in-the-Loop
Wait	Pauses execution for a set time or until a condition is met. Learn more: Loops, Waits, and Human-in-the-Loop
Human	Pauses execution until manual input is received, useful for human-in-the-loop scenarios. Learn more: Loops, Waits, and Human-in-the-Loop
Fork	Executes multiple tasks in parallel, enabling concurrent processing. Learn more: Parallelism and Reusability
Join	Merges the results from parallel tasks, allowing the workflow to proceed together. Learn more: Parallelism and Reusability
Dynamic Fork	Dynamically creates parallel tasks at runtime based on conditions or data. Learn more: Parallelism and Reusability
Sub Workflow	Triggers another workflow within the current workflow for modularity and reusability. Learn more: Parallelism and Reusability
Start Workflow	Starts another workflow asynchronously in parallel with the current workflow. Learn more: Parallelism and Reusability

Wrap up

Orkes Conductor simplifies workflow orchestration by using declarative operators to manage complex logic. With tasks like Switch, Do While, and Fork/Join, you can build flexible, scalable workflows without writing imperative code. This approach accelerates development, reduces maintenance overhead, and enables collaboration across teams. Whether through the visual diagram editor or Conductor SDKs in Java, Python, or other languages, Conductor empowers developers to easily design, manage, and scale workflows.

Mastering control flow with Conductor

With operators, you design the flow, and Conductor handles the execution logic, freeing you to focus on building the features that matter.

Check out how to use Conductor’s operators with these articles:

Implementing Easy-to-Build Workflows with Conductor’s System Tasks

Orkes Developers — Mon, 14 Apr 2025 13:52:00 +0000

This is Part 2 of the series on System Tasks. Check out Part 1, which discusses the pros and cons of using system tasks and alternative solutions.

Whether you’re orchestrating a simple API flow or building complex AI-powered applications, system tasks make it easy to plug functionality into your workflows. In this post, we’ll explore the most common system tasks in Orkes Conductor and demonstrate how you can leverage them to accelerate development and unlock the full potential of Orkes Conductor.

Recap: what are system tasks?

Orkes Conductor is a workflow orchestration platform packed with developer-first features.

System tasks in Orkes Conductor represent reusable business logic components, like HTTP calls and database queries. These tasks run within the Conductor engine, allowing developers to quickly build complex and durable workflows without reinventing the wheel.

Overview of Conductor’s system tasks

With over twenty system tasks and extensive SDK support, you can quickly build out and deploy powerful workflows. Let’s explore example workflows that make use of these most commonly-used tasks:

Event
HTTP
HTTP Poll
Inline
JSON JQ Transform
Wait For Webhook
Alerting tasks
AI tasks

Commonly-used system tasks in Orkes Conductor.

Event

In distributed systems, async communication is key一and that’s where the Event task comes in. It allows your workflow to publish messages to an external event queue or topic, enabling decoupled communication between services. Whether you’re using Kafka, SQS, or another messaging system, Orkes Conductor integrates with many providers, making it easy to trigger downstream processes or notify external systems as part of your workflow.

Example implementation

Imagine a payment workflow that processes a transaction and needs to notify an inventory service once completed. The workflow first makes an HTTP call to a payment API to confirm the transaction. Once confirmed, instead of directly calling the inventory service, an Event task publishes a message to a Kafka topic. The inventory service listens to the Kafka topic, consumes the event, and asynchronously updates stock levels, decoupling the processes for scalability and flexibility.

Here’s the workflow visualized:

Workflow using an Event task.

To create this workflow in Conductor, you must first integrate the required message broker, and then add the relevant system tasks. Here’s the code snippet for creating the workflow in code:

def register_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Define the HTTP task for calling the dummy JSONPlaceholder API
    http_task = HttpTask(
        task_ref_name="fetch_order_details",
        http_input={
            "uri": "https://jsonplaceholder.typicode.com/posts/\${workflow.input.orderId}",
            "method": "GET",
            "headers": {
                "Content-Type": "application/json"
            }
        }
    )

    # 2) Define the Event task using EventTaskInterface
    event_task = EventTaskInterface(
        task_ref_name="notify_inventory",
        event_prefix="kafka",
        event_suffix="quickstart-events"
    )

    # Update input parameters for the event task using data from the HTTP task response
    event_task.input_parameters.update({
        "orderId": "\${fetch_order_details.output.response.body.id}",
        "title": "\${fetch_order_details.output.response.body.title}",
        "body": "\${fetch_order_details.output.response.body.body}"
    })

    # Define the workflow and add the tasks
    workflow = ConductorWorkflow(
        name="inventory_management_workflow",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(http_task)
    workflow.add(event_task)

    # Register the workflow
    workflow.register(True)

    return workflow

Check out the full sample code for the Event Workflow.

HTTP

The HTTP task is used to make calls to remote services exposed over HTTP/HTTPS, a fundamental part of most service-based applications. It provides a clean abstraction that makes it intuitive to declare API calls. This simplifies integration with external systems, eliminating the need for custom workers and manual integration..

Example implementation

In a payment processing system, you need to interact with an external API to charge a customer’s credit card. It sends a POST request via an HTTP task to the payment API with the order ID and amount, along with an authorization token.

If the payment succeeds (status code 200), the workflow continues to update the order status and marks the payment as successful.
If the payment fails (any status code other than 200), the workflow notifies the user of the failure.

The workflow ensures efficient handling of payment success or failure through conditional branching with the use of the Switch task.

Here’s the workflow visualized:

Workflow using an HTTP task.

Here’s the code snippet for creating the workflow in code:

def register_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Define the HTTP task for calling the payment API
    http_task = HttpTask(
        task_ref_name="call_payment_api",
        http_input={
            "uri": "https://httpbin.org/status/200",
            "method": "POST",
            "headers": {
                "Content-Type": "application/json",
                "Authorization": "Bearer \${workflow.input.auth_token}"
            },
            "body": {
                "orderId": "\${workflow.input.orderId}",
                "amount": "\${workflow.input.amount}"
            }
        }
    )

    # 2) Define the SwitchTask with case expression
    switch_task = SwitchTask(
        task_ref_name="check_payment_status",
        case_expression="\${call_payment_api.output.response.statusCode}",
        use_javascript=False
    )

    # 3) Define SetVariableTask for success case
    set_payment_success_variable = SetVariableTask(
        task_ref_name="set_payment_success_variable"
    )
    set_payment_success_variable.input_parameters.update({
        "payment_status": "SUCCESS"
    })

    # Define SetVariableTask for failure case
    set_payment_failure_variable = SetVariableTask(
        task_ref_name="set_payment_failure_variable"
    )
    set_payment_failure_variable.input_parameters.update({
        "payment_status": "FAILURE"
    })

    # Configure the decision logic for SwitchTask
    switch_task.switch_case("200", set_payment_success_variable)
    switch_task.default_case(set_payment_failure_variable)

    # Define the workflow and add the tasks
    workflow = ConductorWorkflow(
        name="payment_workflow",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(http_task)
    workflow.add(switch_task)

    # Register the workflow
    workflow.register(True)

    return workflow

Check out the full sample code for the HTTP Workflow.

HTTP Poll

The HTTP Poll task in Orkes Conductor enables workflows to send HTTP requests to a specified endpoint at regular intervals, continuing until a defined termination condition is met. This is particularly useful for monitoring the status of long-running operations or waiting for external processes to complete.

Example implementation

Consider a scenario where a workflow needs to monitor the status of a file processing job initiated by an external service. The workflow can use the HTTP Poll task to periodically check the status endpoint of the external service until the job is completed.

Here’s the workflow visualized:

Workflow using an HTTP Poll task.

Here’s the code snippet for creating the workflow in code:

def register_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Define the HTTP Poll input
    http_poll_input = HttpPollInput(
        uri="https://httpbin.org/delay/10",
        method="GET",
        accept="application/json",
        termination_condition="\$.output.response.statusCode == 200",
        polling_interval=10
    )

    # 2) Define the HTTP Poll task for checking the file processing status
    http_poll_task = HttpPollTask(
        task_ref_name="check_file_processing_status",
        http_input=http_poll_input
    )

    # 3) Define SetVariableTask for the success case
    set_processing_complete_variable = SetVariableTask(
        task_ref_name="set_processing_complete_variable"
    )
    set_processing_complete_variable.input_parameters.update({
        "processing_status": "COMPLETED"
    })

    # Define SetVariableTask for the failure case
    set_processing_failed_variable = SetVariableTask(
        task_ref_name="set_processing_failed_variable"
    )
    set_processing_failed_variable.input_parameters.update({
        "processing_status": "FAILED"
    })

    # 4) Define SwitchTask to handle success and failure cases
    switch_task = SwitchTask(
        task_ref_name="check_processing_status",
        case_expression="\${check_file_processing_status.output.response.statusCode}",
        use_javascript=False
    )

    # Configure the decision cases for the SwitchTask
    switch_task.switch_case(200, set_processing_complete_variable)
    switch_task.default_case(set_processing_failed_variable)

    # Define the workflow and add the tasks
    workflow = ConductorWorkflow(
        name="file_processing_workflow",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(http_poll_task)
    workflow.add(switch_task)

    # Register the workflow
    workflow.register(True)

    return workflow

Check out the full sample code for the HTTP Poll Workflow.

Inline

The Inline task in Orkes Conductor allows workflows to execute custom JavaScript code during runtime, facilitating dynamic data processing and decision-making without the need for external workers. This task is particularly useful for implementing lightweight logic, such as data transformations, conditional checks, or calculations, directly within the workflow.

Example implementation

In an e-commerce platform, the Inline task can be used to dynamically calculate prices based on various factors like loyalty status and active promotions. Instead of relying on an external service, the task executes custom JavaScript within the workflow to calculate the final price.

Here’s the workflow visualized:

Workflow using an Inline task.

Here’s the code snippet for creating the workflow in code:

def register_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) HTTP task to fetch a random number from random.org
    fetch_random_number_task = HttpTask(
        task_ref_name="fetch_random_number",
        http_input={
            "uri": "https://www.random.org/integers/?num=1&min=1&max=100&col=1&base=10&format=plain&rnd=new",
            "method": "GET",
            "headers": {
                "Content-Type": "application/json"
            }
        }
    )

    # 2) Set variable for base price using the random number
    set_base_price = SetVariableTask(
        task_ref_name="set_base_price"
    )
    set_base_price.input_parameters.update({
        "base_price": "\${fetch_random_number.output.response.body}"
    })

    # 3) Inline task to calculate the final price with discounts
    calculate_price_task = InlineTask(
        task_ref_name="calculate_final_price",
        script="""
            (function() {
                let basePrice = \$.base_price;
                let loyaltyDiscount = \$.loyalty_discount === "gold" ? 0.2 : 0;
                let promotionDiscount = \$.promotion_discount ? 0.1 : 0;
                return basePrice * (1 - loyaltyDiscount - promotionDiscount);
            })();
        """,
        bindings={
            "base_price": "\${workflow.variables.base_price}",
            "loyalty_discount": "\${workflow.input.loyalty_status}",
            "promotion_discount": "\${workflow.input.is_promotion_active}"
        }
    )

    # 4) Set the final calculated price in a workflow variable
    set_price_variable = SetVariableTask(
        task_ref_name="set_final_price_variable"
    )
    set_price_variable.input_parameters.update({
        "final_price": "\${calculate_final_price.output.result}"
    })

    # Define the workflow and add all tasks
    workflow = ConductorWorkflow(
        name="dynamic_pricing_workflow",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(fetch_random_number_task)
    workflow.add(set_base_price)
    workflow.add(calculate_price_task)
    workflow.add(set_price_variable)

    # Register the workflow
    workflow.register(overwrite=True)

    return workflow

Check out the full sample code for the Inline Workflow.

JSON JQ Transform

The JSON JQ Transform task in Orkes Conductor enables workflows to process and transform JSON data using JQ, a powerful command-line tool for querying and manipulating JSON. This task allows you to apply JQ expressions to input JSON, facilitating operations like filtering, mapping, and restructuring data directly within your workflow.

Example implementation

Consider a scenario where a workflow retrieves a list of user data from an external API, but only a subset of fields is required for subsequent processing. The JSON JQ Transform task can be employed to filter and format the API response efficiently.

Here’s the workflow visualized:

Workflow using an JSON JQ Transform task.

Here’s the code snippet for creating the workflow in code:

def register_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # JQ script to transform user data
    jq_script = """
      .users | map({
        username: .login,
        profile_url: .html_url,
        avatar: .avatar_url
      })
    """

    # 1) Create the JSON JQ task to transform incoming user data
    jq_task = JsonJQTask(
        task_ref_name="transform_user_data",
        script=jq_script
    )
    jq_task.input_parameters.update({
        "users": "\${workflow.input.users}"
    })

    # 2) Create a SetVariableTask to store the transformed result
    set_variable_task = SetVariableTask(
        task_ref_name="store_filtered_users"
    )
    set_variable_task.input_parameters.update({
        "filtered_users": "\${transform_user_data.output.result}"
    })

    # Define and register the workflow
    workflow = ConductorWorkflow(
        name="simple_user_data_transform",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.input_parameters = ["users"]
    workflow.add(jq_task)
    workflow.add(set_variable_task)

    # Register the workflow
    workflow.register(overwrite=True)

    return workflow

Check out the full sample code for the JSON JQ Transform Workflow.

Wait For Webhook

The Wait for Webhook task in Orkes Conductor is designed to pause a workflow’s execution until a specific external event is received via a webhook. This integration enables workflows to respond dynamically to real-time events from various platforms, including GitHub, Slack, Stripe, Microsoft Teams, and SendGrid. This is best when the external system can push events out.

Example implementation

A workflow waits for a webhook event indicating that a specific comment has been posted on a GitHub issue. Using a Wait for Webhook task, the workflow pauses until it receives the event with a comment from a specific author, such as “user123”. Once triggered, the workflow stores the comment content in a mock database, allowing subsequent tasks to process the comment, such as analyzing its content or triggering automated responses.

Here’s the workflow visualized:

Workflow using a Wait For Webhook task.

Here’s the code snippet for creating the workflow in code:

def register_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Define the WaitForWebhookTask to pause until a specific comment is received
    wait_for_comment_task = WaitForWebHookTask(
        task_ref_name="wait_for_comment",
        matches={
            "\$['comment']['author']": "user123"
        }
    )

    # 2) Define an HTTP task to store the received comment in a mock database
    store_comment_task = HttpTask(
        task_ref_name="store_comment",
        http_input={
            "uri": "https://jsonplaceholder.typicode.com/posts",
            "method": "POST",
            "headers": {
                "Content-Type": "application/json"
            },
            "body": {
                "comment_body": "\${wait_for_comment.output.comment.body}",
                "author": "user123"
            }
        }
    )

    # Define and register the workflow
    workflow = ConductorWorkflow(
        name="comment_trigger_workflow",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(wait_for_comment_task)
    workflow.add(store_comment_task)

    # Register the workflow
    workflow.register(overwrite=True)

    return workflow

Check out the full sample code for the Wait for Webhook Workflow.

Before running the workflow you must set up a webhook in Conductor for the workflow comment_trigger_workflow and GitHub.

After running the workflow, call this curl command to complete the workflow:

curl -H "Content-Type:application/json" -H "Accept:application/json" \
-H 'key: value' \
-X POST '[INSERT_WEBHOOK_URL]' \
-d '{"comment": {"body": "This is my comment.", "author": "user123"}}'

Alerting Tasks

Alerting tasks are specialized system tasks designed to monitor workflows and trigger alerts based on specific conditions. These tasks enhance observability and enable proactive incident management within your workflows, helping teams stay informed and act quickly when systems deviate from expected behavior. Explore how to use Alerting tasks with our suite of tutorials.

AI Tasks

Orkes Conductor offers a suite of AI tasks that facilitate seamless integration of Large Language Models (LLMs) and vector databases into workflows. These tasks enable functionalities such as text generation, embedding creation, document indexing, and retrieval, allowing developers to build both AI-driven applications and agentic workflows.

By incorporating these AI tasks, users can effectively orchestrate complex AI processes within their workflows, enhancing automation and intelligence in their applications. Explore how to use AI tasks with our suite of tutorials.

Wrap up

Using system tasks in Conductor speeds up development cycles, bringing processes and products from paper to reality in no time at all. In this post, we’ve demonstrated building workflows using system tasks through Python. As a developer-first platform, Orkes Conductor is built for creating workflows in any programming language (Java, Python, JavaScript, C#, Go, Clojure), or even using a visual diagram editor. Learn more about how exactly system tasks make developing workflows so much faster in Part 1.

Unlocking Developer Productivity: Orchestrate Workflows with Built-in System Tasks

Orkes Developers — Mon, 14 Apr 2025 11:20:55 +0000

This is Part 1 of the series on System Tasks. Check out Part 2, which dives into all the use cases and implementation details for Conductor’s most popular system tasks.

Orchestration-driven development can slash time-to-launch into half, especially when combined with built-in, reusable tasks for common cases, like calling an LLM, publishing events to a message broker, or even just calling an HTTP endpoint. Let’s dive deeper into Orkes Conductor’s system tasks and what makes them so powerful in unlocking developer productivity.

What is a workflow engine?

A workflow engine orchestrates complex processes by defining tasks in a declarative format (such as JSON). Unlike traditional in-code approaches, where developers handle retries, logging, and error management, workflow engines like Orkes Conductor automate task execution, abstracting these concerns.

Conductor abstracts away error management and plumbing code, making it effortless to build durable, stateful applications.

How to write workflows using system tasks? Check out our code implementation guide.

A workflow engine decouples execution code from business logic. This yields two time-saving advantages: one, the business logic of tasks and workflows are now reusable, scalable, and more easily maintainable. Two, developers can focus on task dependencies without having to worry about resilience and execution guarantees, which are managed by the engine.

Clear separation of concerns: the workflow engine handles execution, error management, and state tracking; while the business logic resides in the tasks.

Interested to learn more about how Conductor works under the hood? Check out the in-depth blog on durable execution.

Over time, common patterns emerge across applications, like data transformation, webhook integrations, and HTTP calls. This is where system tasks come in to automate business logic into reusable components, further speeding up development time for common use cases.

What are system tasks?

System tasks in Orkes Conductor represent reusable business logic components that combine to form orchestration patterns. These tasks run within the engine, allowing developers to quickly build complex workflows without reinventing the wheel, leading to a flexible, maintainable automation process.

All system tasks in Orkes Conductor.

When building workflows in Orkes Conductor, system tasks are your go-to tools for getting things done—fast. These are built-in, ready-to-use tasks that handle common operations like making HTTP calls, waiting for external events, or even running AI tasks一all without the need to write custom worker tasks. Since they run within Conductor’s JVM, system tasks help streamline development and remove the overhead of managing the custom infrastructure.

System tasks vs worker tasks in Conductor.

Why use system tasks?

System tasks bring clarity and consistency to workflows. Their standardized structure makes it easy to share, read, and reason about workflows across teams一no guesswork, no surprises.

They also make complex operations easier to express, with less boilerplate. Since Conductor handles the execution, you get enterprise-grade resilience without lifting a single finger. With Conductor’s built-in monitoring and debugging tools, it’s easier to track failures, inspect state, and keep things running smoothly.

Benefits

System tasks offer a declarative, reliable way to build workflows. By abstracting common logic into reusable components, they reduce development effort, enhance clarity, and ensure consistent behavior—making them ideal for predictable, maintainable workflow design.

Time to Market: Eliminate the need for custom code, accelerating development for tasks like HTTP calls, database updates, and event handling.
Standardization: Offer consistency across workflows, making them easier to understand than custom worker task implementations.
Scalability: Workflows using system tasks are inherently easier to scale, since execution is managed by Orkes Conductor itself.

What about custom business logic?

Conductor offers all the benefits of workflow orchestration, even for custom business logic. This can be done through two approaches: Inline tasks and Worker tasks.

Inline tasks let you execute lightweight JavaScript code directly within the workflow without the need for external workers. For more complex tasks, you can deploy custom worker tasks. Worker tasks provide full flexibility to implement advanced logic, such as API calls and long-running operations, while still benefiting from Conductor’s automated retries, status updates, and monitoring.

Inline task approach

An Inline task is a type of system task that allows you to execute custom JavaScript code directly within the workflow. It’s ideal for lightweight operations like simple data transformations, conditional checks, or small calculations. Inline tasks are executed within the Conductor JVM, which means they benefit from low latency, no network overhead, and easier debugging.

Find out how to use Inline tasks in Part Two of this blog series.

Key Features:

Executes directly within the Conductor JVM.
Best for lightweight logic and transformations.
No network calls or external dependencies.
Easier to implement and debug but limited to JavaScript.

Worker task approach

A Worker task (also known as Simple task) is an external custom function or service that performs a specific task in a workflow. Written in any language of choice (Python, Java, etc.), it can execute complex business logic, custom algorithms, or long-running operations. Worker tasks run outside the Conductor server, meaning they require additional infrastructure set-up and logging mechanisms.

Key Features:

External execution (outside the Conductor JVM).
Full flexibility to implement complex logic.
Requires task-level logging within the worker code.
Typically involves network calls (e.g., HTTP requests).

System tasks vs Worker tasks

Both system tasks and Worker tasks offer different advantages, and the choice to use one over the other depends on your requirements. Here’s a comparison between the two.

	System tasks	Worker tasks
Advantages	Speed: System tasks run within the Conductor JVM, enabling fast, low-latency execution without network overhead, serialization processing, or external polling. Availability Guarantees: System tasks are always guaranteed to be available and operate within Conductor’s JVM.	Full Control: You can implement more complex or specialized logic that may not be easily achievable using system tasks. Flexibility: Workers can be deployed, restarted, and updated independently from the workflow. Decoupling: Worker crashes are isolated and will not bring the entire system down.
Trade-offs	Tight Coupling: System tasks are harder to scale, redeploy, or maintain independently. Limited Customizability: There is limited leeway for using custom libraries, frameworks, or stacks in system tasks.	Increased Development Effort: Customizing each worker adds development overhead for writing, testing, deploying, and maintaining more code. Monitoring & Debugging Complexity: With workers, you need to implement your own task-level logging within the worker code. Infrastructure Overhead: Workers rely on the availability and health of the external environment in which they're running, which has to be self-maintained.

Wrap up

Orkes Conductor is a workflow engine that offers the best of both worlds: low-code agility when you need to build out common use cases, and full customization control when you need to dive deep into building unique solutions. It’s a development platform made to empower developers in a fast-paced world. In Part Two of this blog series, discover how to use system tasks in Conductor, as we spill the full implementation details.

Building Agentic Workflows with Conductor: Implementation Guide

Orkes Developers — Thu, 10 Apr 2025 11:10:57 +0000

By: Ram Durbha

AI agents are autonomous, but unpredictable and difficult to thoroughly test and audit. How can we introduce governance into agentic systems? Simple: use agentic workflows. An agentic workflow is a structured yet dynamic sequence of tasks, where the key decision-making points are executed by AI agents, with minimal human intervention.

Agentic workflows combine the autonomy of AI agents with the traceability of workflows, bringing the best of both worlds. In highly regulated industries like finance, healthcare, and cybersecurity, agentic workflows provide a safer alternative to AI agents.

In this article, let’s build an agentic workflow using Orkes Conductor, a modern platform for workflow orchestration.

Example: an agentic cybersecurity monitoring system

In cybersecurity, alert fatigue can be catastrophic. The high-volume bombardment of alerts from multiple sources can lead to delays, missed responses, and ultimately, an increased risk of security breaches.

Let’s build an agentic workflow to combat alert fatigue. This workflow should be able to reconstruct attack narratives from disparate sources of IOCs (indicators of compromise). Before we begin building in Conductor, let’s map out the requirements. The agentic cybersecurity process has to:

Collect alerts from multiple sources
Collate related events for reconstructing the chain of events
Prioritize choose follow-up actions

In sum, the agentic system should help paint a picture of the patterns in how an attack developed over time across devices, users, and services so that human reviewers can allocate their energy to understanding and combating threats rather than drowning in alerts. Let’s use an LLM to provide a human-readable timeline of events that constitute a complex attack.

Here’s the high-level flow of the agentic system on the left, and the actual workflow built in Orkes Conductor on the right:

Orkes’ visual workflow editor makes it intuitive to transition from a napkin-top or whiteboard-sketch idea to a production-ready workflow.

Now, here’s how you can create an agentic cybersecurity workflow on your own.

Getting started

Before you begin, make sure you have the following:

An account on Orkes’ Developer Playground
An OpenAI API key

The resources to build the agentic cybersecurity workflow are located in this GitHub repository. Note that all the task workers are mocked in this workflow example but use representative schemas from real-world cyber security tools like those provided by Netscope and TrendMicro.

Step 1: Import workflow definitions

The GitHub repository contains five workflows:

Agentic_Security_Example
Notify-Channels-x-mocked
Security_get_device_id
Vision_one_deep_visibility_hunt
vision_one_device_scan

For each workflow, copy the JSON definition from the repository and add it to your Orkes Developer Playground account in Definitions > Workflow.

Step 2: Configure the OpenAI integration

We will be using OpenAI’s LLMs to power the AI and agentic components of the workflow. In the Developer Playground, go to Integrations and add the OpenAI integration and models. For this workflow, you should add:

gpt-4o (required)
gpt-4o-mini (recommended)
Any other OpenAI models you wish to use

Repeat this procedure for other providers and models you want to use and compare.

Step 3: Add the AI prompt

The GitHub repository also contains the llm_alert_analysis prompt, which analyzes the detected alerts to form an attack narrative. Copy its JSON definition from the repository and add it to your Developer Playground account in AI Prompts. If you are using models other than gpt-4o, make sure to edit your AI prompt and workflow accordingly.

Step 4: Run the workflow

Finally, with the workflow and supporting resources ready, let’s run it to see how it works. By default, the workflow comes baked with sample mock alerts, so it can be directly run without having to configure its inputs.

In Definitions > Workflow, select the Agentic_Security_Example workflow and run it. As each task completes, you can review its output in real time.

Breaking down the workflow

As we track the steps of the Agentic_Security_Example workflow, let's see how Orkes Conductor makes it easy to map the needs of our solution to actual tasks and design patterns:

Conductor's suite of features makes it seamless to build agentic workflows.

Parallel Alert Ingestion: The workflow starts by concurrently fetching malware and malsite alerts using a Fork/Join task, which is Conductor's built-in task for handling parallel data processing.
Alert Processing: Once the raw alert data is ingested, the workflow extracts relevant details from both malware and malsite alerts using JSON JQ Transform tasks. These tasks parse and filter the data, ensuring that only meaningful alert information is passed on in a structured manner for efficient downstream analysis and decision-making.
AI-Powered Analysis: The intelligence core lies in the llm_alert_analysis task, which utilizes Conductor's LLM Text Complete task to send the raw alert data to an LLM (like gpt-4o-mini). The LLM, guided by a sophisticated prompt, analyzes the alerts for threat classification, correlation, context, and risk assessment.
Dynamic Follow-Up Actions: Based on the LLM's output, a Switch task directs the workflow to initiate follow-up actions only if a deep scan is deemed necessary. This demonstrates the agentic nature of the workflow, where the AI's analysis directly influences the subsequent flow of control.
Scalable Remediation: The workflow then uses Dynamic Fork tasks to initiate deep visibility threat hunts and device scans on all potentially affected devices identified by the LLM. This highlights Conductor's ability to scale the response based on the AI's findings.
Notification and Reporting: Finally, the workflow generates a summary and sends notifications through a sub-workflow, ensuring that security teams are informed of the intelligent analysis and any initiated actions.

Going beyond

Now that you know how the workflow works, you can customize it for your own use case. Tweak it, replace the mocks with real data sources, or try creating your own agentic systems in Orkes Conductor.

Advantages of Orkes Conductor for agentic workflows

Orkes Conductor offers several key advantages that make it particularly well-suited for building agentic workflows. It provides all the necessary building blocks and orchestration capabilities to create sophisticated agentic workflows.

Native LLM integration with LLM Text Complete Task: Conductor provides a dedicated LLM Text Complete task that simplifies the integration with various LLM providers like OpenAI. This built-in functionality eliminates the need for complex custom integrations and allows developers to seamlessly incorporate AI-powered text analysis and generation into their workflows.

The Text Complete task is just one of the many system LLM tasks provided in Conductor. The flexibility to choose from these system tasks, built and thoroughly tested for AI use cases, makes it easy to prototype and scale to production.

Dynamic task creation with Dynamic Forks:

For agentic workflows that need to take action based on AI insights, Conductor's Dynamic Fork task is invaluable. It allows the workflow to dynamically create and manage parallel tasks based on the output of previous steps, like the LLM's analysis output in the cybersecurity example. This is crucial for scenarios like initiating scans on multiple affected devices identified by the AI.

Robust workflow versioning:

Building intelligent agents often involves experimentation and iteration. Conductor's robust workflow versioning and management capabilities allow teams to track changes, roll back to previous versions, and maintain complex agentic workflows effectively over time.

Scalability and reliability for critical operations:

Conductor is designed for high-throughput and mission-critical applications. Its architecture ensures the scalability and reliability needed to power agents that operate continuously and handle potentially large volumes of data and events. This is enhanced further by the suite of Security, Governance and Auditing capabilities like RBAC, Open Telemetry and Grafana support.

Flexibility and extensibility:

Conductor's open architecture enables easy integration with a wide range of tools and systems through its various task types (HTTP, Sub Workflow, Event, and more) and ability to execute custom code. This flexibility is essential for building agentic workflows that interact with diverse security ecosystems. Simple workers allow for complex logic, while Inline tasks and JSON JQ Transform tasks allow for well-organized, clean glue code.

Orkes Conductor is a robust and developer-friendly platform made for building and managing the complexities of agentic workflows. Its architecture and features make it ideal for orchestrating complex processes involving AI-driven decision-making.

Wrapping up

Agentic workflows, driven by the reasoning and language processing abilities of LLMs, offer a significant leap forward compared to traditional, static workflows. By integrating LLMs into automated processes, systems can be built to:

Intelligently analyze and correlate diverse sources of information:

LLMs can process and understand natural language descriptions from various sources, enabling sophisticated correlation based on semantic understanding rather than just pattern matching.

Provide rich contextual analysis:

LLMs can leverage their vast knowledge base to provide context around the input content. More importantly, Conductor makes it easier to provide the LLM with additional context through built-in AI Prompts, which can be crafted with placeholder variables that are dynamically replaced during a workflow run.

In the above example, LLMs are used to provide context on detected threats, including information on malware families, attack vectors, and potential impact.

Dynamically adapt to current circumstances:

Agentic workflows enable dynamic decision-making, as shown above, providing automatic adaption to macro and micro changes.

With dozens of one-click-switch AI/LLM integrations, input/output schema enforcement, native human-in-the-loop features, and enterprise RBAC controls, you can seamlessly build agentic workflows in Orkes Conductor for any enterprise use case.

Stay tuned for more agentic examples and tutorials. In the meantime, check out Conductor’s Examples repository for other projects or get hands-on with our free Developer Playground. As always, join the Slack community to connect with fellow developers and the Conductor team.

Orchestrating Long-Running APIs | A Step-by-Step Guide to Asynchronous Execution

Orkes Developers — Fri, 04 Apr 2025 15:04:37 +0000

Handling long-running APIs is a common challenge in modern applications. Whether you’re processing large datasets, generating reports, or interacting with third-party services, some API calls exceed the 30–60 second timeout window allowed by most systems. When an API takes too long to respond, workflows can break—resulting in failed tasks, degraded user experiences, or unnecessary retries.

To address this, we’ll implement a solution using asynchronous orchestration. Instead of waiting for the long-running task to complete, the workflow triggers it asynchronously, stores its progress in an external database, and uses polling to track its status until completion.

In this blog, we’ll walk through a scalable solution using Orkes Conductor—a Unified Application Platform—along with AWS Lambda, Amazon API Gateway, and Amazon DynamoDB.

Why are long-running APIs difficult?

APIs that take several minutes to complete are not uncommon, especially in systems that rely on compute-heavy workloads or external services. However, traditional synchronous API calls aren’t equipped to handle extended processing delays.

Let’s consider a simple HTTP task in Orkes Conductor or any workflow engine. These tasks typically expect a response within a certain timeout window (e.g., 60 seconds). If the external API takes longer than that, the task fails. This limitation affects:

User-facing features–UI elements may hang or fail without feedback.
System performance–Workflow fails, and retries kick in, increasing the load on compute resources and downstream services.
Observability–There’s no clear visibility into the task’s state if the response never arrives.

Stretching timeout windows or blocking workflows isn’t a scalable solution. A better approach is a more resilient workflow architecture pattern—one that embraces asynchronous execution and decouples task initiation from task completion.

Solution: Use asynchronous HTTP invocation with polling

To reliably handle long-running APIs, we’ll implement an asynchronous invocation with a polling pattern using Orkes Conductor and AWS services. This avoids timeouts while maintaining full control and visibility over the process.

At a high level, the pattern involves:

Triggering the long-running task asynchronously
Tracking its execution state in an external store
Polling the task status until it is completed

Implementing asynchronous HTTP invocation with polling

We’ll implement this orchestration pattern using:

AWS Lambda–A serverless computing service that simulates the long-running task.
Amazon DynamoDB–A NoSQL database to store and track the status of each request.
Amazon API Gateway–A REST API layer to expose Lambda endpoints for invocation and polling.
Orkes Conductor–The orchestration platform that orchestrates the asynchronous flow.

How Asynchronous Orchestration Works

The core idea behind this pattern is:

Trigger the long-running task asynchronously

The workflow initiates the long-running process using an HTTP task configured to perform a non-blocking request (e.g., invoking a Lambda function asynchronously via API Gateway). The task immediately returns a requestId, which is used as a unique reference for tracking progress.

Track task progress externally

The Lambda function logs the execution state (e.g., PROCESSING or COMPLETED) in a NoSQL database like Amazon DynamoDB, which serves as a persistent data store for tracking task progress externally.

Poll the task status until completion

Conductor workflow periodically queries the status of the request using an HTTP Poll task. The polling continues until the task is marked as completed in the external system.

This model allows the workflow to remain active and aware of the task’s progress without exceeding timeout thresholds or blocking system resources.

Now that we understand the pattern conceptually, let’s walk through how to implement this using Orkes Conductor and AWS services.

Building a long-running API workflow

In this section, we’ll implement the long-running API orchestration pattern using Orkes Conductor. To follow along, sign in to the free Developer Playground.

Part 1: Set up AWS resources

Step 1: Create an Amazon DynamoDB table for status tracking

Start by creating a table in DynamoDB named lambda-invocation-status with requestId as the partition key.

Creating a table in Amazon DynamoDB

This table stores the execution status of each request.

Step 2: Set up a long-running task in AWS Lambda

Next, we’ll create a Lambda function that:

Accepts a requestId and an action
Simulates a long-running task
Updates the task’s status in DynamoDB

Create an AWS Lambda function using the Python code:

import json
import time
import boto3

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('lambda-invocation-status')

def lambda_handler(event, context):
    request_id = event['requestId']
    action = event.get("action")

    if action == "invoke":
        table.put_item(Item={'requestId': request_id, 'status': 'PROCESSING'})

        # Simulate long processing time
        time.sleep(2)
        process_long_running_task(request_id)

        return {
            "statusCode": 202,
            "body": json.dumps({"requestId": request_id, "status": "PROCESSING"})
        }

    elif action == "status":
        response = table.get_item(Key={'requestId': request_id})
        item = response.get('Item')
        if item:
            return {"statusCode": 200, "body": json.dumps(item)}
        else:
            return {"statusCode": 404, "body": json.dumps({"status": "NOT_FOUND"})}

def process_long_running_task(request_id):
    time.sleep(25)  # Simulate processing time
    table.update_item(
        Key={'requestId': request_id},
        UpdateExpression='SET #s = :val',
        ExpressionAttributeNames={'#s': 'status'},
        ExpressionAttributeValues={':val': 'COMPLETED'}
    )

Adjust the Lambda timeout to 1 minute and attach an IAM role with complete access to DynamoDB.

Creating an AWS Lambda function

Step 3: Set up API Gateway endpoints

Create a REST API in Amazon API Gateway and expose the Lambda function with two endpoints:

POST /invoke-lambda–Triggers the task.

GET /{requestId}–Returns the current task status.

Deploy the API and note the endpoint URL—they’ll be used in the workflow definition.

Part 2: Build the Orkes Conductor workflow

With your AWS resources ready, it’s time to orchestrate the flow in Orkes Conductor. Let’s build a workflow that:

Invokes the Lambda function using an HTTP task.
Polls the status endpoint until the task is completed using the HTTP Poll task.

Step 1: Invoke Lambda asynchronously

Use an HTTP task in Conductor to call the Lambda endpoint via API Gateway, passing a requestId in the body. The Lambda will start the task and return a 202 response.

Calling Lambda endpoint using HTTP task

Step 2: Poll for task completion

Use an HTTP Poll task to query the status endpoint at regular intervals. The polling continues until the external system returns a "COMPLETED" status.

This behavior is configured using the following termination condition in the HTTP Poll task:

(function(){
  return $.output.response.statusCode == 200 &&
         $.output.response.body.body.status == "COMPLETED";
})();

Querying long running task status at regular intervals using HTTP Poll task

Creating Conductor Workflow

Create a workflow in Developer Playground by navigating to Definitions > Workflow. In the Code tab, paste the following code:

{
 "name": "LongRunningAPIWorkflow",
 "description": "Sample workflow",
 "version": 1,
 "tasks": [
   {
     "name": "InvokeLambdaTask",
     "taskReferenceName": "invokeLambda",
     "inputParameters": {
       "http_request": {
         "uri": "https://<your-api-gateway-id>.execute-api.<your-region>.amazonaws.com/test/invoke-lambda",
         "method": "POST",
         "headers": {
           "Content-Type": "application/json",
           "X-Amz-Invocation-Type": "Event"
         },
         "body": {
           "requestId": "${workflow.input.requestId}"
         },
         "accept": "application/json",
         "contentType": "application/json"
       }
     },
     "type": "HTTP"
   },
   {
     "name": "http_poll",
     "taskReferenceName": "http_poll_ref",
     "inputParameters": {
       "http_request": {
         "uri": "https://<your-api-gateway-id>.execute-api.<your-region>.amazonaws.com/test/${workflow.input.requestId}",
         "method": "GET",
         "accept": "application/json",
         "contentType": "application/json",
         "terminationCondition": "(function(){ return $.output.response.statusCode == 200 && $.output.response.body.body.status == \"COMPLETED\";})();",
         "pollingInterval": "10",
         "pollingStrategy": "FIXED",
         "encode": true
       }
     },
     "type": "HTTP_POLL"
   }
 ],
 "inputParameters": [
   "requestId"
 ],
 "schemaVersion": 2
}

Your workflow will look like this:

Long Running API workflow in Orkes Conductor

Before running the workflow, update the URLs in your task definitions with your actual deployed API Gateway endpoints.

In InvokeLambdaTask (HTTP task), replace the URL with your deployed API endpoint for invoking the Lambda:

https://<your-api-gateway-id>.execute-api.<your-region>.amazonaws.com/<stage>/invoke-lambda

In the http_poll task, replace the polling URL with the endpoint for checking Lambda status:

https://<your-api-gateway-id>.execute-api.<your-region>.amazonaws.com/<stage>/${workflow.input.requestId}

Once configured, test the workflow with a sample input:

{
  "requestId": "12345"
}

The requestId serves as a unique identifier for each task instance. It is passed into the workflow, forwarded to the Lambda function, and used to poll the task’s status from DynamoDB.

When the workflow is run, the InvokeLambdaTask triggers the Lambda function. The http_poll task continuously checks the status until it is completed.

Congratulations—you’ve successfully orchestrated a long-running API using Orkes Conductor and AWS. For a complete implementation walkthrough, refer to the tutorial on orchestrating long-running APIs.

Summary

Long-running APIs can easily break synchronous workflows, but with exemplary architecture, you can handle them reliably and at scale.

By combining asynchronous invocation, external status tracking, and polling using Orkes Conductor, you can build resilient and timeout-proof workflows. This pattern can be adapted for a wide range of real-world scenarios, from third-party service orchestration to internal background processing.

Ready to build scalable workflows that can handle long-running operations without breaking? Try out using Orkes Conductor.

–