DBOS | Durable Workflow Orchestration

Build reliable & observable AI agents

Use durable workflows to orchestrate fault-tolerant AI systems with built-in observability, human-in-the-loop control, and reproducible execution. Integrate natively with popular agent frameworks like Pydantic AI, LlamaIndex, and the OpenAI Agents SDK.

Start building

See an example

agent.py

from dbos_openai_agents import DBOSRunner

# Define tool function for the agent
@function_tool
@DBOS.step()
async def get_weather(city: str) -> str:
    """Get the weather for a city."""
    return f"Sunny in {city}"

agent = Agent(name="weather", tools=[get_weather])

# Define the main agent workflow
@DBOS.workflow()
async def run_agent(user_input: str) -> str:
    result = await DBOSRunner.run(agent,
user_input)
    return str(result.final_output)

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

checkout.py

@DBOS.workflow()
def checkout_workflow(items: Items):

    # First, create an order
    order = create_order()

    # Then, reserve inventory
    reserve_inventory(order, items)

    # Next, process payment
    payment_status = process_payment(order, items)

    if payment_status == 'paid':
        # If payment succeeds, fulfill the order
        fulfill_order(order)

    else:
        # If payment fails, release inventory
        # and cancel the order
        undo_reserve_inventory(order, items)
        cancel_order(order)

Scale reliably with durable queues

Use durable queues to simplify fault-tolerant orchestration of thousands of concurrent tasks. Control how many tasks can run concurrently or how often tasks can start.

Start building

See an example

queue.py

# Create a named queue for indexing tasks,
# It runs no more than 10 functions concurrently
queue = Queue("indexing_queue", concurrency=10)

@DBOS.workflow()
def indexing_workflow(urls):
    handles = []

    # Enqueue an index task for each URL
    for url in urls:
        handle = queue.enqueue(index_doc, url)
        handles.append(handle)

    # Wait for each task to complete
    # and get result
    return [h.get_result() for h in handles]

Monitor workflows and queues in real time

Get immediate visibility into workflow health and performance. Inspect workflow executions, track queue metrics, and manage workflows with cancel, resume, and fork.

All with full control over running systems.

Start building

See an example

Dynamic cron schedules made easy

Schedule durable workflows to run exactly once per interval. Create, update, or backfill schedules dynamically without managing external schedulers.

Record a stock's price once a minute, migrate customer data once every hour, or send emails to inactive users once a week.

Start building

See an example

create-schedule.py

@DBOS.workflow()
def customer_workflow(scheduled_time: datetime,
customer_id: str):
    # Code to sync customer data
    ...

def on_customer_registration(customer_id: str):
    # Create a schedule dynamically per user
    # to run the customer_workflow
    DBOS.create_schedule(
        schedule_name=f"cus-{customer_id}-sync",
        workflow_fn=customer_workflow,
        schedule="0 * * * *",  # Runs every hour
        context=customer_id,
    )

Run human-in-the-loop workflows

Run workflows that pause, wait for human input, and resume reliably, even across restarts and redeploys.

Durable sleep and notifications let workflows wait for days or weeks, or until an external signal arrives, before continuing.

Start building

See an example

human-in-the-loop.py

@DBOS.workflow()
def durable_agent(request):
    # Do some work...

    # Upon reaching the step that needs approval,
    # await an approval notification.
    approval = DBOS.recv()

    # If approved, continue execution.
    # Otherwise, raise an exception to stop execution
    if approval is None:
        raise Exception("Agent timed out awaiting")
    elif approval.response == "deny":
        raise Exception("Agent denied approval")
    else:
        print("Agent approved:", agent_status)

@app.post("/agents/{agent_id}/respond")
def respond_to_agent(agent_id, response):
    # Notify an agent it has been approved or denied
    DBOS.send(agent_id, response)

Agentic-Ready Features

Made for reliable agents & workflows

Visualize Workflows & Queues

Monitor and manage performance in real time.

Start for free

Integrations

First-party integrations with agent frameworks like OpenAI Agents, LlamaIndex, and Pydantic AI.

View all integrations

New

MCP Server

Monitor and debug your workflows directly from your preferred coding agent.

Read our quickstart

Workflow Management

Durability makes workflows and queues failure proof. DBOS Conductor gives you fine-grained control over workflow troubleshooting, execution, replay, forking, and alerting--across all of your AI applications.

Read our docs

Secure & Private

DBOS is SOC 2, GDPR and CCPA compliant. DBOS never sees your data and is fully private.

Compare plans

Run Anywhere

Run on-prem or in any cloud. Self-host DBOS Conductor or connect from any environment.

Read the docs

Custom Alerts

Define custom alerts to notify your team of workflow orchestration issues.

Read the docs

Testimonials

Loved by AI, Cloud & Data teams

Helping more devs, AI engineers, and SREs orchestrate fault-tolerant workflows.

"We've been impressed by how lightweight and flexible DBOS is, the speed at which their team ships, and the level of support offered. We, at Yutori, are happy customers and excited to scale with DBOS."

Abhishek Das

CEO & Co-founder

Durable AI Workflow Orchestration

Build reliable software effortlessly

AI agents

Durable execution

Durable queues

Task observability

Cron jobs

Human in the loop

Build reliable & observable AI agents

Orchestrate durable workflows

Scale reliably with durable queues

Monitor workflows and queues in real time

Dynamic cron schedules made easy

Run human-in-the-loop workflows

How it works

It's like putting a PhD. in your code

Embed DBOS

Annotate your code

Observe & manage

Agentic-Ready Features

Made for reliable agents & workflows

Visualize Workflows & Queues

Integrations

MCP Server

Workflow Management

Secure & Private

Run Anywhere

Custom Alerts

Testimonials

Loved by AI, Cloud & Data teams

Insights

Read the latest on workflow durability

Async Python is Secretly Deterministic

Running Durable Workflows from Postgres UDFs

Bridging Languages with Postgres and Workflows

Get started with DBOS