DBOS | Build Reliable Backends 10x Faster, Scale to Millions with 1 Click

Event-driven processing

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

@DBOS.kafka_consumer(config,["alerts-topic"])
@DBOS.workflow()
def process_kafka_alerts(msg: KafkaMessage):
  alerts = msg.value.decode()
  for alert in alerts:
    respond_to_alert(alert)

Create reliable workflows

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

See an example

@DBOS.workflow()
def checkout_workflow(items):
order = create_order()
reserve_inventory(order, items)
payment_status = process_payment(order, items)
if payment_status == 'paid':
fulfill_order(order)
else:
undo_reserve_inventory(order, items)
cancel_order(order)

Cron jobs made easy

Schedule workflows to run at specific times.
Host them completely serverlessly.

See an example

@DBOS.scheduled("0 * * * *") # Run once an hour
@DBOS.workflow()
def run_hourly(scheduled_time: datetime, actual_time: datetime):
   results = search_hackernews("serverless")
   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.

See an example

queue = Queue("indexing_queue")

@DBOS.workflow()
def indexing_workflow(urls: List[HttpUrl]):
  handles: List[WorkflowHandle] = []
  for url in urls:
     handle = queue.enqueue(index_document, url)
     handles.append(handle)
indexed_pages = 0
for handle in handles:
     indexed_pages += handle.get_result()
logger.info(f"Indexed {len(urls)} documents totaling {indexed_pages} pages")

Reliable AI agents

Enhance your AI workflows with DBOS, build reliable AI agents with automatic retries and no limit on how long they can run for.

See an example

@DBOS.step(retries_allowed=True, max_attempts=2)
def process_refund(item, reason):
    return f"Processed refund for item {item}, because {reason}"

@DBOS.step(retries_allowed=True, max_attempts=3)
def apply_discount(amount):
    return f"Applied discount of {amount}%"

refunds_agent = Agent(
    name="Refunds Agent",
    instructions="Help the user with a refund. If the reason is that it was too expensive, offer the user a refund code.",
    functions=[process_refund, apply_discount],
)

Webhooks & notifications

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.

See an example

@slackapp.message()
def handle_message(request: BoltRequest) -> None:
  event_id = request.body["event_id"]
  with SetWorkflowID(event_id):
    DBOS.start_workflow(message_workflow,     request.body["event"])