Ruby SDK

Start here

Goal	Guide
Run a workflow	Run your first workflow
Write a worker	Write your first worker
Build an agent	Coming soon

Featured examples

Category	Maintained upstream example
Workflow and worker	Examples
Agentic workflow	Agentic workflow examples
API journey	Not currently maintained upstream

The agentic-workflow row covers SDK examples that orchestrate LLMs or tools. It is separate from the SDK-authored Conductor Agent quickstart, which is coming soon for this SDK.

Connect to Conductor

For local OSS, set CONDUCTOR_SERVER_URL=http://localhost:8080/api.

For Orkes Developer Edition, set CONDUCTOR_SERVER_URL=https://developer.orkescloud.com/api, CONDUCTOR_AUTH_KEY, and CONDUCTOR_AUTH_SECRET. Keep credentials out of source control.

This SDK reads these environment variables when constructing its standard client configuration.

Features

Maintenance

This SDK is part of the Conductor OSS ecosystem. Conductor OSS remains actively maintained under the Conductor OSS community, with Orkes contributing maintenance, engineering, documentation, and enterprise support.

• Full Feature Parity with Python SDK
• Ruby-Idiomatic Workflow DSL - Clean block-based syntax with 25+ task types
• Worker Framework - Multi-threaded task execution with class-based and block-based workers
• LLM/AI Tasks - Chat completion, embeddings, RAG, image/audio generation
• Orkes Cloud Support - Authentication, secrets, integrations, prompts
• Comprehensive Testing - 400+ unit tests, 110 integration tests

Installation

Add to your Gemfile:

gem 'conductor_ruby'

Or install directly:

gem install conductor_ruby

Quick Start

Hello World

require 'conductor'

# Configuration (reads CONDUCTOR_SERVER_URL from environment)
config = Conductor::Configuration.new

# Create clients
clients = Conductor::Orkes::OrkesClients.new(config)
executor = clients.get_workflow_executor

# Define a worker
class GreetWorker
  include Conductor::Worker::WorkerModule
  worker_task 'greet'

  def execute(task)
    name = get_input(task, 'name', 'World')
    { 'result' => "Hello, #{name}!" }
  end
end

# Build workflow using new DSL
workflow = Conductor.workflow :greetings, version: 1, executor: executor do
  greet = simple :greet, name: wf[:name]
  output result: greet[:result]
end

# Register and execute
workflow.register(overwrite: true)

# Start workers
runner = Conductor::Worker::TaskRunner.new(config)
runner.register_worker(GreetWorker.new)
runner.start

# Execute workflow
result = workflow.execute(input: { 'name' => 'Ruby' }, wait_for_seconds: 30)
puts "Result: #{result.output['result']}"  # => "Hello, Ruby!"

runner.stop

Workflow DSL

The SDK provides a clean, Ruby-idiomatic DSL for building workflows:

workflow = Conductor.workflow :order_processing, version: 1, executor: executor do
  # Access workflow inputs with wf[:param]
  user = simple :get_user, user_id: wf[:user_id]

  # Reference task outputs with task[:field]
  order = simple :validate_order, email: user[:email]

  # HTTP calls
  http :call_api, url: 'https://api.example.com', method: :post, body: { id: order[:id] }

  # Parallel execution
  parallel do
    simple :ship_order, order_id: order[:id]
    simple :send_confirmation, email: user[:email]
  end

  # Conditional branching
  decide order[:region] do
    on 'US' do
      simple :us_shipping
    end
    on 'EU' do
      simple :eu_shipping
    end
    otherwise do
      terminate :failed, 'Unsupported region'
    end
  end

  # Set workflow output
  output tracking: order[:tracking_number], status: 'completed'
end

# Register and execute
workflow.register(overwrite: true)
result = workflow.execute(input: { user_id: 123 }, wait_for_seconds: 60)

Task Methods Reference

Basic Tasks

# Simple task (worker execution)
result = simple :task_name, input1: 'value', input2: wf[:param]

# Inline code execution
jq :transform, query: '.items | map(.name)', input: previous[:data]
javascript :compute, script: 'return inputs.a + inputs.b', a: 1, b: 2

# Set workflow variables
set_variable :save_state, user_id: user[:id], status: 'active'

# Human/manual task
human :approval, display_name: 'Manager Approval', form_template: 'approval_form'

HTTP Tasks

# HTTP request
http :call_api,
  url: 'https://api.example.com/users',
  method: :post,
  headers: { 'Authorization' => 'Bearer ${workflow.secrets.api_token}' },
  body: { name: wf[:name], email: wf[:email] }

