Agent interrupts

Beta

The Agents API is in Beta. It can introduce breaking changes in minor version releases.

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.defineInterrupt() for an interrupt-only tool that never performs work by itself and only asks the client for information.
• ctx.interrupt(metadata) inside a normal tool that can either finish immediately or pause based on runtime conditions.

const askUser = ai.defineInterrupt({
  name: 'ask_user',
  description: 'Ask the user a clarification question.',
  inputSchema: z.object({
    question: z.string(),
    options: z.array(z.string()).min(2).max(5),
  }),
  outputSchema: z.object({
    answer: z.string(),
  }),
});

A normal tool can pause conditionally:

const runShell = ai.defineTool(
  {
    name: 'run_shell',
    description: 'Run a shell command after a safety check.',
    inputSchema: z.object({ command: z.string() }),
  },
  async (input, ctx) => {
    if (isRisky(input.command) && !ctx.resumed?.toolApproved) {
      ctx.interrupt({
        command: input.command,
        reason: 'The command can modify files.',
      });
    }

    return execute(input.command);
  },
);

Receive interrupts

Interrupted tool requests surface on res.interrupts and as tool request chunks while streaming.

const res = await chat.send('Run the migration.');

for (const interrupt of res.interrupts) {
  console.log(interrupt.name);
  console.log(interrupt.input);
}

The agent finish reason is interrupted when the model turn pauses on one or more interrupts.

Respond pattern

Use respond() when the client has the final tool output. The method builds a tool response part. Send that part with chat.resume().

const res = await chat.send('Transfer $50 to Robin.');
const approval = res.interrupts.find((i) => i.name === 'userApproval');

if (approval) {
  const continued = await chat.resume({
    respond: [
      approval.respond({
        approved: true,
        approver: 'alex@example.com',
      }),
    ],
  });

  console.log(continued.text);
}

This pattern is useful for approvals where the tool does not need to run again. The response you provide becomes the tool result.

Restart pattern

Use restart() when the original tool should execute again after the app changes environment or metadata. The method builds a tool request part. Send it with chat.resume().

const res = await chat.send('Run the deployment command.');
const command = res.interrupts.find((i) => i.name === 'run_shell');

if (command) {
  const continued = await chat.resume({
    restart: [command.restart()],
  });

  console.log(continued.text);
}

The AgentInterrupt.restart() convenience preserves the original tool input.

Use respond for approvals where the client supplies the final answer. Use restart when the server-side tool should run again after approval, refreshed credentials, or updated environment state.

Resume validation

The runtime validates resume payloads against session history. A respond entry must match an interrupted tool request by name and ref. A restart entry must match the original tool request, and its input must not be modified. This protects server tools from forged client resume payloads.

Streaming interrupts

const turn = chat.sendStream('Book the hotel.');

for await (const chunk of turn.stream) {
  for (const request of chunk.toolRequests) {
    if (request.metadata?.interrupt) {
      showApproval(request);
    }
  }
}

const res = await turn.response;

Wait for the final response before treating the turn as durably interrupted, because the final response carries the normalized interrupt helpers.

Interrupts are one feature with two transports. The tool side is the same as on Interrupts: genkit.DefineResumableTool, tool.Interrupt, ai.InterruptAs, and the claim through the tool’s Interrupted method, whose Restart and Respond verbs build the resume parts. Only the delivery differs. A plain generate call takes the resume parts through ai.WithResume; an agent takes them through conn.SendResume or AgentInput.Resume, which the agent forwards to the same generate resume path.

Interrupt from a tool

Define a resumable tool with genkit.DefineResumableTool. Its third type parameter is the resume payload: the function receives it as a pointer that is nil on the first call, and that holds the client’s answer when the turn is resumed. The tool pauses by returning tool.Interrupt(ctx, data); on resume it runs again from the top with the payload set. Keep approval and execution logic in one tool, and require the client to explicitly resume the turn.

import (
    "context"
    "errors"
    "fmt"

    "github.com/firebase/genkit/go/ai"
    aix "github.com/firebase/genkit/go/ai/exp"
    "github.com/firebase/genkit/go/ai/tool"
    "github.com/firebase/genkit/go/genkit"
)

