DBOS - Durable Workflow Orchestration

Durable workflows, with endless possibilities

Orchestrate durable workflows

Write your business logic in normal code, with branches, loops, subtasks, and retries. DBOS makes it resilient to any failure.

Start building

See an example

# Define a durable checkout workflow
@DBOS.workflow()
def checkout_workflow(items):
    # Step 1: Create the order
    order = create_order()

    # Step 2: Reserve inventory for the items
    reserve_inventory(order, items)

    # Step 3: Process the payment
    payment_status = process_payment(order, items)

    # Step 4: If paid, fulfill the order
    if payment_status == 'paid':
        fulfill_order(order)
    else:
        # If payment fails, release inventory and cancel the order
        undo_reserve_inventory(order, items)
        cancel_order(order)

Process your events exactly once

Consume events exactly-once, no need to worry about timeouts or offsets.

Start building

See an example

# Listen for new messages on the "alerts-topic" Kafka topic
@DBOS.kafka_consumer(config, ["alerts-topic"])

# Define a durable workflow triggered by each Kafka message
@DBOS.workflow()
def process_kafka_alerts(msg: KafkaMessage):
    # Decode the Kafka message payload
    alerts = msg.value.decode()
    
    # Loop through each alert and respond accordingly
    for alert in alerts:
        respond_to_alert(alert)

Cron jobs made easy

Schedule your durable workflows to run exactly once per time interval. Record a stock's price once a minute, migrate some data once every hour, or send emails to inactive users once a week.

Start building

See an example

# Schedule this workflow to run every hour
@DBOS.scheduled("0 * * * *")

# Define a durable workflow that takes scheduled and actual time
@DBOS.workflow()
def run_hourly(scheduled_time: datetime, actual_time: datetime):
    # Search Hacker News for the keyword "serverless"
    results = search_hackernews("serverless")
    
    # Post each result (comment and URL) to Slack
    for comment, url in results:
        post_to_slack(comment, url)

Resilient data pipelines

Build data pipelines that are reliable and observable by default. DBOS durable queues guarantee all your tasks complete.

Start building

See an example

# Create a named queue for indexing tasks
queue = Queue("indexing_queue")

# Define a durable workflow that indexes a list of URLs
@DBOS.workflow()
def indexing_workflow(urls: List[HttpUrl]):
    handles: List[WorkflowHandle] = []

    # Enqueue a document indexing task for each URL
    for url in urls:
        handle = queue.enqueue(index_document, url)
        handles.append(handle)

    indexed_pages = 0

    # Wait for each indexing task to complete and tally results
    for handle in handles:
        indexed_pages += handle.get_result()

    # Log the total indexed pages
    logger.info(f"Indexed {len(urls)} documents totaling {indexed_pages} pages")

Build reliable AI agents

Use durable workflows to build reliable, fault-tolerant AI agents.

Start building

See an example

@DBOS.workflow()
def agentic_research_workflow(topic, max_iterations):
    research_results = []

    for i in range(max_iterations):
        # Run a query based on the current topic
        research_result = research_query(topic)
        research_results.append(research_result)

        # Stop if the results suggest no further research is needed
        if not should_continue(research_results):
            break

        # Refine the topic for the next research iteration
        topic = generate_next_topic(topic, research_results)

    # Combine all collected results into a final report
    return synthesize_research_report(research_results)

@DBOS.step()
def research_query(topic):
    ...

Handle notifications & webhooks reliably

Effortlessly mix synchronous webhook code with asynchronous event processing. Reliably wait weeks or months for events, then use idempotency and durable execution to process them exactly once.

Start building

See an example

# Listen for incoming Slack messages using the Slack Bolt app
@slackapp.message()
def handle_message(request: BoltRequest) -> None:
    # Use the Slack event ID to set a unique workflow ID
    event_id = request.body["event_id"]
    
    with SetWorkflowID(event_id):
        # Start a durable workflow to process the message
        DBOS.start_workflow(message_workflow, request.body["event"])

Durable workflows

Make code reliable in minutes

Add a few annotations to your code to make it durable.
So if your application crashes or restarts, it automatically resumes your workflows from the last completed step.

Automatic failure recovery

Built-in observability

Durable execution guaranteed

Fetch data inputs

Retrieve raw inputs

Pending

Transform data

Enrich with external APIs

Pending

Store to database

Persist to durable store

Pending

Transform data

Enrich with external APIs

Failed

Fetch data inputs

Retrieve raw inputs

Success

Fetch data inputs

Retrieve raw inputs

Pending

Transform data

Enrich with external APIs

Pending

Store to database

Persist to durable store

Pending

Transform data

Enrich with external APIs

Failed

Transform data

Enrich with external APIs

DBOS Recovering

Fetch data inputs

Retrieve raw inputs

Success

Transform data

Enrich with external APIs

Success

Store to database

Persist to durable store

Success

>_ System Console

Live