Writing

Agent respond() and restart() in Genkit Python: Self-Correcting AI Loops

This post explains what each method does at the wire level, when to use each, and how to build a correction loop that is not vulnerable to infinite retries.

Tool calls fail. Models call tools with the wrong arguments. APIs return unexpected formats. A tool raises an exception and the agent loop stops dead. The naive fix is to put try/except around the tool and return an error string. That works, but it leaves the model with no mechanism to try again or get corrected context. PR #5004 improves respond() and restart() on AgentInterrupt, giving you two precise ways to handle tool failures and correction flows without restarting the entire agent session.

This post explains what each method does at the wire level, when to use each, and how to build a correction loop that is not vulnerable to infinite retries.

The interrupt mechanism

Before respond() and restart() make sense, you need to understand how Genkit tools pause an agent turn.

When a tool raises Interrupt, the model’s current turn stops. The agent framework captures the tool’s partial call, the interrupt payload, and hands you an AgentInterrupt object. The turn is not over; it’s suspended. You can inspect what the model asked for, decide what to do, and then resume the turn by calling either respond() or restart() on the interrupt object.

Here’s a minimal example. A tool pauses when it needs approval:

from genkit import Genkit, Interrupt, ToolRunContext
from genkit.plugins.google_genai import GoogleAI

ai = Genkit(plugins=[GoogleAI()], model='googleai/gemini-flash-latest')

@ai.tool()
async def send_email(to: str, subject: str, body: str, ctx: ToolRunContext) -> dict:
    """Send an email. Pauses for approval on first call."""
    if not ctx.is_resumed():
        raise Interrupt({'to': to, 'subject': subject, 'body': body, 'needs_approval': True})
    # Only reaches here after restart() approves it
    return {'status': 'sent', 'to': to}

The tool raises Interrupt on the first call. ctx.is_resumed() is False on first call and True when the tool runs again after restart(). This is the full contract: raise Interrupt to pause, use restart() to re-run with is_resumed=True, use respond() to provide a result without re-running.

respond(): providing a result without re-running the tool

respond() on AgentInterrupt injects a tool response directly into the conversation history and continues the turn. The tool does not run again. You are supplying the tool’s answer yourself.

From _client.py:

def respond(self, output: OutputT) -> AgentTurn[Any, Any]:
    """Builds a resume turn to answer the interrupt and continue the session."""
    resume_payload = Resume(
        respond=[
            ToolResponsePart(
                tool_response=ToolResponse(
                    name=self.name,
                    ref=self.ref,
                    output=output,
                )
            )
        ]
    )
    return self._session.resume(resume_payload)

The output you pass becomes the tool’s return value in the model’s eyes. The model then continues generating based on that injected result.

When to use respond():

You have a tool that pauses for a human decision (approve/decline). If the human declines, you use respond() to send back a “declined” result, and the model continues with that information. The tool never runs its actual side effect.

# Low-level ai.generate() version of respond
from genkit import respond_to_interrupt

response = await ai.generate(
    messages=messages,
    prompt='Send $250 to Jane for rent.',
    tools=[request_transfer],
)

while response.interrupts:
    interrupt = response.interrupts[0]
    print(f'Transfer requested: {interrupt.tool_request.input}')
    answer = input('Approve? [y/N]: ').strip().lower()

    if answer != 'y':
        # Decline: inject the "declined" result, tool does NOT run again
        decline = respond_to_interrupt(
            {'status': 'declined', 'reason': 'user denied'},
            interrupt=interrupt,
        )
        response = await ai.generate(
            messages=messages,
            resume_respond=decline,
            tools=[request_transfer],
        )
        break

respond_to_interrupt() is the ai.generate() version of AgentInterrupt.respond(). They do the same thing: inject a tool response into the message history and continue.

You can also use respond() to correct a bad tool result. If a tool returned malformed data and the model is confused, you can inject a cleaned-up version:

# Inject a corrected result after noticing the model produced bad output
correction = respond_to_interrupt(
    {'data': cleaned_data, 'note': 'Corrected format'},
    interrupt=interrupt,
)
response = await ai.generate(
    messages=messages,
    resume_respond=correction,
    tools=[my_tool],
)

restart(): re-running the tool with is_resumed=True

restart() causes the tool to run again. This time, ctx.is_resumed() returns True. The tool knows it’s the second run and can behave differently, typically by skipping the pause and executing the actual side effect.

From _client.py:

def restart(self) -> AgentTurn[Any, Any]:
    """Builds a resume turn that re-issues the tool call unchanged."""
    resume_payload = Resume(
        restart=[
            ToolRequestPart(
                tool_request=ToolRequest(
                    name=self.name,
                    ref=self.ref,
                    input=self.input,
                )
            )
        ]
    )
    return self._session.resume(resume_payload)

The input is the same as the original call. Nothing changes on the request side; only ctx.is_resumed() flips to True. You can optionally pass metadata via restart_tool() (the ai.generate() version) to give the tool context about why it’s being restarted:

from genkit import restart_tool

approval_restart = restart_tool(
    tool=request_transfer,
    interrupt=interrupt,
    resumed_metadata={'approved_by': 'admin', 'via': 'web_ui'},
)
response = await ai.generate(
    messages=messages,
    resume_restart=approval_restart,
    tools=[request_transfer],
)

Inside the tool, ctx.resumed_metadata contains whatever you passed:

@ai.tool()
async def request_transfer(body: TransferRequest, ctx: ToolRunContext) -> dict:
    if not ctx.is_resumed():
        raise Interrupt({'summary': f'Wire ${body.amount_usd} to {body.to_account}'})
    meta = ctx.resumed_metadata or {}
    return {'status': 'confirmed', 'approved_by': meta.get('approved_by', 'unknown')}

Pattern: restart loop with backoff for transient errors

restart() is also useful for transient failures. If a tool calls an external API that occasionally 503s, you can pause on failure, then restart after a delay:

import asyncio

@ai.tool()
async def fetch_market_data(ticker: str, ctx: ToolRunContext) -> dict:
    """Fetch market data. Pauses and retries on transient failures."""
    retry_count = (ctx.resumed_metadata or {}).get('retry_count', 0) if ctx.is_resumed() else 0

    try:
        data = await market_api.get(ticker)
        return {'ticker': ticker, 'price': data.price}
    except TransientError as e:
        if retry_count >= 3:
            raise RuntimeError(f'Market data unavailable after {retry_count} retries: {e}')
        raise Interrupt({'error': str(e), 'retry_count': retry_count, 'will_retry': True})

On the caller side, when you see the interrupt with will_retry=True, you wait with backoff before calling restart:

response = await ai.generate(prompt='Get AAPL price', tools=[fetch_market_data])

for attempt in range(4):
    if not response.interrupts:
        break
    interrupt = response.interrupts[0]
    meta = interrupt.metadata.get('interrupt', {})

    if not meta.get('will_retry'):
        break  # permanent failure, stop

    retry_count = meta.get('retry_count', 0)
    wait = 2 ** retry_count  # 1s, 2s, 4s
    await asyncio.sleep(wait)

    restart = restart_tool(
        tool=fetch_market_data,
        interrupt=interrupt,
        resumed_metadata={'retry_count': retry_count + 1},
    )
    response = await ai.generate(
        messages=response.messages,
        resume_restart=restart,
        tools=[fetch_market_data],
    )

A few things worth noticing here. The retry count is stored in resumed_metadata, which means it travels with the tool call through the interrupt mechanism. The tool itself decides when to give up (three retries) by raising a normal RuntimeError instead of Interrupt. And the caller knows it should retry because the interrupt payload explicitly says will_retry=True. If the tool just raised Interrupt with no signal, the caller would not know whether to retry or treat it as a permanent failure.

Pattern: combining respond() and restart() for self-correcting agents

A richer pattern: a tool that produces structured output sometimes returns invalid data. You want to validate the output, and if it’s bad, inject a correction via respond() so the model continues with good data rather than hallucinating on the bad output.

from pydantic import BaseModel, ValidationError

class ProductInfo(BaseModel):
    name: str
    price: float
    sku: str

@ai.tool()
async def lookup_product(query: str, ctx: ToolRunContext) -> dict:
    """Look up product info from the catalog API."""
    if not ctx.is_resumed():
        raw = await catalog_api.search(query)
        # Raise interrupt with raw data for validation
        raise Interrupt({'raw_data': raw, 'query': query})
    # Resumed path: the corrected data was injected via respond()
    return ctx.resumed_metadata or {}


# Caller validates and corrects
response = await ai.generate(
    prompt='Find me a red widget',
    tools=[lookup_product],
)

