Human-in-the-Loop

ctx.interrupt() suspends a run until someone answers: approve, pick one, edit these arguments. A signal says something happened, an interrupt asks.

from pyrula.workflows import workflow, InterruptResponse, InterruptAction

@workflow(name="expense_approval")
async def expense_approval(ctx, amount_cents: int) -> str:
    response: InterruptResponse = await ctx.interrupt(
        "approve_expense",
        payload={"amount_cents": amount_cents},
        expects="approval",
    )

    if response.action == InterruptAction.APPROVE:
        await ctx.step(
            "disburse",
            lambda: disburse(amount_cents, idempotency_key=f"{ctx.run_id}:disburse"),
        )
        return "approved"
    if response.action == InterruptAction.EDIT:
        await ctx.step(
            "disburse_edited",
            lambda: disburse(
                response.value["amount_cents"],
                idempotency_key=f"{ctx.run_id}:disburse_edited",
            ),
        )
        return "approved with edit"
    return "rejected"

The run writes run:interrupted (status interrupted) and frees the worker. A resume emits run:resumed and interrupt returns the response; replay returns the recorded one.

A run waits on one question at a time. If tasks of the body that you started with plain asyncio ask at once, the first question opens and the others are asked after it is answered, one by one. Give concurrent ctx.interrupt calls different event_ids. A body that catches the suspension and returns still leaves the run waiting on the open question.

Answering

await client.resume("expense_approval", run_id, "approve")
await client.resume("expense_approval", run_id, "edit", {"amount_cents": 900})
await client.resume("expense_approval", run_id, "approve", wait_id=wait_id)   # pin the wait

wait_id (on every run:interrupted) pins the answer to one question; without it the client answers whatever the run waits on now. A resume for a resolved or timed-out wait raises RunNotWaitingError. Over HTTP it’s POST .../resume, see Serving over HTTP.

Signature

async def interrupt(
    self,
    event_id: str,
    payload: dict,
    *,
    expects: Literal["approval", "choice", "text", "value"] | None = None,
    options: list[str | dict] | None = None,   # the choices, for expects="choice"
    timeout: float | timedelta | None = None,
    deadline: datetime | None = None,
) -> InterruptResponse: ...

expects tells a UI (and the Cloud console) what kind of answer to collect.

Timeouts

Waits forever by default. Pass timeout= (seconds or timedelta) or deadline= (timezone-aware datetime), not both:

from pyrula.workflows import WaitTimeout

try:
    response = await ctx.interrupt("approve_expense", payload, timeout=86_400)
except WaitTimeout:
    return "no one answered"

Uncaught, WaitTimeout ends the run with code wait_timeout. The deadline is journaled in run:interrupted and reused on replay. If an answer lands first it wins, and the late resume gets RunNotWaitingError.

When the handler fails

If an on_interrupt handler raises, or returns an action the control plane rejects, the run ends in error with code interrupt_handler_failed and your exception reaches the caller. Raise DurableInterrupt to hand the question to someone else and leave the run suspended; to suspend on another exception type, pass suspend_on=(...) to run_and_await.

InterruptResponse

@dataclass
class InterruptResponse:
    action: InterruptAction | str
    value: Any = None

Action	Use for	value
APPROVE	approval, proposed action	none
REJECT	approval, proposed action	none
RESPOND	choice, text, value	the answer
EDIT	proposed action	the edited arguments

To answer in-process, pass on_interrupt= to run_and_await or RunHandle.result(...); the handler returns an InterruptResponse. A lowercase string like "approve" works; an unknown action raises ValueError to the caller and the run stays suspended. On an approval interrupt the bare string "approve" or "reject" takes that branch, on choice/text/value it’s the answer itself. Any other plain value is a respond answer; an explicit InterruptResponse always wins.

Gate irreversible things this way (charges, deletes, outbound email) and keep the approved effect idempotent. For an agent’s tools use @tool(requires_approval=True), see Approval.