# HTTP polling (wait for condition)
http_poll :wait_for_ready,
  url: 'https://api.example.com/status/${workflow.input.job_id}',
  method: :get,
  termination_condition: '$.status == "ready"',
  polling_interval: 5,
  polling_strategy: :fixed

Control Flow

# Parallel execution (fork/join)
parallel do
  simple :branch_a
  simple :branch_b
  simple :branch_c
end

# Conditional branching
decide order[:status] do
  on 'pending' do
    simple :process_pending
  end
  on 'approved' do
    simple :process_approved
  end
  otherwise do
    simple :handle_unknown
  end
end

# Conditional shortcuts
when_true user[:is_premium] do
  simple :apply_discount
end

when_false order[:validated] do
  terminate :failed, 'Order validation failed'
end

# Loop over items
loop_over users[:list], as: :user do
  simple :process_user, user_id: iteration[:user][:id]
end

# Do-while loop
do_while :retry_loop, condition: '${retry_ref.output.success} == false' do
  simple :retry_operation
end

Sub-workflows

# Call another workflow
sub_workflow :process_order,
  workflow_name: 'order_processor',
  version: 2,
  input: { order_id: wf[:order_id] }

# Start workflow (fire-and-forget)
start_workflow :trigger_notification,
  workflow_name: 'send_notifications',
  input: { user_id: user[:id] }

# Inline sub-workflow definition
inline_workflow :nested_process do
  simple :step1
  simple :step2
end

Wait and Events

# Wait for duration
wait :pause, duration: '30s'   # or '5m', '1h', '2d'

# Wait until specific time
wait :scheduled, until: '2024-12-25T00:00:00Z'

# Wait for external webhook
wait_for_webhook :external_callback,
  matches: { 'type' => 'payment', 'order_id' => '${workflow.input.order_id}' }

# Publish event
event :notify, sink: 'conductor:workflow_events', payload: { status: 'completed' }

Termination

# Complete workflow
terminate :success, 'Processing completed successfully'

# Fail workflow
terminate :failed, 'Validation error: missing required field'

Dynamic Tasks

# Dynamic task name (resolved at runtime)
dynamic :run_handler, task_to_execute: wf[:handler_name]

# Dynamic fork (parallel tasks determined at runtime)
dynamic_fork :process_all,
  tasks_input: wf[:items],
  task_name: 'process_item'

LLM/AI Tasks

workflow = Conductor.workflow :ai_assistant, executor: executor do
  # Chat completion (messages auto-converted from simple format)
  response = llm_chat :chat,
    provider: 'openai',
    model: 'gpt-4',
    messages: [
      { role: :system, message: 'You are a helpful assistant.' },
      { role: :user, message: wf[:question] }
    ],
    temperature: 0.7

  # Text completion
  llm_text :complete,
    provider: 'anthropic',
    model: 'claude-3-sonnet',
    prompt: 'Summarize: ${workflow.input.text}'

  # Generate embeddings
  embeddings = llm_embeddings :embed,
    provider: 'openai',
    model: 'text-embedding-3-small',
    text: wf[:document]

  # Store embeddings in vector DB
  llm_store_embeddings :store,
    provider: 'pinecone',
    index: 'documents',
    embeddings: embeddings[:embeddings],
    metadata: { doc_id: wf[:doc_id] }

  # Search embeddings
  llm_search_embeddings :search,
    provider: 'pinecone',
    index: 'documents',
    query: wf[:search_query],
    max_results: 10

  # Generate image
  generate_image :create_image,
    provider: 'openai',
    model: 'dall-e-3',
    prompt: 'A sunset over mountains',
    size: '1024x1024'

  # Generate audio (text-to-speech)
  generate_audio :speak,
    provider: 'openai',
    model: 'tts-1',
    text: response[:content],
    voice: 'nova'

  # MCP (Model Context Protocol) integration
  tools = list_mcp_tools :get_tools, server_name: 'my_mcp_server'

  call_mcp_tool :use_tool,
    server_name: 'my_mcp_server',
    tool_name: 'search_documents',
    arguments: { query: wf[:query] }

  output answer: response[:content]
end

Output References

The DSL uses a clean syntax for referencing outputs:

# Workflow input reference
wf[:user_id]              # => '${workflow.input.user_id}'

# Task output reference
task[:field]              # => '${task_ref.output.field}'
task[:nested][:path]      # => '${task_ref.output.nested.path}'

# Loop iteration references (inside loop_over)
iteration[:current_item]  # Current item being processed
iteration[:index]         # Current index (0-based)
iteration[:user][:name]   # If `as: :user` specified

