Ironflow

Ironflow is a workflow orchestration platform where workflows are imperative Rust code executed by background workers, with persistence, cost tracking, and human approval gates.

Why Ironflow?

• Workflows are Rust code. No YAML, no DSL. Full type safety, IDE support, and compile-time checks.
• Persistent execution. Every step is tracked in a database. Runs survive process restarts.
• Human-in-the-loop. Approval gates pause a run until someone approves or rejects it.
• Cost tracking. Every agent call is metered. Set per-run and monthly budgets.
• Scalable. Add more workers to increase throughput. Workers poll the API for pending runs.

Quick links

• Getting Started – install, configure, run
• Concepts – understand the building blocks
• Guides – step-by-step tutorials
• Architecture – how the pieces fit together
• API Reference (docs.rs) – generated from rustdoc

Installation

Prerequisites

• Rust 1.94+ (see rust-version in Cargo.toml)
• A running PostgreSQL instance (for production; in-memory store available for development)

Add dependencies

Add the crates you need to your Cargo.toml:

[dependencies]
ironflow-engine = "0.1"   # Workflow handler, context, engine
ironflow-api = "0.1"      # REST API server
ironflow-worker = "0.1"   # Background worker
ironflow-store = "0.1"    # Storage backends
ironflow-core = "0.1"     # Shell, agent providers

Minimal project structure

A typical Ironflow project has three parts:

1. A library crate with your workflow handlers
2. A server binary that exposes the API and serves the dashboard
3. A worker binary that executes workflows

my-project/
├── src/
│   └── lib.rs          # Your workflow handlers
├── src/bin/
│   ├── server.rs       # API server
│   └── worker.rs       # Background worker
└── Cargo.toml

See the example server and example worker for complete working code.

Running the Server

The API server exposes a REST API for managing workflows, runs, and steps. It also serves the web dashboard.

Example server

The repository includes a complete example server:

//! ironflow API server example.
//!
//! ```sh
//! cargo run -p ironflow-example-server
//! ```
//!
//! The dashboard is served automatically via the `dashboard` feature in `ironflow-api`.
//!
//! Environment:
//! - `IRONFLOW_ENV` (`production` or `development`, default: development)
//! - `DATABASE_URL` (required in production)
//! - `JWT_SECRET` (required in production, default: dev secret)
//! - `WORKER_TOKEN` (required in production, default: dev token)
//! - `PORT` (default: 3000)
//! - `DASHBOARD_DIR` (optional: overrides the embedded dashboard with a filesystem path)
//! - `ALLOWED_ORIGINS` (comma-separated list; omit to allow same-origin only)
//! - `WEBHOOK_URL` (optional: outbound webhook for run events)
//! - `ARTIFACTS_DIR` (optional: filesystem root for step artifacts; unset
//!   leaves artifacts disabled and the artifact routes answer `501`)
//! - `ARTIFACT_MAX_BYTES` (optional: per-artifact size limit, default 100 MiB)
//! - `IRONFLOW_DEFAULT_RUN_MAX_COST_USD` (optional: default per-run cost cap in
//!   USD, applied when neither the run creation request nor the workflow
//!   handler declares one; unset means no cap)
//! - `IRONFLOW_MONTHLY_COST_LIMIT_USD` (optional: global cost quota for the
//!   current calendar month in UTC; beyond it, creating a run returns
//!   `429 MONTHLY_BUDGET_EXCEEDED` while in-flight runs continue)
//! - `IRONFLOW_SEED` (optional: when set to any value, seeds development data
//!   at startup -- users, runs, steps, API keys)

use std::process;
use std::sync::Arc;

use axum::http::header::{AUTHORIZATION, CONTENT_TYPE};
use axum::http::{HeaderValue, Method};
use tokio::net::TcpListener;
use tokio::spawn;
use tokio_util::sync::CancellationToken;
use tower_http::cors::CorsLayer;
use tracing::{info, warn};
use tracing_subscriber::EnvFilter;

use ironflow_api::config::ServerConfig;
use ironflow_api::reaper::Reaper;
use ironflow_api::routes::{RouterConfig, create_router};
use ironflow_api::sse::SseBroadcaster;
use ironflow_api::state::AppState;
use ironflow_artifacts::blob_store::BlobStore;
use ironflow_artifacts::local::LocalBlobStore;
use ironflow_auth::jwt::JwtConfig;
use ironflow_core::providers::claude::ClaudeCodeProvider;
use ironflow_engine::artifact::DirectArtifactSink;
use ironflow_engine::budget::BudgetConfig;
use ironflow_engine::engine::Engine;
use ironflow_engine::notify::{Event, WebhookSubscriber, WorkflowEventBus};
use ironflow_store::crypto::{KeyRing, SECRET_KEYS_ENV};
use ironflow_store::memory::InMemoryStore;
use ironflow_store::store::Store;
use xtask::seed::{SeedOptions, seed_store};

