Workflows

Orchestrate complex business logic with distributed workflows.

Understanding Workflows

Workflows are sequences of steps that execute business logic. They run asynchronously, can recover from failures, and support parent-child relationships for complex orchestration.

When the automation can be expressed as data instead of worker code (plugin calls, HTTP requests, timers, approval gates), Pipelines describe the same kind of multi-step logic as a declarative JSON DAG with a visual builder. Pipelines run on the workflow engine (every pipeline run is a workflow run named pipeline.<name>), spawn and monitor worker workflows as child runs, and reuse signals and hooks for approvals.

Creating Workflows

Basic Workflow

from vski import step, workflow

@step("fetch-user")
async def fetch_user(user_id):
    # Step 1: Fetch user data
    return await client.collection("_users").get_one(user_id)

@step("generate-email")
async def generate_email(user):
    # Step 2: Generate welcome email
    return generate_welcome_email(user)

@step("send-email")
async def send_email(email):
    # Step 3: Send email
    return await email_service.send(email)

@workflow("send-welcome-email").run
async def send_welcome_email(ctx, payload):
    user_id = payload["userId"]

    user = await fetch_user(user_id)
    email = await generate_email(user)
    await send_email(email)

    return {"success": True, "userId": user_id}

# The decorators register the steps and the workflow automatically:
# there is no separate register call.

Starting a Worker

import asyncio
from vski import WorkflowWorker
from vski.workflows.registry import all_names

worker = WorkflowWorker(client)

# Start a worker for a specific workflow (in the background)
task = asyncio.create_task(worker.start("send-welcome-email"))
await client.wait_for_workflow_ready()

# Start one worker for several workflows at once
task = asyncio.create_task(worker.start(["send-welcome-email", "process-order"]))

# Start a worker for all registered workflows
task = asyncio.create_task(worker.start(all_names()))

# Stop the worker (and its background task) when done
await worker.stop()
await task

Triggering Workflows

# Trigger a workflow with input (a list of positional arguments)
run = await client.workflow.trigger("send-welcome-email", [
    {"userId": "user-123"},
])

print(run["runId"])   # Workflow run ID
print(run["status"])  # "pending", "running", "completed", etc.

Workflow Steps

Basic Step

@step("step-name")
async def step_name():
    # Your logic here
    return {"data": "result"}

result = await step_name()

Steps with Parameters

@step("process-data")
async def process_data(data):
    # Process data
    return {"processed": True}

result = await process_data(data)

Step Error Handling

@step("risky-operation")
async def risky_operation():
    # Might raise an error
    return await risky_operation_call()

try:
    result = await risky_operation()
except Exception as error:
    print("Step failed:", error)
    # Workflow will be marked as failed

Workflow Configuration

Circuit Breaker

Prevent runaway workflows:

@workflow("data-processing", {
    "maxEvents": 100,          # Maximum events
    "executionTimeout": 60000,  # 60 second timeout
}).run
async def data_processing(ctx, payload):
    # Workflow code
    ...

Custom Run ID

run = await client.workflow.trigger("my-workflow", [data], {
    "runId": "custom-run-id-123",
})

Idempotency

Prevent duplicate runs:

run = await client.workflow.trigger("my-workflow", [data], {
    "idempotencyKey": "unique-key-123",
})

Parent-Child Workflows

Creating Child Workflows

@workflow("parent-workflow").run
async def parent_workflow(ctx, payload):
    # Trigger a child workflow linked to this run
    child_run = await client.workflow.trigger(
        "child-workflow", [payload], {"parentRunId": ctx.run_id}
    )

    # Wait for child to complete
    while True:
        child = await client.workflow.get_run(child_run["runId"])
        if child["status"] in ("completed", "failed", "cancelled"):
            break
        await asyncio.sleep(1)

    return {"childRunId": child_run["runId"]}

Accessing Parent Context

