docker run -p 8080:8080 conductoross/conductor:latest

curl -sSL https://raw.githubusercontent.com/conductor-oss/conductor/main/conductor_server.sh | sh

# Installs conductor cli
npm install -g @conductor-oss/conductor-cli

# Start the open source conductor server
conductor server start
# see conductor server --help for all the available commands

pip install conductor-python

from conductor.client.workflow.conductor_workflow import ConductorWorkflow

workflow = ConductorWorkflow(name='greetings', version=1, executor=executor)
greet_task = greet(task_ref_name='greet_ref', name=workflow.input('name'))
workflow >> greet_task
workflow.output_parameters({'result': greet_task.output('result')})
workflow.register(overwrite=True)

from conductor.client.worker.worker_task import worker_task

# register_task_def=True is convenient for local dev quickstarts; in production, manage task definitions separately.
@worker_task(task_definition_name='greet', register_task_def=True)
def greet(name: str) -> str:
    return f'Hello {name}'

from conductor.client.automator.task_handler import TaskHandler
from conductor.client.configuration.configuration import Configuration
from conductor.client.orkes_clients import OrkesClients
from conductor.client.workflow.conductor_workflow import ConductorWorkflow
from conductor.client.worker.worker_task import worker_task


# A worker is any Python function.
@worker_task(task_definition_name='greet', register_task_def=True)
def greet(name: str) -> str:
    return f'Hello {name}'


def main():
    # Configure the SDK (reads CONDUCTOR_SERVER_URL / CONDUCTOR_AUTH_* from env).
    config = Configuration()

    clients = OrkesClients(configuration=config)
    executor = clients.get_workflow_executor()

    # Build a workflow with the >> operator.
    workflow = ConductorWorkflow(name='greetings', version=1, executor=executor)
    greet_task = greet(task_ref_name='greet_ref', name=workflow.input('name'))
    workflow >> greet_task
    workflow.output_parameters({'result': greet_task.output('result')})
    workflow.register(overwrite=True)

    # Start polling for tasks (one worker subprocess per worker function).
    with TaskHandler(configuration=config, scan_for_annotated_workers=True) as task_handler:
        task_handler.start_processes()

        # Run the workflow and get the result.
        run = executor.execute(name='greetings', version=1, workflow_input={'name': 'Conductor'})
        print(f'result: {run.output["result"]}')
        print(f'execution: {config.ui_host}/execution/{run.workflow_id}')


if __name__ == '__main__':
    main()

python quickstart.py

from conductor.client.worker.worker_task import worker_task

# Sync worker — for CPU-bound work (uses ThreadPoolExecutor)
@worker_task(task_definition_name='process_image', thread_count=4)
def process_image(image_url: str) -> dict:
    import PIL.Image, io, requests
    img = PIL.Image.open(io.BytesIO(requests.get(image_url).content))
    img.thumbnail((256, 256))
    return {'width': img.width, 'height': img.height}


# Async worker — for I/O-bound work (uses AsyncTaskRunner, no thread overhead)
@worker_task(task_definition_name='fetch_data', thread_count=50)
async def fetch_data(url: str) -> dict:
    import httpx
    async with httpx.AsyncClient() as client:
        resp = await client.get(url)
    return resp.json()

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

config = Configuration()
with TaskHandler(configuration=config, scan_for_annotated_workers=True) as task_handler:
    task_handler.start_processes()
    task_handler.join_processes()  # blocks forever (workers poll continuously)

from conductor.client.workflow.conductor_workflow import ConductorWorkflow
from conductor.client.workflow.task.http_task import HttpTask
from conductor.client.workflow.task.wait_task import WaitTask

workflow = ConductorWorkflow(name='order_pipeline', version=1, executor=executor)

# Custom worker task
validate = validate_order(task_ref_name='validate', order_id=workflow.input('order_id'))

