Overview

This skill produces a secrets checklist — a table of every secret name, where it came from in Prefect, and where it must be re-entered in Orchestra. It never asks for actual secret values and never writes them to any file. Prefect's encrypted Secret block store has no bulk export mechanism, so most secrets must be re-entered by hand in Orchestra — the exception is a value that already lives in AWS Secrets Manager or Azure Key Vault, which an Orchestra connection can read in place via its Credential Store.

Parameter Mapping

Hard rule (enforced always):

NEVER ask the user to paste real secret values into this chat, and never write real secret values into any file. Produce only a checklist of secret NAMES and where to enter them.

Orchestra YAML Structure

Secrets themselves do not appear in YAML. The YAML only shows where the secret will be consumed:

# Python task reading a secret from env var (set on the connection)
integration: PYTHON
integration_job: PYTHON_EXECUTE_SCRIPT
connection: python_git_prod_12345   # env vars set on this connection in Orchestra UI
parameters:
  command: 'python scripts/call_api.py'
  python_version: '3.12'

# scripts/call_api.py — reads secret at runtime from env
import os
import requests

token = os.getenv("MY_API_TOKEN")   # set on the Orchestra Python connection
response = requests.get("https://api.example.com/data", headers={"Authorization": f"Bearer {token}"})

Conversion Steps

• Search the codebase for: Secret.load(, .get_secret_value(), SecretStr, SecretDict, prefect.blocks.system.Secret, AwsSecret
• For each hit: record the secret name and how it is used (SaaS credential, env var in task, etc.)
• Separate Prefect Variables (Variable.get()) from Secret blocks — Variables are non-sensitive and go to inputs: or env_vars:, not Orchestra secrets
• Produce the secrets checklist (see template below)
• For each row marked "Orchestra connection field": follow prefect-connections-to-orchestra to create the connection and enter the value in the credential field — in the UI, or through the public API / Orchestra MCP, feeding each value straight from wherever you read it out of Prefect rather than staging it in a file
• For each row marked "Python connection env var": open the Orchestra Python connection → Advanced → Environment Variables → add the key
• For a secret already held in an external store: if that store is AWS Secrets Manager or Azure Key Vault, the value can stay where it is — create the Orchestra connection to the store, then set Credential Store on the target connection and identify the secret (Secret ARN for AWS Secrets Manager, secret name for Azure Key Vault) instead of re-entering it. Confirm the secret's JSON keys match the integration's expected field names. Any other store (Hashicorp Vault, GCP Secret Manager) still needs a manual re-entry
• After all secrets are entered, verify each pipeline runs successfully in a test environment before promoting to production

Before / After Example

Prefect (before)

from prefect.blocks.system import Secret
from prefect import task, flow
from pydantic import SecretStr

@task
def call_external_api():
    token = Secret.load("my-api-token").get()
    response = requests.get("https://api.example.com", headers={"Authorization": f"Bearer {token}"})
    return response.json()

@task
def write_to_snowflake(data):
    # SnowflakeCredentials block holds SecretStr password
    creds = SnowflakeCredentials.load("sf-prod")
    conn = creds.get_client()
    conn.execute("INSERT INTO ...")

Orchestra YAML (after)

pipeline:
  stage-fetch:
    tasks:
      call-external-api:
        integration: PYTHON
        integration_job: PYTHON_EXECUTE_SCRIPT
        connection: python_git_prod_12345   # MY_API_TOKEN set on this connection
        parameters:
          command: 'python scripts/call_external_api.py'
          python_version: '3.12'
          set_outputs: true
        depends_on: []

  stage-load:
    tasks:
      write-to-snowflake:
        integration: SNOWFLAKE
        integration_job: SNOWFLAKE_RUN_QUERY
        connection: snowflake_prod_67890   # password entered in Orchestra UI
        parameters:
          query: "INSERT INTO ..."
        depends_on:
          - call-external-api

Secrets checklist output (example):

| Secret name        | Source Prefect block/pattern          | Orchestra target                                        | Action                                    |
|--------------------|---------------------------------------|---------------------------------------------------------|-------------------------------------------|
| SNOWFLAKE_PASSWORD | SnowflakeCredentials "sf-prod".password | Snowflake connection "snowflake_prod_67890" → password  | Re-enter manually in Orchestra UI         |
| MY_API_TOKEN       | Secret.load("my-api-token")           | Python connection "python_git_prod_12345" → env var MY_API_TOKEN | Re-enter in Python connection JSON   |
| GH_DEPLOY_KEY      | GitHubCredentials "gh-deploy".token   | Python connection "python_git_prod_12345" → deploy key  | Paste private key in Orchestra UI         |

Gotchas

• Prefect's encrypted Secret block store has no bulk export to Orchestra — every secret Prefect itself holds must be retrieved one at a time; plan for this in your migration timeline. The bottleneck is the Prefect side, not the Orchestra side: connections can be created through the public API or MCP, so a migration script can create each connection as it recovers the value, and only the recovery stays manual. Secrets already sitting in AWS Secrets Manager or Azure Key Vault escape the re-key entirely, by being read in place through the connection's Credential Store
• Prefect Variable.get() values are not encrypted — route to inputs: or env_vars: in the Orchestra pipeline, not to secrets
• SecretStr fields inside credential blocks follow prefect-connections-to-orchestra, not this skill — they are stored on the connection, not as standalone secrets
• An external credential store holds one JSON object per connection — in AWS Secrets Manager or Azure Key Vault the secret's value must be a JSON object containing exactly the keys the target integration expects, and every required one rather than only the sensitive fields. The keys are the parameter IDs from that integration's Authentication parameters table, not friendly names — a Snowflake connection wants snowflake_account, snowflake_user, snowflake_password and the rest of its required parameters, not account / user / password. In Azure Key Vault, set that whole JSON object as the secret's value rather than splitting it into one secret per field. One Prefect secret rarely maps to one store secret: a flow that loads a password on its own still needs the non-secret parameters gathered into the same JSON object
• A mismatch is caught when the connection is saved — Orchestra reads the secret at that point and rejects it if the store credential is invalid, the secret can't be read, or its JSON doesn't match the integration's parameters, so this is checked up front rather than failing mid-run. From then on it reads the latest version each time it uses the connection, so rotation in the store needs no change in Orchestra — worth saying in the checklist, since it's the part of the old Prefect workflow that disappears
• A credential store only appears once its own connection exists — the Credential Store dropdown offers AWS Secrets Manager or Azure Key Vault only when a connection to that store has already been created, so sequence the store connection before the connections that read from it. Both store integrations are in beta and may not be enabled on the account, and a vault whose networking is restricted to selected networks has to allow Orchestra's IPs through before any secret can be read — confirm both before building a migration plan around this route. See Azure Key Vault
• Never store secrets in task env_vars: in the YAML file itself — only reference connection-level env vars or use Orchestra's secrets integration
• If a Secret block is used purely for observability (e.g. masking values in logs) with no downstream consumer, it can be dropped entirely

References

• https://docs.prefect.io/v3/develop/blocks
• https://docs.getorchestra.io/docs/core-concepts/tasks/connections
• Orchestra external secret storage: https://docs.getorchestra.io/docs/core-concepts/tasks/connections#external-secret-storage
• Managing connections programmatically (public API and MCP): https://docs.getorchestra.io/docs/core-concepts/tasks/connections#can-i-manage-connections-programmatically

Adding Alerts

See prefect-alerts-to-orchestra for all notification patterns.