aceteam-workflow-engine 2.0.0rc13

A powerful workflow orchestration engine for composing complex computational tasks from modular, type-safe nodes

AceTeam Workflow Engine

A powerful, modular workflow orchestration system designed for composing complex computational tasks from smaller, configurable steps. This engine powers the workflow functionality in AceTeam.ai and is now available as an open-source package.

Overview

The Workflow Engine enables you to:

• Define workflows as directed acyclic graphs (DAGs)
• Chain node-based tasks with type-safe data passing
• Persist and retrieve node outputs using various storage backends
• Execute workflows programmatically or via API

Installation

pip install aceteam-workflow-engine

Claude Code skill (optional)

The wengine skill is distributed through the AceTeam marketplace (its source lives here, in plugins/wengine/). If you use Claude Code, install it so Claude can help you explore, compose, validate, and run workflows:

/plugin marketplace add aceteam-ai/marketplace
/plugin install wengine@aceteam

Example

import asyncio

from workflow_engine import IntegerValue, Workflow
import workflow_engine.nodes
from workflow_engine.contexts import LocalContext
from workflow_engine.execution import TopologicalExecutionAlgorithm

context = LocalContext()
algorithm = TopologicalExecutionAlgorithm()

# Load and run a workflow
with open("examples/addition.json") as f:
    workflow = Workflow.model_validate_json(f.read())

result = asyncio.run(algorithm.execute(
    context=context,
    workflow=workflow,
    input={"c": IntegerValue(-256)},
)) # {'sum': 1811}

Check the examples directory for more sample workflows in JSON form:

• addition.json: basic arithmetic operations
• append.json: text file manipulation
• error.json: graceful error handling

Key Features

Core Functionality

• Graph-Based Execution: Workflows are executed as DAGs with automatic dependency resolution
• Type-Safe Data Flow: Data passing between nodes is validated using MIME types
• Flexible Storage: Supports multiple storage backends (Supabase, Local, In-Memory)
• Error Handling: Robust error propagation and logging system
• Versioning: Built-in support for workflow versioning

Workflow Structure

Workflows are defined as JSON with:

• input_node: Defines the workflow's input schema using InputNode
• inner_nodes: The processing nodes that perform computations
• output_node: Defines the workflow's output schema using OutputNode
• edges: Connect node outputs to inputs (including to/from input/output nodes)

Node Types

• InputNode: Special node defining workflow input fields and their types
• OutputNode: Special node defining workflow output fields and their types
• Processing Nodes: Execute computational tasks with configurable parameters

Storage Backends

• Supabase: Primary storage backend for production use
• Local: File-system based storage for development
• In-Memory: Lightweight storage for testing

Value Type Casting

The workflow engine supports automatic type casting between Value types. The graph below shows all available casting paths:

Value types serialize to JSON Schema for validation and type resolution. See Values for how the schema system works and limitations around deeply nested generics.

Architecture

src/workflow_engine/
├── contexts/          # Storage backend implementations
│   ├── in_memory.py   # In-memory storage
│   └── local.py       # Local file system storage
├── core/              # Core workflow components
│   ├── context.py     # Execution context
│   ├── data.py        # Data handling
│   ├── edge.py        # Edge definitions
│   ├── execution.py   # Execution logic
│   ├── file.py        # File handling
│   ├── node.py        # Node base classes
│   └── workflow.py    # Workflow definitions
├── execution/         # Execution strategies
│   └── topological.py # DAG-based execution
├── nodes/             # Node implementations
│   ├── arithmetic.py  # Math operations
│   ├── constant.py    # Constant values
│   ├── json.py        # JSON operations
│   └── text.py        # Text operations
└── utils/             # Helper utilities

Development

Setup

# Using uv (recommended)
uv sync
uv run pre-commit install

Testing

uv run pytest  # Runs both unit and integration tests

Documentation

• Getting Started - Installation and first workflow
• Architecture - Module structure and design decisions
• Nodes - Built-in node reference
• Values - Value type system and casting rules
• Execution - Execution algorithms, retry, and rate limiting
• Contexts - Storage backends and lifecycle hooks
• Migrations - Node versioning and migration system
• Workflow Loading - Safe workflow loading with migration
• CLI - The wengine command-line reference
• Node Distribution - Installing nodes from packages, git, and paths via engine.yaml
• Examples
• Changelog
• Contributing

Contributing

We welcome contributions! Please see our Contributing Guide for details.

License

MIT License

About

This workflow engine is developed and maintained by Adanomad Consulting and powers the workflow functionality in AceTeam.ai. For commercial support or consulting, please contact us at [e-mail].