Edda - Norse mythology poetic narratives that preserve ancient sagas and legends
Lightweight durable execution framework - no separate server required
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/
- ✨ 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_failuredecorator - 🌐 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
- 🤖 MCP Integration: Expose durable workflows as AI tools via Model Context Protocol
- 🌍 ASGI/WSGI Support: Deploy with your preferred server (uvicorn, gunicorn, uWSGI)
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
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:
wait_timer(duration_seconds): Wait for a relative durationwait_until(until_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 wait_timer(ctx, duration_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.
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 API<br/>Clients]
CE[CloudEvents<br/>Producer]
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 Database<br/>PostgreSQL/MySQL<br/>SQLite: single-process only)]
API -->|"workflow.start()<br/>(Direct Invocation)"| W1
API -->|"workflow.start()<br/>(Direct Invocation)"| W2
CE -->|"POST /<br/>(CloudEvents)"| W1
CE -->|"POST /<br/>(CloudEvents)"| W3
W1 <-->|Workflow<br/>State| DB
W2 <-->|Workflow<br/>State| DB
W3 <-->|Workflow<br/>State| 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()andwait_timer()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
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?
- Activities already executed return cached results from history
- Workflow resumes from the last checkpoint
- No manual intervention required
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
# All extras (PostgreSQL, MySQL, Viewer UI)
uv add edda-framework --extra postgresql --extra mysql --extra viewerInstall 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.gitInstall specific version or branch:
# Specific tag/release
uv add git+https://github.com/i2y/[email protected]
pip install git+https://github.com/i2y/[email protected]
# 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 postgresqlforasyncpgdriver- Recommended for production
- MySQL: Add
--extra mysqlforaiomysqldriver- Recommended for production
- Viewer UI: Add
--extra viewerfor workflow visualization
| Database | Use Case | Multi-Pod Support | Production Ready | Notes |
|---|---|---|---|---|
| SQLite | Development, testing, single-process deployments | ❌ No | 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.
If you want to contribute to Edda or modify the framework itself:
# Clone repository
git clone https://github.com/i2y/edda.git
cd kairo
uv sync --all-extrasRun Edda's test suite:
# Run tests
uv run pytest
# Run with coverage
uv run pytest --cov=eddaActivity: 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). See MIGRATION_GUIDE_ACTIVITY_ID.md for details.
Edda ensures workflow progress is never lost through deterministic replay:
- Activity results are recorded in a history table
- On crash recovery, workflows resume from the last checkpoint
- Already-executed activities return cached results from history
- 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 result2Key guarantees:
- Activities execute exactly once (results cached in history)
- Workflows can survive arbitrary crashes
- No manual checkpoint management required
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 userRetry 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:
- Activity-level policy (
@activity(retry_policy=...)) - Application-level policy (
EddaApp(default_retry_policy=...)) - Framework default (5 attempts, exponential backoff)
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 called on workflow failure (reverse order)
print(f"Cancelled reservation for {item_id}")
return {"cancelled": True}
@activity
@on_failure(cancel_reservation)
async def reserve_inventory(ctx: WorkflowContext, item_id: str):
print(f"Reserved {item_id}")
return {"reserved": True}
@workflow
async def order_workflow(ctx: WorkflowContext, item1: str, item2: str):
await reserve_inventory(ctx, item1) # Step 1: Reserve item1
await reserve_inventory(ctx, item2) # Step 2: Reserve item2
await charge_payment(ctx) # Step 3: If this fails...
# Compensation runs: cancel item2 → cancel item1 (reverse order)Multiple workers can safely process workflows using database-based exclusive control. This means:
- Edda uses database-based locks (not Redis or ZooKeeper)
- Each workflow instance runs on only one worker at a time
- If a worker crashes, another worker automatically resumes
- No additional infrastructure required
# Worker 1 and Worker 2 can run simultaneously
# Only one will acquire the lock for each workflow instance
app = EddaApp(
db_url="postgresql://localhost/workflows", # Shared database for coordination
service_name="order-service"
)Features:
- Each workflow instance runs on only one worker at a time (automatic coordination)
- Automatic stale lock cleanup (5-minute timeout)
- Crashed workflows automatically resume on any available worker
Type-safe workflows with automatic validation:
from pydantic import BaseModel, Field
from edda import workflow, WorkflowContext
class OrderItem(BaseModel):
item_id: str
quantity: int = Field(..., ge=1)
price: float = Field(..., gt=0)
@workflow
async def process_order(ctx: WorkflowContext, items: list[OrderItem]) -> dict:
# Automatic validation before workflow starts
total = sum(item.price * item.quantity for item in items)
return {"total": total, "item_count": len(items)}Activities are automatically transactional by default, ensuring atomicity:
from edda import activity, send_event_transactional, WorkflowContext
@activity # Automatically transactional
async def create_order(ctx: WorkflowContext, order_id: str):
# All operations in a single transaction:
# 1. Activity execution
# 2. History recording
# 3. Event publishing (outbox table)
await send_event_transactional(
ctx,
event_type="order.created",
event_source="order-service",
event_data={"order_id": order_id}
)
return {"order_id": order_id}Custom Database Operations - Use ctx.session for your database operations:
@activity # Edda manages the transaction
async def process_payment(ctx: WorkflowContext, order_id: str, amount: float):
# Access Edda-managed session (same database as Edda)
session = ctx.session
# Your business logic
payment = Payment(order_id=order_id, amount=amount)
session.add(payment)
# Edda event (same transaction)
await send_event_transactional(
ctx,
event_type="payment.processed",
event_source="payment-service",
event_data={"order_id": order_id, "amount": amount}
)
# Edda automatically commits: your data + Edda's outbox (atomic!)
return {"payment_id": f"PAY-{order_id}"}Edda provides optional event-driven capabilities for workflows that need to wait for external events.
Native support for CloudEvents protocol:
from edda import EddaApp
app = EddaApp(
db_url="sqlite:///workflow.db",
service_name="order-service",
outbox_enabled=True # Enable transactional outbox
)
# Accepts CloudEvents at any HTTP pathCloudEvents handling:
- All HTTP requests (any path) are accepted as CloudEvents
- Events without matching workflow handlers are silently discarded
- Special endpoint:
POST /cancel/{instance_id}for workflow cancellation - Automatic CloudEvents validation and parsing
- Works with CloudEvents-compatible systems (Knative Eventing, CloudEvents SDKs, etc.)
CloudEvents HTTP Binding compliance:
- 202 Accepted: Event accepted for asynchronous processing (success)
- 400 Bad Request: CloudEvents parsing/validation error (non-retryable)
- 500 Internal Server Error: Internal error (retryable)
- Error responses include
error_typeandretryableflags for client retry logic
Workflows can wait for external events or timers without consuming worker resources. While waiting, the workflow state is persisted to the database and can be resumed by any available worker when the event arrives or timer expires:
from edda import workflow, wait_event, send_event, WorkflowContext
@workflow
async def payment_workflow(ctx: WorkflowContext, order_id: str):
# Send payment request event
await send_event("payment.requested", "payment-service", {"order_id": order_id})
# Wait for payment completion event (process-releasing)
payment_event = await wait_event(ctx, "payment.completed")
return payment_event.datawait_timer() for time-based waiting:
from edda import wait_timer
@workflow
async def order_with_timeout(ctx: WorkflowContext, order_id: str):
# Create order
await create_order(ctx, order_id)
# Wait 60 seconds for payment
await wait_timer(ctx, duration_seconds=60)
# Check payment status
return await check_payment(ctx, order_id)Multi-worker continuation behavior:
wait_event()releases the workflow lock atomically- Event delivery acquires the lock and resumes on any available worker
- Safe for multi-pod/multi-container environments (K8s, Docker Compose, etc.)
- No worker is blocked while waiting for events or timers
For technical details, see Multi-Worker Continuations.
Edda runs as an ASGI application:
# Run standalone
uvicorn demo_app:application --port 8001Mounting to existing ASGI apps:
You can mount EddaApp to any path in existing ASGI frameworks:
from fastapi import FastAPI
from edda import EddaApp
# Create FastAPI app
api = FastAPI()
# Create Edda app
edda_app = EddaApp(db_url="sqlite:///workflow.db")
# Mount Edda at /workflows path
api.mount("/workflows", edda_app)
# Now Edda handles all requests under /workflows/*
# POST /workflows/any-path -> CloudEvents handler
# POST /workflows/cancel/{instance_id} -> CancellationThis works with any ASGI framework (Starlette, FastAPI, Quart, etc.)
For WSGI environments (gunicorn, uWSGI, Flask, Django), use the WSGI adapter:
from edda import EddaApp
from edda.wsgi import create_wsgi_app
# Create Edda app
edda_app = EddaApp(db_url="sqlite:///workflow.db")
# Convert to WSGI
wsgi_application = create_wsgi_app(edda_app)Running with WSGI servers:
# With Gunicorn
gunicorn demo_app:wsgi_application --workers 4
# With uWSGI
uwsgi --http :8000 --wsgi-file demo_app.py --callable wsgi_applicationSync Activities: For WSGI environments or legacy codebases, you can write synchronous activities:
from edda import activity, WorkflowContext
@activity
def process_payment(ctx: WorkflowContext, amount: float) -> dict:
# Sync function - automatically executed in thread pool
# No async/await needed!
return {"status": "paid", "amount": amount}
@workflow
async def payment_workflow(ctx: WorkflowContext, order_id: str) -> dict:
# Workflows still use async (for deterministic replay)
result = await process_payment(ctx, 99.99)
return resultPerformance note: ASGI servers (uvicorn, hypercorn) are recommended for better performance with Edda's async architecture. WSGI support is provided for compatibility with existing infrastructure and users who prefer synchronous programming.
Edda integrates with the Model Context Protocol (MCP), allowing AI assistants like Claude to interact with your durable workflows as long-running tools.
from edda.integrations.mcp import EddaMCPServer
from edda import WorkflowContext, activity
# Create MCP server
server = EddaMCPServer(
name="Order Service",
db_url="postgresql://user:pass@localhost/orders",
)
@activity
async def process_payment(ctx: WorkflowContext, amount: float):
return {"status": "paid", "amount": amount}
@server.durable_tool(description="Process customer order")
async def process_order(ctx: WorkflowContext, order_id: str):
await process_payment(ctx, 99.99)
return {"status": "completed", "order_id": order_id}
# Deploy with uvicorn
if __name__ == "__main__":
import uvicorn
uvicorn.run(server.asgi_app(), host="0.0.0.0", port=8000)Each @durable_tool automatically generates four MCP tools:
- Main tool (
process_order): Starts the workflow, returns instance ID - Status tool (
process_order_status): Checks workflow progress with completed activity count and suggested poll interval - Result tool (
process_order_result): Gets final result when completed - Cancel tool (
process_order_cancel): Cancels workflow if running or waiting, executes compensation handlers
This enables AI assistants to work with workflows that take minutes, hours, or even days to complete, with full control over the workflow lifecycle.
Define reusable prompt templates that can access workflow state:
from mcp.server.fastmcp.prompts.base import UserMessage
from mcp.types import TextContent
@server.prompt(description="Analyze a workflow execution")
async def analyze_workflow(instance_id: str) -> UserMessage:
"""Generate analysis prompt for a specific workflow."""
instance = await server.storage.get_instance(instance_id)
history = await server.storage.get_history(instance_id)
text = f"""Analyze this workflow:
**Status**: {instance['status']}
**Activities**: {len(history)}
**Result**: {instance.get('output_data')}
Please provide insights and optimization suggestions."""
return UserMessage(content=TextContent(type="text", text=text))AI clients can use these prompts to generate context-aware analysis of your workflows.
For detailed documentation, see MCP Integration Guide.
Extend Edda with custom observability without coupling to specific tools:
from edda import EddaApp
class MyHooks:
async def on_workflow_start(self, instance_id, workflow_name, input_data):
print(f"Workflow {workflow_name} started: {instance_id}")
async def on_workflow_complete(self, instance_id, workflow_name, result):
print(f"Workflow {workflow_name} completed")
async def on_activity_complete(self, instance_id, activity_id, activity_name, result, cache_hit):
print(f"Activity {activity_name} completed (cache_hit={cache_hit})")
app = EddaApp(
db_url="sqlite:///workflow.db",
service_name="my-service",
hooks=MyHooks()
)MyHooks implements the WorkflowHooks Protocol through structural subtyping. See integration examples in the examples directory.
Edda supports both JSON (dict) and binary (bytes) data for event storage and transport, allowing you to choose based on your needs.
For debugging and human-readable logs, use JSON dict format:
from google.protobuf import json_format
from edda import send_event, wait_event
# Send: Protobuf → JSON dict
msg = OrderCreated(order_id="123", amount=99.99)
await send_event("order.created", "orders", json_format.MessageToDict(msg))
# Receive: JSON dict → Protobuf
event = await wait_event(ctx, "payment.completed")
payment = json_format.ParseDict(event.data, PaymentCompleted())Benefits:
- ✅ Human-readable in database and logs
- ✅ Easy debugging and troubleshooting
- ✅ Full Viewer UI compatibility
- ✅ CloudEvents Structured Content Mode compatible
For maximum performance and zero storage overhead, Edda stores binary data directly in database BLOB columns:
from edda import send_event, wait_event
# Send binary data (e.g., Protobuf)
msg = OrderCreated(order_id="123", amount=99.99)
await send_event("order.created", "orders", msg.SerializeToString()) # bytes → BLOB
# Receive binary data
event = await wait_event(ctx, "payment.completed")
payment = PaymentCompleted()
payment.ParseFromString(event.data) # bytes from BLOBBenefits:
- ✅ Zero storage overhead (100 bytes → 100 bytes, not 133 bytes with base64)
- ✅ Maximum performance (no encoding/decoding)
- ✅ Native BLOB storage (SQLite, PostgreSQL, MySQL)
- ✅ CloudEvents Binary Content Mode compatible
Both modes are equally valid for production use:
-
JSON Mode: Human-readable, excellent observability, Viewer UI support
- Use when debugging, monitoring, and data inspection are priorities
-
Binary Mode: Zero serialization overhead, smaller storage
- Use when payload size or serialization performance are critical
- Ideal for high-throughput scenarios (>1000 events/sec)
Both modes are first-class citizens - choose based on your specific requirements, not environment.
- Getting Started: Installation and setup guide
- Core Concepts: Learn about workflows, activities, and durable execution
- Examples: See Edda in action with real-world examples
- FastAPI Integration: Integrate with FastAPI (direct invocation + CloudEvents)
- Transactional Outbox: Reliable event publishing with guaranteed delivery
- Viewer UI: Visualize and monitor your workflows
- Lifecycle Hooks: Add observability and monitoring with custom hooks
- CloudEvents HTTP Binding: CloudEvents specification compliance and error handling
This project is licensed under the MIT License - see the LICENSE file for details.
- GitHub Issues: https://github.com/i2y/edda/issues
- Documentation: https://github.com/i2y/edda#readme