ombra 0.1.2

Minimal code, maximum automation for AI workflows

Ombra

Minimal code, maximum automation for AI workflows.

Write regular Python functions. Add two decorators. Get FastAPI endpoints, execution history, and time travel debugging automatically.

from ombra import workflow, step

@step()
async def fetch_data(query: str) -> dict:
    return {"query": query, "results": ["item1", "item2", "item3"]}

@step()
async def process_data(data: dict) -> list:
    return [item.upper() for item in data["results"]]

@workflow(name="demo", description="Demo workflow")
async def demo_workflow(query: str) -> str:
    data = await fetch_data(query)
    processed = await process_data(data)
    return f"Result: {', '.join(processed)}"

That's it. From these 15 lines of code, you automatically get:

• ⚡ FastAPI endpoint: POST /workflows/demo/execute
• 🔄 Time travel: Resume from any step when debugging
• 📊 Web UI: View execution history and step-by-step details
• 💾 Automatic checkpointing: All inputs/outputs saved to SQLite

Why Ombra?

Most workflow frameworks make you write framework code. Ombra lets you write normal Python and generates everything else automatically.

Test = Production: Test your AI workflows using FastAPI's /docs interface. The same endpoints you test locally become your production API. No translation layer, no separate testing harness.

Quick Start

Installation

pip install ombra

uv add ombra

Requires Python ≥3.12

Run Your First Workflow

1. Create a file workflows/demo.py:

from ombra import workflow, step

@step()
async def fetch_data(query: str) -> dict:
    return {"query": query, "results": ["item1", "item2", "item3"]}

@step()
async def process_data(data: dict) -> list:
    return [item.upper() for item in data["results"]]

@workflow(name="demo", description="Demo workflow")
async def demo_workflow(query: str) -> str:
    data = await fetch_data(query)
    processed = await process_data(data)
    return f"Result: {', '.join(processed)}"

2. Create app.py:

from ombra import create_app

app = create_app()

3. Run it:

uvicorn app:app --reload

4. Open your browser:
   • API Docs: http://localhost:8000/docs
   • Web UI: http://localhost:8000/workflows

Test Your Workflow

Go to http://localhost:8000/docs and find your workflow's endpoint:

POST /workflows/demo/execute

Click "Try it out", enter your parameters, and execute. Your workflow runs and all steps are automatically checkpointed.

What you test in /docs is exactly what runs in production. No surprises.

Time Travel: Debug AI Workflows

The killer feature for AI development.

The Problem

You're building a 5-step LLM workflow. Step 4 fails because of a prompt issue. Without Ombra, you have to:

1. Re-run steps 1-3 (wasting time and money on API calls)
2. Try to manually cache results (error-prone)
3. Hope nothing changed in the earlier steps

The Solution

# Original execution failed at step 4
POST /workflows/my_workflow/execute
# Execution ID: abc123

# Fix your code, then resume from step 4
POST /workflows/my_workflow/resume?execution_id=abc123&step_name=step_4

Steps 1-3 use cached results. Only step 4 and beyond run again.

This is huge for:

• LLM calls: Don't re-run expensive model calls when debugging downstream steps
• API rate limits: Don't waste rate limit quota on steps you've already tested
• Slow operations: Skip web scraping, data processing, etc. when iterating

See It in Action

After an execution, visit the Web UI at http://localhost:8000/workflows. Click on any execution to see:

• Step-by-step execution log
• Inputs and outputs for each step
• Timestamps and duration
• "Resume from here" button for each step

Real-World Example: AI Content Pipeline

from ombra import workflow, step
import httpx
from openai import AsyncOpenAI

client = AsyncOpenAI()

@step()
async def scrape_webpage(url: str) -> str:
    """Fetch webpage content."""
    async with httpx.AsyncClient() as http:
        response = await http.get(url)
        return response.text

@step()
async def extract_key_points(html: str) -> list[str]:
    """Extract key information using LLM."""
    response = await client.chat.completions.create(
        model="gpt-4",
        messages=[{
            "role": "user",
            "content": f"Extract key points from this HTML:\n\n{html}"
        }]
    )
    return response.choices[0].message.content.split("\n")

