or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

Files

docs

client-api.mdconfiguration.mdcontext-utilities.mdcore-workflows.mddeployments.mdindex.mdruntime-context.mdstate-management.mdvariables.md

runtime-context.mddocs/

0

# Runtime Context Access

1

2

Prefect provides runtime context modules that allow flows and tasks to access information about their current execution environment. These modules provide dynamic access to deployment parameters, flow run metadata, and task run details during execution.

3

4

## Capabilities

5

6

### Deployment Runtime Context

7

8

Access current deployment information including parameters, ID, and version during flow execution.

9

10

```python

11

import prefect.runtime.deployment

12

```

13

14

```python { .api }

15

# Available attributes on prefect.runtime.deployment:

16

id: str # The deployment's unique ID

17

name: str # The deployment's name

18

version: str # The deployment's version

19

flow_run_id: str # The current flow run ID for this deployment

20

parameters: Dict[str, Any] # Parameters passed to this deployment run

21

```

22

23

**Usage Example:**

24

25

```python

26

from prefect import flow

27

import prefect.runtime.deployment

28

29

@flow

30

def my_flow():

31

# Access deployment information

32

deployment_id = prefect.runtime.deployment.id

33

params = prefect.runtime.deployment.parameters

34

version = prefect.runtime.deployment.version

35

36

print(f"Running deployment {deployment_id} version {version}")

37

print(f"Parameters: {params}")

38

```

39

40

### Flow Run Runtime Context

41

42

Access current flow run information including metadata, parameters, and execution details.

43

44

```python

45

import prefect.runtime.flow_run

46

```

47

48

```python { .api }

49

# Available attributes on prefect.runtime.flow_run:

50

id: str # The flow run's unique ID

51

tags: List[str] # The flow run's set of tags

52

scheduled_start_time: datetime # Expected scheduled start time

53

name: str # Name of the flow run

54

flow_name: str # Name of the flow

55

flow_version: str # Version of the flow

56

parameters: Dict[str, Any] # Parameters passed to this run

57

parent_flow_run_id: Optional[str] # ID of parent flow run (if subflow)

58

parent_deployment_id: Optional[str] # ID of parent deployment (if any)

59

root_flow_run_id: str # ID of root flow run in hierarchy

60

run_count: int # Number of times this flow run has been executed

61

```

62

63

**Usage Example:**

64

65

```python

66

from prefect import flow

67

import prefect.runtime.flow_run

68

69

@flow

70

def my_flow():

71

# Access flow run context

72

flow_run_id = prefect.runtime.flow_run.id

73

tags = prefect.runtime.flow_run.tags

74

parameters = prefect.runtime.flow_run.parameters

75

run_count = prefect.runtime.flow_run.run_count

76

77

print(f"Flow run {flow_run_id} (attempt {run_count})")

78

print(f"Tags: {tags}, Parameters: {parameters}")

79

```

80

81

### Task Run Runtime Context

82

83

Access current task run information and metadata during task execution.

84

85

```python

86

import prefect.runtime.task_run

87

```

88

89

```python { .api }

90

# Available attributes on prefect.runtime.task_run:

91

id: str # The task run's unique ID

92

name: str # Name of the task run

93

task_key: str # Unique key identifying the task

94

flow_run_id: str # ID of the parent flow run

95

task_inputs: Dict[str, Any] # Input parameters for the task

96

run_count: int # Number of times this task run has been executed

97

```

98

99

**Usage Example:**

100

101

```python

102

from prefect import task

103

import prefect.runtime.task_run

104

105

@task

106

def my_task():

107

# Access task run context

108

task_run_id = prefect.runtime.task_run.id

109

task_name = prefect.runtime.task_run.name

110

task_key = prefect.runtime.task_run.task_key

111

run_count = prefect.runtime.task_run.run_count

112

113

print(f"Task {task_name} ({task_key}) run {run_count}")

114

return f"Executed task run {task_run_id}"

115

```

116

117

## Common Patterns

118

119

### Conditional Logic Based on Runtime Context

120

121

```python

122

from prefect import flow, task

123

import prefect.runtime.deployment

124

import prefect.runtime.flow_run

125

126

@task

127

def process_data(data_source: str):

128

# Use deployment parameters to configure processing

129

config = prefect.runtime.deployment.parameters.get("processing_config", {})

130

131

if "debug" in prefect.runtime.flow_run.tags:

132

print(f"Debug mode: processing {data_source} with config {config}")

133

134

return f"Processed {data_source}"

135

136

@flow

137

def data_pipeline():

138

# Access runtime context for dynamic behavior

139

env = prefect.runtime.deployment.parameters.get("environment", "prod")

140

data_source = f"s3://bucket/{env}/data.csv"

141

142

return process_data(data_source)

143

```

144

145

### Testing with Mock Runtime Context

146

147

Runtime attributes can be mocked using environment variables:

148

149

```python

150

import os

151

152

# Mock deployment context

153

os.environ["PREFECT__RUNTIME__DEPLOYMENT__ID"] = "test-deployment-id"

154

os.environ["PREFECT__RUNTIME__DEPLOYMENT__PARAMETERS"] = '{"env": "test"}'

155

156

# Mock flow run context

157

os.environ["PREFECT__RUNTIME__FLOW_RUN__ID"] = "test-flow-run-id"

158

os.environ["PREFECT__RUNTIME__FLOW_RUN__TAGS"] = '["test", "debug"]'

159

```

160

161

**Note:** Runtime context modules return empty values when not executing within a Prefect flow run context.