#[tokio::main]
async fn main() {
    tracing_subscriber::fmt()
        .with_env_filter(
            EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| "info,ironflow=debug".parse().expect("valid filter")),
        )
        .init();

    let config = ServerConfig::from_env().unwrap_or_else(|e| {
        eprintln!("{e}");
        process::exit(1);
    });

    let mut store = InMemoryStore::new();

    let key_ring = KeyRing::from_env().unwrap_or_else(|e| {
        eprintln!("invalid secret key configuration: {e}");
        process::exit(1);
    });

    let has_key_ring = key_ring.is_some();
    match key_ring {
        Some(ring) => {
            info!(
                active_version = ring.active_version(),
                configured_versions = ?ring.versions(),
                "secret store enabled"
            );
            store.set_key_ring(ring);
        }
        None => {
            info!("{SECRET_KEYS_ENV} not set, secret store disabled");
        }
    }

    let store: Arc<dyn Store> = Arc::new(store);

    // A secret encrypted with a key that is no longer configured is
    // unreadable. Fail here rather than at the first workflow that needs it.
    if has_key_ring {
        let status = store.secret_key_status().await.unwrap_or_else(|e| {
            eprintln!("cannot read secret key versions: {e}");
            process::exit(1);
        });

        if !status.is_consistent() {
            let missing: Vec<String> = status.missing.iter().map(|v| v.to_string()).collect();
            eprintln!(
                "secret key versions present in database but missing from configuration: {}\n\
                 set {SECRET_KEYS_ENV} to include them, or rotate before removing a key",
                missing.join(", ")
            );
            process::exit(1);
        }
    }
    if std::env::var("IRONFLOW_SEED").is_ok() {
        info!("IRONFLOW_SEED set, seeding development data...");
        let seed_opts = SeedOptions {
            force: false,
            artifacts_dir: config.artifacts_dir.clone(),
        };
        seed_store(&*store, &seed_opts).await.unwrap_or_else(|e| {
            warn!("seed skipped: {e}");
        });
    }

    let provider = Arc::new(ClaudeCodeProvider::new());

    let jwt_config = Arc::new(JwtConfig {
        secret: config.jwt_secret.clone(),
        access_token_ttl_secs: 900,
        refresh_token_ttl_secs: 604800,
        cookie_domain: None,
        cookie_secure: config.is_production,
    });

    let budget = BudgetConfig::from_env();
    info!(
        default_run_max_cost_usd = ?budget.default_run_max_cost_usd,
        monthly_cost_limit_usd = ?budget.monthly_cost_limit_usd,
        "cost guardrails loaded"
    );

    let mut engine = Engine::new(store.clone(), provider).with_budget_config(budget);
    ironflow_workflows::register_all(&mut engine).expect("failed to register workflows");

    // Artifacts stay off until a storage root is configured. The API and any
    // in-process run then share the same backend, so a file a step produces is
    // downloadable from the same server that stored it.
    let blob_store: Option<Arc<dyn BlobStore>> = config.artifacts_dir.as_ref().map(|dir| {
        info!(
            dir = %dir.display(),
            max_bytes = config.artifact_max_bytes,
            "artifact storage enabled"
        );
        Arc::new(LocalBlobStore::new(dir).max_bytes(config.artifact_max_bytes))
            as Arc<dyn BlobStore>
    });

    if let Some(ref blob) = blob_store {
        engine.set_artifact_sink(Arc::new(DirectArtifactSink::new(
            blob.clone(),
            store.clone(),
        )));
    }

    if let Some(ref webhook_url) = config.webhook_url {
        info!(url = %webhook_url, "registering webhook subscriber");
        engine.subscribe(
            WebhookSubscriber::new(webhook_url),
            &[Event::RUN_STATUS_CHANGED, Event::STEP_FAILED],
        );
    }

    let sse_broadcaster = SseBroadcaster::new();
    let event_sender = sse_broadcaster.sender();
    engine.subscribe(sse_broadcaster, Event::ALL);

    let event_bus = WorkflowEventBus::new();
    engine.set_event_bus(event_bus.clone());

    let engine = Arc::new(engine);

    let cors = build_cors(&config);

    let mut state = AppState::new(
        store.clone(),
        engine.clone(),
        jwt_config,
        config.worker_token.clone(),
        event_sender,
    )
    .with_event_bus(event_bus);
    if let Some(blob) = blob_store {
        state = state.with_blob_store(blob);
    }

    // Without the reaper, a run whose worker dies stays Running forever.
    let shutdown = CancellationToken::new();
    spawn(Reaper::new(store, engine).run(shutdown.clone()));
    let router_config = RouterConfig {
        dashboard_dir: config.dashboard_dir.clone(),
        rate_limit_auth: config.rate_limit_auth,
        rate_limit_general: config.rate_limit_general,
    };
    let app = create_router(state, router_config)
        .layer(cors)
        .into_make_service();

    let addr = format!("0.0.0.0:{}", config.port);
    let listener = TcpListener::bind(&addr).await.expect("bind address");

    info!("==============================================");
    info!("  ironflow server on http://{addr}");
    info!(
        "  environment: {}",
        if config.is_production {
            "production"
        } else {
            "development"
        }
    );
    info!("==============================================");

    axum::serve(listener, app)
        .with_graceful_shutdown(async move {
            tokio::signal::ctrl_c().await.expect("ctrl+c handler");
            info!("shutting down...");
            shutdown.cancel();
        })
        .await
        .expect("serve");
}

