Skip to main content

How to Implement Human Approvals

To collect human input or approvals during a Flyte workflow execution, you can use the Human-in-the-Loop (HITL) plugin. This allows you to pause a task, serve an interactive form or API endpoint, and resume execution once a human provides the required data.

Basic Implementation​

The following example demonstrates how to create an event that requests an integer from a user and waits for the response.

import flyteplugins.hitl as hitl
from flytekit import task

@task
async def process_with_human_input() -> int:
    # Step 1: Create the event (this automatically serves the HITL app)
    event = await hitl.new_event.aio(
        "integer_input_event",
        data_type=int,
        scope="run",
        prompt="What should I add to the result?",
    )
    
    # Step 2: Wait for human input
    # This method polls object storage and is crash-resilient
    y = await event.wait.aio()
    
    return y + 10

Key Components​

Creating an Event​

The hitl.new_event function (a wrapper for Event.create) initializes the interaction. When called, the SDK:

  1. Starts a FastAPI application using flyte.serve if it isn't already running.
  2. Generates a unique request_id.
  3. Writes request metadata to object storage (e.g., S3 or local /tmp/flyte/hitl).

Waiting for Input​

The event.wait() method pauses the task execution. It uses a polling mechanism that checks object storage for a response file. Because the state is stored in object storage rather than memory, the task can crash and restart without losing the pending request; it will simply resume polling upon restart.

Synchronous Usage​

If you are working in a non-async task, you can use the synchronous versions of these methods:

import flyteplugins.hitl as hitl
from flytekit import task

@task
def sync_human_approval() -> bool:
    # Create event synchronously
    event = hitl.new_event(
        "approval_event",
        data_type=bool,
        prompt="Do you approve this deployment?"
    )
    
    # Wait synchronously
    approved = event.wait()
    return approved

Interaction Methods​

Once an event is created, humans can provide input through two primary interfaces provided by the Event object:

  1. Web Form: Accessible via event.form_url. This is automatically rendered in the Flyte UI as a Flyte Report using the show_form override.
  2. JSON API: Accessible via event.api_url. This allows for programmatic submissions using the HITLSubmissionTyped schema.

Example of a manual API submission using curl:

curl -X POST http://<app-endpoint>/submit/json \
  -H "Content-Type: application/json" \
  -d '{
    "request_id": "your-request-id",
    "value": 42,
    "data_type": "int",
    "response_path": "s3://path/to/response.json"
  }'

Supported Data Types​

The HITL plugin supports the following basic types for the data_type parameter:

  • int
  • float
  • bool
  • str

If a complex type is provided, the system defaults to treating it as a JSON-serialized str.

Troubleshooting and Requirements​

  • Cluster Environment: The HITL plugin requires a cluster environment that supports flyte.serve. It uses flyte.init_in_cluster to set up the FastAPI application environment.
  • Organization Configuration: The plugin relies on the _U_ORG_NAME environment variable to generate unique subdomains for the HITL application (e.g., hitl-event-app-myorg-myproject-development).
  • Timeouts: By default, wait() will timeout after 3600 seconds (1 hour). You can adjust this by passing timeout_seconds to new_event.
  • Polling Interval: The default polling interval is 5 seconds, configurable via poll_interval_seconds.