Interrupts

The interrupt system enables human-in-the-loop workflows. An agent can pause mid-loop, surface a question or approval request to the user, and resume from exactly where it stopped once a response is provided.

How it works

Raising an interrupt from a hook

Block a destructive tool call until the user approves:

from elsai import Agent
from elsai.hooks import BeforeToolCallEvent, HookProvider, hook

class ApprovalHook(HookProvider):
    def __init__(self, app_name: str):
        self.app_name = app_name

    @hook
    def approve(self, event: BeforeToolCallEvent) -> None:
        if event.tool_use["name"] != "delete_files":
            return

        approval = event.interrupt(
            f"{self.app_name}-approval",
            reason={"paths": event.tool_use["input"]["paths"]},
        )

        if approval.lower() != "y":
            event.cancel_tool = "User denied permission"

agent = Agent(hooks=[ApprovalHook("myapp")])

Raising an interrupt from a tool

Tools can interrupt via ToolContext:

from elsai import tool
from elsai.types.tools import ToolContext

@tool(context=True)
def delete_files(paths: list[str], tool_context: ToolContext) -> bool:
    """Delete files at the given paths after user confirmation."""
    approval = tool_context.interrupt(
        "myapp-approval",
        reason={"paths": paths},
    )
    if approval.lower() != "y":
        return False
    # proceed with deletion
    return True

Handling the interrupt response

Loop until stop_reason is no longer "interrupt":

result = agent(f"Delete files: {paths}")

while result.stop_reason == "interrupt":
    responses = []
    for interrupt in result.interrupts:
        if interrupt.name == "myapp-approval":
            user_input = input(f"Delete {interrupt.reason['paths']}? (y/N): ")
            responses.append({
                "interruptResponse": {
                    "interruptId": interrupt.id,
                    "response": user_input,
                }
            })
    result = agent(responses)

print(result)

Interrupt payload

Field	Type	Description
interrupt.id	str	Unique ID — required when sending a response
interrupt.name	str	App-defined name used to route responses
interrupt.reason	dict	JSON-serialisable context sent to the caller

Persisting interrupts across sessions

Combine with a SessionManager so state survives process restarts:

from elsai.session import FileSessionManager

agent = Agent(
    hooks=[ApprovalHook("myapp")],
    session_manager=FileSessionManager(
        session_id="myapp-session",
        storage_dir="./sessions",
    ),
)

Avoiding repeated prompts

Cache the approval in agent state so the user is only asked once per session:

@hook
def approve(self, event: BeforeToolCallEvent) -> None:
    if event.agent.state.get(f"{self.app_name}-trusted") == "true":
        return  # already approved this session

    approval = event.interrupt(f"{self.app_name}-approval", reason={})

    if approval.lower() == "y":
        event.agent.state.set(f"{self.app_name}-trusted", "true")
    else:
        event.cancel_tool = "User denied permission"

Multi-agent interrupts

Interrupts work in Swarm via BeforeNodeCallEvent:

from elsai.hooks import BeforeNodeCallEvent

class SwarmApprovalHook(HookProvider):
    @hook
    def approve(self, event: BeforeNodeCallEvent) -> None:
        if event.node_id != "cleanup":
            return
        approval = event.interrupt(
            "myapp-approval",
            reason={"node": event.node_id},
        )
        if approval.lower() != "y":
            event.cancel = "User denied node execution"

Key behaviours

• All hooks on an interrupted event execute before returning to the caller.
• Concurrent tool calls run to completion — results from finished tools are preserved when an interrupt fires mid-batch.
• A single hook or tool can raise multiple interrupts sequentially.