Edda

Edda

Edda - Norse mythology poetic narratives that preserve ancient sagas and legends

> Lightweight durable execution framework - no separate server required

[](https://opensource.org/licenses/MIT) [](https://www.python.org/downloads/) [](https://i2y.github.io/edda/) [](https://deepwiki.com/i2y/edda)

Overview

Edda is a lightweight durable execution framework for Python that runs as a library in your application - no separate workflow server required. It provides automatic crash recovery through deterministic replay, allowing long-running workflows to survive process restarts and failures without losing progress.

Perfect for: Order processing, distributed transactions (Saga pattern), AI agent orchestration, and any workflow that must survive crashes.

For detailed documentation, visit https://i2y.github.io/edda/

Key Features

• ✨ Lightweight Library: Runs in your application process - no separate server infrastructure
• 🔄 Durable Execution: Deterministic replay with workflow history for automatic crash recovery
• 🎯 Workflow & Activity: Clear separation between orchestration logic and business logic
• 🔁 Saga Pattern: Automatic compensation on failure with @on_failure decorator
• 🌐 Multi-worker Execution: Run workflows safely across multiple servers or containers
• 🔒 Pydantic Integration: Type-safe workflows with automatic validation
• 📦 Transactional Outbox: Reliable event publishing with guaranteed delivery
• ☁️ CloudEvents Support: Native support for CloudEvents protocol
• ⏱️ Event & Timer Waiting: Free up worker resources while waiting for events or timers, resume on any available worker
• 📬 Channel-based Messaging: Actor-model style communication with competing (job queue) and broadcast (fan-out) modes
• ⚡ Instant Notifications: PostgreSQL LISTEN/NOTIFY for near-instant event delivery (optional)
• 🤖 MCP Integration: Expose durable workflows as AI tools via Model Context Protocol
• 🧠 Mirascope Integration: Durable LLM calls
• 🦙 LlamaIndex Integration: Make LlamaIndex Workflows durable with crash recovery
• 📊 pydantic-graph Integration: Durable graph-based workflows (experimental)
• 🌍 ASGI/WSGI Support: Deploy with your preferred server (uvicorn, gunicorn, uWSGI)

Use Cases

Edda excels at orchestrating long-running workflows that must survive failures:

• 🏢 Long-Running Jobs: Order processing, data pipelines, batch jobs - from minutes to days, weeks, or even months
• 🔄 Distributed Transactions: Coordinate microservices with automatic compensation (Saga pattern)
• 🤖 AI Agent Workflows: Orchestrate multi-step AI tasks (LLM calls, tool usage, long-running inference)
• 📡 Event-Driven Workflows: React to external events with guaranteed delivery and automatic retry

Business Process Automation

Edda's waiting functions make it ideal for time-based and event-driven business processes:

• 📧 User Onboarding: Send reminders if users haven't completed setup after N days
• 🎁 Campaign Processing: Evaluate conditions and notify winners after campaign ends
• 💳 Payment Reminders: Send escalating reminders before payment deadlines
• 📦 Scheduled Notifications: Shipping updates, subscription renewals, appointment reminders

Waiting functions:

• sleep(seconds): Wait for a relative duration
• sleep_until(target_time): Wait until an absolute datetime (e.g., campaign end date)
• wait_event(event_type): Wait for external events (near real-time response)

@workflow
async def onboarding_reminder(ctx: WorkflowContext, user_id: str):
    await sleep(ctx, seconds=3*24*60*60)  # Wait 3 days
    if not await check_completed(ctx, user_id):
        await send_reminder(ctx, user_id)

Key benefit: Workflows never lose progress - crashes and restarts are handled automatically through deterministic replay.

Architecture

Edda runs as a lightweight library in your applications, with all workflow state stored in a shared database:

%%{init: {'theme':'base', 'themeVariables': {'primaryTextColor':'#1a1a1a', 'secondaryTextColor':'#1a1a1a', 'tertiaryTextColor':'#1a1a1a', 'textColor':'#1a1a1a', 'nodeTextColor':'#1a1a1a'}}}%%
graph TB
    subgraph ext["External Systems"]
        API[REST APIClients]
        CE[CloudEventsProducer]
    end

    subgraph cluster["Your Multiple Instances"]
        subgraph pod1["order-service Pod 1"]
            W1[Edda Workflow]
        end
        subgraph pod2["order-service Pod 2"]
            W2[Edda Workflow]
        end
        subgraph pod3["order-service Pod 3"]
            W3[Edda Workflow]
        end
    end

    DB[(Shared DatabasePostgreSQL/MySQLSQLite: single-process only)]

    API -->|"workflow.start()(Direct Invocation)"| W1
    API -->|"workflow.start()(Direct Invocation)"| W2
    CE -->|"POST /(CloudEvents)"| W1
    CE -->|"POST /(CloudEvents)"| W3

    W1 |WorkflowState| DB
    W2 |WorkflowState| DB
    W3 |WorkflowState| DB

    style DB fill:#e1f5ff
    style W1 fill:#fff4e6
    style W2 fill:#fff4e6
    style W3 fill:#fff4e6

Key Points:

• Multiple workers can run simultaneously across different pods/servers
• Each workflow instance runs on only one worker at a time (automatic coordination)
• wait_event() and sleep() free up worker resources while waiting, resume on any worker when event arrives or timer expires
• Automatic crash recovery with stale lock cleanup and workflow auto-resume

Quick Start

from edda import EddaApp, workflow, activity, WorkflowContext

@activity
async def process_payment(ctx: WorkflowContext, amount: float):
    # Durable execution - automatically recorded in history
    print(f"Processing payment: ${amount}")
    return {"status": "paid", "amount": amount}

@workflow
async def order_workflow(ctx: WorkflowContext, order_id: str, amount: float):
    # Workflow orchestrates activities with automatic retry on crash
    result = await process_payment(ctx, amount)
    return {"order_id": order_id, **result}

# Simplified example - production code needs:
# 1. await app.initialize() before starting workflows
# 2. try-finally with await app.shutdown() for cleanup
# 3. PostgreSQL or MySQL for multi-process/multi-pod deployments
app = EddaApp(db_url="sqlite:///workflow.db")

# Start workflow
instance_id = await order_workflow.start(order_id="ORD-123", amount=99.99)

What happens on crash?

1. Activities already executed return cached results from history
2. Workflow resumes from the last checkpoint
3. No manual intervention required

Installation

Install Edda from PyPI using uv:

# Basic installation (includes SQLite support)
uv add edda-framework

# With PostgreSQL support
uv add edda-framework --extra postgresql

# With MySQL support
uv add edda-framework --extra mysql

# With Viewer UI
uv add edda-framework --extra viewer

# With PostgreSQL instant notifications (LISTEN/NOTIFY)
uv add edda-framework --extra postgres-notify

# With LlamaIndex Workflow integration
uv add edda-framework --extra llamaindex

# With pydantic-graph integration (experimental)
uv add edda-framework --extra graph

# All extras (PostgreSQL, MySQL, Viewer UI)
uv add edda-framework --extra postgresql --extra mysql --extra viewer

Installing from GitHub (Development Versions)

Install the latest development version directly from GitHub:

# Using uv (latest from main branch)
uv add git+https://github.com/i2y/edda.git

# Using pip
pip install git+https://github.com/i2y/edda.git

Install specific version or branch:

# Specific tag/release
uv add git+https://github.com/i2y/edda.git@v0.1.0
pip install git+https://github.com/i2y/edda.git@v0.1.0

# Specific branch
uv add git+https://github.com/i2y/edda.git@feature-branch
pip install git+https://github.com/i2y/edda.git@feature-branch

# With extras (PostgreSQL, Viewer)
uv add "git+https://github.com/i2y/edda.git[postgresql,viewer]"
pip install "git+https://github.com/i2y/edda.git[postgresql,viewer]"

Database Drivers:

• SQLite: Included by default (via aiosqlite)
• Single-process deployments only (supports multiple async workers within one process, not multiple processes/pods)
• PostgreSQL: Add --extra postgresql for asyncpg driver
• Recommended for production
• MySQL: Add --extra mysql for aiomysql driver
• Recommended for production
• Viewer UI: Add --extra viewer for workflow visualization

Database Selection Guide

| Database | Use Case | Multi-Pod Support | Production Ready | Notes | |----------|----------|-------------------|------------------|-------| | SQLite | Development, testing, single-process deployments | ❌ No | ⚠️ Limited | Supports multiple async workers within one process, but not multiple processes/pods (K8s, Docker Compose with multiple replicas) | | PostgreSQL | Production, multi-process/multi-pod systems | ✅ Yes | ✅ Yes | Recommended for production - Full support for database-based exclusive control and concurrent workflows | | MySQL | Production with existing MySQL infrastructure | ✅ Yes | ✅ Yes | Suitable for production - Good choice if you already use MySQL |

Important: For multi-process or multi-pod deployments (K8s, Docker Compose with multiple replicas, etc.), you must use PostgreSQL or MySQL. SQLite supports multiple async workers within a single process, but its table-level locking makes it unsuitable for multi-process/multi-pod scenarios.

> Tip: For PostgreSQL, install the postgres-notify extra for near-instant event delivery using LISTEN/NOTIFY instead of polling.

Database Schema Migration

Automatic Migration (Default)

Edda automatically applies database migrations at startup. No manual commands needed:

from edda import EddaApp

# Migrations are applied automatically
app = EddaApp(db_url="postgresql://user:pass@localhost/dbname")

This is safe in multi-worker environments - Edda handles concurrent startup gracefully.

Manual Migration with dbmate (Optional)

For explicit schema control, you can disable auto-migration and use dbmate:

# Disable auto-migration
app = EddaApp(
    db_url="postgresql://...",
    auto_migrate=False  # Use dbmate-managed schema
)

# Install dbmate
brew install dbmate  # macOS

# Add schema submodule
git submodule add https://github.com/durax-io/schema.git schema

# Run migration manually
DATABASE_URL="postgresql://user:pass@localhost/dbname" dbmate -d ./schema/db/migrations/postgresql up

> Note: Edda's auto-migration uses the same SQL files as dbmate, maintaining full compatibility.

Development Installation

If you want to contribute to Edda or modify the framework itself:

# Clone repository
git clone https://github.com/i2y/edda.git
cd edda
uv sync --all-extras

Running Tests

Run Edda's test suite:

# Run tests
uv run pytest

# Run with coverage
uv run pytest --cov=edda

Core Concepts

Workflows and Activities

Activity: A unit of work that performs business logic. Activity results are recorded in history.

Workflow: Orchestration logic that coordinates activities. Workflows can be replayed from history after crashes.

from edda import workflow, activity, WorkflowContext

@activity
async def send_email(ctx: WorkflowContext, email: str, message: str):
    # Business logic - this will be recorded
    print(f"Sending email to {email}")
    return {"sent": True}

@workflow
async def user_signup(ctx: WorkflowContext, email: str):
    # Orchestration logic
    await send_email(ctx, email, "Welcome!")
    return {"status": "completed"}

Activity IDs: Activities are automatically identified with IDs like "send_email:1" for deterministic replay. Manual IDs are only needed for concurrent execution (e.g., asyncio.gather).

Durable Execution

Edda ensures workflow progress is never lost through deterministic replay:

1. Activity results are recorded in a history table
2. On crash recovery, workflows resume from the last checkpoint
3. Already-executed activities return cached results from history
4. New activities continue from where the workflow left off

@workflow
async def long_running_workflow(ctx: WorkflowContext, user_id: str):
    # Activity 1: Recorded in history
    result1 = await create_user(ctx, user_id)

    # If process crashes here, activity won't re-execute on restart

    # Activity 2: Continues from history on restart
    result2 = await send_welcome_email(ctx, result1["email"])

    return result2

Key guarantees:

• Activities execute exactly once (results cached in history)
• Workflows can survive arbitrary crashes
• No manual checkpoint management required

Automatic Activity Retry

Activities automatically retry with exponential backoff when errors occur, improving reliability without manual error handling:

from edda import activity, WorkflowContext

@activity
async def call_external_api(ctx: WorkflowContext, url: str):
    # Automatically retries up to 5 times with exponential backoff
    # Delays: 1s, 2s, 4s, 8s, 16s
    response = await httpx.get(url, timeout=10)
    return response.json()

Default retry policy:

• 5 attempts (including initial)
• Exponential backoff: 1s, 2s, 4s, 8s, 16s between attempts
• Max delay: 60 seconds
• Total duration: 5 minutes maximum

Custom retry policies for specific activities:

from edda import activity, RetryPolicy, WorkflowContext

@activity(retry_policy=RetryPolicy(
    max_attempts=3,
    initial_interval=0.5,
    backoff_coefficient=2.0,
    max_interval=10.0,
    max_duration=60.0
))
async def flaky_operation(ctx: WorkflowContext, data: dict):
    # Custom: 3 attempts, delays 0.5s, 1s, 2s
    return await external_service.process(data)

Application-level default policy:

from edda import EddaApp, RetryPolicy

app = EddaApp(
    db_url="sqlite:///workflow.db",
    default_retry_policy=RetryPolicy(
        max_attempts=10,
        initial_interval=2.0
    )
)

Non-retryable errors with TerminalError:

from edda import activity, TerminalError, WorkflowContext

@activity
async def validate_user(ctx: WorkflowContext, user_id: str):
    user = await get_user(user_id)
    if user is None:
        # Immediately fail without retry (user doesn't exist)
        raise TerminalError(f"User {user_id} not found")
    return user

Retry metadata for observability:

Retry information is automatically embedded in activity history for monitoring:

{
    "event_type": "ActivityCompleted",
    "event_data": {
        "activity_name": "call_external_api",
        "result": {...},
        "retry_metadata": {
            "total_attempts": 3,
            "total_duration_ms": 7200,
            "last_error": {...},
            "exhausted": False,
            "errors": [...]
        }
    }
}

Policy resolution order:

1. Activity-level policy (@activity(retry_policy=...))
2. Application-level policy (EddaApp(default_retry_policy=...))
3. Framework default (5 attempts, exponential backoff)

Compensation (Saga Pattern)

When a workflow fails, Edda automatically executes compensation functions for already-executed activities in reverse order. This implements the Saga pattern for distributed transaction rollback.

Key behavior:

• Compensation functions run in reverse order of activity execution
• Only already-executed activities are compensated
• If Activity A and B completed, then C fails → B and A compensations run (in that order)

from edda import activity, on_failure, compensation, workflow, WorkflowContext

@compensation
async def cancel_reservation(ctx: WorkflowContext, item_id: str):
    # Automatically calle

…

## Source & license

This open-source MCP server is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [i2y](https://github.com/i2y)
- **Source:** [i2y/edda](https://github.com/i2y/edda)
- **License:** MIT
- **Homepage:** https://i2y.github.io/edda/

Install and usage instructions live in the source repository linked above.