Procedural Child Task Spawning

Hatchet supports the dynamic creation of child tasks during a parent task’s execution. This powerful feature enables:

Complex, reusable task hierarchies - Break down complex tasks into simpler, reusable components
Fan-out parallelism - Scale out to multiple parallel tasks dynamically
Dynamic task behavior - Create loops and conditional branches at runtime
Agent-based tasks - Support AI agents that can create new tasks based on analysis results or loop until a condition is met

Creating Parent and Child Tasks

To implement child task spawning, you first need to create both parent and child task definitions.

First, we’ll declare a couple of tasks for the parent and child:

class ParentInput(BaseModel):
    n: int = 100


class ChildInput(BaseModel):
    a: str


parent_wf = hatchet.workflow(name="FanoutParent", input_validator=ParentInput)
child_wf = hatchet.workflow(name="FanoutChild", input_validator=ChildInput)


@parent_wf.task(execution_timeout=timedelta(minutes=5))
async def spawn(input: ParentInput, ctx: Context) -> dict[str, Any]:
    print("spawning child")

    result = await child_wf.aio_run_many(
        [
            child_wf.create_bulk_run_item(
                input=ChildInput(a=str(i)),
                options=TriggerWorkflowOptions(
                    additional_metadata={"hello": "earth"}, key=f"child{i}"
                ),
            )
            for i in range(input.n)
        ]
    )

    print(f"results {result}")

    return {"results": result}

We also created a step on the parent task that spawns the child tasks. Now, we’ll add a couple of steps to the child task:

examples/fanout/worker.py

@child_wf.task()
def process(input: ChildInput, ctx: Context) -> dict[str, str]:
    print(f"child process {input.a}")
    return {"status": input.a}


@child_wf.task(parents=[process])
def process2(input: ChildInput, ctx: Context) -> dict[str, str]:
    process_output = ctx.task_output(process)
    a = process_output["status"]

    return {"status2": a + "2"}

And that’s it! The fanout parent will run and spawn the child, and then will collect the results from its steps.

import { hatchet } from "../hatchet-client";
 
// Child task definition
export const child = hatchet.task({
  name: "child",
  fn: (input) => {
    return {
      Value: input.N,
    };
  },
});
 
// Parent task definition
export const parent = hatchet.task({
  name: "parent",
  fn: async (input, ctx) => {
    const n = input.N;
    const promises = [];
 
    // Spawn multiple child tasks in parallel
    for (let i = 0; i < n; i++) {
      promises.push(ctx.runChild(child, { N: i }));
    }
 
    // Wait for all child tasks to complete
    const childRes = await Promise.all(promises);
 
    // Sum the results
    const sum = childRes.reduce((acc, curr) => acc + curr.Value, 0);
 
    return {
      Result: sum,
    };
  },
});

import (
	"sync"
 
	v1 "github.com/hatchet-dev/hatchet/pkg/v1"
	"github.com/hatchet-dev/hatchet/pkg/v1/task"
	"github.com/hatchet-dev/hatchet/pkg/v1/workflow"
	"github.com/hatchet-dev/hatchet/pkg/worker"
)
 
// Child task input and output types
type ChildInput struct {
	N int `json:"n"`
}
 
type ValueOutput struct {
	Value int `json:"value"`
}
 
type ChildResult struct {
	One ValueOutput `json:"one"`
}
 
// Child workflow definition
func Child(hatchet *v1.HatchetClient) workflow.WorkflowDeclaration[ChildInput, ChildResult] {
	child := v1.WorkflowFactory[ChildInput, ChildResult](
		workflow.CreateOpts[ChildInput]{
			Name: "child",
		},
		hatchet,
	)
 
	child.Task(
		task.CreateOpts[ChildInput]{
			Name: "one",
			Fn: func(input ChildInput, ctx worker.HatchetContext) (*ValueOutput, error) {
				return &ValueOutput{
					Value: input.N,
				}, nil
			},
		},
	)
 
	return child
}
 
// Parent task input and output types
type ParentInput struct {
	N int `json:"n"`
}
 
type SumOutput struct {
	Result int `json:"result"`
}
 
type ParentResult struct {
	Sum *SumOutput `json:"sum,omitempty"`
}
 
