dbos-python
Guide for building reliable, fault-tolerant Python applications with DBOS durable workflows. Use when adding DBOS to existing Python code, creating workflows and steps, or using queues for concurrency control.
About this skill
This skill equips an AI agent with comprehensive best practices and guidelines for developing robust, fault-tolerant Python applications leveraging DBOS durable workflows. It covers key aspects from integrating DBOS into existing Python projects, designing workflows and individual steps, and managing concurrency using queues. The agent can use this knowledge to assist developers in building resilient systems, generating correct DBOS-compliant code, and understanding advanced features like workflow communication (events, messages, streams), application configuration, and external client interactions. This empowers the agent to serve as a knowledgeable assistant for Python developers working with the DBOS framework.
Best use case
Assisting developers in designing and implementing reliable, fault-tolerant Python applications. Generating code snippets and architectural advice adhering to DBOS best practices. Guiding through DBOS integration, workflow creation, concurrency control, and testing.
Guide for building reliable, fault-tolerant Python applications with DBOS durable workflows. Use when adding DBOS to existing Python code, creating workflows and steps, or using queues for concurrency control.
The AI agent provides accurate, contextually relevant advice, code examples, or architectural recommendations for building Python applications with DBOS, resulting in more reliable and fault-tolerant software designs and implementations.
Practical example
Example input
How do I create a durable workflow in Python using DBOS for a multi-step order processing system, ensuring fault tolerance and idempotency?
Example output
To create a durable workflow for an order processing system with DBOS, you'd define a class decorated with `@dbos_workflow()`. Each processing step (e.g., payment, inventory update, shipping) should be a `@transactional_function()` within the workflow. This ensures that each step is executed exactly once, even if failures occur. Here's a basic structure:
```python
from dbos_core import Workflow, WorkflowContext, transactional_function, dbos_workflow
@dbos_workflow()
class OrderProcessingWorkflow(Workflow):
def __init__(self, ctx: WorkflowContext):
super().__init__(ctx)
@transactional_function()
def _process_payment(self, order_id: str, amount: float) -> bool:
"""Simulates processing a payment. Idempotent by design."""
self.ctx.logger.info(f"Processing payment for order {order_id}, amount {amount}")
# Add actual payment gateway interaction logic here
# Ensure this function is idempotent (e.g., check if payment already processed)
return True # Or False on failure
@transactional_function()
def _update_inventory(self, order_id: str, items: dict) -> bool:
"""Simulates updating inventory. Idempotent."""
self.ctx.logger.info(f"Updating inventory for order {order_id}, items {items}")
# Add actual inventory system interaction logic here
# Ensure idempotency
return True
@transactional_function()
def _send_confirmation_email(self, order_id: str, email: str) -> bool:
"""Sends a confirmation email. Can be retried."""
self.ctx.logger.info(f"Sending confirmation email for order {order_id} to {email}")
# Add email sending logic here
return True
def run(self, order_details: dict) -> str:
order_id = order_details['id']
self.ctx.logger.info(f"Starting order workflow for {order_id}")
if not self._process_payment(order_id, order_details['total_amount']):
self.ctx.logger.error(f"Payment failed for {order_id}")
return "ORDER_PAYMENT_FAILED"
if not self._update_inventory(order_id, order_details['items']):
self.ctx.logger.error(f"Inventory update failed for {order_id}")
# Compensation or manual intervention might be needed here
return "ORDER_INVENTORY_FAILED"
self._send_confirmation_email(order_id, order_details['customer_email'])
self.ctx.logger.info(f"Order {order_id} processed successfully.")
return "ORDER_PROCESSED_SUCCESSFULLY"
# To invoke this workflow from your DBOS application:
# workflow_instance = OrderProcessingWorkflow(dbos_runtime_context) # Context provided by DBOS
# result = await workflow_instance.run({
# 'id': 'ORD-2023-001',
# 'total_amount': 99.99,
# 'items': {'itemA': 1, 'itemB': 2},
# 'customer_email': 'customer@example.com'
# })
# print(result)
```
**Key DBOS Principles Applied:**
* **Durable Workflow:** The `run` method orchestrates the steps, and DBOS ensures its state is preserved across failures.
* **Transactional Functions:** Each `@transactional_function()` (`_process_payment`, `_update_inventory`, `_send_confirmation_email`) is executed within a transaction, providing atomicity and exactly-once semantics. If a step fails, DBOS can retry it or roll back, maintaining data consistency.
* **Idempotency:** It's crucial that your business logic within `@transactional_function()` is idempotent, meaning executing it multiple times has the same effect as executing it once (e.g., only debiting a payment once even if the function is retried).When to use this skill
- When an AI agent needs to provide guidance or generate code for: - Integrating DBOS into new or existing Python applications. - Designing and implementing durable workflows and individual steps. - Managing concurrent operations using DBOS queues. - Setting up inter-workflow communication via events, messages, or streams. - Configuring and launching DBOS-powered Python applications. - Developing external applications that interact with DBOS using DBOSClient. - Crafting test strategies for DBOS applications.
When not to use this skill
- When the task does not involve Python development, durable workflows, or the DBOS framework specifically. This skill is not for direct execution of Python code by the agent, but rather a knowledge base for guiding and generating code for Python development with DBOS.
Installation
Claude Code / Cursor / Codex
Manual Installation
- Download SKILL.md from GitHub
- Place it in
.claude/skills/dbos-python/SKILL.mdinside your project - Restart your AI agent — it will auto-discover the skill
How dbos-python Compares
| Feature / Agent | dbos-python | Standard Approach |
|---|---|---|
| Platform Support | Claude | Limited / Varies |
| Context Awareness | High | Baseline |
| Installation Complexity | easy | N/A |
Frequently Asked Questions
What does this skill do?
Guide for building reliable, fault-tolerant Python applications with DBOS durable workflows. Use when adding DBOS to existing Python code, creating workflows and steps, or using queues for concurrency control.
Which AI agents support this skill?
This skill is designed for Claude.
How difficult is it to install?
The installation complexity is rated as easy. You can find the installation instructions above.
Where can I find the source code?
You can find the source code on GitHub using the link provided at the top of the page.
Related Guides
AI Agents for Coding
Browse AI agent skills for coding, debugging, testing, refactoring, code review, and developer workflows across Claude, Cursor, and Codex.
Best AI Skills for Claude
Explore the best AI skills for Claude and Claude Code across coding, research, workflow automation, documentation, and agent operations.
ChatGPT vs Claude for Agent Skills
Compare ChatGPT and Claude for AI agent skills across coding, writing, research, and reusable workflow execution.
SKILL.md Source
# DBOS Python Best Practices
Guide for building reliable, fault-tolerant Python applications with DBOS durable workflows.
## When to Use
Reference these guidelines when:
- Adding DBOS to existing Python code
- Creating workflows and steps
- Using queues for concurrency control
- Implementing workflow communication (events, messages, streams)
- Configuring and launching DBOS applications
- Using DBOSClient from external applications
- Testing DBOS applications
## Rule Categories by Priority
| Priority | Category | Impact | Prefix |
|----------|----------|--------|--------|
| 1 | Lifecycle | CRITICAL | `lifecycle-` |
| 2 | Workflow | CRITICAL | `workflow-` |
| 3 | Step | HIGH | `step-` |
| 4 | Queue | HIGH | `queue-` |
| 5 | Communication | MEDIUM | `comm-` |
| 6 | Pattern | MEDIUM | `pattern-` |
| 7 | Testing | LOW-MEDIUM | `test-` |
| 8 | Client | MEDIUM | `client-` |
| 9 | Advanced | LOW | `advanced-` |
## Critical Rules
### DBOS Configuration and Launch
A DBOS application MUST configure and launch DBOS inside its main function:
```python
import os
from dbos import DBOS, DBOSConfig
@DBOS.workflow()
def my_workflow():
pass
if __name__ == "__main__":
config: DBOSConfig = {
"name": "my-app",
"system_database_url": os.environ.get("DBOS_SYSTEM_DATABASE_URL"),
}
DBOS(config=config)
DBOS.launch()
```
### Workflow and Step Structure
Workflows are comprised of steps. Any function performing complex operations or accessing external services must be a step:
```python
@DBOS.step()
def call_external_api():
return requests.get("https://api.example.com").json()
@DBOS.workflow()
def my_workflow():
result = call_external_api()
return result
```
### Key Constraints
- Do NOT call `DBOS.start_workflow` or `DBOS.recv` from a step
- Do NOT use threads to start workflows - use `DBOS.start_workflow` or queues
- Workflows MUST be deterministic - non-deterministic operations go in steps
- Do NOT create/update global variables from workflows or steps
## How to Use
Read individual rule files for detailed explanations and examples:
```
references/lifecycle-config.md
references/workflow-determinism.md
references/queue-concurrency.md
```
## References
- https://docs.dbos.dev/
- https://github.com/dbos-inc/dbos-transact-pyRelated Skills
n8n-expression-syntax
Validate n8n expression syntax and fix common errors. Use when writing n8n expressions, using {{}} syntax, accessing $json/$node variables, troubleshooting expression errors, or working with webhook data in workflows.
mermaid-expert
Create Mermaid diagrams for flowcharts, sequences, ERDs, and architectures. Masters syntax for all diagram types and styling.
mcp-builder-ms
Use this skill when building MCP servers to integrate external APIs or services, whether in Python (FastMCP) or Node/TypeScript (MCP SDK).
makepad-deployment
CRITICAL: Use for Makepad packaging and deployment. Triggers on: deploy, package, APK, IPA, 打包, 部署, cargo-packager, cargo-makepad, WASM, Android, iOS, distribution, installer, .deb, .dmg, .nsis, GitHub Actions, CI, action, marketplace
macos-menubar-tuist-app
Build, refactor, or review SwiftUI macOS menubar apps that use Tuist.
kaizen
Guide for continuous improvement, error proofing, and standardization. Use this skill when the user wants to improve code quality, refactor, or discuss process improvements.
issues
Interact with GitHub issues - create, list, and view issues.
hugging-face-tool-builder
Your purpose is now is to create reusable command line scripts and utilities for using the Hugging Face API, allowing chaining, piping and intermediate processing where helpful. You can access the API directly, as well as use the hf command line tool.
git-pushing
Stage all changes, create a conventional commit, and push to the remote branch. Use when explicitly asks to push changes ("push this", "commit and push"), mentions saving work to remote ("save to github", "push to remote"), or completes a feature and wants to share it.
git-hooks-automation
Master Git hooks setup with Husky, lint-staged, pre-commit framework, and commitlint. Automate code quality gates, formatting, linting, and commit message enforcement before code reaches CI.
gh-review-requests
Fetch unread GitHub notifications for open PRs where review is requested from a specified team or opened by a team member. Use when asked to "find PRs I need to review", "show my review requests", "what needs my review", "fetch GitHub review requests", or "check team review queue".
fp-types-ref
Quick reference for fp-ts types. Use when user asks which type to use, needs Option/Either/Task decision help, or wants fp-ts imports.