Yes — in Prefect 3.x you can build a “human-in-the-loop” approval pattern that sends an email, pauses, waits for an external callback, then resumes the flow with the provided input. High-level options - pause_flow_run: Keeps the flow process alive and polls for input; good for short waits. - suspend_flow_run: Puts the flow into a Suspended/Paused state and frees infrastructure; best for longer waits. You must explicitly resume it. Core APIs - Pausing/suspending:

pause_flow_run(wait_for_input=..., key="...")

or

suspend_flow_run(wait_for_input=..., key="...")

- Programmatic resume:

resume_flow_run(flow_run_id, run_input={...})

- Getting the current run id:

get_run_context().flow_run.id

Docs: - Interactive (human-in-the-loop) workflows - REST API: Resume flow run - prefect-email integration Example: email + suspend + external approval callback Flow code (suspend and wait for a boolean approval, then continue):

from uuid import UUID
from prefect import flow, suspend_flow_run, get_run_logger
from prefect.context import get_run_context
from prefect_email.message import email_send_message
from prefect_email.credentials import EmailServerCredentials

@flow
def approval_flow(to_email: str):
    logger = get_run_logger()

    # Get the current flow run id for the callback link
    run_id = str(get_run_context().flow_run.id)

    # Build approval links to your external app
    approve_url = f"<https://myapp.example.com/approve?run_id={run_id}&approved=true>"
    reject_url = f"<https://myapp.example.com/approve?run_id={run_id}&approved=false>"

    # Send the email (configure a Block named "my-smtp-creds" in the UI first)
    creds = EmailServerCredentials.load("my-smtp-creds")
    email_send_message(
        subject="Approval needed",
        msg=f"Please approve this run.\nApprove: {approve_url}\nReject: {reject_url}",
        email_server_credentials=creds,
        email_to=to_email,
    )

    # Suspend and wait for input; frees infra while waiting
    # Use a key so the external callback can target this input
    approved: bool | None = suspend_flow_run(
        wait_for_input=bool,
        key="approval",
        timeout=7 * 24 * 3600  # 7 days
    )

    if approved:
        <http://logger.info|logger.info>("Approved, continuing workflow...")
        # ... do the thing ...
    else:
        <http://logger.info|logger.info>("Rejected (or no response in time).")
        # ... handle rejection ...

if __name__ == "__main__":
    approval_flow("<mailto:approver@example.com|approver@example.com>")

External approval endpoint (FastAPI example):

from uuid import UUID
from fastapi import FastAPI, HTTPException, Query
from prefect import resume_flow_run

app = FastAPI()

@app.get("/approve")
def approve(
    run_id: UUID = Query(...),
    approved: bool = Query(...),
    token: str = Query(...)  # validate a shared secret or signed JWT!
):
    # TODO: validate token to prevent unauthorized approvals
    try:
        # Provide input keyed by the same `key` used in suspend_flow_run(...)
        resume_flow_run(flow_run_id=run_id, run_input={"approval": approved})
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))
    return {"status": "ok"}

cURL example using the REST API (non-Python systems)

curl -X POST \
  -H "Authorization: Bearer $PREFECT_API_KEY" \
  -H "Content-Type: application/json" \
  "$PREFECT_API_URL/api/flow-runs/<flow_run_id>/resume" \
  -d '{"run_input": {"approval": true}}'

Notes and tips - Use suspend_flow_run for long waits; it tears down infra and you must call resume_flow_run to continue. - For short waits, pause_flow_run works similarly and will poll until input is available or timeout. - The

key

argument lets you correlate inputs when there are multiple pauses in a flow.

- You can collect structured inputs by passing a Pydantic model instead of `bool`:

from pydantic import BaseModel

  class Approval(BaseModel):
      approved: bool
      comment: str | None = None

  decision = suspend_flow_run(wait_for_input=Approval, key="approval")
  # External app would then pass:
  # resume_flow_run(run_id, run_input={"approval": {"approved": true, "comment": "LGTM"}})

- If you prefer manual approval via the Prefect UI, users can click Resume on the flow run and enter input directly. Questions to tailor this for you - Which external app will issue the callback (and in what language)? - How long might approvals take (minutes vs. hours/days)? That determines pause vs. suspend. - Do you want a simple boolean approval, or a structured payload (e.g., comment, amount, approver id)? - Will you host your own approval endpoint, or should we use a Prefect Automation + Webhook pattern?

@Marvin Suggest me production approach, my callback will be from teams, slack, gmail, Outlook. And wait will be for around 1 week

Great use case. For a 1-week wait with approvals coming from Teams, Slack, Gmail, and Outlook, here’s a production-ready pattern: Recommended architecture - Use suspend_flow_run: It pauses the flow and frees compute until you resume it. Perfect for a 1-week SLA. - Central “approval callback” service: Host a small HTTPS endpoint (e.g., FastAPI on Cloud Run, Lambda + API Gateway, Azure Functions) that all channels can call via a link/button. The service verifies the request and calls resume_flow_run(flow_run_id, run_input=...). - Signed, single-use links: Embed a signed token with run_id, key, and optional metadata in approve/reject links to prevent spoofing. Store/validate nonces for idempotency. - Channel-specific UX: - Gmail/Outlook: Include approve/reject links in the email body directly. Avoid parsing replies; links are simpler and more reliable. Outlook Actionable Messages/Adaptive Cards are possible but add overhead. - Slack/Teams: Use message buttons that open your approval URL (Action.OpenUrl in Teams Adaptive Card; Slack Block Kit button with url). You can also support slash commands if you prefer. Core flow pattern (Prefect 3.x) - Suspend and wait for typed input with a key - Send approval links via email (and optionally through Slack/Teams bots) - External callback resumes the flow with a structured payload Flow example