@workflow("child-workflow").run
async def child_workflow(ctx, payload):
    # Access the parent run ID via the run record
    run = await client.workflow.get_run(ctx.run_id)
    print("Parent run ID:", run.get("parentRunId"))

    # Child workflow logic

Parallel Execution

Fan-Out Pattern

Execute multiple steps in parallel using ctx.parallel():

async def _process_item(item):
    return process_item(item)

def process_step(item):
    """Wrap one item as a durable step with a unique step ID."""
    return step(f"process-{item['id']}", lambda: _process_item(item))

@workflow("process-items").run
async def process_items(ctx, payload):
    items = payload["items"]

    # Process items in parallel
    results = await ctx.parallel([process_step(item) for item in items])

    return {"results": results}

Fan-In Pattern

Process in parallel and then aggregate results:

@step("aggregate")
async def aggregate(results):
    return aggregate_results(results)

@workflow("aggregate-results").run
async def aggregate_results_workflow(ctx, payload):
    items = payload["items"]

    # Process in parallel
    results = await ctx.parallel([process_step(item) for item in items])

    # Aggregate results
    summary = await aggregate(results)

    return summary

Signals

Signals allow workflows to wait for external input, enabling approval workflows, human-in-the-loop patterns, and dynamic branching.

Waiting for Signals

Pause workflow execution until a signal is received:

@step("request-expense")
async def request_expense(amount):
    return {"status": "pending", "amount": amount}

@step("process-payment")
async def process_payment():
    return {"paid": True}

@step("notify-rejection")
async def notify_rejection():
    return {"notified": True}

@workflow("approval-workflow").run
async def approval_workflow(ctx, payload):
    amount = payload["amount"]

    # Request approval
    await request_expense(amount)

    # Wait for the approval signal:
    # {"approved": bool, "reason": str (optional)}
    approval = await ctx.wait_for_signal("manager-approval")

    if approval["approved"]:
        await process_payment()
        return {"status": "approved"}

    await notify_rejection()
    return {"status": "rejected", "reason": approval.get("reason")}

Sending Signals

External systems can send signals to resume workflows:

# From an external service
await client.workflow.send_signal(run_id, "manager-approval", {
    "approved": True,
    "comment": "Approved for processing",
})

Sleeping and Delays

Use ctx.sleep() to pause workflow execution for a specified duration:

async def _process(item):
    return process_item(item)

@workflow("rate-limited-workflow").run
async def rate_limited_workflow(ctx, payload):
    items = payload["items"]

    for i, item in enumerate(items):
        # A unique step ID per pass keeps each iteration durable
        await step(f"process-item-{i}", _process)(item)

        # Wait 1 second between items to respect rate limits
        await ctx.sleep("1s")

    return {"processed": len(items)}

Duration formats:

• Milliseconds: 5000 or "5000ms"
• Seconds: "5s" or 10
• Minutes: "10m"
• Hours: "1h"

Workflow Hooks

Creating Approval Hooks

@step("publish-document")
async def publish_document_step(document_id):
    return publish_document(document_id)

@workflow("document-approval").run
async def document_approval(ctx, payload):
    document_id = payload["documentId"]

    # Create an approval hook: the token is the secret an external
    # caller presents to execute it
    hook = await client.workflow.hooks.create({
        "runId": ctx.run_id,
        "signalName": "document-approval",
        "token": "approve-" + ctx.run_id,
        "metadata": {"documentId": document_id},
    })

    # Wait for approval: executing the hook delivers its payload to the
    # run as the "document-approval" signal
    approval = await ctx.wait_for_signal("document-approval")

    if not approval["approved"]:
        raise RuntimeError("Document rejected")

    # Continue workflow
    await publish_document_step(document_id)

    return {"published": True}

Executing Hooks

# External endpoint receives hook execution
result = await client.workflow.hooks.execute("hook-token", {
    "approved": True,
    "comment": "Approved for publication",
})