/// Build CORS layer from config.
///
/// - If `allowed_origins` is set: only those origins are permitted (comma-separated).
/// - If unset: no extra origins are allowed (same-origin only).
///
/// Credentials (cookies) are always allowed so JWT cookies work cross-origin.
fn build_cors(config: &ServerConfig) -> CorsLayer {
    let methods = vec![Method::GET, Method::POST, Method::PUT, Method::DELETE];
    let headers = vec![AUTHORIZATION, CONTENT_TYPE];

    match config.allowed_origins {
        Some(ref raw) => {
            let origins: Vec<HeaderValue> = raw
                .split(',')
                .map(str::trim)
                .filter(|s| !s.is_empty())
                .filter_map(|s| match s.parse::<HeaderValue>() {
                    Ok(v) => Some(v),
                    Err(err) => {
                        warn!(origin = s, %err, "ignoring invalid CORS origin");
                        None
                    }
                })
                .collect();

            info!(?origins, "CORS: allowing configured origins");

            CorsLayer::new()
                .allow_origin(origins)
                .allow_methods(methods)
                .allow_headers(headers)
                .allow_credentials(true)
        }
        None => {
            info!("CORS: no ALLOWED_ORIGINS set, same-origin only");

            CorsLayer::new()
                .allow_methods(methods)
                .allow_headers(headers)
        }
    }
}

Environment variables

Running

cargo run -p ironflow-example-server

The server starts on http://localhost:3000. The dashboard is available at the root URL.

Running a Worker

Workers poll the API for pending runs, acquire leases, and execute workflow handlers.

Example worker

//! ironflow worker example.
//!
//! ```sh
//! cargo run -p ironflow-example-worker
//! ```
//!
//! Environment:
//! - `API_URL` (default: http://localhost:3000)
//! - `WORKER_TOKEN` (default: dev token)
//! - `CONCURRENCY` (default: 2)
//! - `POLL_INTERVAL_SECS` (default: 2)

use std::env;
use std::sync::Arc;
use std::time::Duration;

use tracing::info;
use tracing_subscriber::EnvFilter;

use ironflow_core::providers::claude::ClaudeCodeProvider;
use ironflow_worker::WorkerBuilder;
use ironflow_workflows::handlers;

#[tokio::main]
async fn main() {
    tracing_subscriber::fmt()
        .with_env_filter(
            EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| "info,ironflow=debug".parse().expect("valid filter")),
        )
        .init();

    let api_url = env::var("API_URL").unwrap_or_else(|_| "http://localhost:3000".to_string());
    let worker_token =
        env::var("WORKER_TOKEN").unwrap_or_else(|_| "ironflow-dev-worker-token".to_string());
    let concurrency: usize = env::var("CONCURRENCY")
        .ok()
        .and_then(|c| c.parse().ok())
        .unwrap_or(2);
    let poll_interval: u64 = env::var("POLL_INTERVAL_SECS")
        .ok()
        .and_then(|p| p.parse().ok())
        .unwrap_or(2);

    let mut builder = WorkerBuilder::new(&api_url, &worker_token)
        .provider(Arc::new(ClaudeCodeProvider::new()))
        .concurrency(concurrency)
        .poll_interval(Duration::from_secs(poll_interval));

    // Same list as the server: one source of truth for both binaries.
    for handler in handlers() {
        builder = builder.register(handler);
    }

    let worker = builder.build().expect("failed to build worker");

    info!("==============================================");
    info!("  ironflow worker");
    info!("  API: {api_url}");
    info!("  Concurrency: {concurrency}");
    info!("==============================================");

    if let Err(e) = worker.run().await {
        tracing::error!("worker error: {e}");
    }
}

Environment variables

Running

cargo run -p ironflow-example-worker

Scaling

To increase throughput, start multiple workers. Each worker polls independently and acquires leases on runs, so no coordination is needed beyond the API server.

WorkflowHandler

A WorkflowHandler is the core abstraction in Ironflow. It defines a named workflow as imperative Rust code.

The trait

pub trait WorkflowHandler: Send + Sync {
    fn name(&self) -> &str;
    fn description(&self) -> &str;
    fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a>;

