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. Interrupt subclasses BaseException, so a broad except Exception: in tool code does not swallow it.
ToolApproval middleware from genkit_middleware to pause before any tool outside an allowlist runs, without changing tool code.

from decimal import Decimal

from pydantic import BaseModel

from genkit.exp import Genkit
from genkit.exp.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('gemini-flash-latest'),
    system='Always call userApproval before calling transfer_money.',
    tools=[user_approval, transfer_money],
    store=InMemorySessionStore(),
)

Inside a normal tool, accept ctx: ToolContext from genkit to inspect ctx.is_resumed() and ctx.resumed_metadata:

from typing import Any

from pydantic import BaseModel

from genkit import Interrupt, ToolContext

class RunShellInput(BaseModel):
    command: str

@ai.tool()
async def run_shell(input: RunShellInput, ctx: ToolContext) -> dict[str, Any]:
    """Run a shell command after checking approval metadata."""
    if 'rm ' in input.command and not (ctx.resumed_metadata or {}).get('tool_approved'):
        raise Interrupt({
            'command': input.command,
            'reason': 'Command deletes files and requires human approval.',
        })
    return {'command': input.command, 'exit_code': 0}

To gate existing tools without changing their code, add ToolApproval. Tools in allowed_tools run immediately; every other tool call pauses:

from genkit_middleware import ToolApproval

approval_agent = ai.define_agent(
    name='approvalAgent',
    model=GoogleAI.gemini('gemini-flash-latest'),
    system='You are a banking assistant.',
    tools=[transfer_money],
    use=[ToolApproval(allowed_tools=[])],
    store=InMemorySessionStore(),
)

Receive interrupts

Interrupted tool requests surface as AgentInterrupt items on res.interrupts at turn end, and as tool request chunks while streaming. Each item exposes name, ref, and input.

from genkit.exp.agent import AgentFinishReason

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

print(res.finish_reason == AgentFinishReason.INTERRUPTED)
# => True
for interrupt in res.interrupts:
    print(interrupt.name, interrupt.input)
    # => userApproval {'action': 'transfer', 'details': '$500 to savings'}

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.

approvals = [i for i in res.interrupts if i.name == 'userApproval']

if approvals:
    continued = await chat.resume(
        respond=[approvals[0].respond({'approved': True, 'note': 'Confirmed by user'})]
    )
    print(continued.text)

Restart pattern

Use restart() when the server-side tool should run again after approval. Pass resumed_metadata to hand the approval to the tool, which reads it from ctx.resumed_metadata. ToolApproval lets a restarted call through when resumed_metadata contains 'tool_approved': True.

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

if res.interrupts:
    continued = await chat.resume(
        restart=[i.restart(resumed_metadata={'tool_approved': True}) for i in res.interrupts]
    )
    print(continued.text)

The restart helper keeps the original tool call ref and input, so the tool runs again with exactly the arguments the model sent.

Resume validation

Before a resume reaches the model, the runtime checks every respond and restart entry against the tool requests recorded in session history, matching by name and ref from the newest message back. A restart must also carry the exact input of the request it matches. This stops a client from forging a tool call or re-running a tool with arguments the model never sent.

A rejected resume does not raise. The turn finishes with finish_reason=AgentFinishReason.FAILED, and the response’s error.status is 'INVALID_ARGUMENT':

from genkit import Part

continued = await chat.resume(
    respond=[Part.from_tool_response(name='userApproval', ref='forged-ref', output={})]
)

print(continued.finish_reason, continued.error.status)
# => failed INVALID_ARGUMENT