type TransferInput struct {
    To     string  `json:"to"`
    Amount float64 `json:"amount"`
}

type TransferOutput struct {
    ConfirmationID string `json:"confirmationId"`
}

// TransferInterrupt is what the tool says when it pauses. The input already
// says what the model asked for, so this holds only the tool's reason.
type TransferInterrupt struct {
    Reason string `json:"reason"`
}

// Confirmation is the resume payload the client sends back to approve or
// reject the paused transfer.
type Confirmation struct {
    Approved bool `json:"approved"`
}

transferMoney := genkit.DefineResumableTool(g, "transferMoney",
    "Transfer money after user approval.",
    func(ctx context.Context, input TransferInput, confirm *Confirmation) (TransferOutput, error) {
        if confirm == nil {
            return TransferOutput{}, tool.Interrupt(ctx, TransferInterrupt{
                Reason: "Approval is required before transferring money.",
            })
        }
        if !confirm.Approved {
            return TransferOutput{}, errors.New("transfer rejected by user")
        }

        return TransferOutput{ConfirmationID: "txn-123"}, nil
    },
)

DefineResumableTool returns *ai.ResumableToolAction[In, Out, Res], so the tool above has type *ai.ResumableToolAction[TransferInput, TransferOutput, Confirmation]. Write the type out when the tool must outlive the enclosing function:

type App struct {
    Transfer *ai.ResumableToolAction[TransferInput, TransferOutput, Confirmation]
}

The banker agent in the basic-agents sample runs this end to end: its tool interrupts with a discriminated payload, and the CLI asks the user and resumes the turn.

Collect interrupts from the stream

Interrupts are model tool request parts that are paused, which part.IsInterrupt() reports.

var interrupts []*ai.Part

for chunk, err := range conn.Receive() {
    if err != nil {
        return fmt.Errorf("stream turn: %w", err)
    }
    if chunk.ModelChunk != nil {
        interrupts = append(interrupts, chunk.ModelChunk.Interrupts()...)
    }
    if chunk.TurnEnd != nil {
        break
    }
}

Claim each part with the tool that raised it. The claim decodes the input to the tool’s input type, and ai.InterruptAs[T] decodes the data the tool paused with:

call, ok := transferMoney.Interrupted(interrupts[0])
if ok {
    meta, _ := ai.InterruptAs[TransferInterrupt](call.Part)
    fmt.Printf("Approve transfer of %.2f to %s? (%s)\n", call.Input.Amount, call.Input.To, meta.Reason)
}

Interrupted reports false for an interrupt of another tool and for a hold that middleware such as ToolApproval raised. Claim holds as shown in Answer middleware holds.

Recover interrupts from a snapshot

(*ai.ModelResponse).Interrupts() and chunk.ModelChunk.Interrupts() cover the live paths. From a stored snapshot there is no response object, so filter the messages yourself: take the newest message whose Role is ai.RoleModel and keep the parts for which part.IsInterrupt() reports true. *ai.Part implements MarshalJSON and UnmarshalJSON, so an interrupt part can be persisted in your own queue, reloaded, and claimed as is.

// BankState is the agent's custom state type.
type BankState struct {
    Balance float64 `json:"balance,omitempty"`
}

func pendingInterrupts(snap *aix.SessionSnapshot[BankState]) []*ai.Part {
    if snap == nil || snap.State == nil {
        // State is nil on a pending snapshot: detached work is still running.
        return nil
    }

    for i := len(snap.State.Messages) - 1; i >= 0; i-- {
        msg := snap.State.Messages[i]
        if msg.Role != ai.RoleModel {
            continue
        }

        var pending []*ai.Part
        for _, part := range msg.Content {
            if part.IsInterrupt() {
                pending = append(pending, part)
            }
        }
        return pending
    }
    return nil
}

Stop at the first model message. Resume validation only searches the most recent model response, so parts from an earlier turn are rejected as stale.

Resume an interrupted turn