    // Optional methods
    fn category(&self) -> Option<&str> { None }
    fn input_schema(&self) -> Option<Value> { None }
    fn default_labels(&self) -> HashMap<String, String> { HashMap::new() }
    fn source_code(&self) -> Option<&str> { None }
}

Example: a greeting workflow

use std::collections::HashMap;

use ironflow_engine::config::ShellConfig;
use ironflow_engine::context::WorkflowContext;
use ironflow_engine::handler::{HandlerFuture, WorkflowHandler, input_schema_for};
use schemars::JsonSchema;
use serde::Deserialize;
use serde_json::Value;

/// Input payload for the greeting workflow.
///
/// Derives [`JsonSchema`] so the dashboard can render a dynamic form.
#[derive(Deserialize, JsonSchema)]
struct GreetingInput {
    /// Person to greet.
    name: String,
    /// Greeting language (en, fr, es).
    #[serde(default = "default_language")]
    language: String,
    /// Number of times to repeat the greeting.
    #[serde(default = "default_repeat")]
    repeat: u32,
    /// Whether to output in uppercase.
    #[serde(default)]
    uppercase: bool,
}

fn default_language() -> String {
    "en".to_string()
}

fn default_repeat() -> u32 {
    1
}

pub struct Greeting;

impl WorkflowHandler for Greeting {
    fn name(&self) -> &str {
        "greeting"
    }

    fn category(&self) -> Option<&str> {
        Some("examples")
    }

    fn input_schema(&self) -> Option<Value> {
        Some(input_schema_for::<GreetingInput>())
    }

    fn default_labels(&self) -> HashMap<String, String> {
        HashMap::from([("project".to_string(), "ironflow".to_string())])
    }

    fn description(&self) -> &str {
        "A demo workflow that greets someone. \
         Shows how input_schema generates a dynamic form in the dashboard."
    }

    fn source_code(&self) -> Option<&str> {
        Some(include_str!("greeting.rs"))
    }

    fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
        Box::pin(async move {
            let input: GreetingInput = ctx.input().await?;

            let greeting = match input.language.as_str() {
                "fr" => format!("Bonjour, {} !", input.name),
                "es" => format!("Hola, {}!", input.name),
                _ => format!("Hello, {}!", input.name),
            };

            let mut message = (0..input.repeat)
                .map(|_| greeting.as_str())
                .collect::<Vec<_>>()
                .join("\n");

            if input.uppercase {
                message = message.to_uppercase();
            }

            ctx.shell("greet", ShellConfig::new(&format!("echo '{message}'")))
                .await?;

            Ok(())
        })
    }
}

Key points

• name() must be unique across all registered handlers. It identifies the workflow in the API and the database.
• execute() receives a WorkflowContext to create steps. Steps are persisted as they complete.
• input_schema() returns a JSON Schema derived from a #[derive(JsonSchema)] struct. The dashboard renders it as a dynamic form.
• source_code() optionally embeds the handler source for display in the dashboard.

Registration

Handlers are registered in the Engine before starting the server or worker:

let mut engine = Engine::new(store, provider);
engine.register(Box::new(Greeting))?;

See Writing a Workflow for a step-by-step guide.

Steps

A Step is an atomic unit of work within a Run. Each step is persisted in the database with its input, output, status, cost, duration, and token counts.

Step kinds

Shell steps

let output = ctx.shell("build", ShellConfig::new("cargo build")).await?;
if output.is_success() {
    // continue
}

HTTP steps

let response = ctx.http("fetch-data", HttpConfig::get("https://api.example.com/data")).await?;

Agent steps

let result = ctx.agent("analyze", AgentStepConfig::new("Analyze this log file")).await?;

Step status lifecycle

Steps follow this state machine:

stateDiagram-v2
    [*] --> Pending
    Pending --> Running
    Running --> Completed
    Running --> Failed
    Pending --> Skipped

Every step transition is recorded. Failed steps report their error in the step output.

Operations

An Operation is the extensibility mechanism for custom step types. Operations let you integrate external services (GitLab, Slack, any HTTP API) as tracked steps.

The trait

use std::future::Future;
use std::pin::Pin;

use ironflow_engine::error::EngineError;
use serde_json::Value;

pub trait Operation: Send + Sync {
    fn kind(&self) -> &str;
    fn execute(&self) -> Pin<Box<dyn Future<Output = Result<Value, EngineError>> + Send + '_>>;
    fn input(&self) -> Option<Value> { None }
}

• kind() returns a short identifier (e.g. "slack", "gitlab") stored in the database
• execute() runs the operation and returns JSON output
• input() optionally returns structured input for observability

Using an operation in a workflow

Operations are invoked via ctx.operation(), which takes a step name and a reference to the operation:

use ironflow_engine::context::WorkflowContext;

let slack = SlackNotify::new(&webhook_url);
ctx.operation("notify-team", &slack).await?;

