Can context secrets be passed from agent to dask executors during flow run? Sorry if im repeating this question. Im trying to get some credentials available on the dask workers, I have set envs like

PREFECT__CONTEXT__SECRETS__DB_URI

on the agent. Is this supposed to be available as prefect.context.secrets["DB_URI"] when the flow is run on dask worker without any other setup?

Are you on Prefect Cloud or Prefect Server?

prefect server

Thanks. In that case you indeed have to use local secrets. Generally, local secrets are retrieved from the agent environment so if you set it within the agent (e.g. when starting it), then the secret should be available within your task (regardless of whether the task run gets executed locally or on Dask). Example:

$ prefect agent docker start --env PREFECT__CONTEXT__SECRETS__CRED="CRED_VALUE"

And if normal PrefectSecret doesn’t work, you may try the `EnvVarSecret`:

from prefect import task, Flow
from prefect.tasks.secrets import EnvVarSecret

@task(log_stdout=True)
def print_value(x):
    print(x)

with Flow("Example") as flow:
    secret = EnvVarSecret("CRED")
    print_value(secret)

flow.run() # prints the value of the "CRED" environment variable

But if this doesn’t work, you can set it directly in your config.toml on the agent:

[context.secrets]
MYSECRET = "MY SECRET VALUE"

im using prefect's helm chart to deploy this. I dont think (know how) I'll be able to pass --env to agent start command directly. How and where do I set value for CRED if I'm using

EnvVarSecret

?

You would put it in the env variables section:

---
apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    app: prefect-agent
  name: prefect-agent
spec:
  replicas: 1
  selector:
    matchLabels:
      app: prefect-agent
  template:
    metadata:
      labels:
        app: prefect-agent
    spec:
      containers:
      - args:
        - prefect agent kubernetes start
        command:
        - /bin/bash
        - -c
        env:
        - name: PREFECT__CONTEXT__SECRETS__CRED
          value: 'CRED_VALUE'

great, this makes CRED available to tasks using prefect.context.secrets.CRED OR EnvVarSecret("CRED")?

it should, yes

cool, thanks Anna

the executor is retrieved from storage (rather than from the backend), so you don’t need to provide it at registration. As long as your storage contains the Dask address, this should work

so post registration of a flow, I have to modify the flow in the storage if theres a change in the executor URI? 😕 how do I go about that?

if your Dask address changes, you can just update this in your flow storage. You can e.g. use script storage and just upload your changed file to your GCS bucket - you could even do it as part of your CI/CD pipeline

okay, one more question that I have is, whats the right way for create_flow_run to refer to a particular flow from a list of many flows? I want to trigger certain flows from apis. Any way to trigger it by name? does prefect Client support listing flows?

The create_flow_run task does support triggering a flow run by flow name already. To list specific flows and their names, you can use the GraphQL API, e.g.

query {
  flow(where: { name: { _ilike: "%dask%" } }) {
    id
    name
    version
  }
}

thanks

I see. If setting this on the agent startup command works, I would use that, there is nothing that speaks against it imo. But if you feel like it’s a deal breaker for you, feel free to open an issue in the Server repository and we could investigate more.