Executing a hook emits the hook’s signal to its run, with the hook’s metadata merged into the payload.

Workflow Events

Emitting Events

import time

@step("process-data")
async def process_data_step():
    # Process data
    ...

@workflow("track-events").run
async def track_events(ctx, payload):
    await client.workflow.create_event(
        ctx.run_id,
        event_type="step-started",
        payload={"step": "process-data", "timestamp": int(time.time() * 1000)},
    )

    await process_data_step()

    await client.workflow.create_event(
        ctx.run_id,
        event_type="step-completed",
        payload={"step": "process-data", "timestamp": int(time.time() * 1000)},
    )

    return {"success": True}

Querying Events

events = await client.workflow.list_events("run-id")

for event in events:
    print(f"{event['eventType']}: {event['payload']}")

Monitoring Workflows

Get Workflow Status

run = await client.workflow.get_run("run-id")

print(run["status"])      # pending, running, completed, failed, cancelled
print(run["startedAt"])
print(run["completedAt"])
print(run["error"])       # Error details if failed

List Workflow Runs

runs = await client.workflow.list_runs(params={
    "workflowName": "send-welcome-email",
    "status": "completed",
    "limit": 20,
})

print(runs)

Get Child Workflows

children = await client.workflow.get_child_workflows("parent-run-id")

Get Parent Workflow

parent = await client.workflow.get_parent_workflow("child-run-id")

Canceling Workflows

await client.workflow.cancel_run("run-id", reason="No longer needed")

Practical Examples

Data Processing Pipeline

@step("fetch-user")
async def fetch_user(user_id):
    return await client.collection("_users").get_one(user_id)

@step("enrich-data")
async def enrich_data(user):
    return await external_service.enrich(user)

@step("save-data")
async def save_data(user_id, enriched):
    return await client.collection("user_profiles").update(user_id, enriched)

@step("send-notification")
async def send_notification(email):
    return await notification_service.send(email, "Profile updated")

@workflow("process-user-data").run
async def process_user_data(ctx, payload):
    user_id = payload["userId"]

    # Step 1: Fetch user
    user = await fetch_user(user_id)

    # Step 2: Enrich data
    enriched = await enrich_data(user)

    # Step 3: Save enriched data
    await save_data(user_id, enriched)

    # Step 4: Send notification
    await send_notification(user["email"])

    return {"success": True}

Order Processing

@step("validate-order")
async def validate_order(order_id):
    return await validate_order_service(order_id)

@step("process-payment")
async def process_payment(order):
    return await payment_service.charge(order)

@step("update-inventory")
async def update_inventory(items):
    return await inventory_service.reserve(items)

@step("ship-order")
async def ship_order(order):
    return await shipping_service.create(order)

@step("send-confirmation")
async def send_confirmation(email):
    return await email_service.send(email, "Order confirmed")

@workflow("process-order").run
async def process_order(ctx, payload):
    order_id = payload["orderId"]

    # Step 1: Validate order
    order = await validate_order(order_id)

    # Step 2: Process payment
    payment = await process_payment(order)

    # Step 3: Update inventory
    await update_inventory(order["items"])

    # Step 4: Ship order
    await ship_order(order)

    # Step 5: Send confirmation
    await send_confirmation(order["email"])

    return {"shipped": True}

Batch Processing

@step("get-batch")
async def get_batch(batch_id):
    return await client.collection("batch_items").get_list(
        1, 1000, filter=f"batchId = '{batch_id}' && status = 'pending'"
    )

async def _process(item):
    result = await process_item(item)
    await client.collection("batch_items").update(
        item["id"], {"status": "completed", "result": result}
    )
    return result

def process_step(item):
    return step(f"process-{item['id']}", lambda: _process(item))

@step("aggregate")
async def aggregate(items, results):
    return {
        "total": len(items),
        "succeeded": sum(1 for r in results if r.get("success")),
        "failed": sum(1 for r in results if not r.get("success")),
    }