while response.interrupts:
    interrupt = response.interrupts[0]
    meta = interrupt.metadata.get('interrupt', {})
    raw = meta.get('raw_data', {})

    try:
        product = ProductInfo.model_validate(raw)
        # Valid: inject the validated result
        correction = respond_to_interrupt(product.model_dump(), interrupt=interrupt)
    except ValidationError as e:
        # Invalid: inject a cleaned-up fallback
        corrected = {
            'name': raw.get('title', 'Unknown product'),
            'price': float(raw.get('cost', 0)),
            'sku': raw.get('id', 'UNKNOWN'),
        }
        correction = respond_to_interrupt(corrected, interrupt=interrupt)

    response = await ai.generate(
        messages=response.messages,
        resume_respond=correction,
        tools=[lookup_product],
    )

The tool never executes the actual catalog API call in its normal path; it always pauses and hands the raw data to the caller for validation. The caller validates against the Pydantic schema and injects either the validated result or a corrected fallback. The model continues with clean data in either case.

Gotcha: infinite loops in respond()

A real trap: if the model responds to your injected result by calling the same tool again (because the injected result still looks like it needs action), you’ll loop indefinitely. Every respond() call can trigger another tool call that respond()s again.

The safest guard is a turn counter:

MAX_RESPOND_CYCLES = 5
cycle = 0

while response.interrupts and cycle < MAX_RESPOND_CYCLES:
    cycle += 1
    interrupt = response.interrupts[0]
    # ... validate and inject correction ...
    response = await ai.generate(
        messages=response.messages,
        resume_respond=correction,
        tools=[my_tool],
    )

if cycle >= MAX_RESPOND_CYCLES:
    raise RuntimeError(f'Exceeded {MAX_RESPOND_CYCLES} correction cycles — check tool output format')

Another signal to watch: if interrupt.tool_request.name on successive cycles is the same tool and the interrupt payload looks identical, the model is stuck. Break the cycle explicitly rather than injecting another correction that produces the same result.

Contrast with abort_signal

abort_signal (article #18) is for cancellation. If the user clicks stop, an external timer fires, or a Cloud Run instance is being shut down, you set the abort signal and the current turn is cancelled. No tool result is injected. The turn ends.

respond() and restart() are for continuation. The turn is paused, not cancelled. You provide information (or permission to retry) and the turn continues from where it stopped. These are different states: cancellation terminates the turn, interrupt/respond/restart extends it.

A confused mental model here leads to wrong behavior. If you use respond() when you meant to abort (say, because the user declined a transfer), the model receives a “declined” result and keeps talking, now explaining why the transfer was declined. That might be correct behavior for a banking agent. If you abort instead, the model output stops and you get no explanation. Which one you want depends on your UX. The framework gives you both options; you choose.

respond() and restart() on AgentSession

The examples above use ai.generate() directly. When working with AgentSession (the higher-level agent API), the same concepts apply but through AgentInterrupt on the stream:

from genkit.agent import InMemoryLatestStateStore, AgentInit

store = InMemoryLatestStateStore()
agent = ai.define_agent(name='transferAgent', model='googleai/gemini-flash-latest', store=store)

session = agent.connect(AgentInit())
turn = session.send('Wire $100 to Alice')

async for chunk in turn.stream:
    if chunk.tool_request and chunk.raw and turn.interrupt:
        interrupt = turn.interrupt
        print(f'Tool paused: {interrupt.name}, input: {interrupt.input}')

        approval = input('Approve? [y/N]: ').strip().lower()
        if approval == 'y':
            # restart: tool runs again with is_resumed=True
            next_turn = interrupt.restart()
        else:
            # respond: inject "declined", tool does NOT run again
            next_turn = interrupt.respond({'status': 'declined'})

        # consume the next turn
        async for next_chunk in next_turn.stream:
            if next_chunk.text:
                print(next_chunk.text, end='', flush=True)
        await next_turn.output

output = await turn.output
print(output.text)
await session.close()

AgentInterrupt.respond() and AgentInterrupt.restart() return AgentTurn objects you consume the same way as the original turn. The session state (message history, custom state) is updated automatically after the resumed turn completes.

When to use which

If you’re deciding how to handle a paused tool:

Use respond() when you want to supply the tool’s result yourself, without the tool running again. Human-in-the-loop rejections, validation corrections, and fallback data injection all fit here.

Use restart() when you want the tool to run again, this time knowing it has approval or corrected context. Human-in-the-loop approvals, transient error retries, and permission escalations all fit here.

The distinction is: does the tool need to execute its side effect? Yes: restart. No: respond.