# Built-in HTTP task — call any API, no worker needed
charge_payment = HttpTask(task_ref_name='charge_payment', http_input={
    'uri': 'https://api.stripe.com/v1/charges',
    'method': 'POST',
    'headers': {'Authorization': ['Bearer ${workflow.input.stripe_key}']},
    'body': {'amount': '${validate.output.amount}'}
})

# Built-in Wait task — pause the workflow for 10 seconds
cool_down = WaitTask(task_ref_name='cool_down', wait_for_seconds=10)

# Another custom worker task
notify = send_notification(task_ref_name='notify', message='Order complete')

# Chain with >> operator
workflow >> validate >> charge_payment >> cool_down >> notify

# Execute synchronously and wait for the result
result = workflow.execute(workflow_input={'order_id': 'ORD-123', 'stripe_key': 'sk_test_...'})
print(result.output)

from typing import Union
from conductor.client.worker.worker_task import worker_task
from conductor.client.context.task_context import get_task_context, TaskInProgress

@worker_task(task_definition_name='batch_job')
def batch_job(batch_id: str) -> Union[dict, TaskInProgress]:
    ctx = get_task_context()
    ctx.add_log(f"Processing batch {batch_id}, poll #{ctx.get_poll_count()}")

    if ctx.get_poll_count() < 3:
        # Not done yet — re-queue and check again in 30 seconds
        return TaskInProgress(callback_after_seconds=30, output={'progress': ctx.get_poll_count() * 33})

    # Done after 3 polls
    return {'status': 'completed', 'batch_id': batch_id}

from conductor.client.automator.task_handler import TaskHandler
from conductor.client.configuration.configuration import Configuration
from conductor.client.configuration.settings.metrics_settings import MetricsSettings

config = Configuration()
metrics = MetricsSettings(directory='/tmp/conductor-metrics', http_port=8000)

with TaskHandler(configuration=config, metrics_settings=metrics, scan_for_annotated_workers=True) as task_handler:
    task_handler.start_processes()
    task_handler.join_processes()

# Prometheus-compatible endpoint
curl http://localhost:8000/metrics

from conductor.client.configuration.configuration import Configuration
from conductor.client.http.models import StartWorkflowRequest, RerunWorkflowRequest, TaskResult
from conductor.client.orkes_clients import OrkesClients

config = Configuration()
clients = OrkesClients(configuration=config)
workflow_client = clients.get_workflow_client()
task_client = clients.get_task_client()
executor = clients.get_workflow_executor()

# Start async (returns workflow ID immediately)
workflow_id = executor.start_workflow(StartWorkflowRequest(name='my_workflow', input={'key': 'value'}))

# Execute sync (blocks until workflow completes)
result = executor.execute(name='my_workflow', version=1, workflow_input={'key': 'value'})

# Lifecycle management
workflow_client.pause_workflow(workflow_id)
workflow_client.resume_workflow(workflow_id)
workflow_client.terminate_workflow(workflow_id, reason='no longer needed')
workflow_client.retry_workflow(workflow_id)          # retry from last failed task
workflow_client.restart_workflow(workflow_id)         # restart from the beginning
workflow_client.rerun_workflow(workflow_id,           # rerun from a specific task
    RerunWorkflowRequest(re_run_from_task_id=task_id))

# Send a signal to a waiting workflow (complete a WAIT task externally)
task_client.update_task(TaskResult(
    workflow_instance_id=workflow_id,
    task_id=wait_task_id,
    status='COMPLETED',
    output_data={'approved': True}
))

# Search workflows
results = workflow_client.search(query='status IN (RUNNING) AND correlationId = "order-123"')

# Automated multi-turn chat
python examples/agentic_workflows/llm_chat.py

# Multi-agent debate
python examples/agentic_workflows/multiagent_chat.py --topic "renewable energy"

# RAG pipeline
pip install "markitdown[pdf]"
python examples/rag_workflow.py document.pdf "What are the key findings?"