Skip to main content
This deterministic example shows one Launchpad workflow calling another workflow while reusing the same TaskContext. It is useful when a larger workflow needs to delegate a focused sub-process without losing the parent workflow’s accumulated state. The example ships as NestedWorkflow in app/launchpad/workflows/examples/nested_workflow/ and is registered as WorkflowRegistry.NESTED_WORKFLOW.

​
What it demonstrates

  • Parent workflow composition with await ReplyDraftWorkflow().run_async(context=task_context)
  • A child workflow that writes output into the same TaskContext
  • A runnable example that does not require LLM, tracing, database, or vector-search credentials
  • The Workflow.run_async(context=...) path added for workflow composition

​
Workflow graph

​
Parent workflow

from launchpad.core.schema import NodeConfig, WorkflowSchema
from launchpad.core.workflow import Workflow
from launchpad.workflows.examples.nested_workflow.nodes.run_reply_workflow_node import (
    RunReplyWorkflowNode,
)
from launchpad.workflows.examples.nested_workflow.nodes.send_reply_node import (
    SendReplyNode,
)
from launchpad.workflows.examples.nested_workflow.schema import (
    NestedWorkflowEventSchema,
)


class NestedWorkflow(Workflow):
    workflow_schema = WorkflowSchema(
        description="Parent workflow that delegates reply drafting to a child workflow.",
        event_schema=NestedWorkflowEventSchema,
        start=RunReplyWorkflowNode,
        nodes=[
            NodeConfig(
                node=RunReplyWorkflowNode,
                connections=[SendReplyNode],
                description="Calls ReplyDraftWorkflow with the shared context.",
            ),
        ],
    )

​
Delegating to the child workflow

The parent node passes the current task context into the child workflow. Because the context is shared, DraftReplyNode can save its output and SendReplyNode can read it later in the parent flow. 
class RunReplyWorkflowNode(Node):
    class OutputType(BaseModel):
        delegated: bool
        workflow: str

    async def process(self, task_context: TaskContext) -> TaskContext:
        await ReplyDraftWorkflow().run_async(context=task_context)
        self.save_output(self.OutputType(delegated=True, workflow="ReplyDraftWorkflow"))
        return task_context

​
Run it

uv run playground/nested_workflow.py
The script loads app/launchpad/workflows/examples/nested_workflow/request_examples/billing_question.json, runs WorkflowRegistry.NESTED_WORKFLOW.value(), and prints the final TaskContext.