Build resume parts from the claimed call, then send them with conn.SendResume. Use Restart to re-execute the tool, delivering the client’s typed answer to its resume parameter. The tool runs again from the top, this time with a non-nil payload. The claim already checked the part, so the verbs cannot fail.

if err := conn.SendResume(&aix.ToolResume{
    Restart: []*ai.Part{call.Restart(Confirmation{Approved: true})},
}); err != nil {
    return fmt.Errorf("send resume: %w", err)
}

Use Respond when the result should be the tool output and the tool should not run again, such as supplying a precomputed result:

if err := conn.SendResume(&aix.ToolResume{
    Respond: []*ai.Part{call.Respond(TransferOutput{ConfirmationID: "manual-approval"})},
}); err != nil {
    return fmt.Errorf("send resume: %w", err)
}

call.RestartWithInput(input, resume) re-executes the tool with revised input, and goes in Restart too.

A resumed turn can interrupt again. Streaming clients should handle interrupts in a loop: receive until TurnEnd, resolve any interrupts, send resume, then receive the continuation.

The same resumable-tool API works outside an agent. The basic-tool-interrupts sample runs the pause-and-resume cycle across two ordinary generate turns instead.

Resume without a live connection

An AgentConnection is a convenience, not a requirement. aix.AgentInput carries the same payload on its Resume field, and Agent.Run accepts invocation options alongside a full AgentInput, so one call resumes a session from any process:

out, err := agent.Run(ctx, &aix.AgentInput{
    Resume: &aix.ToolResume{Restart: []*ai.Part{call.Restart(Confirmation{Approved: true})}},
}, aix.WithSessionID[BankState](sessionID))

An AgentInput carrying only Resume, with no Message, is valid. The two fields of aix.ToolResume match the verbs:

Field	Type	Holds parts from	Effect
Restart	[]*ai.Part	call.Restart, call.RestartWithInput	Re-runs the tool with the typed resume payload attached.
Respond	[]*ai.Part	call.Respond	Supplies the tool output directly; the tool does not run again.

Both may appear in one payload. Over HTTP the same object rides in data.resume.

Resume from another process

The process that paused the turn does not need to be the one that answers it. Read the session’s latest snapshot, find the pending interrupt parts, claim them with the tool, then run:

func approveTransfer(
    ctx context.Context,
    agent *aix.Agent[BankState],
    transferMoney *ai.ResumableToolAction[TransferInput, TransferOutput, Confirmation],
    sessionID string,
) error {
    snap, err := agent.GetLatestSnapshot(ctx, sessionID)
    if err != nil {
        return fmt.Errorf("read latest snapshot: %w", err)
    }
    if snap.FinishReason != aix.AgentFinishReasonInterrupted {
        return fmt.Errorf("session %q is not waiting on an interrupt", sessionID)
    }

    var restarts []*ai.Part
    for _, part := range pendingInterrupts(snap) {
        call, ok := transferMoney.Interrupted(part)
        if !ok {
            continue
        }
        fmt.Printf("Approve transfer of %.2f to %s?\n", call.Input.Amount, call.Input.To)
        restarts = append(restarts, call.Restart(Confirmation{Approved: true}))
    }
    if len(restarts) == 0 {
        return errors.New("no pending interrupts for this tool")
    }

    out, err := agent.Run(ctx, &aix.AgentInput{
        Resume: &aix.ToolResume{Restart: restarts},
    }, aix.WithSessionID[BankState](sessionID))
    if err != nil {
        return fmt.Errorf("resume session: %w", err)
    }

    fmt.Println(out.Message.Text())
    return nil
}

The server re-validates the payload with aix.ValidateResumeAgainstHistory before it reaches the model, so a forged restart is rejected.

Durability

An interrupted turn is an ordinary completed turn. With a session store it writes a snapshot whose Status is completed and whose FinishReason is interrupted, so it is a resume point like any settled turn; see the status table on Background execution. Read FinishReason, not Status, to tell an interrupt apart from a plain finished turn.