// Parent workflow definition
func Parent(hatchet *v1.HatchetClient) workflow.WorkflowDeclaration[ParentInput, ParentResult] {
	child := Child(hatchet)
	parent := v1.WorkflowFactory[ParentInput, ParentResult](
		workflow.CreateOpts[ParentInput]{
			Name: "parent",
		},
		hatchet,
	)
 
	parent.Task(
		task.CreateOpts[ParentInput]{
			Name: "sum",
			Fn: func(input ParentInput, ctx worker.HatchetContext) (*SumOutput, error) {
				sum := 0
 
				// Use a WaitGroup to coordinate parallel child workflows
				var wg sync.WaitGroup
				var mu sync.Mutex
				var firstErr error
 
				// Launch child workflows in parallel
				results := make([]*ChildResult, 0, input.N)
				wg.Add(input.N)
				for j := 0; j < input.N; j++ {
					go func(index int) {
						defer wg.Done()
 
						// Spawn and run a child workflow
						result, err := child.RunAsChild(ctx, ChildInput{N: 1})
 
						mu.Lock()
						defer mu.Unlock()
 
						if err != nil && firstErr == nil {
							firstErr = err
							return
						}
 
						if firstErr == nil {
							results = append(results, result)
						}
					}(j)
				}
 
				// Wait for all goroutines to complete
				wg.Wait()
 
				// Check if any errors occurred
				if firstErr != nil {
					return nil, firstErr
				}
 
				// Sum results from all children
				for _, result := range results {
					sum += result.One.Value
				}
 
				return &SumOutput{
					Result: sum,
				}, nil
			},
		},
	)
 
	return parent
}

Running Child Tasks

To spawn and run a child task from a parent task, use the appropriate method for your language:

# Inside a parent task
child_result = child_task.run(child_input)

// Inside a parent task
const childResult = await ctx.runChild(childTask, childInput);

// Inside a parent task
result, err := childWorkflow.RunAsChild(ctx, childInput)

Parallel Child Task Execution

As shown in the examples above, you can spawn multiple child tasks in parallel:

# Run multiple child workflows concurrently with asyncio
import asyncio
 
async def run_child_workflows(n: int) -> list[dict[str, Any]]:
	return await child.aio_run_many([
		child.create_bulk_run_item(
			options=TriggerWorkflowOptions(
				input=ChildInput(n=i),
			)
		)
		for i in range(n)
	])
 
# In your parent task
child_results = await run_child_workflows(input.n)

// Run multiple child tasks in parallel
const promises = [];
for (let i = 0; i < n; i++) {
  promises.push(ctx.runChild(child, { N: i }));
}
const childResults = await Promise.all(promises);

// Run multiple child tasks in parallel using goroutines
var wg sync.WaitGroup
var mu sync.Mutex
results := make([]*ChildResult, 0, n)
 
wg.Add(n)
for i := 0; i < n; i++ {
    go func(index int) {
        defer wg.Done()
        result, err := child.RunAsChild(ctx, ChildInput{N: index})
 
        if err == nil {
            mu.Lock()
            results = append(results, result)
            mu.Unlock()
        }
    }(i)
}
wg.Wait()

Use Cases for Child Workflows

Child workflows are ideal for:

Dynamic fan-out processing - When the number of parallel tasks is determined at runtime
Reusable workflow components - Create modular workflows that can be reused across different parent workflows
Resource-intensive operations - Spread computation across multiple workers
Agent-based systems - Allow AI agents to spawn new workflows based on their reasoning
Long-running operations - Break down long operations into smaller, trackable units of work

Error Handling with Child Workflows

When working with child workflows, it’s important to properly handle errors. Here are patterns for different languages:

try:
    child_result = child.run(ChildInput(a="foobar"))
except Exception as e:
    # Handle error from child workflow
    print(f"Child workflow failed: {e}")
    # Decide how to proceed - retry, skip, or fail the parent

try {
  const childResult = await ctx.runChild(child, { N: i });
} catch (error) {
  // Handle error from child workflow
  console.error(`Child workflow failed: ${error}`);
  // Decide how to proceed - retry, skip, or fail the parent
}

result, err := child.RunAsChild(ctx, ChildInput{N: i})
if err != nil {
    // Handle error from child workflow
    fmt.Printf("Child workflow failed: %v\n", err)
    // Decide how to proceed - retry, skip, or fail the parent
}