apflow
Task orchestration framework for AI applications
About
More than a traditional task list, apflow is a powerful orchestration framework that builds complex task trees with dependencies, priorities, and unified execution. Serves as a Modular AI Orchestration Layer that runs Standalone or Embedded, seamlessly integrating AI agent support. The core is pure orchestration with no LLM dependencies — CrewAI integration is optional. Features built-in task scheduling with support for cron expressions, intervals, daily/weekly/monthly triggers, and external scheduler integration. The framework includes an extensive executor ecosystem supporting HTTP/REST APIs, SSH remote execution, Docker containers, gRPC services, WebSocket communication, MCP integration, and LLM-based task tree generation.
Features
Workflow Examples
Visualize how tasks are organized in trees and how dependencies control execution order
Sequential Pipeline with Task Tree
Demonstrates both task tree organization (parent-child) and execution dependencies. The tree organizes tasks hierarchically, while dependencies control when tasks execute.
Install
pip install apflowQuick Start
from apflow import TaskManager, TaskTreeNode, create_session
# Create database session and task manager
db = create_session()
task_manager = TaskManager(db)
# Create a task using built-in executor
task = await task_manager.task_repository.create_task(
name="system_info_executor", # Built-in executor ID
user_id="user_123",
priority=2,
inputs={"resource": "cpu"} # Get CPU information
)
# Build task tree and execute
task_tree = TaskTreeNode(task)
await task_manager.distribute_task_tree(task_tree)
# Get result
result = await task_manager.task_repository.get_task_by_id(task.id)
print(f"Status: {result.status}")
print(f"Result: {result.result}")