Implementing an operation

use std::future::Future;
use std::pin::Pin;

use ironflow_engine::error::EngineError;
use ironflow_engine::operation::Operation;
use serde_json::{Value, json};

pub struct SlackNotify {
    webhook_url: String,
    message: String,
}

impl Operation for SlackNotify {
    fn kind(&self) -> &str {
        "slack-notify"
    }

    fn input(&self) -> Option<Value> {
        Some(json!({ "message": self.message }))
    }

    fn execute(&self) -> Pin<Box<dyn Future<Output = Result<Value, EngineError>> + Send + '_>> {
        Box::pin(async move {
            // Send to Slack webhook using self.webhook_url
            Ok(json!({ "ok": true }))
        })
    }
}

Built-in vs custom

Built-in step types (Shell, Http, Agent, Approval) have dedicated methods on WorkflowContext. Operations are for everything else – they give you a typed extension point without modifying the engine.

See Writing an Operation for a step-by-step guide.

Engine & Worker

Engine

The Engine is the in-memory registry that maps workflow names to handlers and orchestrates run execution. It holds references to the Store (persistence), the Provider (agent backends), and the event publisher.

let mut engine = Engine::new(store, provider);
engine.register(Box::new(MyWorkflow))?;

The Engine is used by both the API server (for metadata and describe endpoints) and the Worker (for execution).

Worker

A Worker is a background process that:

1. Polls the API for pending runs
2. Acquires a lease on a run
3. Executes the workflow handler via the Engine
4. Refreshes the lease periodically during execution
5. Reports the result back to the API

let worker = WorkerBuilder::new(&api_url, &worker_token)
    .provider(Arc::new(ClaudeCodeProvider::new()))
    .concurrency(2)
    .poll_interval(Duration::from_secs(2))
    .register(Box::new(MyWorkflow))
    .build()?;

worker.run().await?;

Lease & Reaper

Workers hold a time-limited lease on each run they execute. If a worker crashes or is evicted, the lease expires and the Reaper (a background task in the API server) detects the orphaned run and requeues it.

Scaling

Workers are stateless. Add more workers to increase throughput. Each worker polls independently – no coordination is needed beyond the API server.

Approval Gates

An approval gate pauses a workflow run until a human approves or rejects it. This enables human-in-the-loop workflows like deploy pipelines where production deploys require sign-off.

How it works

1. The handler calls ctx.approval() with a prompt message
2. The run transitions to AwaitingApproval
3. The worker releases the run and moves on to other work
4. A human calls POST /api/v1/runs/:id/approve or POST /api/v1/runs/:id/reject
5. On approval, the run is requeued. A worker picks it up, replays completed steps from cache, skips the approved gate, and continues execution
6. On rejection, the run transitions to Failed

Example

//! Deploy workflow with human approval gate before production.

use ironflow_engine::config::{ApprovalConfig, ShellConfig};
use ironflow_engine::context::WorkflowContext;
use ironflow_engine::handler::{HandlerFuture, WorkflowHandler};

/// Deploy pipeline that requires human approval before shipping to production.
///
/// 1. **build** -- compile the project
/// 2. **test** -- run the test suite
/// 3. **deploy-staging** -- deploy to staging environment
/// 4. **approval gate** -- pause and wait for human approval
/// 5. **deploy-production** -- resumes after approval via step replay
///
/// After approval (POST /api/v1/runs/:id/approve), the engine
/// re-executes the handler: completed steps return cached output,
/// the approved gate is skipped, and execution continues with
/// deploy-production.
///
/// If the approval is rejected, the run transitions to `Failed`.
/// If cancelled, the run transitions to `Cancelled`.
pub struct DeployApproval;

impl WorkflowHandler for DeployApproval {
    fn name(&self) -> &str {
        "deploy-approval"
    }

    fn description(&self) -> &str {
        "Deploy pipeline with human approval gate before production. \
         Demonstrates ctx.approval() for human-in-the-loop workflows."
    }

    fn source_code(&self) -> Option<&str> {
        Some(include_str!("deploy_approval.rs"))
    }

    fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
        Box::pin(async move {
            // Step 1: Build
            ctx.shell(
                "build",
                ShellConfig::new("echo 'Compiling...' && sleep 0.2 && echo 'Build OK'"),
            )
            .await?;

            // Step 2: Test
            ctx.shell(
                "test",
                ShellConfig::new("echo 'Running tests...' && sleep 0.3 && echo '87 tests passed'"),
            )
            .await?;

            // Step 3: Deploy to staging
            ctx.shell(
                "deploy-staging",
                ShellConfig::new(
                    "echo 'Deploying to staging...' && sleep 0.2 && echo 'Staging live'",
                ),
            )
            .await?;

            // Step 4: Human approval gate
            // The run pauses here and transitions to AwaitingApproval.
            // A human must call POST /api/v1/runs/:id/approve to continue,
            // or POST /api/v1/runs/:id/reject to fail the run.
            ctx.approval(
                "prod-approval",
                ApprovalConfig::new("Staging looks good. Deploy to production?")
                    .with_timeout_seconds(3600),
            )
            .await?;

            // Step 5: Deploy to production (only reached after approval)
            ctx.shell(
                "deploy-production",
                ShellConfig::new(
                    "echo 'Deploying to production...' && sleep 0.3 && echo 'Production live'",
                ),
            )
            .await?;

            Ok(())
        })
    }
}

