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 defineCustomAgent() when you need full control over the turn logic:

Running multiple model calls sequentially inside a single user turn.
Choosing models, prompts, or tools dynamically at runtime.
Implementing custom loops (e.g. planner-executor or self-correction).
Emitting status updates or artifacts during a turn.

Runtime contract

A custom agent receives a SessionRunner and an AgentFnOptions object:

final customAgent = ai.defineCustomAgent(
  name: 'myAgent',
  fn: (sess, options) async {
    // Custom logic...
  },
);

sess (the SessionRunner) manages the session’s active turns, messages, custom state, and artifacts.
options (the AgentFnOptions) provides sendChunk(chunk) to stream model chunks back to the client, along with a cancel token and ambient request context.

Call sess.run((input, turnContext) async => TurnResult) to process each input turn. The runner appends the user input to the message history automatically. For server-managed agents, turnContext.snapshotId is reserved beforehand so external state matches the snapshot.

Multi-step example

This simplified multi-step researcher is modeled directly on the research_agent.dart sample:

final researchAgent = ai.defineCustomAgent(
  name: 'researchAgent',
  // A loose JSON map is a good fit for ad-hoc status/progress state.
  stateSchema: SchemanticType.map(
    SchemanticType.string(),
    SchemanticType.dynamicSchema(),
  ),
  fn: (sess, options) async {
    Message? lastMessage;

    await sess.run((input, turnContext) async {
      final userText = input.message?.content.firstOrNull?.text ?? '';

      // Step 1: Decompose the question into subquestions.
      // The updater receives the typed state (Map<String, dynamic>? here).
      sess.updateCustom((state) {
        final s = state ?? <String, dynamic>{};
        s['status'] = 'Decomposing question...';
        return s;
      });

      final decompose = await ai.generate(
        model: liteModel,
        prompt: 'Break this question into two sub-questions:\n$userText',
        outputFormat: 'json',
        outputSchema: SchemanticType.list(SchemanticType.string()),
      );
      final subQuestions = (decompose.output ?? [userText])
          .map((q) => q.toString())
          .toList();

      // Step 2: Research each subquestion
      final answers = [];
      for (final q in subQuestions) {
        sess.updateCustom((state) {
          final s = state ?? <String, dynamic>{};
          s['status'] = 'Researching: $q';
          return s;
        });

        final research = await ai.generate(prompt: q);
        answers.add({'question': q, 'answer': research.text});
      }

      // Step 3: Synthesize and stream the final response
      sess.updateCustom((state) {
        final s = state ?? <String, dynamic>{};
        s['status'] = 'Synthesizing final response...';
        return s;
      });

      final synthesis = ai.generateStream(
        prompt: 'Synthesize these findings:\n$answers',
      );

      await for (final chunk in synthesis) {
        options.sendChunk(AgentStreamChunk(modelChunk: chunk.rawChunk));
      }

      final finalRes = await synthesis.onResult;
      lastMessage = finalRes.message;
      if (lastMessage != null) {
        sess.addMessages([lastMessage!]);
      }

      return null;
    });

    return AgentResult(
      message:
          lastMessage ??
          Message(
            role: Role.model,
            content: [TextPart(text: 'Research complete.')],
          ),
    );
  },
  store: InMemorySessionStore(),
);

Failure and recovery

If the per-turn callback throws an exception, the runtime marks the turn as failed and resolves the action with finishReason: 'failed'. The response carries the last-good state or snapshot ID, allowing the client to safely retry without continuing from broken or corrupted partial states.