from uuid import UUID
import os, time
from pydantic import BaseModel
from prefect import flow, suspend_flow_run, get_run_logger
from prefect.context import get_run_context
from prefect_email.message import email_send_message
from prefect_email.credentials import EmailServerCredentials

class Approval(BaseModel):
    approved: bool
    channel: str | None = None     # "email"|"slack"|"teams"|...
    actor: str | None = None       # who approved
    comment: str | None = None

APP_BASE = os.getenv("APP_BASE", "<https://approvals.example.com>")  # your callback base URL

def approval_links(run_id: str, key: str="approval"):
    # token should be a signed value that encodes run_id/key and expires
    # for brevity, we insert a placeholder
    token = f"SIGNED.{run_id}.{key}"
    approve = f"{APP_BASE}/approve?token={token}&approved=true"
    reject  = f"{APP_BASE}/approve?token={token}&approved=false"
    return approve, reject

@flow
def hitl_flow(to_email: str):
    logger = get_run_logger()
    run_id = str(get_run_context().flow_run.id)

    approve_url, reject_url = approval_links(run_id)

    # Send email with clickable approve/reject links
    creds = EmailServerCredentials.load("my-smtp-creds")
    email_send_message(
        subject="Approval needed",
        msg=(
            "Please review and approve this run.\n\n"
            f"Approve: {approve_url}\n"
            f"Reject:  {reject_url}\n"
        ),
        email_server_credentials=creds,
        email_to=to_email,
    )

    # Suspend for up to 7 days and wait for structured input
    decision: Approval | None = suspend_flow_run(
        wait_for_input=Approval,
        key="approval",
        timeout=7 * 24 * 3600,  # 7 days
    )

    if decision and decision.approved:
        <http://logger.info|logger.info>(f"Approved by {decision.actor or decision.channel}. Proceeding.")
        # ... continue downstream steps ...
    else:
        <http://logger.info|logger.info>("Rejected or no response. Taking fallback path.")
        # ... handle rejection or timeout ...

Callback service (FastAPI) example - Verifies the token, extracts run_id and key, then resumes the flow with a structured payload ``` import os from uuid import UUID from fastapi import FastAPI, HTTPException, Query from prefect import resume_flow_run SECRET = os.getenv("APPROVAL_TOKEN_SECRET", "change-me") def verify_and_decode(token: str) -> dict: # Replace with proper JWT/HMAC verification and expiration checks # Expected to return {"run_id": "...", "key": "..."} try: _, run_id, key = token.split(".") return {"run_id": run_id, "key": key} except Exception:

raise HTTPException(status_code=400, detail="Invalid token") app = FastAPI() @app.get("/approve") def approve(token: str = Query(...), approved: bool = Query(...), channel: str | None = Query(None), actor: str | None = Query(None), comment: str | None = Query(None)): data = verify_and_decode(token) run_id = UUID(data["run_id"]) key = data["key"] run_input = { key: { "approved": approved, "channel": channel, "actor": actor, "comment": comment, } } try: resume_flow_run(flow_run_id=run_id, run_input=run_input) except Exception as e: # If already resumed or timed out, treat as idempotent if "NotPausedError" in str(e): return {"status": "already_resumed"} raise HTTPException(status_code=400, detail=str(e)) return {"status": "ok"} ``` How each channel fits in - Gmail/Outlook: The email content already includes approve/reject links that open your endpoint. This is the most reliable, minimal-integration approach. - Slack: - Send a message with Block Kit buttons linking to your approve/reject URLs (they can include channel=slack in the query). - Optional: A slash command could call your endpoint with run_id and decision, but link buttons are simpler and work on mobile. - Microsoft Teams: - Use an Adaptive Card with Action.OpenUrl buttons to your approve/reject URLs, or a bot message with links. - If you already have a Teams bot, it can add “approve/reject” buttons that hit the same endpoint. Operational best practices - Security: - Sign the token (HMAC/JWT) with expiry. Include run_id, key, and maybe a nonce. Validate on the server. - Use TLS and store secrets in a vault. - Idempotency: - Resume once per run/key. If called again, return 200 with “already_resumed”. - Observability: - Log who approved (actor) and channel in run_input. - Optionally write a Prefect markdown artifact with the approval links for easy retrieval from the UI. - Timeouts: - The suspend timeout is 7 days here. If approval arrives after that, resume_flow_run will fail—handle gracefully in the callback and inform the approver. - Scale: - Many concurrent suspended runs are fine; they do not hold compute. - No-code alternative: - If you don’t want to host a service, use Slack/Teams workflow + Zapier/Make to call the Prefect Resume REST API directly with run_input. Just be sure to secure the call with your Prefect API key and restrict access. Key docs - Interactive (human-in-the-loop) workflows: docs - REST API (resume flow run): docs - prefect-email: docs A couple of questions to tailor this: - Are you on Prefect Cloud or self-hosted (and where will you host the callback API)? - Do you need just boolean approval, or should we capture actor/comment and channel too? - Do you want Slack/Teams messages sent automatically as well as email? If so, which platform(s) should we integrate first?