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 genkitx.DefineCustomAgent when the prompt-backed loop does not fit:

The agent must call multiple models in one turn.
The workflow chooses models or tools dynamically.
You need custom retry, planning, or validation around each turn.
The agent emits artifacts or state updates outside a normal model stream.
You need the reserved turn snapshot ID before work starts.

Runtime contract

The custom function receives:

ctx is the invocation context, which is not the transport connection. Before a detach, a client disconnect cancels it. After a detach it stays live and the turn keeps running; only Agent.Abort or the process exiting cancels it. Every genkit.GenerateStream call made with this context follows the same rule.
resp aix.Responder streams model chunks and artifacts to the client.
sess *aix.SessionRunner[State] manages turns, messages, custom state, artifacts, and snapshots.

SessionRunner methods

SessionRunner[State] adds two methods of its own and embeds *aix.Session[State] for everything else:

Method	Purpose
`Run(ctx, fn) error`	Loops over the invocation’s inputs, calling `fn` once per turn.
`Result() *AgentResult`	The last message and artifacts currently recorded.
`State() *SessionState[State]`	The whole session state.
`SessionID() string`	The session this invocation belongs to.
`Messages() []*ai.Message`	Conversation history.
`AddMessages(...*ai.Message)`	Appends messages.
`SetMessages([]*ai.Message)`	Replaces history wholesale.
`UpdateMessages(func([]ai.Message) []ai.Message)`	Atomic read-modify-write on history.
`Custom() State`	Typed custom state.
`UpdateCustom(func(State) State)`	Atomic update; emits a custom patch chunk.
`Artifacts() []*Artifact`	Artifacts recorded so far.
`AddArtifacts(...*Artifact)`	Appends artifacts.
`UpdateArtifacts(func([]Artifact) []Artifact)`	Atomic read-modify-write on artifacts.

The three Update* callbacks run while the session lock is held. Do not call another Session method or send on a Responder from inside one.

Turn loop semantics

sess.Run(ctx, fn) loops over the invocation’s input channel, calling fn once per turn, and returns only when the invocation ends or fn returns an error. It is not a single turn, so put per-turn timeouts and retries inside fn, not around Run. Each turn runs in its own trace span. The runner adds the user message to the session before fn, then emits a TurnEnd chunk and writes a snapshot when a store exists.

fn returns (*aix.TurnResult, error). TurnResult has one field, FinishReason. Returning nil reports no finish reason and the framework infers nothing.

When fn returns a bare error, Run discards the turn: it emits TurnEnd with aix.AgentFinishReasonFailed, writes no snapshot, stops looping, and returns the error, so the previous turn’s snapshot stays the resume point. Return a non-nil TurnResult together with the error to commit the turn instead. The session as fn left it then persists as a failed snapshot carrying the error, which a caller can resume; do that only when the messages you added end at a turn seam, with every tool request answered, since that is what the next model call needs. A turn that ends because the invocation’s context was cancelled lands as aborted on the same terms. Either way, return the error to resolve the invocation, or call Run again to keep serving inputs on the same connection.

aix.AgentResult, the agent function’s return value, has three fields: Message, Artifacts, and FinishReason. Leave FinishReason empty to accept the last turn’s reason.

Custom agent example

import (
    "context"
    "fmt"

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

coder := genkitx.DefineCustomAgent(g, "coder",
    func(ctx context.Context, resp aix.Responder, sess *aix.SessionRunner[CoderState]) (*aix.AgentResult, error) {
        err := sess.Run(ctx, func(ctx context.Context, input *aix.AgentInput) (*aix.TurnResult, error) {
            turn := aix.TurnContextFromContext(ctx)

            sess.UpdateCustom(func(state CoderState) CoderState {
                state.Status = "Generating answer"
                state.LastSnapshotID = turn.SnapshotID
                return state
            })

            for chunk, err := range genkit.GenerateStream(ctx, g,
                ai.WithModelName("googleai/gemini-flash-latest"),
                ai.WithSystem("You are a concise coding assistant."),
                ai.WithMessages(sess.Messages()...),
            ) {
                if err != nil {
                    // A bare error discards the turn. Return a TurnResult with
                    // it to commit the messages added so far as a failed snapshot.
                    return nil, fmt.Errorf("stream model: %w", err)
                }
                if chunk.Done {
                    sess.AddMessages(chunk.Response.Message)
                    return &aix.TurnResult{
                        FinishReason: aix.AgentFinishReason(chunk.Response.FinishReason),
                    }, nil
                }
                resp.SendModelChunk(chunk.Chunk)
            }

            // No TurnResult: report no finish reason. The framework infers
            // nothing from a nil result.
            return nil, nil
        })
        if err != nil {
            // sess.Run stopped looping on the first callback error. Returning it
            // resolves the invocation as failed with the last-good state; calling
            // sess.Run again would keep serving inputs instead.
            return nil, fmt.Errorf("run turn: %w", err)
        }
        return sess.Result(), nil
    },
    aix.WithSessionStore(store),
    aix.WithDescription[CoderState]("Concise coding assistant"),
)

sess.Result() is a convenience that returns the last message and artifacts currently recorded in the session.

The coder agent in the basic-agents sample is this shape running alongside five agents built the ordinary way, so the extra work a custom agent takes on is easy to compare.

Turn context

aix.TurnContextFromContext(ctx) returns read-only turn metadata:

SnapshotID is the snapshot ID reserved before this turn runs. It is empty for client-managed agents.
ParentSnapshotID is the snapshot this turn continues from.
TurnIndex is the zero-based turn number within the invocation.

Use this when external resources need to line up with the snapshot that will later be saved.

Responder behavior

Responder.SendModelChunk(chunk) streams token-level model output. Responder.SendArtifact(artifact) streams an artifact and records it in the session. Send methods return promptly when the work context is canceled. Their session side effects are applied before they return, so snapshots and sess.Result() observe them.

Failure and detach behavior

If the per-turn callback returns an error, the invocation resolves as a failed AgentOutput with structured error details and a resume point: the failed turn’s own snapshot when the callback committed it, otherwise the previous turn’s. When a client detaches, chunks after detach are not forwarded, but session side effects such as artifacts still apply to the final snapshot.

Next steps

Sessions and state covers custom state, artifacts, and state transforms.
Background execution covers detach, pending snapshots, and abort.
Agent error handling covers the failure channels a custom agent resolves into.