Jump to content

Connect SuperML | Leeroopedia MCP: Equip your AI agents with best practices, code verification, and debugging knowledge. Powered by Leeroo — building Organizational Superintelligence. Contact us at founders@leeroo.com.

Implementation:PrefectHQ Prefect Pause Flow Run

From Leeroopedia


Metadata
Sources Prefect
Domains Orchestration, Human_In_The_Loop
Last Updated 2026-02-09 00:00 GMT

Overview

Concrete function for pausing flow execution and collecting typed human input via UI forms provided by the Prefect library.

Description

The pause_flow_run function suspends a running flow and waits for input from a human via the Prefect UI. It accepts a RunInput subclass that defines the form fields, and returns the populated input when a human submits the form. Supports timeout configuration and initial data for pre-populating the form.

Code Reference

  • Repository: https://github.com/PrefectHQ/prefect
  • File: src/prefect/flow_runs.py (L334) for pause_flow_run, examples/ai_database_cleanup_with_approval.py (L96-108) for usage
  • Signature:
def pause_flow_run(
    wait_for_input: Optional[Type[RunInput]] = None,
    timeout: int = 3600,
) -> RunInput:
    """Pause the current flow run and wait for human input.

    Args:
        wait_for_input: RunInput subclass defining the form.
        timeout: Seconds to wait before timing out.
    Returns:
        Populated RunInput instance with human-provided values.
    """
  • Import: from prefect.flow_runs import pause_flow_run; from prefect.input import RunInput

I/O Contract

Inputs

  • wait_for_input (Type[RunInput], optional) — form class defining the input fields
  • timeout (int, optional) — wait timeout in seconds, default 3600

Outputs

  • RunInput instance with populated form fields

Usage Example

from pydantic import Field
from prefect import flow
from prefect.flow_runs import pause_flow_run
from prefect.input import RunInput

class CleanupApproval(RunInput):
    approve: bool = Field(default=False)
    notes: str = Field(default="", description="Why approve/reject?")

@flow(name="human-approval")
def get_human_approval(preview: str, count: int) -> tuple[bool, str]:
    approval = pause_flow_run(
        wait_for_input=CleanupApproval.with_initial_data(
            description=f"**Preview ({count} runs):**\n{preview}"
        ),
        timeout=3600,
    )
    return approval.approve, approval.notes

Related Pages

Page Connections

Double-click a node to navigate. Hold to expand connections.
Principle
Implementation
Heuristic
Environment