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

# Task Context

> Learn how the TaskContext provides stateful data management throughout workflow execution, enabling seamless data sharing between nodes.

<Info>
  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.
</Info>

## TaskContext Class

The `TaskContext` class serves as the central data container for workflow execution:

```python theme={null}
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:**

```python theme={null}
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:**

```python theme={null}
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.

```python theme={null}
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