# Resonate Human In The Loop Pattern Python

> Implement human-in-the-loop workflows in Python — durable functions that suspend on ctx.promise() until an external actor (webhook, UI, CLI, operator) resolves or rejects. Use when a workflow must wait on an action that doesn't originate from another worker, such as approval gates, manual review, or out-of-band data.

- Skill: `resonatehq/resonate-human-in-the-loop-pattern-python` (Agent Skill)
- Install (CLI): `npx skillmds@latest add resonatehq/resonate-human-in-the-loop-pattern-python`
- Raw SKILL.md: https://api.skillmd.com/api/skills/resonatehq/resonate-human-in-the-loop-pattern-python/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Integrations & APIs
- License: Apache-2.0
- Author: resonatehq (https://skillmd.com/u/resonatehq)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/resonatehq/resonate-human-in-the-loop-pattern-python

---


# Resonate Human-in-the-Loop Pattern — Python

## Overview

A human-in-the-loop workflow suspends on a durable promise that is resolved (or rejected) by something outside the Resonate worker set — a person clicking an Approve button, a webhook from a third-party system, an operator running a CLI command. The worker doesn't poll or sleep; it opens a `ctx.promise()` and awaits it, and Resonate wakes it when the promise settles.

## When to use

- Approval gates in business workflows (expense approval, content moderation, deploy gate)
- Waiting on third-party callbacks (Stripe webhooks, DocuSign signature events)
- Operator-driven unblock steps (break-glass in incident runbooks)
- Any workflow step where the data or decision comes from outside the Resonate worker

## Basic shape

```python
from __future__ import annotations
import asyncio, os, time
from typing import TYPE_CHECKING
from resonate.resonate import Resonate
from resonate.types import Value

if TYPE_CHECKING:
    from resonate.context import Context

r = Resonate(url=os.environ.get("RESONATE_URL", "http://localhost:8001"))

async def notify_reviewer(ctx: Context, order_id: str, amount: int, approval_id: str) -> str:
    # Side effects live in leaves — this prints/notifies exactly once
    print(f"  order {order_id} (${amount}) needs review; promise id: {approval_id!r}")
    return approval_id

async def ship_order(ctx: Context, order_id: str) -> str:
    return f"shipped-{order_id}"

async def cancel_order(ctx: Context, order_id: str) -> str:
    return f"canceled-{order_id}"

async def fulfill_order(ctx: Context, order_id: str, amount: int) -> str:
    # Open the approval promise; its id is deterministic (derived from the workflow id)
    approval = ctx.promise()
    approval_id = await approval.id()

    # Notify the reviewer via a leaf (side effects belong in leaves)
    await ctx.run(notify_reviewer, order_id=order_id, amount=amount, approval_id=approval_id)

    # Suspend here — the worker holds no state while waiting
    # This can be seconds, hours, or days; no process needs to stay alive
    decision_raw = await approval

    if decision_raw.get("approve"):
        return await ctx.run(ship_order, order_id=order_id)
    return await ctx.run(cancel_order, order_id=order_id)
```

From outside the worker (a webhook, a UI route, a CLI):

```python
from resonate.types import Value

# Resolve the promise — wakes the suspended workflow
await r.promises.resolve(approval_id, Value(data={"approve": True, "note": "looks good"}))

# Or reject it
await r.promises.reject(approval_id, Value(data={"reason": "policy violation"}))
```

`r.promises.resolve/reject` are `async` — always `await` them.

## The canonical SDK example

The SDK's `examples/human-in-the-loop` shows the full pattern with a `ReviewerInbox` dependency (an `asyncio.Future`) so the simulated reviewer learns the promise id the instant the leaf creates it — without polling:

```python
import asyncio
from resonate.resonate import Resonate
from resonate.types import Value

class ReviewerInbox:
    def __init__(self) -> None:
        self.approval_id: asyncio.Future[str] = asyncio.get_running_loop().create_future()

    def publish(self, approval_id: str) -> None:
        if not self.approval_id.done():
            self.approval_id.set_result(approval_id)

async def notify_reviewer(ctx, order_id: str, amount: int, approval_id: str) -> str:
    print(f"  order {order_id} (${amount}) pending; id: {approval_id!r}")
    ctx.get_dependency(ReviewerInbox).publish(approval_id)
    return approval_id

async def fulfill_order(ctx, order_id: str, amount: int) -> str:
    approval = ctx.promise()
    approval_id = await approval.id()
    await ctx.run(notify_reviewer, order_id=order_id, amount=amount, approval_id=approval_id)
    decision_raw = await approval
    if decision_raw.get("approve"):
        return await ctx.run(ship_order, order_id=order_id, note=decision_raw.get("note", ""))
    return await ctx.run(cancel_order, order_id=order_id, note=decision_raw.get("note", ""))

async def main() -> None:
    r = Resonate(url="http://localhost:8001")
    inbox = ReviewerInbox()
    r.with_dependency(inbox)
    r.register(fulfill_order)
    try:
        wid = f"fulfill-{time.time_ns()}"
        handle = r.run(wid, fulfill_order, "order-42", 199)

        # Simulated reviewer: wait for the inbox, then resolve
        async def simulate_reviewer() -> None:
            approval_id = await inbox.approval_id
            await r.promises.resolve(approval_id, Value(data={"approve": True, "note": "ok"}))

        reviewer = asyncio.create_task(simulate_reviewer())
        out = await handle.result()
        await reviewer
        print(f"OK: {out}")
    finally:
        await r.stop()

asyncio.run(main())
```

## Promise IDs are deterministic

The orchestrator and the external resolver need to agree on the promise ID. When you call `ctx.promise()` without an explicit id, the SDK assigns a deterministic id derived from the workflow's own promise id (e.g., `{workflow_id}.1` for the first promise). Your leaf captures and publishes this id so the external resolver knows where to settle.

If you need a portable, human-readable id, pass a keyword argument (not yet in the v0.7.0 public API for `ctx.promise`; derive it from workflow inputs and embed it in the leaf instead). The safe pattern is always: open the promise, await its id, pass the id to a leaf for notification.

## Passing context to the resolver

Pass data to the resolver via a leaf that publishes the promise id alongside any routing metadata:

```python
async def notify_for_travel(
    ctx: Context, traveler_id: str, destination: str, approval_id: str
) -> str:
    # In a real system: Slack message, email, dashboard row
    print(f"  travel approval needed: {traveler_id} → {destination} [{approval_id}]")
    return approval_id
```

The reviewer sees the id and calls `r.promises.resolve(approval_id, Value(data=decision))`.

## Webhook-driven resolution

A typical FastAPI webhook route that resolves a promise:

```python
from fastapi import FastAPI, Request
from resonate.resonate import Resonate
from resonate.types import Value

app = FastAPI()
r = Resonate(url="http://localhost:8001")

@app.post("/webhooks/docusign")
async def docusign_callback(req: Request) -> dict:
    body = await req.json()
    envelope_id = body["envelopeId"]
    status = body["status"]

    if status == "completed":
        await r.promises.resolve(
            f"docusign:{envelope_id}",
            Value(data={"signed": True, "signed_at": body["completedAt"]}),
        )
    elif status in {"declined", "voided"}:
        await r.promises.reject(
            f"docusign:{envelope_id}",
            Value(data={"reason": status}),
        )

    return {"ok": True}
```

Note: the worker that is awaiting `ctx.promise()` is woken as soon as the webhook resolves.

## Multiple human approvers (parallel)

Fan out to independent approvers; wait for all to respond:

```python
async def multi_approver(ctx: Context, request_id: str, count: int) -> dict:
    # Open N approval promises in sequence (each gets a deterministic id)
    approvals = [ctx.promise() for _ in range(count)]
    ids = [await a.id() for a in approvals]

    # Notify each approver via a leaf
    for i, approval_id in enumerate(ids):
        await ctx.run(notify_approver, request_id, i, approval_id)

    # Await all decisions
    decisions = [await a for a in approvals]
    approved_count = sum(1 for d in decisions if d.get("approve"))

    if approved_count == count:
        return {"status": "approved"}
    return {"status": "rejected", "approved": approved_count, "of": count}
```

## Distinct Python idioms

- **`Value(data=...)`** — `r.promises.resolve/reject` wrap the payload in a `Value` object. Import from `resonate.types`.
- **`r.promises.*` are all `async`** — always `await` them.
- **FastAPI `async def` routes** work naturally with `await r.promises.resolve(...)`.
- **`ctx.get_dependency(ReviewerInbox)`** — dependencies are fetched by type, not string key.
- **`try/except` on `await approval`** — handles timeout/rejection naturally with Python's exception model.

## Avoid: polling

Bad:

```python
async def bad_approval(ctx: Context, order_id: str) -> dict:
    # Polling defeats the point
    while True:
        status = await ctx.run(check_approval_status, order_id)
        if status:
            break
        await ctx.sleep(timedelta(seconds=30))
    return {"status": "approved"}
```

`ctx.promise()` gives you wake-on-event; polling wastes checkpoints and loses the ability to pass decision data into the workflow.

## Related skills

- `resonate-basic-ephemeral-world-usage-python` — `r.promises.resolve/reject/get` from the ephemeral world (webhook handler side)
- `resonate-basic-durable-world-usage-python` — `ctx.promise`, `ctx.get_dependency`
- `resonate-saga-pattern-python` — sagas with human-gated steps
- `durable-execution` — foundational replay semantics; promise-wait survives crashes