Configuration

ApprovalConfig::new("Deploy to production?")
    .with_timeout_seconds(3600)  // Auto-reject after 1 hour

The timeout is optional. Without it, the run waits indefinitely.

Step replay

After an approval, the engine re-executes the handler from the beginning. Completed steps return their cached output immediately – they do not re-run. The approved gate is skipped, and execution resumes with the next step.

Writing a Workflow

This guide walks through creating a workflow handler from scratch.

1. Define your input

If your workflow accepts input, define a struct with Deserialize and JsonSchema:

use schemars::JsonSchema;
use serde::Deserialize;

#[derive(Deserialize, JsonSchema)]
struct DeployInput {
    environment: String,
    version: String,
}

The JsonSchema derive lets the dashboard render a dynamic form for triggering the workflow.

2. Implement WorkflowHandler

use ironflow_engine::config::ShellConfig;
use ironflow_engine::context::WorkflowContext;
use ironflow_engine::handler::{HandlerFuture, WorkflowHandler, input_schema_for};
use serde_json::Value;

pub struct Deploy;

impl WorkflowHandler for Deploy {
    fn name(&self) -> &str {
        "deploy"
    }

    fn description(&self) -> &str {
        "Deploy a version to an environment"
    }

    fn input_schema(&self) -> Option<Value> {
        Some(input_schema_for::<DeployInput>())
    }

    fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
        Box::pin(async move {
            let input: DeployInput = ctx.input().await?;

            ctx.shell(
                "build",
                ShellConfig::new(&format!("echo 'Building {}'", input.version)),
            ).await?;

            ctx.shell(
                "deploy",
                ShellConfig::new(&format!(
                    "echo 'Deploying {} to {}'",
                    input.version, input.environment
                )),
            ).await?;

            Ok(())
        })
    }
}

3. Register in your handlers list

pub fn handlers() -> Vec<Box<dyn WorkflowHandler>> {
    vec![
        Box::new(Deploy),
        // ... other handlers
    ]
}

Both the server and the worker must register the same handlers. The recommended pattern is a shared handlers() function in a library crate.

4. Complete example

The greeting workflow in the examples directory demonstrates all features:

use std::collections::HashMap;

use ironflow_engine::config::ShellConfig;
use ironflow_engine::context::WorkflowContext;
use ironflow_engine::handler::{HandlerFuture, WorkflowHandler, input_schema_for};
use schemars::JsonSchema;
use serde::Deserialize;
use serde_json::Value;

/// Input payload for the greeting workflow.
///
/// Derives [`JsonSchema`] so the dashboard can render a dynamic form.
#[derive(Deserialize, JsonSchema)]
struct GreetingInput {
    /// Person to greet.
    name: String,
    /// Greeting language (en, fr, es).
    #[serde(default = "default_language")]
    language: String,
    /// Number of times to repeat the greeting.
    #[serde(default = "default_repeat")]
    repeat: u32,
    /// Whether to output in uppercase.
    #[serde(default)]
    uppercase: bool,
}

fn default_language() -> String {
    "en".to_string()
}

fn default_repeat() -> u32 {
    1
}

pub struct Greeting;

impl WorkflowHandler for Greeting {
    fn name(&self) -> &str {
        "greeting"
    }

    fn category(&self) -> Option<&str> {
        Some("examples")
    }

    fn input_schema(&self) -> Option<Value> {
        Some(input_schema_for::<GreetingInput>())
    }

    fn default_labels(&self) -> HashMap<String, String> {
        HashMap::from([("project".to_string(), "ironflow".to_string())])
    }

    fn description(&self) -> &str {
        "A demo workflow that greets someone. \
         Shows how input_schema generates a dynamic form in the dashboard."
    }

    fn source_code(&self) -> Option<&str> {
        Some(include_str!("greeting.rs"))
    }

    fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
        Box::pin(async move {
            let input: GreetingInput = ctx.input().await?;

            let greeting = match input.language.as_str() {
                "fr" => format!("Bonjour, {} !", input.name),
                "es" => format!("Hola, {}!", input.name),
                _ => format!("Hello, {}!", input.name),
            };

            let mut message = (0..input.repeat)
                .map(|_| greeting.as_str())
                .collect::<Vec<_>>()
                .join("\n");

            if input.uppercase {
                message = message.to_uppercase();
            }

            ctx.shell("greet", ShellConfig::new(&format!("echo '{message}'")))
                .await?;

            Ok(())
        })
    }
}

