Skip to main content

Documentation Index

Fetch the complete documentation index at: https://launchpad.datalumina.com/llms.txt

Use this file to discover all available pages before exploring further.

The task context is a stateful Pydantic model used throughout the workflow. It provides a single reference point accessible from any node, so relevant data can be stored and retrieved as needed.

​
TaskContext Class

The TaskContext class serves as the central data container for workflow execution: 
class TaskContext(BaseModel):
    event: Any
    nodes: Dict[str, Any] = Field(default_factory=dict)
    metadata: Dict[str, Any] = Field(default_factory=dict)
    should_stop: bool = Field(default=False)
    trace_id: str | None = Field(default=None)

    def update_node(self, node_name: str, **kwargs):
        self.nodes[node_name] = {**self.nodes.get(node_name, {}), **kwargs}

    def stop_workflow(self) -> None:
        self.should_stop = True
TaskContext attributes:
  • event — the original triggering event, parsed against the workflow’s event_schema.
  • nodes — results from each node’s execution, keyed by class name.
  • metadata — workflow-level metadata (the workflow orchestrator stores the node registry here under metadata["nodes"]).
  • should_stop — set to True via stop_workflow() to halt execution cleanly after the current node finishes.
  • trace_id — Langfuse trace ID captured on entry when enable_tracing=True; useful when you want to log or surface the trace URL back to the caller.

​
The Event Attribute

The event attribute in TaskContext serves as the main entry point for workflow input data. When a workflow is initialized, the provided event is parsed according to the event_schema specified within the WorkflowSchema. This mechanism supports many event formats because each workflow defines its own event schema. You can run multiple workflows for different inputs without changing the shared workflow infrastructure. Example WorkflowSchema: 
class ExampleWorkflow(Workflow):
    workflow_schema = WorkflowSchema(
        description="",
        event_schema=ExampleEventSchema,
        start=InitialNode,
        nodes=[
            NodeConfig(
                node=InitialNode,
                connections=[],
                description="",
                concurrent_nodes=[],
            ),
        ],
    )

​
Type Hinting

By default, the event attribute of TaskContext has the type Any, which means you won’t get autocomplete or type checking when accessing its fields or methods. However, since each workflow defines its own event_schema, you already know the expected structure of event within that workflow. To benefit from IDE features like autocomplete and static type checking, explicitly type the event attribute when retrieving it from the TaskContext. This makes your code more readable and helps catch errors earlier. Example: 
event: ExampleEventSchema = task_context.event
Implementation steps:
  1. Define Event Schema - Create a Pydantic model for your event structure
  2. Configure Workflow - Set the event_schema in your WorkflowSchema
  3. Type the Event - Cast the event to your schema type in node processing
  4. Enjoy Type Safety - Get full IDE support and compile-time error checking

​
Stopping a workflow mid-flight

Any node can halt the run after it returns by calling task_context.stop_workflow(). The orchestrator checks should_stop between nodes and exits the loop cleanly, preserving everything saved so far. Typical uses: a guardrail node detects a prompt injection, or a router decides there is nothing left to do. 
class GuardrailNode(Node):
    async def process(self, task_context: TaskContext) -> TaskContext:
        if self.is_blocked(task_context.event):
            task_context.stop_workflow()
        return task_context

​
Best practices

  • Event schema design: Use specific, well-described field names
  • Node result storage: Store results under the node class name for consistency
  • Metadata usage: Keep workflow-level configuration in metadata
  • Type safety: Cast the event to your schema type in node processing