conductor-skills

Writing Conductor Workers

Workers execute SIMPLE tasks in a workflow. They poll the Conductor server for tasks, execute business logic, and report results back. Workers are stateless and idempotent.

Before you scaffold — check the built-ins first

Most “I need a worker for X” requests are actually built-in tasks. Walk SKILL.md Rule 6’s table before writing any worker code — if a built-in matches (LLM call, Kafka publish, PDF render, vector index/search, sub-workflow trigger, wait, human approval, JQ transform, fork/join, etc.), use it. A custom worker is the right answer only when no built-in covers the operation (e.g. an internal API, a proprietary system, business logic that doesn’t fit a generic task).

Scaffolding flow (SKILL.md Rule 7)

Confirm no built-in fits. Name the closest candidate and why it doesn’t work.
Ask the language. Supported officially: Java, Go, Python, TypeScript/JavaScript, .NET (C#), Rust, Ruby. Don’t assume — defaults shift across teams.
WebFetch the SDK repo README (table below) before writing code. The SDKs evolve fast — annotations, runner classes, package paths, and entry points have changed across major versions. Pin the version and install command from what the README currently says, not from memory.
Scaffold from the pattern below, match the worker’s task type to the SIMPLE task’s name exactly, and include the idempotency note.

How workers work

Workflow reaches a SIMPLE task
Task is placed in a queue
Worker polls the queue, picks up the task
Worker executes logic using task’s inputData
Worker returns result (COMPLETED or FAILED) with outputData
Workflow continues to the next task

SDKs — canonical repos (WebFetch these before scaffolding)

All official SDKs live under the conductor-oss GitHub org with the *-sdk naming convention:

Language	Repo (WebFetch the README)	Package / Install
Python	github.com/conductor-oss/python-sdk	PyPI: `conductor-python` — `pip install conductor-python`
JavaScript / TypeScript	github.com/conductor-oss/javascript-sdk	npm: `@io-orkes/conductor-javascript` — `npm install @io-orkes/conductor-javascript`
Java	github.com/conductor-oss/java-sdk	Maven: `org.conductoross:conductor-client` (and `conductor-client-spring` for Spring Boot)
Go	github.com/conductor-oss/go-sdk	Go module path: `go get github.com/conductor-sdk/conductor-go` (the source repo is at `conductor-oss/go-sdk` but the Go import path retains the historical `conductor-sdk/conductor-go` name)
.NET (C#)	github.com/conductor-oss/csharp-sdk	NuGet: `conductor-csharp` — `dotnet add package conductor-csharp`
Ruby	github.com/conductor-oss/ruby-sdk	Gem: `conductor_ruby` (early version — confirm patterns from the repo)
Rust	github.com/conductor-oss/rust-sdk	crates.io: `conductor-rust` (early version — confirm patterns from the repo)

All SDKs connect via the same env vars: CONDUCTOR_SERVER_URL, CONDUCTOR_AUTH_KEY, CONDUCTOR_AUTH_SECRET.

Why the WebFetch step matters. The Python/JS/Java/Go patterns below are stable across recent versions but the API surface still drifts (e.g. Python switched runner classes; Java’s @WorkerTask annotation gained options; Go module path moved org without changing the import). The .NET / Ruby / Rust SDKs are younger and their APIs shift more — fetch the README before scaffolding rather than trust an example below.

Python

pip install conductor-python

Define a worker

from conductor.client.worker.worker_task import worker_task

@worker_task(task_definition_name='process_order')
def process_order(order_id: str, amount: float) -> dict:
    # Your business logic here
    return {'status': 'processed', 'order_id': order_id, 'total': amount * 1.1}

Function parameters are automatically mapped from the task’s inputParameters. The return value becomes the task’s outputData.

Start workers

from conductor.client.automator.task_handler import TaskHandler
from conductor.client.configuration.configuration import Configuration

config = Configuration()  # reads CONDUCTOR_SERVER_URL, CONDUCTOR_AUTH_KEY, CONDUCTOR_AUTH_SECRET
with TaskHandler(configuration=config, scan_for_annotated_workers=True) as handler:
    handler.start_processes()
    # Workers poll until stopped

JavaScript / TypeScript

npm install @io-orkes/conductor-javascript

Define and run a worker

import {
  orkesConductorClient,
  TaskManager,
} from "@io-orkes/conductor-javascript";

const client = await orkesConductorClient({
  serverUrl: "http://localhost:8080/api",
});

const taskManager = new TaskManager(client, [
  {
    taskType: "process_order",
    execute: async ({ inputData }) => {
      return {
        status: "COMPLETED",
        outputData: {
          status: "processed",
          order_id: inputData.order_id,
        },
      };
    },
  },
]);

taskManager.startPolling();

Java

Gradle:

implementation 'org.conductoross:conductor-client:5.0.0'

Maven:

<dependency>
    <groupId>org.conductoross</groupId>
    <artifactId>conductor-client</artifactId>
    <version>5.0.0</version>
</dependency>

Option A: Implement Worker interface

public class ProcessOrderWorker implements Worker {
    @Override
    public String getTaskDefName() {
        return "process_order";
    }

    @Override
    public TaskResult execute(Task task) {
        String orderId = (String) task.getInputData().get("order_id");
        TaskResult result = new TaskResult(task);
        result.setStatus(TaskResult.Status.COMPLETED);
        result.addOutputData("status", "processed");
        result.addOutputData("order_id", orderId);
        return result;
    }
}

Option B: @WorkerTask annotation

public class Workers {
    @WorkerTask("process_order")
    public Map<String, Object> processOrder(@InputParam("order_id") String orderId) {
        return Map.of("status", "processed", "order_id", orderId);
    }
}

Start workers

ConductorClient client = ConductorClient.builder()
    .basePath("http://localhost:8080/api")
    .build();

TaskClient taskClient = new TaskClient(client);
new TaskRunnerConfigurer.Builder(taskClient, List.of(new ProcessOrderWorker()))
    .withThreadCount(10)
    .build()
    .init();

Go

go get github.com/conductor-sdk/conductor-go

Define and run a worker

package main

import (
    "fmt"
    "time"
    "github.com/conductor-sdk/conductor-go/sdk/client"
    "github.com/conductor-sdk/conductor-go/sdk/model"
    "github.com/conductor-sdk/conductor-go/sdk/worker"
)

func ProcessOrder(task *model.Task) (interface{}, error) {
    orderId := fmt.Sprintf("%v", task.InputData["order_id"])
    return map[string]interface{}{
        "status":   "processed",
        "order_id": orderId,
    }, nil
}

func main() {
    apiClient := client.NewAPIClientFromEnv()
    taskRunner := worker.NewTaskRunnerWithApiClient(apiClient)
    taskRunner.StartWorker("process_order", ProcessOrder, 1, time.Millisecond*100)
    // Blocks and polls until stopped
    select {}
}

Linking workers to workflows

In your workflow definition, use "type": "SIMPLE" and set "name" to the task type your worker polls for:

{
  "name": "process_order",
  "taskReferenceName": "process_order_ref",
  "type": "SIMPLE",
  "inputParameters": {
    "order_id": "${workflow.input.order_id}",
    "amount": "${workflow.input.amount}"
  }
}

The worker registered for task type process_order will automatically pick up this task when the workflow reaches it.

Best practices

Idempotent: Workers may receive the same task on retry. Design for safe re-execution.
Timeouts: Set responseTimeoutSeconds on task definitions so stuck tasks get rescheduled.
Error handling: Return FAILED status with a reasonForIncompletion message for graceful failures. Return FAILED_WITH_TERMINAL_ERROR to fail the task without retrying.
Scaling: Run multiple worker instances for throughput. Each polls independently.
Domain isolation: Use task domains to route tasks to specific worker groups (e.g. region-specific workers).

This site is open source. Improve this page.