Official Ruby SDK for Conductor OSS - a durable workflow orchestration engine.
- 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
Add to your Gemfile:
gem'conductor_ruby'Or install directly:
gem install conductor_rubyrequire'conductor'# Configuration (reads CONDUCTOR_SERVER_URL from environment)config=Conductor::Configuration.new# Create clientsclients=Conductor::Orkes::OrkesClients.new(config)executor=clients.get_workflow_executor# Define a workerclassGreetWorkerincludeConductor::Worker::WorkerModuleworker_task'greet'defexecute(task)name=get_input(task,'name','World'){'result'=>"Hello, #{name}!"}endend# Build workflow using new DSLworkflow=Conductor.workflow:greetings,version: 1,executor: executordogreet=simple:greet,name: wf[:name]outputresult: greet[:result]end# Register and executeworkflow.register(overwrite: true)# Start workersrunner=Conductor::Worker::TaskRunner.new(config)runner.register_worker(GreetWorker.new)runner.start# Execute workflowresult=workflow.execute(input: {'name'=>'Ruby'},wait_for_seconds: 30)puts"Result: #{result.output['result']}"# => "Hello, Ruby!"runner.stopThe SDK provides a clean, Ruby-idiomatic DSL for building workflows:
workflow=Conductor.workflow:order_processing,version: 1,executor: executordo# 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 callshttp:call_api,url: 'https://api.example.com',method: :post,body: {id: order[:id]}# Parallel executionparalleldosimple:ship_order,order_id: order[:id]simple:send_confirmation,email: user[:email]end# Conditional branchingdecideorder[:region]doon'US'dosimple:us_shippingendon'EU'dosimple:eu_shippingendotherwisedoterminate:failed,'Unsupported region'endend# Set workflow outputoutputtracking: order[:tracking_number],status: 'completed'end# Register and executeworkflow.register(overwrite: true)result=workflow.execute(input: {user_id: 123},wait_for_seconds: 60)# Simple task (worker execution)result=simple:task_name,input1: 'value',input2: wf[:param]# Inline code executionjq:transform,query: '.items | map(.name)',input: previous[:data]javascript:compute,script: 'return inputs.a + inputs.b',a: 1,b: 2# Set workflow variablesset_variable:save_state,user_id: user[:id],status: 'active'# Human/manual taskhuman:approval,display_name: 'Manager Approval',form_template: 'approval_form'# HTTP requesthttp: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# Parallel execution (fork/join)paralleldosimple:branch_asimple:branch_bsimple:branch_cend# Conditional branchingdecideorder[:status]doon'pending'dosimple:process_pendingendon'approved'dosimple:process_approvedendotherwisedosimple:handle_unknownendend# Conditional shortcutswhen_trueuser[:is_premium]dosimple:apply_discountendwhen_falseorder[:validated]doterminate:failed,'Order validation failed'end# Loop over itemsloop_overusers[:list],as: :userdosimple:process_user,user_id: iteration[:user][:id]end# Do-while loopdo_while:retry_loop,condition: '${retry_ref.output.success} == false'dosimple:retry_operationend# Call another workflowsub_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 definitioninline_workflow:nested_processdosimple:step1simple:step2end# Wait for durationwait:pause,duration: '30s'# or '5m', '1h', '2d'# Wait until specific timewait:scheduled,until: '2024-12-25T00:00:00Z'# Wait for external webhookwait_for_webhook:external_callback,matches: {'type'=>'payment','order_id'=>'${workflow.input.order_id}'}# Publish eventevent:notify,sink: 'conductor:workflow_events',payload: {status: 'completed'}# Complete workflowterminate:success,'Processing completed successfully'# Fail workflowterminate:failed,'Validation error: missing required field'# 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'workflow=Conductor.workflow:ai_assistant,executor: executordo# 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 completionllm_text:complete,provider: 'anthropic',model: 'claude-3-sonnet',prompt: 'Summarize: ${workflow.input.text}'# Generate embeddingsembeddings=llm_embeddings:embed,provider: 'openai',model: 'text-embedding-3-small',text: wf[:document]# Store embeddings in vector DBllm_store_embeddings:store,provider: 'pinecone',index: 'documents',embeddings: embeddings[:embeddings],metadata: {doc_id: wf[:doc_id]}# Search embeddingsllm_search_embeddings:search,provider: 'pinecone',index: 'documents',query: wf[:search_query],max_results: 10# Generate imagegenerate_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) integrationtools=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]}outputanswer: response[:content]endThe DSL uses a clean syntax for referencing outputs:
# Workflow input referencewf[:user_id]# => '${workflow.input.user_id}'# Task output referencetask[: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 processediteration[:index]# Current index (0-based)iteration[:user][:name]# If `as: :user` specifiedThe 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 variablesexport CONDUCTOR_SERVER_URL=http://localhost:8080/api
# For Orkes Cloud:# export CONDUCTOR_AUTH_KEY=your_key# export CONDUCTOR_AUTH_SECRET=your_secret# Run hello worldcd 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.rbclassImageProcessorincludeConductor::Worker::WorkerModuleworker_task'process_image',poll_interval: 1,thread_count: 4defexecute(task)url=get_input(task,'image_url')# Process image...result=Conductor::Http::Models::TaskResult.completeresult.add_output_data('processed_url',processed_url)result.log('Image processed successfully')resultendendworker=Conductor::Worker.define('simple_task')do |task|
input=task.input_data['value']{result: input * 2}# Return hash for automatic TaskResultendrunner=Conductor::Worker::TaskRunner.new(config)runner.register_worker(ImageProcessor.new)runner.register_worker(worker)runner.start(threads: 4)# Graceful shutdowntrap('INT'){runner.stop}sleepwhilerunner.running?export CONDUCTOR_SERVER_URL=http://localhost:8080/api
export CONDUCTOR_AUTH_KEY=your_key # For Orkes Cloudexport CONDUCTOR_AUTH_SECRET=your_secret # For Orkes Cloudconfig=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 | 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. |
clients=Conductor::Orkes::OrkesClients.new(config)workflow_client=clients.get_workflow_clienttask_client=clients.get_task_clientmetadata_client=clients.get_metadata_clientscheduler_client=clients.get_scheduler_clientprompt_client=clients.get_prompt_clientsecret_client=clients.get_secret_clientauthorization_client=clients.get_authorization_clientworkflow_executor=clients.get_workflow_executor# Unit tests
bundle exec rspec spec/conductor/
# Integration tests (requires Conductor server)
CONDUCTOR_SERVER_URL=http://localhost:8080/api bundle exec rspec spec/integration/- Ruby 2.6+ (Ruby 3+ recommended)
- Conductor OSS 3.x or Orkes Cloud
faraday ~> 2.0- HTTP clientfaraday-net_http_persistent ~> 2.0- Connection poolingfaraday-retry ~> 2.0- Automatic retriesconcurrent-ruby ~> 1.2- Thread-safe concurrency
- Fork the repository
- Create your feature branch (
git checkout -b feature/amazing-feature) - Run tests (
bundle exec rspec) - Commit your changes (
git commit -m 'Add amazing feature') - Push to the branch (
git push origin feature/amazing-feature) - Open a Pull Request
Apache 2.0 - see LICENSE for details.