@step()
async def summarize_points(points: list[str]) -> str:
    """Generate a summary from key points."""
    response = await client.chat.completions.create(
        model="gpt-4",
        messages=[{
            "role": "user",
            "content": f"Summarize these points:\n" + "\n".join(points)
        }]
    )
    return response.choices[0].message.content

@step()
async def validate_summary(summary: str) -> dict:
    """Check if summary meets quality criteria."""
    word_count = len(summary.split())
    return {
        "summary": summary,
        "valid": word_count >= 50 and word_count <= 200,
        "word_count": word_count
    }

@workflow(
    name="content_pipeline",
    description="Scrape, extract, summarize, and validate content"
)
async def content_pipeline(url: str) -> dict:
    html = await scrape_webpage(url)
    points = await extract_key_points(html)
    summary = await summarize_points(points)
    result = await validate_summary(summary)
    return result

This is just normal Python code. No framework cruft.

Now you can:

• Test it in /docs
• View execution history in the Web UI
• If validate_summary fails, resume from there without re-running the expensive LLM calls
• Deploy the same API you tested locally

How It Works

@step() Decorator

Marks a function as a workflow step. When executed:

1. Saves inputs to SQLite checkpoint
2. Runs your function
3. Saves outputs to SQLite checkpoint
4. Links checkpoint to current execution

In replay mode, returns cached results instead of re-running.

@workflow() Decorator

Creates a workflow entry point. When you apply it:

1. Registers workflow in global registry
2. Generates Pydantic model from function signature
3. Creates FastAPI endpoint: POST /workflows/{name}/execute
4. Saves schema to database for persistence

Auto-Discovery

On startup, Ombra scans:

• workflows/ directory
• src/workflows/ directory

All .py files are imported, triggering @workflow decorators to register themselves.

Database

Everything is stored in .ombra_executions.db (SQLite):

• Executions (status, timing, inputs/outputs)
• Checkpoints (step inputs/outputs)
• Step execution order
• Workflow schemas

Project Structure

your-project/
├── workflows/              # Your workflow files (auto-discovered)
│   ├── demo.py
│   ├── content_pipeline.py
│   └── steps/              # Shared step functions
│       └── common_steps.py
├── app.py                  # FastAPI app entry point
├── .ombra_executions.db    # SQLite database (auto-created)
└── pyproject.toml

Configuration

Custom Database Path

from ombra import workflow

@workflow(
    name="my_workflow",
    description="Workflow with custom DB",
    db_path="/custom/path/executions.db"
)
async def my_workflow(x: int) -> int:
    return x * 2

Custom Discovery Paths

from ombra.web.app import create_app

app = create_app(discovery_paths=["my_workflows", "shared/workflows"])

API Reference

Decorators

@step()

@step()
async def my_step(arg: str) -> dict:
    return {"result": arg}

• Must be applied to async functions
• Automatically checkpoints inputs/outputs
• Participates in time travel

@workflow(name: str, description: str, db_path: str = ".ombra_executions.db")

@workflow(name="my_workflow", description="Does something cool")
async def my_workflow(x: int, y: str) -> dict:
    # Your code here
    return result

• Must be applied to async functions
• Generates FastAPI endpoint automatically
• Supports type hints for validation

Generated Endpoints

For each workflow, Ombra generates:

POST /workflows/{name}/execute

• Executes workflow with given parameters
• Returns execution ID and result
• Example: {"execution_id": "abc123", "result": {...}}

POST /workflows/{name}/resume

• Query params: execution_id, step_name
• Resumes from specified step using cached results
• Returns new execution ID and result

GET /workflows/{name}/executions

• Lists all executions for a workflow
• Returns execution history with status and timing

Web UI Routes

• GET /workflows - Execution history dashboard
• GET /executions/{id} - Detailed view of single execution
• GET /docs - FastAPI auto-generated API documentation

Development

Running Locally

# Install dependencies
pip install -e .

# Run with auto-reload
uvicorn app:app --reload

# Or use the Python API
python -c "from ombra import create_app; import uvicorn; uvicorn.run(create_app(), host='0.0.0.0', port=8000)"

Building the UI (Optional)

The Web UI uses Tailwind CSS. If you want to customize styles:

# Install Node.js dependencies
npm install

# Watch for changes (development)
npm run watch:css

# Build for production
npm run build:css

The compiled CSS is committed to the repo, so end users don't need Node.js.

License

MIT