Procedural Child Workflow Spawning

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

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

Creating Parent and Child Workflows

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

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

examples/fanout/worker.py

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 task on the parent workflow that spawns the child workflows. Now, we’ll add a couple of tasks to the child workflow:

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 tasks.

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

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

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

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

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

Parallel Child Workflow Execution

As shown in the examples above, you can spawn multiple child workflows 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 workflows 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 workflows 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
}