Short answer: Not directly. In Prefect 3.x, pull steps run before your flow code executes and they don’t have access to flow run parameters. They can, however, read environment variables/job variables provided to the job. So you can’t conditionally install packages in a pull step based on flow parameters, but you can do it based on job variables/env vars. Two workable patterns: 1) Use job variables/env vars with a pull step - Set a job variable that the worker exposes as an environment variable (e.g., EXTRA_PKG). - Add a pull step that conditionally installs the package based on that env var. Example prefect.yaml snippet:

pull:
  - prefect.deployments.steps.pull.git_clone:
      repository: <https://github.com/your-org/your-repo>
      branch: main
  - prefect.deployments.steps.utility.run_shell_script:
      script: |
        if [ -n "${EXTRA_PKG}" ]; then
          echo "Installing ${EXTRA_PKG}..."
          python -m pip install "${EXTRA_PKG}"
        else
          echo "No EXTRA_PKG specified; skipping."
        fi
      expand_env_vars: true

Then, provide

EXTRA_PKG

via job variables on your deployment (varies by worker type). For a Process worker you can set env via job variables; with Python deploy this looks like:

flow.from_source("<https://github.com/your-org/your-repo>", entrypoint="flows/my_flow.py:flow").deploy(
  name="my-deploy",
  work_pool_name="my-pool",
  job_variables={
    "env": {"EXTRA_PKG": "polars==1.9.0"}
  },
)

Docs: Customize job variables 2) Install conditionally inside the flow (based on flow parameters) - If you must key off flow parameters, do the install inside your flow before imports that need it:

from prefect import flow
import sys, subprocess

@flow
def my_flow(extra_pkg: str | None = None):
    if extra_pkg:
        subprocess.check_call([sys.executable, "-m", "pip", "install", extra_pkg])
    # rest of your flow

Note: On-the-fly installs can slow runs and reduce reproducibility; prefer pre-baked images or the job-variable approach when possible. Other options - Create separate deployments (or images) per dependency set and choose the appropriate deployment at run time. Quick clarification to tailor the best solution: - When you say “job parameters,” do you mean flow run parameters (the inputs to your flow) or job variables (infrastructure-level fields/env for the worker)? - Which worker type are you using (Process, Docker, Kubernetes)? The exact job_variables shape varies by worker. Related docs: - Deploy via Python - Job variables: Customize job variables

Thanks for clarifying — you mean flow run parameters. In Prefect 3.x, pull steps run before your flow is loaded and before parameters are available, so they cannot read flow parameters. That’s why you can’t conditionally install a package in pull steps based on flow params. Viable ways to achieve your goal: - Duplicate the value into job variables at run time (recommended for pull-step installs) - When you kick off the run, pass the package name both as a flow parameter and as a job variable (exposed as an env var). Your pull step reads the env var and installs. - Example kickoff via API:

from prefect.deployments import run_deployment

  run_deployment(
      "my-flow/my-deploy",
      parameters={"extra_pkg": "<git+ssh://github.com/org/customlib.git#egg=customlib>"},
      job_variables={"env": {"EXTRA_PKG": "<git+ssh://github.com/org/customlib.git#egg=customlib>"}},
  )

prefect.yaml pull steps:

pull:
    - prefect.deployments.steps.pull.git_clone:
        repository: <https://github.com/your-org/your-repo>
        branch: main
    - prefect.deployments.steps.utility.run_shell_script:
        script: |
          if [ -n "${EXTRA_PKG}" ]; then
            echo "Installing ${EXTRA_PKG}..."
            python -m pip install "${EXTRA_PKG}"
          else
            echo "No EXTRA_PKG specified; skipping."
          fi
        expand_env_vars: true

Docs: Customize job variables, Deploy via Python - Install inside the flow based on the parameter (simple, but less reproducible)

from prefect import flow
  import sys, subprocess

  @flow
  def my_flow(extra_pkg: str | None = None):
      if extra_pkg:
          subprocess.check_call([sys.executable, "-m", "pip", "install", extra_pkg])
      # proceed with imports/logic that require the package

Runtime context (if needed): prefect.runtime Notes and trade-offs: - Security: Installing arbitrary packages from parameters can be risky — consider validating/allowlisting inputs. - Reproducibility: On-the-fly installs can slow runs and make environments non-deterministic. If possible, prefer pinned wheels/private index and/or prebuilt images per variant. Quick question so I can tailor the best approach: - How are you triggering runs (UI/manual, API, or from another flow)? If you can start runs via API or a parent flow, passing both

parameters

and

job_variables

per run is the cleanest way to drive pull-step installs.