Build reliable & observable AI agents

from dbos_openai_agents import DBOSRunner

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

# Create an agentic workflow
@DBOS.workflow()
async def run_agent(user_input: str) -> str:
    agent = Agent(name="weather", tools=[get_weather])
    result = await DBOSRunner.run(agent, user_input)
    return str(result.final_output)

Orchestrate durable workflows

@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

# 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

Dynamic cron schedules made easy

@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

@DBOS.workflow()
def durable_agent(request):
    # Upon reaching a 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)

“In EDI, you can’t have things fall between the cracks. You need the ability to trace it end to end, and we don’t think we could do that without DBOS.”

CTO

"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."

CEO & Co-founder

“DBOS makes it dead simple for our engineering team to build reliable, low-latency, AI-powered user experiences."

CTO and Co-founder

"I love the design of DBOS. If you're a gamer, it's like having a “save point” in your programs. If a function fails, a new function can start, picking up at the last checkpoint."

CEO & Co-Founder

“DBOS is really simple to manage. It gives us durable orchestration and observability without a lot of coding and without having to run additional infrastructure.”

CEO and Co-founder

‍“We looked at Google Workflows, Temporal, Airflow and Celery but DBOS was the one framework that solved many pieces of the puzzle while giving us implementation flexibility.”

Engineer

"DBOS makes it impossible to screw up."

CTO and Co-founder

“In EDI, you can’t have things fall between the cracks. You need the ability to trace it end to end, and we don’t think we could do that without DBOS.”

CTO

"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."

CEO & Co-founder

“DBOS makes it dead simple for our engineering team to build reliable, low-latency, AI-powered user experiences."

CTO and Co-founder

"I love the design of DBOS. If you're a gamer, it's like having a “save point” in your programs. If a function fails, a new function can start, picking up at the last checkpoint."

CEO & Co-Founder

“DBOS is really simple to manage. It gives us durable orchestration and observability without a lot of coding and without having to run additional infrastructure.”

CEO and Co-founder

‍“We looked at Google Workflows, Temporal, Airflow and Celery but DBOS was the one framework that solved many pieces of the puzzle while giving us implementation flexibility.”

Engineer

"DBOS makes it impossible to screw up."

CTO and Co-founder

“In EDI, you can’t have things fall between the cracks. You need the ability to trace it end to end, and we don’t think we could do that without DBOS.”

CTO

"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."

CEO & Co-founder

“DBOS makes it dead simple for our engineering team to build reliable, low-latency, AI-powered user experiences."

CTO and Co-founder

"I love the design of DBOS. If you're a gamer, it's like having a “save point” in your programs. If a function fails, a new function can start, picking up at the last checkpoint."

CEO & Co-Founder

“DBOS is really simple to manage. It gives us durable orchestration and observability without a lot of coding and without having to run additional infrastructure.”

CEO and Co-founder

‍“We looked at Google Workflows, Temporal, Airflow and Celery but DBOS was the one framework that solved many pieces of the puzzle while giving us implementation flexibility.”

Engineer

"DBOS makes it impossible to screw up."

CTO and Co-founder