Writing an Operation

Operations let you extend Ironflow with custom step types for integrating external services.

1. Implement the Operation trait

use std::env;
use std::future::Future;
use std::pin::Pin;

use ironflow_engine::error::EngineError;
use ironflow_engine::operation::Operation;
use serde_json::{Value, json};

pub struct SlackNotify {
    webhook_url: String,
    message: String,
}

impl SlackNotify {
    pub fn new(message: &str) -> Self {
        let webhook_url = env::var("SLACK_WEBHOOK_URL")
            .expect("SLACK_WEBHOOK_URL env var required");
        Self {
            webhook_url,
            message: message.to_string(),
        }
    }
}

impl Operation for SlackNotify {
    fn kind(&self) -> &str {
        "slack-notify"
    }

    fn input(&self) -> Option<Value> {
        Some(json!({ "message": self.message }))
    }

    fn execute(&self) -> Pin<Box<dyn Future<Output = Result<Value, EngineError>> + Send + '_>> {
        Box::pin(async move {
            let client = reqwest::Client::new();
            let resp = client
                .post(&self.webhook_url)
                .json(&json!({ "text": self.message }))
                .send()
                .await
                .map_err(|e| EngineError::OperationFailed {
                    kind: "slack-notify".to_string(),
                    message: e.to_string(),
                })?;

            Ok(json!({ "status": resp.status().as_u16() }))
        })
    }
}

2. Use it in a workflow

let notifier = SlackNotify::new("Deploy complete!");
ctx.operation("notify-team", &notifier).await?;

The step is tracked in the database like any other step, with its input, output, and status.

Parallel Execution

Ironflow supports running multiple steps in parallel within a workflow.

Using ctx.parallel()

Pass a list of step configurations to ctx.parallel(). All steps run concurrently and the method returns when all complete:

//! Demo workflow showcasing parallel execution and conditional branching.

use ironflow_engine::config::{ShellConfig, StepConfig};
use ironflow_engine::context::WorkflowContext;
use ironflow_engine::handler::{HandlerFuture, WorkflowHandler};

/// Simulated CI pipeline that demonstrates DAG features:
///
/// 1. **build** (sequential)
/// 2. **test-unit + test-integration + lint** (parallel)
/// 3. **deploy** or **notify-failure** (conditional branch on test results)
pub struct CiPipeline;

impl WorkflowHandler for CiPipeline {
    fn name(&self) -> &str {
        "ci-pipeline"
    }

    fn description(&self) -> &str {
        "Simulated CI pipeline with parallel tests and conditional deploy. \
         Demonstrates ctx.parallel() and native Rust if/else branching."
    }

    fn source_code(&self) -> Option<&str> {
        Some(include_str!("ci_pipeline.rs"))
    }

    fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
        Box::pin(async move {
            // Step 1: Build
            let build = ctx
                .shell(
                    "build",
                    ShellConfig::new(
                        "echo 'Compiling project...' && sleep 0.2 && echo 'Build successful'",
                    ),
                )
                .await?;

            if !build.is_success() {
                ctx.shell(
                    "notify-build-failure",
                    ShellConfig::new("echo 'BUILD FAILED - notifying team'"),
                )
                .await?;
                return Ok(());
            }

            // Step 2: Parallel tests + lint
            let results = ctx
                .parallel(
                    vec![
                        (
                            "test-unit",
                            StepConfig::Shell(ShellConfig::new(
                                "echo 'Running unit tests...' && sleep 0.3 && echo '42 tests passed'",
                            )),
                        ),
                        (
                            "test-integration",
                            StepConfig::Shell(ShellConfig::new(
                                "echo 'Running integration tests...' && sleep 0.5 && echo '12 tests passed'",
                            )),
                        ),
                        (
                            "lint",
                            StepConfig::Shell(ShellConfig::new(
                                "echo 'Running linter...' && sleep 0.1 && echo 'No warnings'",
                            )),
                        ),
                    ],
                    true,
                )
                .await?;

            // Step 3: Conditional deploy
            let all_passed = results.iter().all(|r| r.output.is_success());

            if all_passed {
                ctx.shell(
                    "deploy",
                    ShellConfig::new("echo 'Deploying to production...' && sleep 0.2 && echo 'Deployed successfully'"),
                )
                .await?;
            } else {
                ctx.shell(
                    "notify-test-failure",
                    ShellConfig::new("echo 'TESTS FAILED - deployment skipped'"),
                )
                .await?;
            }

            Ok(())
        })
    }
}

How it works

• All steps in a parallel() call start at the same time
• The method returns a Vec<ParallelResult> with outputs in the same order as the input
• If fail_fast is true (the second argument), the remaining steps are cancelled when one fails
• If fail_fast is false, all steps run to completion regardless of individual failures

