PipelineContext: Runtime Context for Tasks

PipelineContext is a typed dataclass that the pipeline framework automatically builds and injects into any task that declares a ctx parameter. It carries the user, dataset, and per-item context for the current pipeline run, and provides an extras dict for custom state.

Fields

Field	Type	Description
`user`	`Any`	The user that triggered the pipeline. Used for access control and provenance.
`dataset`	`Any`	The resolved dataset object for the current run.
`data_item`	`Any`	The individual data item being processed in this pipeline execution.
`pipeline_name`	`Optional[str]`	The name passed to `run_pipeline` or `run_tasks`.
`extras`	`Dict[str, Any]`	Arbitrary key/value state you can pass into the pipeline and read in any task. Defaults to `{}`.

How injection works

The framework inspects every task function’s signature at construction time. If it finds a parameter named ctx, it passes the current PipelineContext as that argument when the task runs. Matching is by parameter name, not by type annotation.

from cognee.modules.pipelines.models.PipelineContext import PipelineContext

# ctx is injected automatically — no manual wiring needed
async def my_task(data, ctx: PipelineContext = None):
    if ctx:
        print(ctx.user)
        print(ctx.dataset)
        print(ctx.pipeline_name)

Tasks that do not declare ctx simply receive no context and are unaffected.

Using `extras` for custom pipeline state

Pass a dict as the context argument to run_pipeline (or extras to run_tasks). Every task in the pipeline can read those values from ctx.extras.

from cognee.modules.pipelines import Task, run_pipeline
from cognee.modules.pipelines.models.PipelineContext import PipelineContext

async def score_items(data, ctx: PipelineContext = None):
    multiplier = ctx.extras.get("score_multiplier", 1) if ctx else 1
    return [item * multiplier for item in data]

async for _ in run_pipeline(
    tasks=[Task(score_items)],
    data=[1, 2, 3],
    datasets=["my_dataset"],
    pipeline_name="scoring_pipeline",
    context={"score_multiplier": 10},   # becomes ctx.extras
):
    pass

ctx.extras is always a plain dict — it is never None.

Examples and details

Accessing user and dataset in a task

The user and dataset fields are most useful when you need to write provenance records or apply per-tenant logic:

async def store_result(data_points, ctx: PipelineContext = None):
    user    = ctx.user    if ctx else None
    dataset = ctx.dataset if ctx else None
    data_item = ctx.data_item if ctx else None

    for dp in data_points:
        await write_to_store(dp, user_id=user.id, dataset_id=dataset.id)
    return data_points

The built-in add_data_points task already does this automatically, so you typically only need to read these fields when writing your own storage tasks.

Extras persist across chained tasks

All tasks in the same pipeline run share the same PipelineContext object, so values written during construction remain available in every downstream task:

async def filter_task(data, ctx: PipelineContext = None):
    threshold = ctx.extras.get("min_threshold", 0) if ctx else 0
    return [x for x in data if x >= threshold]

async def label_task(data, ctx: PipelineContext = None):
    prefix = ctx.extras.get("label_prefix", "") if ctx else ""
    return [f"{prefix}{x}" for x in data]

# Both tasks receive the same extras
async for _ in run_pipeline(
    tasks=[Task(filter_task), Task(label_task)],
    data=[1, 5, 10],
    datasets=["demo"],
    context={"min_threshold": 4, "label_prefix": "item_"},
):
    pass

Making ctx optional

Always default ctx to None so the task can also be called directly in tests or scripts without a running pipeline:

async def my_task(data, ctx: PipelineContext = None):
    name = ctx.pipeline_name if ctx else "standalone"
    ...

Dataset routing: which dataset does add_data_points write to?

The built-in add_data_points discovers which dataset to attribute nodes and edges to entirely from ctx.dataset:

async def add_data_points(data_points, ..., ctx: PipelineContext = None):
    user      = ctx.user      if ctx else None
    data_item = ctx.data_item if ctx else None
    dataset   = ctx.dataset   if ctx else None

    if user and dataset and data_item:
        # writes dataset-level provenance (dataset_id, data_id)
        # alongside the graph and vector writes

When you call run_pipeline(..., datasets=["my_dataset"]), the pipeline resolves each name (or UUID) into a Dataset object and places it on ctx.dataset, so every task — including add_data_points — receives the resolved dataset automatically. You never pass the dataset to the task itself.Called outside a pipeline (await add_data_points(points) with ctx=None), user, dataset, and data_item are all None, so the provenance block is skipped: the nodes and edges are still written to the graph and vector stores, but no dataset-level provenance is recorded. Run the task through run_pipeline/run_tasks with a datasets=[...] argument whenever you need per-dataset attribution.

Tasks

Learn how tasks are defined and composed

Pipelines

See how tasks and context flow through pipeline runs

Custom Tasks & Pipelines

Step-by-step guide to building your own pipeline

Getting Started

Core Concepts

Setup Configuration

Guides

Examples

CLI

Rust SDK

TypeScript SDK

OSS

PipelineContext

PipelineContext: Runtime Context for Tasks

Fields

How injection works

Using `extras` for custom pipeline state

Examples and details

Tasks

Pipelines

Custom Tasks & Pipelines

​PipelineContext: Runtime Context for Tasks

​Fields

​How injection works

​Using extras for custom pipeline state

​Examples and details

Tasks

Pipelines

Custom Tasks & Pipelines

PipelineContext: Runtime Context for Tasks

Fields

How injection works

Using `extras` for custom pipeline state

Examples and details