Human in the loop

This feature is in preview. The API may change before general availability. To try it out, contact support@intunedhq.com and we’ll enable it for your workspace.

Pause an automation and wait for a person. Use human_in_the_loop to ask a question and get a structured answer back, or hand_over_control_to_human to give an operator the live browser and let them finish the task themselves. Both helpers call an endpoint on your system, which you register in Intuned.json. Intuned doesn’t provide the operator UI—you decide who gets asked and how.

Time spent waiting on a person does not count toward the attempt’s request_timeout. The timeout you pass to these helpers is the only limit on the wait. See Human in the loop for the full request and callback contract.

Available helpers

human_in_the_loop — Ask a person for data and wait for the answer
hand_over_control_to_human — Hand the live browser to a person

Function reference

human_in_the_loop

from intuned_runtime import human_in_the_loop

async def human_in_the_loop(
    *,
    prompt: str,
    data_schema: type[BaseModel] | dict[str, Any],
    timeout: int = 3600,
    supporting_material: list[SupportingMaterial] | None = None,
    metadata: dict[str, Any] | None = None,
) -> Any: ...

Sends a request to your configured human-in-the-loop endpoint and waits. Returns the data your system posts back to the callback URL, validated against data_schema. Parameters

str

required

What you need from the person, in plain language. Shown to whoever handles the request.

type[BaseModel] | dict[str, Any]

required

The shape of the answer you expect. Sent to your endpoint as JSON Schema, and used to validate the callback payload before the automation resumes. Accepts a Pydantic model or a JSON Schema dictionary, like extract_structured_data.

int

default:"3600"

How long to wait for an answer, in seconds. After this, the call raises HumanInTheLoopTimeoutError.

list[SupportingMaterial] | None

Context to help the person answer—a screenshot of the page they’re deciding about, extracted text, or a link. See SupportingMaterial.

dict[str, Any] | None

Arbitrary JSON passed through to your endpoint untouched. Use it to route the request to the right queue, team, or end user.

Returns Returns the validated answer. When you pass a Pydantic model, you get an instance of it; with a JSON Schema dictionary, you get a dict. Raises

HumanInTheLoopTimeoutError — No valid answer arrived within timeout.
NonRetryableError — No human-in-the-loop endpoint is configured for the project.

Example

from enum import Enum
from intuned_runtime import human_in_the_loop
from playwright.async_api import Page
from pydantic import BaseModel


class Category(str, Enum):
    INVOICE = "invoice"
    STATEMENT = "statement"
    OTHER = "other"


class Classification(BaseModel):
    category: Category
    confirmed: bool
    note: str | None = None


async def classify(page: Page, params: dict | None = None, **_kwargs):
    screenshot = await page.screenshot()

    answer = await human_in_the_loop(
        prompt="Which category does this document belong to?",
        data_schema=Classification,
        timeout=7200,
        supporting_material=[
            {"type": "image", "buffer": screenshot},
            {"type": "text", "text": f"Document ID: {params['document_id']}"},
        ],
        metadata={"document_id": params["document_id"], "queue": "billing-review"},
    )

    # answer is a Classification instance
    return {"category": answer.category.value, "reviewed": answer.confirmed}

hand_over_control_to_human

from intuned_runtime import hand_over_control_to_human

async def hand_over_control_to_human(
    *,
    prompt: str,
    timeout: int = 3600,
    supporting_material: list[SupportingMaterial] | None = None,
    metadata: dict[str, Any] | None = None,
) -> None: ...

Sends a request to your configured handoff endpoint with a stream URL, and waits for the person to finish. The stream URL renders the live browser session in an iframe or a browser tab, with full keyboard and mouse control and a Complete button in the top-right corner. The call returns when the person selects Complete.

Call this as the last step of your automation. Once control is handed over, the person can navigate, submit forms, and change page state—anything your code does afterward runs against a page you no longer control.

Requirements

The project must run in headful mode ("headful": true in Intuned.json). There’s no browser to stream otherwise.
A handoffEndpoint must be configured. See Configure your endpoints.

Parameters

str

required

What the person needs to do once they take over. Shown alongside the stream.

int

default:"3600"

How long the session stays open, in seconds. The remaining time is shown to the person in the stream UI. When it expires, the stream closes and the call raises HumanInTheLoopTimeoutError.

list[SupportingMaterial] | None

Context to show next to the stream. See SupportingMaterial.

dict[str, Any] | None

Arbitrary JSON passed through to your endpoint untouched.

Returns Returns None when the person selects Complete. Raises

HumanInTheLoopTimeoutError — The person didn’t complete the session within timeout.
NonRetryableError — No handoff endpoint is configured, or the project isn’t running in headful mode.

Example

from intuned_runtime import hand_over_control_to_human, tracing_paused
from playwright.async_api import Page


async def submit_application(page: Page, params: dict | None = None, **_kwargs):
    await page.goto("https://portal.example.com/applications/new")
    await page.get_by_label("Applicant ID").fill(params["applicant_id"])

    # Everything is filled in except the identity check, which only a person can clear.
    async with tracing_paused():
        await hand_over_control_to_human(
            prompt=(
                "Complete the identity verification step, then submit the "
                "application and select Complete."
            ),
            timeout=1800,
            metadata={"applicant_id": params["applicant_id"]},
        )

A handoff streams a live browser to a person who may type a password or a one-time code. The trace keeps recording while they work, so wrap the handoff in tracing_paused when the session could involve credentials.

`SupportingMaterial`

class TextMaterial(TypedDict):
    type: Literal["text"]
    text: str


class ImageUrlMaterial(TypedDict):
    type: Literal["image"]
    url: str


class ImageBufferMaterial(TypedDict):
    type: Literal["image"]
    buffer: bytes


SupportingMaterial = TextMaterial | ImageUrlMaterial | ImageBufferMaterial

Optional context delivered to your endpoint with the request.

TextMaterial — Plain text. Use it for extracted values, a page excerpt, or an explanation of why the automation stopped.
ImageUrlMaterial — An image at a URL your endpoint can fetch.
ImageBufferMaterial — An image as raw bytes. Buffers are uploaded by the runtime and delivered to your endpoint as short-lived signed URLs, so a page.screenshot() result can be passed straight through.

Human in the loop — Endpoint configuration, request payloads, and the callback contract.
Intuned.json reference — humanInTheLoop project configuration.
Pause tracing — Keep credentials out of traces during a handoff.
Non-retryable errors — Stop a Run instead of asking a person twice.

Getting started

Intuned Agent

Web Tasks

Platform features

Learn

How-To

Integrations

References

Resources

Human in the loop

Available helpers

Function reference

human_in_the_loop

hand_over_control_to_human

`SupportingMaterial`

​Available helpers

​Function reference

​human_in_the_loop

​hand_over_control_to_human

​SupportingMaterial

​Related

Available helpers

Function reference

human_in_the_loop

hand_over_control_to_human

`SupportingMaterial`

Related