Examples

The examples/ directory contains comprehensive examples:

Example	Description
helloworld/	Simplest complete example - worker + workflow + execution
workflow_dsl.rb	Comprehensive new DSL showcase
simple_worker.rb	Worker patterns: class-based, block-based, error handling
kitchensink.rb	All major task types using new DSL
dynamic_workflow.rb	Create and execute workflows at runtime
workflow_ops.rb	Lifecycle operations: pause, resume, restart, retry
agentic_workflows/	LLM chat and AI workflow examples

Run examples:

# Set environment variables
export CONDUCTOR_SERVER_URL=<YOUR-CLUSTER-URL>/api
# For Orkes Cloud:
# export CONDUCTOR_AUTH_KEY=your_key
# export CONDUCTOR_AUTH_SECRET=your_secret

# Run hello world
cd examples/helloworld && bundle exec ruby helloworld.rb

# Run DSL showcase
bundle exec ruby examples/workflow_dsl.rb

# Run kitchen sink
bundle exec ruby examples/kitchensink.rb

Worker Framework

Class-Based Workers

class ImageProcessor
  include Conductor::Worker::WorkerModule

  worker_task 'process_image', poll_interval: 1, thread_count: 4

  def execute(task)
    url = get_input(task, 'image_url')
    # Process image...

    result = Conductor::Http::Models::TaskResult.complete
    result.add_output_data('processed_url', processed_url)
    result.log('Image processed successfully')
    result
  end
end

Block-Based Workers

worker = Conductor::Worker.define('simple_task') do |task|
  input = task.input_data['value']
  { result: input * 2 }  # Return hash for automatic TaskResult
end

Running Workers

runner = Conductor::Worker::TaskRunner.new(config)
runner.register_worker(ImageProcessor.new)
runner.register_worker(worker)
runner.start(threads: 4)

# Graceful shutdown
trap('INT') { runner.stop }
sleep while runner.running?

Configuration

Environment Variables

export CONDUCTOR_SERVER_URL=<YOUR-CLUSTER-URL>/api
export CONDUCTOR_AUTH_KEY=your_key        # For Orkes Cloud
export CONDUCTOR_AUTH_SECRET=your_secret  # For Orkes Cloud

Programmatic

config = Conductor::Configuration.new(
  server_api_url: 'https://play.orkes.io/api',
  auth_key: 'your_key',
  auth_secret: 'your_secret',
  auth_token_ttl_min: 45,
  verify_ssl: true
)

API Coverage

Resource APIs (17 classes)

API	Description
WorkflowResourceApi	Workflow execution and management
TaskResourceApi	Task polling and updates
MetadataResourceApi	Workflow/task definitions
SchedulerResourceApi	Scheduled workflows
EventResourceApi	Event handlers
WorkflowBulkResourceApi	Bulk operations
PromptResourceApi	AI prompt templates
SecretResourceApi	Secret management
IntegrationResourceApi	External integrations
+ 8 more	Authorization, Users, Groups, Roles, etc.

High-Level Clients (9 classes)

clients = Conductor::Orkes::OrkesClients.new(config)

workflow_client = clients.get_workflow_client
task_client = clients.get_task_client
metadata_client = clients.get_metadata_client
scheduler_client = clients.get_scheduler_client
prompt_client = clients.get_prompt_client
secret_client = clients.get_secret_client
authorization_client = clients.get_authorization_client
workflow_executor = clients.get_workflow_executor

Testing

# Unit tests
bundle exec rspec spec/conductor/

# Integration tests (requires Conductor server)
CONDUCTOR_SERVER_URL=<YOUR-CLUSTER-URL>/api bundle exec rspec spec/integration/

Requirements

• Ruby 2.6+ (Ruby 3+ recommended)
• Conductor OSS 3.x or Orkes Cloud

Dependencies

• faraday ~> 2.0 - HTTP client
• faraday-net_http_persistent ~> 2.0 - Connection pooling
• faraday-retry ~> 2.0 - Automatic retries
• concurrent-ruby ~> 1.2 - Thread-safe concurrency

Contributing

1. Fork the repository
2. Create your feature branch (git checkout -b feature/amazing-feature)
3. Run tests (bundle exec rspec)
4. Commit your changes (git commit -m 'Add amazing feature')
5. Push to the branch (git push origin feature/amazing-feature)
6. Open a Pull Request

License

Apache 2.0 - see LICENSE for details.

Links

• Conductor OSS
• Orkes Cloud
• Documentation
• Python SDK
• Community Slack