feat: implement top 5 improvements from AP+Flow evaluation

1. task.create_tree MCP tool — AI agents can create entire task trees
   in a single call (TaskCreateTreeModule in task_modules.py)

2. Distributed wired into CLI — apflow serve --cluster, apflow worker
   commands. DistributedRuntime initialized in create_app(cluster=True)

3. End-to-end example — examples/quickstart.py shows full workflow:
   register executors → create tree → execute → get results

4. DAG support documented — README explains parent_id (tree) vs
   dependencies (fan-in DAG pattern) with examples

5. Plugin discovery via entry points — [project.entry-points."apflow.executors"]
   allows third-party packages to register executors without import ordering

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

Author: tercel

README.md

@@ -62,6 +62,8 @@ serve(app.registry, name="apflow")
 # Or from the command line
 apflow serve              # A2A HTTP server
 apflow serve --explorer   # With Explorer UI
+apflow serve --cluster    # Distributed cluster mode
+apflow worker --db ...    # Start worker node
 apflow mcp                # MCP server (for Claude/Cursor)
 apflow info               # Show registered modules
 ```
@@ -70,14 +72,25 @@ apflow info               # Show registered modules
 
 ### Task Orchestration
 
-Dependency graph execution with priority scheduling, parallel execution, and result aggregation.
+Dependency graph execution with priority scheduling, parallel execution, and result aggregation. Supports both tree structure (`parent_id`) and DAG patterns (`dependencies` for fan-in).
 
 ```python
+# Tree: sequential pipeline
 tasks = [
     {"id": "fetch", "name": "Fetch Data", "priority": 1},
     {"id": "process", "name": "Process", "parent_id": "fetch", "priority": 2},
     {"id": "notify", "name": "Notify", "parent_id": "process", "priority": 3},
 ]
