Agent interrupts

Interrupts let a tool pause execution and return a tool request to the client. The client can approve, reject, provide missing data, refresh credentials, or ask the user a question, then resume the turn.

Use interrupts when the model can decide that outside input is needed but the tool should not proceed automatically. Common cases include human approval, missing user choices, risky operations, payments, external auth, and actions that need a fresh environment check.

Define an interrupt

ai.define_interrupt() for an interrupt-only tool that never performs work by itself and only asks the client for information.
raise Interrupt(metadata) inside a normal tool that can either finish immediately or pause based on runtime conditions.

from decimal import Decimal

from pydantic import BaseModel

from genkit import Genkit
from genkit.agent import InMemorySessionStore
from genkit_google_genai import GoogleAI

ai = Genkit(plugins=[GoogleAI()])

class UserApprovalInput(BaseModel):
    action: str
    details: str

user_approval = ai.define_interrupt(
    name='userApproval',
    description='Ask the user for approval before proceeding with a sensitive action.',
    input_schema=UserApprovalInput,
)

class TransferMoneyInput(BaseModel):
    amount: Decimal
    to_account: str

@ai.tool()
async def transfer_money(input: TransferMoneyInput) -> str:
    """Transfer money after approval."""
    return f'Transferred ${input.amount} to {input.to_account}.'

banking_agent = ai.define_agent(
    name='bankingAgent',
    model='googleai/gemini-flash-latest',
    system='If the user wants to transfer money, ALWAYS use userApproval before transfer_money.',
    tools=[user_approval, transfer_money],
    store=InMemorySessionStore(),
)

A normal tool can pause conditionally:

from genkit import Interrupt, ToolRunContext

@ai.tool()
async def run_shell(input: RunShellInput, ctx: ToolRunContext) -> dict:
    """Run a shell command after a safety check."""
    if is_risky(input.command) and not (ctx.resumed_metadata or {}).get('tool_approved'):
        raise Interrupt({
            'command': input.command,
            'reason': 'The command can modify files.',
        })
    return execute(input.command)

Receive interrupts

Interrupted tool requests surface on res.interrupts at turn end, and as tool request chunks while streaming.

res = await chat.send('Transfer $500 to savings.')

for interrupt in res.interrupts:
    print(interrupt.name)
    print(interrupt.input)

When an interrupt occurs, the turn finishes with finish_reason=AgentFinishReason.INTERRUPTED.

Respond pattern

Use respond() when the client already has the final tool output. Pass the part to chat.resume(respond=...) (or chat.resume_stream(respond=...) for streaming). The tool does not run again on the server.

res = await chat.send('Transfer $500 to savings.')
approvals = [i for i in res.interrupts if i.name == 'userApproval']

if approvals:
    continued = await chat.resume(
        respond=[approvals[0].respond({'approved': True, 'feedback': 'Looks good!'})]
    )
    print(continued.text)

Restart pattern

Use restart() when the server-side tool should run again after approval or a corrected request.

res = await chat.send('Deploy code to production.')
deploy_interrupts = [i for i in res.interrupts if i.name == 'deployApproval']

if deploy_interrupts:
    continued = await chat.resume(
        restart=[deploy_interrupts[0].restart()]
    )
    print(continued.text)

The restart helper preserves the original tool input and forces the server-side tool function to execute again with those parameters.

Resume validation

To prevent client-side spoofing, the runtime validates that every respond and restart entry matches an active paused tool call by name and unique reference ID.