Custom orchestration

Most Genkit agents should use the standard prompt-backed loop. Custom orchestration is for cases where the application must control turn processing directly while still using the Agents API for sessions, snapshots, streaming, HTTP transport, and background execution. If you need complete ownership of the backend contract instead, a Genkit flow with direct generate() calls may be a better fit.

When to use a custom agent

Use ai.defineCustomAgent() when the workflow needs one of these behaviors:

Several model calls in one user turn.
Dynamic model selection or custom stopping rules.
Planner and executor loops with application decisions between calls.
Manual message history management.
Custom state and artifact updates before the final response.
Custom stream chunks that do not come directly from a single model call.

Runtime contract

A custom agent receives a SessionRunner and an options object:

async (sess, { sendChunk, abortSignal, context }) => {
  // Custom loop.
};

Use sess.run(async (input, turnContext) => {}) to process each input turn. The runner adds input.message to history before the callback runs. For server-managed agents, turnContext.snapshotId is reserved before the turn starts and reused when the turn snapshot is saved.

The runner exposes helpers for the parts of session state you are most likely to need. Use getState() for the full state, getMessages() and addMessages() for conversation history, getCustom() and updateCustom(fn) for typed application state, and getArtifacts() and addArtifacts(artifacts) for generated outputs.

Use updateCustom(fn) for progress and control data that the UI should react to during a turn. Use addArtifacts() when the agent produces a named output the user may inspect later, such as a report, patch, or JSON result. Custom state should stay compact because it is part of the conversation state that gets snapshotted or returned to the client.

Multi-step example

export const researchAgent = ai.defineCustomAgent(
  {
    name: 'researchAgent',
    description:
      'Plans research, answers subquestions, and synthesizes results.',
    stateSchema: ResearchStateSchema,
    store,
  },
  async (sess, { sendChunk, abortSignal }) => {
    let finalMessage;

    await sess.run(async (input, turnContext) => {
      const userText =
        input.message?.content.find((part) => part.text)?.text ?? '';
      const priorMessages = sess.getMessages();

      sess.updateCustom((state) => ({
        ...state,
        status: 'Decomposing question',
        turn: turnContext.turnIndex,
      }));

      const plan = await ai.generate({
        model: liteModel,
        prompt: `Break this question into three subquestions:\n${userText}`,
        output: { format: 'json', schema: z.array(z.string()).length(3) },
        abortSignal,
      });

      const subQuestions = plan.output ?? [userText];

      sess.updateCustom((state) => ({
        ...state,
        subQuestions,
        status: 'Researching',
      }));

      const answers = [];
      for (const question of subQuestions) {
        const answer = await ai.generate({
          prompt: `Answer in two paragraphs:\n${question}`,
          abortSignal,
        });
        answers.push({ question, answer: answer.text });
      }

      sess.updateCustom((state) => ({
        ...state,
        answers,
        status: 'Synthesizing',
      }));

      const stream = ai.generateStream({
        messages: priorMessages,
        prompt: `Synthesize these findings:\n${JSON.stringify(answers)}`,
        abortSignal,
      });

      for await (const chunk of stream.stream) {
        sendChunk({ modelChunk: chunk });
      }

      const response = await stream.response;
      finalMessage = response.message;

      if (response.message) {
        sess.addMessages([response.message]);
      }

      sess.addArtifacts([
        {
          name: `research-${turnContext.snapshotId}.json`,
          parts: [{ text: JSON.stringify(answers) }],
        },
      ]);

      return { finishReason: response.finishReason };
    });

    return {
      message: finalMessage,
      artifacts: sess.getArtifacts(),
      finishReason: sess.lastTurnFinishReason,
    };
  },
);

Use input.message for the current user message. The custom handler is not passed an input.messages array. Read history from sess.getMessages().

Failure and recovery

If the per-turn callback throws, the runtime ends the turn, emits its turn end, and resolves the invocation with finishReason: 'failed', or 'aborted' when the caller stopped the run. By default the turn rolls back. For client-managed agents, the response carries the last-good state. For server-managed agents, the response carries the last-good snapshot ID. This lets clients retry without preserving partial failed-turn mutations.

To keep the work a failed turn already did, throw CommittedTurnError from genkit/beta instead. The runner commits the session as the callback left it and reports it as the resume point. With a session store, it saves that state as a failed snapshot (aborted when the caller stopped the run) with the error on it; without one, the response carries that state. Throw it only when the session is at a point the turn can continue from.

A client retries a committed turn by sending an input with no message and no resume; see Resume after a failure or a stop. The runner passes that input to the callback as is, so handle it by running the turn again on the conversation already in the session:

import { CommittedTurnError } from 'genkit/beta';

await sess.run(async (input) => {
  // No message and no resume: a retry of a committed turn.
  const retry = !input.message && !input.resume;
  try {
    await runSteps(sess, { retry, abortSignal });
  } catch (e) {
    // Commit the steps that finished so the client can retry from here.
    throw new CommittedTurnError(e);
  }
});

Streaming custom data

sendChunk({ modelChunk }) forwards model chunks. sess.updateCustom() emits a customPatch chunk. sess.addArtifacts() records artifacts, while sendChunk({ artifact }) can stream an artifact chunk explicitly when the UI needs immediate visibility.