+
+# DAG: fan-in pattern (task depends on multiple predecessors)
+tasks = [
+    {"id": "a", "name": "Step A", "priority": 1},
+    {"id": "b", "name": "Step B", "priority": 1},
+    {"id": "merge", "name": "Merge Results", "priority": 2,
+     "parent_id": "a",
+     "dependencies": [{"id": "a", "required": True}, {"id": "b", "required": True}]},
+]
+
 tree = await task_creator.create_task_tree_from_array(tasks)
 await task_manager.distribute_task_tree(tree)
 ```

examples/quickstart.py

@@ -0,0 +1,100 @@
+"""
+apflow Quickstart — End-to-End Example
+
+This example shows the complete workflow:
+1. Register custom executors
+2. Create a task tree
+3. Execute it
+4. Get results
+
+Run: python examples/quickstart.py
+"""
+
+import asyncio
+from apflow.adapters.function_executor import function_executor
+from apflow.app import create_app
+
+
+# Step 1: Register custom executors using @function_executor
+@function_executor(
+    id="fetch_price",
+    description="Fetch the price of a product",
+    input_schema={
+        "type": "object",
+        "properties": {"product": {"type": "string"}},
+        "required": ["product"],
+    },
+)
+async def fetch_price(inputs: dict) -> dict:
+    """Simulate fetching a product price."""
+    prices = {"widget": 9.99, "gadget": 19.99, "doohickey": 4.99}
+    product = inputs["product"]
+    return {"product": product, "price": prices.get(product, 0.0)}
+
+
+@function_executor(
+    id="calculate_total",
+    description="Calculate total from fetched prices",
+)
+async def calculate_total(inputs: dict) -> dict:
+    """Aggregate results from dependency tasks."""
+    total = sum(
+        v.get("price", 0) for k, v in inputs.items() if isinstance(v, dict) and "price" in v
+    )
+    return {"total": round(total, 2), "item_count": len(inputs)}
+
+
+async def main():
+    # Step 2: Create the app (bootstraps session, TaskManager, Registry)
+    app = create_app(connection_string="sqlite:///:memory:")
+
+    print(f"Registered modules: {len(list(app.registry.list()))}")
+    for m in sorted(app.registry.list()):
+        print(f"  {m}")
+
+    # Step 3: Create a task tree
+    #   fetch_widget ──┐
+    #                   ├──→ calculate_total
+    #   fetch_gadget ──┘
+    tasks = [
+        {
+            "id": "fetch_widget",
+            "name": "Fetch Widget Price",
+            "priority": 1,
+            "inputs": {"product": "widget"},
+            "params": {"executor_id": "fetch_price"},
+        },
+        {
+            "id": "fetch_gadget",
+            "name": "Fetch Gadget Price",
+            "priority": 1,
+            "inputs": {"product": "gadget"},
+            "params": {"executor_id": "fetch_price"},
+        },
+        {
+            "id": "total",
+            "name": "Calculate Total",
+            "priority": 2,
+            "parent_id": "fetch_widget",
+            "dependencies": [
+                {"id": "fetch_widget", "required": True},
+                {"id": "fetch_gadget", "required": True},
+            ],
+            "params": {"executor_id": "calculate_total"},
+        },
+    ]
+
+    tree = await app.task_creator.create_task_tree_from_array(tasks)
+    print(f"\nTask tree created: root={tree.task.id}")
+
+    # Step 4: Execute
+    await app.task_manager.distribute_task_tree(tree)
+
+    # Step 5: Get results
+    for task_id in ["fetch_widget", "fetch_gadget", "total"]:
+        task = await app.task_repository.get_task_by_id(task_id)
+        print(f"  {task.name}: status={task.status}, result={task.result}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

pyproject.toml

@@ -43,6 +43,10 @@ dependencies = [
 [project.scripts]
 apflow = "apflow.cli:main"
 
+[project.entry-points."apflow.executors"]
+# Third-party packages register executors here:
+# my_executor = "my_package.executors:register"
+
 # Optional dependencies
 [project.optional-dependencies]
 # Documentation (MkDocs with Material theme)

src/apflow/app.py

@@ -45,6 +45,7 @@ def __init__(
 def create_app(
     connection_string: Optional[str] = None,
     namespace: str = "apflow",
+    cluster: bool = False,
 ) -> ApflowApp:
     """Create and initialize the full apflow application stack.
 
@@ -114,6 +115,18 @@ def create_app(
     )
     logger.info(f"apcore Registry populated ({len(list(registry.list()))} modules)")
 
+    # Start distributed runtime if cluster mode
+    if cluster:
+        try:
+            from apflow.core.distributed.config import DistributedConfig
+            from apflow.core.distributed.runtime import DistributedRuntime
+
+            dist_config = DistributedConfig.from_env()
+            DistributedRuntime(dist_config)  # Starts background tasks
+            logger.info(f"Distributed runtime enabled (node: {dist_config.node_id})")
+        except Exception as e:
+            logger.warning(f"Distributed runtime failed to initialize: {e}")
+
     return ApflowApp(
         session=session,
         task_manager=task_manager,

src/apflow/bridge/registry_setup.py

@@ -24,6 +24,7 @@
     TaskCreateModule,
     TaskDeleteModule,
     TaskExecuteModule,
+    TaskCreateTreeModule,
     TaskGetModule,
     TaskListModule,
 )
@@ -76,6 +77,7 @@ def create_apflow_registry(
     # 2. Register task management modules
     task_modules = {
         "task.create": TaskCreateModule(task_creator, task_repository),
+        "task.create_tree": TaskCreateTreeModule(task_creator, task_repository),
         "task.execute": TaskExecuteModule(task_manager),
         "task.list": TaskListModule(task_repository),
         "task.get": TaskGetModule(task_repository),

src/apflow/bridge/scanner_bridge.py

@@ -49,6 +49,28 @@ def discover_executor_modules() -> list[ExecutableTaskModuleAdapter]:
     except ImportError:
         pass
 
+    # Discover executors from entry points (third-party packages)
+    try:
+        from importlib.metadata import entry_points
+
+        eps = entry_points(group="apflow.executors")
+        seen_ids = {a.executor_id for a in adapters}
+        for ep in eps:
+            try:
+                register_fn = ep.load()
+                register_fn()  # Expected to call @function_executor or @executor_register
+                # Re-scan function executors after registration
+                for eid, ecls in get_function_executor_classes().items():
+                    if eid not in seen_ids:
+                        adapter = _create_adapter_from_class(eid, ecls)
+                        if adapter is not None:
+                            adapters.append(adapter)
+                            seen_ids.add(eid)
+            except Exception as e:
+                logger.warning(f"Failed to load executor entry point '{ep.name}': {e}")
+    except Exception:
+        pass
+
     logger.info(f"Discovered {len(adapters)} executor modules for apcore registration")
     return adapters
 

src/apflow/bridge/task_modules.py

@@ -244,3 +244,98 @@ async def execute(self, inputs: dict[str, Any], context: Any = None) -> dict[str
 
         await self._repo.delete_task(task_id)
         return {"task_id": task_id, "deleted": True}
+
+
+_TASK_CREATE_TREE_INPUT = {
+    "type": "object",
+    "properties": {
+        "tasks": {
+            "type": "array",
+            "minItems": 1,
+            "description": "Array of task definitions. Use parent_id to build tree, dependencies for DAG ordering.",
+            "items": {
+                "type": "object",
+                "properties": {
+                    "id": {"type": "string", "description": "Optional task ID (auto-generated if omitted)"},
+                    "name": {"type": "string", "minLength": 1, "description": "Task name (required)"},
+                    "parent_id": {"type": "string", "description": "Parent task ID for tree structure"},
+                    "priority": {
+                        "type": "integer",
+                        "minimum": 0,
+                        "maximum": 3,
+                        "default": 2,
+                        "description": "0=urgent, 1=high, 2=normal, 3=low",
+                    },
+                    "inputs": {"type": "object", "description": "Task input parameters"},
+                    "params": {"type": "object", "description": "Executor init parameters"},
+                    "dependencies": {
+                        "type": "array",
+                        "items": {"type": "object"},
+                        "description": "Task dependencies for DAG ordering [{id: 'task_id', required: true}]",
+                    },
+                    "token_budget": {"type": "integer", "minimum": 0},
+                    "cost_policy": {"type": "string"},
+                    "max_attempts": {"type": "integer", "minimum": 1, "maximum": 100, "default": 3},
+                },
+                "required": ["name"],
+            },
+        },
+    },
+    "required": ["tasks"],
+}
+
+_TASK_CREATE_TREE_OUTPUT = {
+    "type": "object",
+    "properties": {
+        "root_task_id": {"type": "string", "description": "ID of the first root task"},
+        "task_count": {"type": "integer", "description": "Total tasks created"},
+        "task_ids": {
+            "type": "array",
+            "items": {"type": "string"},
+            "description": "All created task IDs",
+        },
+    },
+}
+
+
+class TaskCreateTreeModule:
+    """Create a complete task tree from an array of task definitions in one call."""
+
+    description = (
+        "Create a multi-step task workflow from an array of task definitions. "
+        "Use parent_id for tree structure and dependencies for execution ordering. "
+        "Tasks without parent_id are root tasks. Multiple roots are allowed."
+    )
+
+    def __init__(self, task_creator: Any, task_repository: Any) -> None:
+        self._creator = task_creator
+        self._repo = task_repository
+        self.input_schema = _make_schema(_TASK_CREATE_TREE_INPUT)
+        self.output_schema = _make_schema(_TASK_CREATE_TREE_OUTPUT)
+
+    async def execute(self, inputs: dict[str, Any], context: Any = None) -> dict[str, Any]:
+        tasks = inputs.get("tasks", [])
+        if not tasks:
+            raise ValueError("tasks array must be non-empty")
+
+        for t in tasks:
+            if not t.get("name"):
+                raise ValueError("Each task must have a non-empty 'name'")
+
+        tree = await self._creator.create_task_tree_from_array(tasks)
+
+        # Collect all task IDs from the tree
+        task_ids: list[str] = []
+
+        def _collect_ids(node: Any) -> None:
+            task_ids.append(node.task.id)
+            for child in node.children:
+                _collect_ids(child)
+
+        _collect_ids(tree)
+
+        return {
+            "root_task_id": tree.task.id,
+            "task_count": len(task_ids),
+            "task_ids": task_ids,
+        }

src/apflow/cli.py

@@ -32,6 +32,9 @@ def cli() -> None:
 @click.option("--metrics", is_flag=True, help="Enable /metrics endpoint")
 @click.option("--cors", default=None, help="CORS origins (comma-separated)")
 @click.option("--db", default=None, help="Database connection string")
+@click.option(
+    "--cluster", is_flag=True, help="Enable distributed cluster mode (requires PostgreSQL)"
+)
 @click.option("--log-level", default=None, help="Log level (DEBUG/INFO/WARNING/ERROR)")
 def serve(
     host: str,
@@ -41,12 +44,13 @@ def serve(
     metrics: bool,
     cors: Optional[str],
     db: Optional[str],
+    cluster: bool,
     log_level: Optional[str],
 ) -> None:
     """Start A2A HTTP server (internal network service)."""
     from apflow.app import create_app
 
-    app = create_app(connection_string=db)
+    app = create_app(connection_string=db, cluster=cluster)
 
     cors_origins = [s.strip() for s in cors.split(",")] if cors else None
 
@@ -148,6 +152,37 @@ def info() -> None:
         click.echo(f"Registry: error ({e})")
 
 
+@cli.command()
+@click.option("--db", required=True, help="PostgreSQL connection string (required for cluster)")
+@click.option("--node-id", default=None, help="Worker node ID (auto-generated if omitted)")
+@click.option("--log-level", default=None, help="Log level")
+def worker(db: str, node_id: Optional[str], log_level: Optional[str]) -> None:
+    """Start a distributed worker node (requires PostgreSQL)."""
+    import asyncio
+
+    click.echo(f"Starting worker node: {node_id or 'auto'}")
+
+    from apflow.app import create_app
+
+    create_app(connection_string=db, cluster=True)
+
+    try:
+        from apflow.core.distributed.config import DistributedConfig
+        from apflow.core.distributed.worker import WorkerRuntime
+
+        config = DistributedConfig.from_env()
+        if node_id:
+            config.node_id = node_id
+
+        worker_rt = WorkerRuntime(config)
+        click.echo(f"Worker {config.node_id} running (Ctrl+C to stop)")
+        asyncio.run(worker_rt.start())
+    except ImportError:
+        click.echo("Error: distributed module not available", err=True)
+    except KeyboardInterrupt:
+        click.echo("Worker stopped")
+
+
 def main() -> None:
     """Entry point for apflow CLI."""
     cli()