Background execution

Background execution lets a client submit work, disconnect, and return later. It requires server-managed state because the server needs a snapshot to track progress, liveness, completion, failure, and cancellation.

Use background execution for work that may outlive the request or the browser tab, such as report generation, long research tasks, multi-step planning, or tool-heavy workflows. Keep normal foreground streaming for short turns where the user is actively waiting and foreground cancellation is enough.

Store support matters for background work. See Session stores for which stores support snapshot status changes and aborting detached work.

Server requirements

Configure a store and expose companion endpoints when serving a remote agent:

from fastapi import FastAPI
from genkit.agent import FileSessionStore
from genkit_fastapi import serve_agent

report_agent = ai.define_agent(
    name='reportAgent',
    model='googleai/gemini-flash-latest',
    system='Create detailed research reports.',
    store=FileSessionStore('.sessions'),
)

app = FastAPI()
app.include_router(serve_agent(report_agent), prefix='/api')

The runtime writes a pending snapshot and refreshes its heartbeat while the background work processes the turn.

Detach from a turn

Submit a background task from the client using detach() on the AgentChat:

chat = report_agent.chat(session_id='report-123')
task = await chat.detach('Write the quarterly market report.')

# Save snapshot_id so you can poll or abort it later
snapshot_id = task.snapshot_id

Poll or wait

Use poll() to yield status snapshots until the task reaches a terminal status:

from genkit.agent import SnapshotStatus

async for snapshot in task.poll(interval=1.5):
    print('Current status:', snapshot.status)
    if snapshot.status == SnapshotStatus.COMPLETED:
        print(snapshot.state)

Use wait() to block until completion:

final_snapshot = await task.wait(interval=1.5)
if final_snapshot.status == SnapshotStatus.FAILED:
    print('Task failed:', final_snapshot.error.message if final_snapshot.error else None)

Terminal statuses are completed, failed, aborted, and expired.

Reconnect by snapshot ID

To reconnect and inspect or resume a detached task from a different client process, read the stored snapshot ID and load it:

snapshot = await report_agent.get_snapshot(snapshot_id=snapshot_id)

if snapshot and snapshot.status == SnapshotStatus.COMPLETED:
    chat = await report_agent.load_chat(snapshot_id=snapshot_id)
    res = await chat.send('Summarize this report.')
    print(res.text)

Only completed snapshots can be resumed.

Abort work

Cancel a pending task from the client using task.abort():

await task.abort()

Or abort directly by snapshot ID from the agent handle:

await report_agent.abort(snapshot_id)

Aborting sets the pending snapshot status to aborted. The background worker stops the turn when it observes that change. Long-running tools should check ctx.abort_signal.is_set() on ToolRunContext so they can exit cleanly.