Skip to content

Interrupts & human-in-the-loop

Sometimes the agent shouldn't decide alone. A human approves the $4,000 refund. A release manager signs off before the deploy goes out. Policy requires a named person between recommendation and action.

Tulip treats human approval as a tool the model can call — same shape as any other tool, except it surfaces a question to your app and resumes when the human responds.

For consequential actions the same pause is enforced by the runtime, not offered to the model — a policy answering require_human holds the action at the admission gate.

The shape

The SDK ships a built-in ask_user tool. The model calls it like any other tool; instead of returning a value, the run pauses — the agent yields an InterruptEvent and stops. Your app reads the question off the event, gets an answer, and calls agent.resume(answer) to continue.

from tulip.agent import Agent
from tulip.core.events import InterruptEvent
from tulip.tools.decorator import tool

@tool(idempotent=True)
def issue_refund(order_id: str, amount: float) -> dict:
    return billing.refund(order_id, amount)

agent = Agent(
    model="anthropic:claude-sonnet-4-6",
    tools=[lookup_order, issue_refund],
    # ``ask_user`` is auto-registered only in explicit-completion mode; the
    # default is "auto", where it is absent and the prompt below would ask
    # for a tool the agent does not have.
    completion_mode="explicit",
    system_prompt=(
        "You are a customer-support agent. "
        "Always call ask_user for approval before issue_refund."
    ),
)

async for event in agent.run("Refund order ord-4821 if the return is eligible."):
    if isinstance(event, InterruptEvent):
        answer = input(f"{event.question} ")   # or Slack / web / email
        async for resumed in agent.resume(answer):
            print(resumed)

When the model calls ask_user, the runtime captures the question and yields an InterruptEvent(question=..., options=..., interrupt_id=...), then pauses. Your app surfaces the question to a human and threads the answer back via agent.resume(...). You can also expose your own pause-for-input by calling interrupt(...) from inside a tool body — ask_user is just the built-in wrapper.

Three ways the human responds

Synchronous — read from stdin

The simplest case for CLI agents and demos: write your tool to call input("[y/N] ") directly. The thread blocks until the human types.

@tool
def cli_approval(reason: str) -> dict:
    answer = input(f"{reason}\nApprove? [y/N] ").strip().lower()
    return {"approved": answer == "y", "reason": reason}

Async — checkpointer-mediated

For long-running workflows, the agent yields an InterruptEvent and pauses when the model calls ask_user. A separate process (browser, Slack action, email link) eventually resumes. agent.resume(...) is an async generator, so you iterate it — you don't await it:

async for event in agent.resume("approved"):
    handle(event)

The loop threads the response into the next Think and continues streaming events.

Steering — a second model votes

Not strictly human-in-the-loop, but lives in the same family. The SteeringHook runs an LLM-as-judge on every tool call before it fires:

from tulip.hooks.builtin.steering import SteeringHook
from tulip.models import get_model

agent = Agent(
    ...,
    hooks=[SteeringHook(
        # A model instance, not a provider string — the hook calls
        # ``.complete()`` on whatever it is given.
        model=get_model("anthropic:claude-sonnet-4-6"),
        policy="Reject any tool call that doesn't match the user's stated request.",
    )],
)

When the judge votes "no", the call is rejected and the agent re-plans. This is policy enforcement, not human review — but it's the same shape: a checkpoint between Think and Execute.

Cancelling a run mid-flight

Three ways to stop a running agent without waiting for the termination algebra to fire:

  1. Hook raises to short-circuit the loop. Any hook callback can raise to abort the run. Useful for budget guards.
class BudgetGuard(HookProvider):
    async def on_iteration_start(
        self, iteration: int, state: AgentState
    ) -> None:
        if state.total_tokens_used > 100_000:
            raise RuntimeError("token budget exceeded")
  1. Caller cancels the task. Standard asyncio cancellation:
run = asyncio.create_task(agent.run(prompt))
# ... later
run.cancel()
  1. agent.cancel(). Sets a flag the runner polls between nodes; the loop exits at the next safe point with TerminateEvent(reason="cancelled") (and result.stop_reason == "cancelled"). State still flushes to the checkpointer first, so the conversation can resume cleanly later.

In the agent.cancel() case the loop emits a final TerminateEvent(reason="cancelled") so your downstream observability gets a clean signal.

What you don't lose on cancel

Cancelled runs still persist state to the checkpointer. The thread_id retains the conversation up to the moment of cancel. You can resume later with the same thread, inspect the state for debugging, or branch off a new thread from the partial conversation.

See also