---
title: context
sidebarTitle: context
---

# `prefect.context`



Async and thread safe models for passing runtime context data.

These contexts should never be directly mutated by the user.

For more user-accessible information about the current run, see [`prefect.runtime`](https://docs.prefect.io/v3/api-ref/python/prefect-runtime-flow_run).


## Functions

### `serialize_context` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L63" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
serialize_context(asset_ctx_kwargs: Union[dict[str, Any], None] = None) -> dict[str, Any]
```


Serialize the current context for use in a remote execution environment.

Optionally provide asset_ctx_kwargs to create new AssetContext, that will be used
in the remote execution environment. This is useful for TaskRunners, who rely on creating the
task run in the remote environment.


### `hydrated_context` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L98" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
hydrated_context(serialized_context: Optional[dict[str, Any]] = None, client: Union[PrefectClient, SyncPrefectClient, None] = None) -> Generator[None, Any, None]
```

### `get_run_context` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L768" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
get_run_context() -> Union[FlowRunContext, TaskRunContext]
```


Get the current run context from within a task or flow function.

**Returns:**
- A `FlowRunContext` or `TaskRunContext` depending on the function type.

**Raises:**
- `RuntimeError`: If called outside of a flow or task run.


### `get_settings_context` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L804" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
get_settings_context() -> SettingsContext
```


Get the current settings context which contains profile information and the
settings that are being used.

Generally, the settings that are being used are a combination of values from the
profile and environment. See `prefect.context.use_profile` for more details.


### `tags` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L821" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
tags(*new_tags: str) -> Generator[set[str], None, None]
```


Context manager to add tags to flow and task run calls.

Tags are always combined with any existing tags.

**Examples:**

```python
from prefect import tags, task, flow
@task
def my_task():
    pass
```

Run a task with tags

```python
@flow
def my_flow():
    with tags("a", "b"):
        my_task()  # has tags: a, b
```

Run a flow with tags

```python
@flow
def my_flow():
    pass
with tags("a", "b"):
    my_flow()  # has tags: a, b
```

Run a task with nested tag contexts

```python
@flow
def my_flow():
    with tags("a", "b"):
        with tags("c", "d"):
            my_task()  # has tags: a, b, c, d
        my_task()  # has tags: a, b
```

Inspect the current tags

```python
@flow
def my_flow():
    with tags("c", "d"):
        with tags("e", "f") as current_tags:
             print(current_tags)
with tags("a", "b"):
    my_flow()
# {"a", "b", "c", "d", "e", "f"}
```


### `use_profile` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L888" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
use_profile(profile: Union[Profile, str], override_environment_variables: bool = False, include_current_context: bool = True) -> Generator[SettingsContext, Any, None]
```


Switch to a profile for the duration of this context.

Profile contexts are confined to an async context in a single thread.

**Args:**
- `profile`: The name of the profile to load or an instance of a Profile.
- `override_environment_variable`: If set, variables in the profile will take
precedence over current environment variables. By default, environment
variables will override profile settings.
- `include_current_context`: If set, the new settings will be constructed
with the current settings context as a base. If not set, the use_base settings
will be loaded from the environment and defaults.


### `root_settings_context` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L940" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
root_settings_context() -> SettingsContext
```


Return the settings context that will exist as the root context for the module.

The profile to use is determined with the following precedence
- Command line via 'prefect --profile &lt;name&gt;'
- Environment variable via 'PREFECT_PROFILE'
- Profiles file via the 'active' key


## Classes

### `ContextModel` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L159" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>


A base model for context data that forbids mutation and extra data while providing
a context manager


**Methods:**

#### `get` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L194" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
get(cls: type[Self]) -> Optional[Self]
```

Get the current context instance


#### `model_copy` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L198" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
model_copy(self: Self) -> Self
```

Duplicate the context model, optionally choosing which fields to include, exclude, or change.

**Attributes:**
- `include`: Fields to include in new model.
- `exclude`: Fields to exclude from new model, as with values this takes precedence over include.
- `update`: Values to change/add in the new model. Note\: the data is not validated before creating
the new model - you should trust this data.
- `deep`: Set to `True` to make a deep copy of the model.

**Returns:**
- A new model instance.


#### `serialize` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L219" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
serialize(self, include_secrets: bool = True) -> dict[str, Any]
```

Serialize the context model to a dictionary that can be pickled with cloudpickle.


### `SyncClientContext` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L228" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>


A context for managing the sync Prefect client instances.

Clients were formerly tracked on the TaskRunContext and FlowRunContext, but
having two separate places and the addition of both sync and async clients
made it difficult to manage. This context is intended to be the single
source for sync clients.

The client creates a sync client, which can either be read directly from
the context object OR loaded with get_client, inject_client, or other
Prefect utilities.

with SyncClientContext.get_or_create() as ctx:
    c1 = get_client(sync_client=True)
    c2 = get_client(sync_client=True)
    assert c1 is c2
    assert c1 is ctx.client


**Methods:**

#### `get` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L194" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
get(cls: type[Self]) -> Optional[Self]
```

Get the current context instance


#### `get_or_create` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L277" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
get_or_create(cls) -> Generator[Self, None, None]
```

#### `model_copy` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L198" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
model_copy(self: Self) -> Self
```

Duplicate the context model, optionally choosing which fields to include, exclude, or change.

**Attributes:**
- `include`: Fields to include in new model.
- `exclude`: Fields to exclude from new model, as with values this takes precedence over include.
- `update`: Values to change/add in the new model. Note\: the data is not validated before creating
the new model - you should trust this data.
- `deep`: Set to `True` to make a deep copy of the model.

**Returns:**
- A new model instance.


#### `serialize` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L219" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
serialize(self, include_secrets: bool = True) -> dict[str, Any]
```

Serialize the context model to a dictionary that can be pickled with cloudpickle.


### `AsyncClientContext` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L286" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>


A context for managing the async Prefect client instances.

Clients were formerly tracked on the TaskRunContext and FlowRunContext, but
having two separate places and the addition of both sync and async clients
made it difficult to manage. This context is intended to be the single
source for async clients.

The client creates an async client, which can either be read directly from
the context object OR loaded with get_client, inject_client, or other
Prefect utilities.

with AsyncClientContext.get_or_create() as ctx:
    c1 = get_client(sync_client=False)
    c2 = get_client(sync_client=False)
    assert c1 is c2
    assert c1 is ctx.client


**Methods:**

#### `get` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L194" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
get(cls: type[Self]) -> Optional[Self]
```

Get the current context instance


#### `get_or_create` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L335" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
get_or_create(cls) -> AsyncGenerator[Self, None]
```

#### `model_copy` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L198" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
model_copy(self: Self) -> Self
```

Duplicate the context model, optionally choosing which fields to include, exclude, or change.

**Attributes:**
- `include`: Fields to include in new model.
- `exclude`: Fields to exclude from new model, as with values this takes precedence over include.
- `update`: Values to change/add in the new model. Note\: the data is not validated before creating
the new model - you should trust this data.
- `deep`: Set to `True` to make a deep copy of the model.

**Returns:**
- A new model instance.


#### `serialize` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L219" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
serialize(self, include_secrets: bool = True) -> dict[str, Any]
```

Serialize the context model to a dictionary that can be pickled with cloudpickle.


### `RunContext` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L344" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>


The base context for a flow or task run. Data in this context will always be
available when `get_run_context` is called.

**Attributes:**
- `start_time`: The time the run context was entered
- `client`: The Prefect client instance being used for API communication


**Methods:**

#### `get` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L194" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
get(cls: type[Self]) -> Optional[Self]
```

Get the current context instance


#### `model_copy` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L198" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
model_copy(self: Self) -> Self
```

Duplicate the context model, optionally choosing which fields to include, exclude, or change.

**Attributes:**
- `include`: Fields to include in new model.
- `exclude`: Fields to exclude from new model, as with values this takes precedence over include.
- `update`: Values to change/add in the new model. Note\: the data is not validated before creating
the new model - you should trust this data.
- `deep`: Set to `True` to make a deep copy of the model.

**Returns:**
- A new model instance.


#### `serialize` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L365" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
serialize(self: Self, include_secrets: bool = True) -> dict[str, Any]
```

#### `serialize` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L219" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
serialize(self, include_secrets: bool = True) -> dict[str, Any]
```

Serialize the context model to a dictionary that can be pickled with cloudpickle.


### `EngineContext` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L373" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>


The context for a flow run. Data in this context is only available from within a
flow run function.

**Attributes:**
- `flow`: The flow instance associated with the run
- `flow_run`: The API metadata for the flow run
- `task_runner`: The task runner instance being used for the flow run
- `run_results`: A mapping of result ids to run states for this flow run
- `log_prints`: Whether to log print statements from the flow run
- `parameters`: The parameters passed to the flow run
- `detached`: Flag indicating if context has been serialized and sent to remote infrastructure
- `result_store`: The result store used to persist results
- `persist_result`: Whether to persist the flow run result
- `task_run_dynamic_keys`: Counter for task calls allowing unique keys
- `observed_flow_pauses`: Counter for flow pauses
- `events`: Events worker to emit events


**Methods:**

#### `serialize` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L427" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
serialize(self: Self, include_secrets: bool = True) -> dict[str, Any]
```

#### `serialize` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L365" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
serialize(self: Self, include_secrets: bool = True) -> dict[str, Any]
```

### `TaskRunContext` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L453" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>


The context for a task run. Data in this context is only available from within a
task run function.

**Attributes:**
- `task`: The task instance associated with the task run
- `task_run`: The API metadata for this task run


**Methods:**

#### `serialize` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L474" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
serialize(self: Self, include_secrets: bool = True) -> dict[str, Any]
```

#### `serialize` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L365" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
serialize(self: Self, include_secrets: bool = True) -> dict[str, Any]
```

### `AssetContext` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L497" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>


The asset context for a materializing task run. Contains all asset-related information needed
for asset event emission and downstream asset dependency propagation.

**Attributes:**
- `direct_asset_dependencies`: Assets that this task directly depends on (from task.asset_deps)
- `downstream_assets`: Assets that this task will create/materialize (from MaterializingTask.assets)
- `upstream_assets`: Assets from upstream task dependencies
- `materialized_by`: Tool that materialized the assets (from MaterializingTask.materialized_by)
- `task_run_id`: ID of the associated task run
- `materialization_metadata`: Metadata for materialized assets


**Methods:**

#### `add_asset_metadata` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L578" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
add_asset_metadata(self, asset_key: str, metadata: dict[str, Any]) -> None
```

Add metadata for a materialized asset.

**Args:**
- `asset_key`: The asset key
- `metadata`: Metadata dictionary to add

**Raises:**
- `ValueError`: If asset_key is not in downstream_assets


#### `asset_as_related` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L621" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
asset_as_related(asset: Asset) -> dict[str, str]
```

Convert Asset to event related format.


#### `asset_as_resource` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L599" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
asset_as_resource(asset: Asset) -> dict[str, str]
```

Convert Asset to event resource format.


#### `emit_events` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L636" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
emit_events(self, state: State) -> None
```

Emit asset events


#### `from_task_and_inputs` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L522" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
from_task_and_inputs(cls, task: 'Task[Any, Any]', task_run_id: UUID, task_inputs: Optional[dict[str, set[Any]]] = None, copy_to_child_ctx: bool = False) -> 'AssetContext'
```

Create an AssetContext from a task and its resolved inputs.

**Args:**
- `task`: The task instance
- `task_run_id`: The task run ID
- `task_inputs`: The resolved task inputs (TaskRunResult objects)
- `copy_to_child_ctx`: Whether this context should be copied on a child AssetContext

**Returns:**
- Configured AssetContext


#### `get` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L194" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
get(cls: type[Self]) -> Optional[Self]
```

Get the current context instance


#### `model_copy` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L198" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
model_copy(self: Self) -> Self
```

Duplicate the context model, optionally choosing which fields to include, exclude, or change.

**Attributes:**
- `include`: Fields to include in new model.
- `exclude`: Fields to exclude from new model, as with values this takes precedence over include.
- `update`: Values to change/add in the new model. Note\: the data is not validated before creating
the new model - you should trust this data.
- `deep`: Set to `True` to make a deep copy of the model.

**Returns:**
- A new model instance.


#### `related_materialized_by` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L629" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
related_materialized_by(by: str) -> dict[str, str]
```

Create a related resource for the tool that performed the materialization


#### `serialize` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L700" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
serialize(self: Self, include_secrets: bool = True) -> dict[str, Any]
```

Serialize the AssetContext for distributed execution.


#### `serialize` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L219" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
serialize(self, include_secrets: bool = True) -> dict[str, Any]
```

Serialize the context model to a dictionary that can be pickled with cloudpickle.


#### `update_tracked_assets` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L679" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
update_tracked_assets(self) -> None
```

Update the flow run context with assets that should be propagated downstream.


### `TagsContext` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L712" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>


The context for `prefect.tags` management.

**Attributes:**
- `current_tags`: A set of current tags in the context


**Methods:**

#### `get` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L723" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
get(cls) -> Self
```

#### `get` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L194" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
get(cls: type[Self]) -> Optional[Self]
```

Get the current context instance


#### `model_copy` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L198" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
model_copy(self: Self) -> Self
```

Duplicate the context model, optionally choosing which fields to include, exclude, or change.

**Attributes:**
- `include`: Fields to include in new model.
- `exclude`: Fields to exclude from new model, as with values this takes precedence over include.
- `update`: Values to change/add in the new model. Note\: the data is not validated before creating
the new model - you should trust this data.
- `deep`: Set to `True` to make a deep copy of the model.

**Returns:**
- A new model instance.


#### `serialize` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L219" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
serialize(self, include_secrets: bool = True) -> dict[str, Any]
```

Serialize the context model to a dictionary that can be pickled with cloudpickle.


### `SettingsContext` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L730" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>


The context for a Prefect settings.

This allows for safe concurrent access and modification of settings.

**Attributes:**
- `profile`: The profile that is in use.
- `settings`: The complete settings model.


**Methods:**

#### `get` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L750" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
get(cls) -> Optional['SettingsContext']
```

#### `get` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L194" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
get(cls: type[Self]) -> Optional[Self]
```

Get the current context instance


#### `model_copy` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L198" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
model_copy(self: Self) -> Self
```

Duplicate the context model, optionally choosing which fields to include, exclude, or change.

**Attributes:**
- `include`: Fields to include in new model.
- `exclude`: Fields to exclude from new model, as with values this takes precedence over include.
- `update`: Values to change/add in the new model. Note\: the data is not validated before creating
the new model - you should trust this data.
- `deep`: Set to `True` to make a deep copy of the model.

**Returns:**
- A new model instance.


#### `serialize` <sup><a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/context.py#L219" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>

```python
serialize(self, include_secrets: bool = True) -> dict[str, Any]
```

Serialize the context model to a dictionary that can be pickled with cloudpickle.