Conditional branching

Since workflows are Rust code, conditional logic is just if/else:

let results = ctx.parallel(steps, true).await?;
let all_passed = results.iter().all(|r| r.output.is_success());

if all_passed {
    ctx.shell("deploy", ShellConfig::new("echo 'Deploying'")).await?;
} else {
    ctx.shell("notify", ShellConfig::new("echo 'Tests failed'")).await?;
}

No special DSL for branching – Rust control flow works directly.

Transports

Transports control where agent steps execute. By default, agents run on the local machine via ClaudeCodeProvider. Ironflow ships additional transports for running agents in isolated environments.

Available transports

Docker transport

Executes agent commands inside a running Docker container via docker exec:

//! Docker transport example.
//!
//! Executes `claude` inside a running Docker container via `docker exec`.
//! The container must already be running and have the `claude` binary installed.
//!
//! # Usage
//!
//! ```sh
//! DOCKER_CONTAINER=claude-worker cargo run --bin docker-transport
//! ```

use std::env;

use ironflow_core::prelude::*;
use ironflow_core::providers::claude::DockerProvider;

#[tokio::main]
async fn main() -> Result<(), OperationError> {
    let container = env::var("DOCKER_CONTAINER").expect("DOCKER_CONTAINER env var required");

    let provider = DockerProvider::new(&container).working_dir("/workspace");

    let result = Agent::new()
        .prompt("What is 2 + 2?")
        .max_budget_usd(0.10)
        .run(&provider)
        .await?;

    println!("Response: {}", result.text());
    println!("Model: {}", result.model().unwrap_or("unknown"));
    println!("Duration: {}ms", result.duration_ms());

    Ok(())
}

SSH transport

Connects to a remote host via SSH:

//! SSH transport example.
//!
//! Connects to a remote host via SSH and runs `claude` there.
//! Requires the `claude` binary to be installed on the remote host.
//!
//! # Usage
//!
//! ```sh
//! SSH_HOST=build-server SSH_USER=deploy SSH_PASSWORD=secret cargo run --bin ssh-transport
//! ```

use std::env;

use ironflow_core::prelude::*;
use ironflow_core::providers::claude::SshProvider;
use ironflow_core::providers::claude::ssh::HostKeyPolicy;

#[tokio::main]
async fn main() -> Result<(), OperationError> {
    let host = env::var("SSH_HOST").expect("SSH_HOST env var required");
    let user = env::var("SSH_USER").expect("SSH_USER env var required");
    let password = env::var("SSH_PASSWORD").expect("SSH_PASSWORD env var required");

    // For production, use HostKeyPolicy::Fingerprint or HostKeyPolicy::KnownHostsFile
    let provider = SshProvider::new(&host, &user)
        .password(&password)
        .host_key_policy(HostKeyPolicy::AcceptAll);

    let result = Agent::new()
        .prompt("What is 2 + 2?")
        .max_budget_usd(0.10)
        .run(&provider)
        .await?;

    println!("Response: {}", result.text());
    println!("Model: {}", result.model().unwrap_or("unknown"));
    println!("Duration: {}ms", result.duration_ms());

    Ok(())
}

Kubernetes transport

Two modes are available:

• Ephemeral – creates a pod for each agent call, deletes it when done
• Persistent – reuses a long-lived pod for multiple calls

See the examples/transports/ directory for complete Kubernetes examples.

Choosing a transport

• Development: use ClaudeCodeProvider (local). No setup needed.
• CI/CD pipelines: Docker or Kubernetes for isolation.
• Remote build servers: SSH for machines you already manage.
• Multi-tenant production: Kubernetes ephemeral pods for strong isolation between tenants.

Architecture Overview

Ironflow follows a client-server architecture with background workers for execution.

Components

graph TD
    Dashboard[Web Dashboard] --> API[API Server]
    CLI[CLI] --> SDK[Rust SDK]
    MCP[MCP Server] --> SDK
    SDK --> API
    API --> Store[(Database)]
    API --> Artifacts[(Blob Store)]
    Worker1[Worker 1] --> API
    Worker2[Worker 2] --> API
    Worker1 --> Provider[Agent Provider]
    Worker2 --> Provider

Crate map

Request flow

1. A client (dashboard, CLI, SDK, or webhook) sends a request to the API
2. The API validates authentication, creates a Run in the Store, and returns it
3. A Worker polls the API, acquires a lease on the Run, and executes the handler
4. The handler calls steps (ctx.shell(), ctx.agent(), etc.), each persisted as they complete
5. Events are published via SSE for real-time updates
6. On completion or failure, the Worker reports the result back to the API

Data flow

Runs and steps are stored in PostgreSQL (or in-memory for development). Artifacts (files produced by steps) are stored in a separate blob store (local filesystem by default). The two are linked by artifact metadata on each step.