@step("update-batch")
async def update_batch(batch_id, summary):
    return await client.collection("batches").update(
        batch_id, {"status": "completed", "summary": summary}
    )

@workflow("batch-process").run
async def batch_process(ctx, payload):
    batch_id = payload["batchId"]

    # Step 1: Get batch items
    page = await get_batch(batch_id)
    items = page["items"]

    # Step 2: Process in parallel (ctx.parallel keeps each step durable)
    results = await ctx.parallel([process_step(item) for item in items])

    # Step 3: Aggregate results
    summary = await aggregate(items, results)

    # Step 4: Update batch
    await update_batch(batch_id, summary)

    return summary

Scheduled Reports

from datetime import datetime, timezone

@step("fetch-data")
async def fetch_data(report_type):
    return await fetch_report_data(report_type)

@step("generate")
async def generate(data, report_type):
    return await report_service.generate(data, report_type)

@step("save")
async def save_report(report, report_type):
    return await client.collection("reports").create({
        "type": report_type,
        "url": report["url"],
        "generatedAt": datetime.now(timezone.utc).isoformat(),
    })

@step("notify")
async def notify(report_id):
    return await notification_service.notify(report_id)

@workflow("generate-report").run
async def generate_report(ctx, payload):
    report_type = payload["type"]

    # Step 1: Fetch data
    data = await fetch_data(report_type)

    # Step 2: Generate report
    report = await generate(data, report_type)

    # Step 3: Save report
    saved = await save_report(report, report_type)

    # Step 4: Notify recipients
    await notify(saved["id"])

    return {"reportId": saved["id"]}

Error Handling

Retry Failed Steps

async def _potentially_failing():
    return await potentially_failing_operation()

# retries=3: retry up to 3 times (a 1s-per-attempt backoff is built in)
retryable_step = step("retryable-step", _potentially_failing, {"retries": 3})

result = await retryable_step()

Step Error Callbacks

There is no onError step option; handle failures inside the step function and re-raise to record them on the run:

async def _process_data():
    try:
        return await process_data()
    except Exception as error:
        # Log error
        print("Step failed:", error)
        # Send alert
        await alert_service.send(error)
        # Re-raise so retries still apply and the failure is recorded
        raise

await step("process-data", _process_data)()

Workflow-Level Error Handling

import time

@step("risky-operation")
async def risky_operation():
    return await risky_operation_call()

@workflow("error-handling").run
async def error_handling(ctx, payload):
    try:
        await risky_operation()
    except Exception as error:
        # Log error as a run event
        await client.workflow.create_event(
            ctx.run_id,
            event_type="error",
            payload={"error": str(error), "timestamp": int(time.time() * 1000)},
        )

        # Continue or re-raise
        raise

Best Practices

1. Idempotent steps - Make steps idempotent to handle retries
2. Descriptive step names - Use clear names for debugging
3. Proper timeouts - Set reasonable execution timeouts
4. Error handling - Handle errors gracefully
5. Logging - Emit events for tracking
6. Testing - Test workflows thoroughly
7. Monitoring - Monitor workflow execution
8. Clean up - Clean up old workflow runs
9. Use children - Use child workflows for complex processes
10. Parallelize - Use parallel execution when possible

API Endpoints

Method	Endpoint	Description
POST	/api/workflows/runs	Trigger workflow
GET	/api/workflows/runs/:id	Get workflow run
GET	/api/workflows/runs	List workflow runs
PATCH	/api/workflows/runs/:id	Update workflow run
DELETE	/api/workflows/runs/:id	Cancel workflow
GET	/api/workflows/runs/:id/children	Get child workflows
GET	/api/workflows/runs/:id/parent	Get parent workflow
POST	/api/workflows/events	Emit event
GET	/api/workflows/runs/:id/events	List events
POST	/api/workflows/hooks	Create hook
POST	/api/workflows/hooks/:token	Execute hook