There is no expiry on a paused turn. The pending and expired statuses apply only to detached background work. The only constraint is that a resume must target tool requests in the snapshot’s most recent model message, so answer the interrupt before sending an unrelated user message on the same session.

Resume validation

The runtime validates a resume payload against session history before it reaches the model. A respond entry must match an interrupted tool request by name and ref. A restart entry must match the original request and carry its input. A restart built with RestartWithInput carries the revised input and records the original one it replaced, and it is the recorded original that must match. This protects server tools from forged client payloads that change the input of the call the model made.

Only the most recent model response is searched. An entry naming a tool request from an earlier turn is rejected as stale, separately from one naming a request that never existed. Refs are only unique within a single model response, so searching further back would let a caller validate a restart whose input was forged from an already-settled call. A client that accumulates resume entries across turns and resends them all must send only the entries for the pending response.

genkitx.DefineAgent and genkitx.DefinePromptAgent run this check for you. A custom agent that accepts AgentInput.Resume from untrusted callers should run it itself before forwarding the payload:

if input.Resume != nil {
    if err := aix.ValidateResumeAgainstHistory(input.Resume, sess.Messages()); err != nil {
        return nil, err
    }
}

The check accepts a revised input. A server that must not let callers revise a tool’s input rejects restart parts whose Restart.OriginalInput is set, as shown in Resuming from an untrusted caller.

Next steps

• Interrupts covers the same tool API outside an agent turn.
• Sessions and state explains snapshots and the FinishReason field a paused turn records.
• Background execution covers detaching from a turn and reading its snapshot later.

Define an interrupt

An interrupt pauses execution and prompts the client for external validation or missing inputs. Use ai.defineInterrupt() for an interrupt-only tool that always pauses, or a regular tool that returns .interrupt() when it should pause only under certain conditions. The example below uses the second form:

import 'package:genkit/genkit.dart';
import 'package:genkit/experimental.dart';
import 'package:schemantic/schemantic.dart';

part 'banking_agent.g.dart';

@Schema()
abstract class $UserApprovalInput {
  @Field(description: 'The action to be approved')
  String get action;
  @Field(description: 'Details about the action')
  String get details;
}

final userApproval = ai.defineTool(
  name: 'userApproval',
  description: 'Ask the user for approval before proceeding with a sensitive action.',
  inputSchema: UserApprovalInput.$schema,
  // No outputSchema needed: the output is provided by the client on resume
  fn: (input, ctx) async => .interrupt(),
);

final bankingAgent = ai.defineAgent(
  name: 'bankingAgent',
  system: 'If the user wants to transfer money, ALWAYS use userApproval.',
  tools: [userApproval, transferMoney],
);

Receive interrupts

Interrupted tool requests surface on res.interrupts at turn end, or can be checked as tool requests inside stream chunks.

final res = await chat.send(text: 'Transfer $500 to savings.');

for (final interrupt in res.interrupts) {
  print(interrupt.name);
  print(interrupt.input);
}

When an interrupt occurs, the agent’s final turn finishReason resolves as interrupted.

Respond pattern

Use respond() when the client has final tool outputs ready. This builder returns a ToolResponsePart that you pass to chat.resume() via its respond parameter. This resolves the tool call directly without re-executing the tool function on the server.

final res = await chat.send(text: 'Transfer $500 to savings.');
final approvals = res.interrupts.where((i) => i.name == 'userApproval').toList();

if (approvals.isNotEmpty) {
  final continued = await chat.resume(
    respond: [
      approvals.first.respond({'approved': true, 'feedback': 'Looks good!'}),
    ],
  );
  print(continued.text);
}

Restart pattern

Use restart() when the tool itself should execute again on the server after state, parameters, or external configurations are corrected by the user.

final res = await chat.send(text: 'Deploy code to production.');
final deployInterrupts = res.interrupts.where((i) => i.name == 'deployApproval').toList();

if (deployInterrupts.isNotEmpty) {
  final continued = await chat.resume(
    restart: [
      deployInterrupts.first.restart(),
    ],
  );
  print(continued.text);
}

The restart builder 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.

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