Chat clients as steps with Microsoft.Extensions.AI

Inngest runs a function by calling it again after every step, so a call to a model that is not in a step is made again on every call, and paid for again. The Inngest.Extensions.AI package puts the calls of an IChatClient of Microsoft.Extensions.AI in steps: each call to the model, and each tool the model asks for, is made once, saved with the run and handed back from it on every replay. If a later step fails, its retry does not call the model again.

For a model call you make yourself, AIWrapAsync and InferAsync do the same one call at a time; see Calls to AI models. This package does it for a whole chat pipeline, tool loop included.

Setting it up

dotnet add package Inngest.Extensions.AI

The package builds on Microsoft.Extensions.AI. On net8.0 that brings System.Text.Json 10 with it, which is newer than the one the runtime ships, so an app on net8.0 that adds the package runs on the newer one.

It adds two stages to a ChatClientBuilder. Stages added first are the outermost, so the stage that runs the tools goes first and calls the stage that makes the model call a step, once for each round:

using Microsoft.Extensions.AI;

builder.Services
    .AddChatClient(model)
    .UseInngestFunctionInvocation() // the tools the model asks for, each a step
    .UseInngestSteps()              // each call to the model, a step
    .UseOpenTelemetry()             // sees only the calls that reach the model
    .UseLogging();

UseInngestFunctionInvocation takes the place of UseFunctionInvocation. It adds a FunctionInvokingChatClient, so its callback changes the same properties, such as MaximumIterationsPerRequest and AllowConcurrentInvocation. A pipeline with UseFunctionInvocation instead still has its model calls saved, but runs its tools in the function itself, on every call, and a tool that charges a card or sends mail then does it again on every replay.

Then use the client in a function as you would anywhere else:

using Microsoft.Extensions.AI;

var reply = await chat.GetResponseAsync(
    $"Write a short thank-you note for order {context.Event.Data.OrderId}.",
    cancellationToken: cancellationToken);

await context.Step.RunAsync("send-note", ct => email.SendAsync(context.Event.Data.CustomerId, reply.Text, ct));

Outside a run, and inside the handler of a step, where steps cannot be nested, the stages change nothing, so the same client serves a controller, a unit test and a function.

What the steps are

Call	Step	Input the Inngest UI shows	Saved with the run
GetResponseAsync	chat, then chat:1, chat:2, …	{"messages": [...], "options": {...}}	The ChatResponse as JSON, and a digest of the input
GetStreamingResponseAsync	The same ids	The same	The whole response, as JSON, and the digest
A tool call	tool <name>, then tool <name>:1, …	{"name": ..., "callId": ..., "arguments": {...}}	{"result": ..., "inputDigest": ...}

All of them are AI wrap steps, so the Inngest UI shows them as calls to a model with their input. The ids follow the rules of any step id: the same call gets the same id on every replay, and the SDK numbers repeats in the order they are made.

The digest is a hash of the input, not the input, written as "inngest": {"inputDigest": ...} next to the response, which readers of a ChatResponse skip. A replay compares it with the input of the call that gets the result, to catch the mix-up that Naming a step describes.

The messages, options and responses are written with AIJsonUtilities.DefaultOptions of Microsoft.Extensions.AI, which knows AIContent and its subtypes. Text, reasoning, function calls and results, usage and the finish reason all make it through, and the app registers nothing for them in its JsonSerializerOptions. Those options have source-generated contracts, so this works under Native AOT too. The SDK itself only handles JsonElement.

What JSON cannot hold does not come back, the same as with the caching clients of Microsoft.Extensions.AI: RawRepresentation of responses, messages and content is gone, and the values in AdditionalProperties come back as JsonElements. The call that makes a step gets the saved response too, as the result of every step is read back from JSON, so the code between calls sees the same on the first call and on replays.

Naming a step

Without a name, the model calls of a run are chat, chat:1 and so on, in the order they reach the client. That order is the same on every replay when nothing but steps is awaited before the calls: one after the other, or together with Task.WhenAll over calls that the function makes at once. It is not when something else is awaited first, such as a database query, an HttpClient request or a Task.Delay. Calls like that reach the client in the order those finish, which differs from one replay to the next, so which of them gets chat and which chat:1 changes, and each can be handed the response that was made for another.

Name each such call, so that its id does not depend on timing:

using Microsoft.Extensions.AI;

string[] userIds = ["u-1", "u-2", "u-3"];

// Each call waits for a lookup of its own, so the calls reach the client in the order the lookups finish.
var notes = await Task.WhenAll(userIds.Select(async userId =>
{
    var profile = await profiles.LoadAsync(userId, cancellationToken);
    return await chat.GetResponseAsync(
        $"Write a short thank-you note for {profile.Name}.",
        new ChatOptions().WithInngestStepId($"note-{userId}"),
        cancellationToken);
}));

WithInngestStepId puts the name in ChatOptions.AdditionalProperties, as a string, and the client takes it out before the options reach the model, so a provider never sends it on. A name gets :1, :2 for repeats like any id. Naming calls is also how to tell them apart in the Inngest UI.

To catch the mix-up, each saved result carries a digest of the input it was made for. When a replay hands a call a result with another digest than its own input has, the client logs a warning that names the step and WithInngestStepId. The warning comes from a logger that is not silenced while the run replays, so it appears once for each such step in each call of the function. It is a warning and not an exception, because a prompt with a part that changes between replays, such as the time, would then fail the run on every replay: it gets the warning on every replay, and the fix is to name the call, to keep the changing part out of the prompt, or to ignore the warning. An input edited in the Inngest UI does not change the digest, which is that of the input the code asked for.

Tool steps are named for the tool, and a model that calls the same tool twice in one response gets the ids in the order of its calls, so they need no name. Conversations that run side by side would claim the same tool ids in no fixed order, so a conversation that you name carries its name into its tool steps: with WithInngestStepId("plan") the steps are plan, plan tool get_weather, plan:1. The result of a tool is saved with a digest of its arguments and checked in the same way.

Streaming

A streaming call inside a run is one step that reads the model’s whole stream, saves the complete response and then hands it out as updates. The consumer gets the updates when the model has finished, not while it writes. A stream that nobody enumerates is not a call, as with any IChatClient.

To show the text while the model writes it, publish it to a realtime topic from the step:

using Microsoft.Extensions.AI;

var tokens = new RealtimeTopic($"order:{context.Event.Data.OrderId}", "tokens");

var reply = await chat
    .GetStreamingResponseAsync("Summarize the order.", new ChatOptions().WithInngestRealtimeStream(tokens), cancellationToken)
    .ToChatResponseAsync(cancellationToken);

Subscribers receive the text as a stream, as PublishStreamAsync sends it. A model sends an update every few characters, and Inngest takes at most 1,000 chunks of up to 1 KB in a stream, so the text is gathered into chunks. A chunk goes out when it is full, or when the text has waited 100 milliseconds, and the wait grows to a second once the stream has 500 chunks, so that a long answer does not use them up. A text that still outgrows the limit, which is about a megabyte, is published up to it and the rest is left out, with a warning in the log. The model call is not affected.

This is why the step publishes through IInngestRealtime and does not use the PublishStreamAsync step tool: a step cannot be called inside the handler of another, and the model call is the step that reads the stream. A publishing step of its own would also have to save the text, which leaves out the function calls, usage and finish reason of the response, and the two steps could disagree after a failure.

If the model fails in the middle of its stream, the step is retried, and the retry publishes the stream again from the start. Inngest ends a publish request that stops early like one that finished, so subscribers see the first stream end where the model failed, with the text up to there, and then a second stream with the whole text. The streams have different StreamIds and the same RunId, so a page that shows the text starts over at each StreamStart of its run. A replay publishes nothing. Failing to publish is logged as a warning and does not fail the model call, which would otherwise be made, and paid for, again.

Tools

A tool that the model asks for is a step named tool <name>. The arguments are its input, and the result is saved as {"result": ...}, so a replay hands the model the result without running the tool. The result comes back as a JsonElement, which is what a function made with AIFunctionFactory returns anyway; a tool without a result returns null.

A tool that throws is retried like any step, up to the function’s retries, and when they are used up FunctionInvokingChatClient reports the failure to the model, as it does for any tool that throws. An ArgumentException is the exception to the retries: the model chose the arguments, so the same ones fail again, and the step fails at once. Throw NonRetriableException from a tool to skip the retries for another reason.

The tools run in the handler of their step, outside the async flow of the function, so state that the function keeps in async-local variables is not visible to them. FunctionInvokingChatClient.CurrentContext is set there, so a tool can still ask the client to stop calling the model by setting Terminate, and a replay stops where the run did.

A tool that changes the context of its call, which FunctionInvokingChatClient.CurrentContext gives it, such as its Options or Messages, changes it only when the step runs. A replay hands back the saved result and does not run the tool, so the context stays as the client made it. Only Terminate is saved with the result and set again.

Tools the model asks for in one response are steps like any others made together: set AllowConcurrentInvocation and they run as parallel steps, each in a call of its own. Without it they run one after the other, each a step, and every step is a call to your function, unless checkpointing runs consecutive steps in one request.

Letting Inngest call the model

InngestInferChatClient is an IChatClient whose calls Inngest makes. Each call is an inference step: Inngest sends the request to a provider that speaks the OpenAI chat completions API, and no worker waits for the answer. The client turns the messages, tools and options into the request body and reads the response back into a ChatResponse, with its tool calls and usage.

using Microsoft.Extensions.AI;

IChatClient chat = new ChatClientBuilder(new InngestInferChatClient(
        new Uri("https://api.openai.com/v1/chat/completions"), "gpt-4o-mini", apiKey))
    .UseInngestFunctionInvocation()
    .Build();

It is a step already, so it needs no UseInngestSteps; UseInngestFunctionInvocation still makes the tools steps. Like every inference step it works inside a run only and throws elsewhere. Inngest stores the request with the run, the API key included. A streaming call returns the saved response as updates once the model has finished, as Inngest does not stream the answer to the SDK. It speaks the OpenAI chat format only; for another provider’s format, call InferAsync yourself.

Strict structured outputs are asked for as Microsoft’s adapter for the OpenAI client takes them: with a strict additional property, true or false, on the ChatOptions or on an AIFunction, which wins for that function. Without it, the request has no strict and the schemas of the response format and of the tools go out as Microsoft.Extensions.AI made them. With true, they are changed into what strict mode takes, which is no additional properties, every property required and no keywords OpenAI does not support, such as minLength, which move into the description, and the request says "strict": true.

An answer that is an error, in an error object, or that has no choices fails the call with what the provider said, as a NonRetriableException: Inngest saved the answer with the run, so a retry would read the same one. A refusal comes back as an ErrorContent with the code Refusal, and goes out as refusal when the assistant’s message is sent again.

The mapping leaves out reasoning content, author names, data content that is not an image, and the additional properties of messages and content. Of the options it sends ModelId, Instructions, Temperature, TopP, FrequencyPenalty, PresencePenalty, MaxOutputTokens, Seed, StopSequences, ResponseFormat, Tools, ToolMode and AllowMultipleToolCalls, and no others. MaxOutputTokens goes out as max_completion_tokens, which the current models take, and not as max_tokens.

Editing the input

The Inngest UI can rerun a step with an edited input. For a model call, the edited messages and options are what the model is called with, and the tools and the request factory of the call are kept as the code passed them, as they are not JSON. For a tool call, the edited arguments are what the tool is called with. An edit that cannot be read, such as one that is not an object with messages, or with arguments, or that names a kind of content that does not exist, fails the step at once, as reading it again would fail the same way. Values in the additionalProperties of an edit that are booleans, numbers or strings are handed to the model as a bool, a long or double, and a string, as a provider tests for those, and objects and arrays stay JsonElements.

Without an edit, the model gets the messages and options the caller passed, not a copy read back from the recorded JSON, so what JSON cannot hold, such as the raw representations of a provider, still reaches it.

What the input costs

Inngest stores the input of a step as plain JSON, which the encryption middleware does not cover, and the messages of a conversation repeat in the input of every model call of a run. Where the prompts must not be stored in the clear, or the conversations grow too large for that, leave the input out:

using Microsoft.Extensions.AI;

builder
    .UseInngestFunctionInvocation(configureSteps: options => options.RecordInput = false)
    .UseInngestSteps(options => options.RecordInput = false);

The steps are then plain run steps. The results are saved either way, and so are the model’s answers and the tools’ results, which is where the sensitive data usually is as well, and so is a digest of each input, which is a hash of it and not the input. A step whose input cannot be written as JSON, such as one with an additional property of a type the app has no contract for under Native AOT, runs without its input and logs a warning once.

Testing

The test host of Inngest.Testing runs these steps like any others. The simplest test gives the pipeline a fake IChatClient as the model, runs the function and counts the calls: the model is called once per step however many times the host replays the function. A step can also be answered by id, with the JSON of a ChatResponse; the steps of InngestInferChatClient are answered with StepMocks.Infer and the provider’s response:

using Inngest.Testing;
using Microsoft.Extensions.AI;

var answer = new ChatResponse(new ChatMessage(ChatRole.Assistant, "Thank you for your order!"));
var mocks = new StepMocks().Run("chat", JsonSerializer.SerializeToElement(answer, AIJsonUtilities.DefaultOptions));

var run = await host.RunAsync("send-note", @event, mocks, cancellationToken);

run.GetStep("chat") and run.GetStep("tool get_weather") show the input and output of each step.

What it does not do

• It wraps IChatClient only, and InngestInferChatClient is for the OpenAI chat format only. Other Microsoft.Extensions.AI clients, such as IEmbeddingGenerator, are not steps; wrap them with RunAsync.
• The updates of a stream arrive after the model has finished, not as it writes; publish to a realtime topic for that.
• Calls made inside a step handler are not steps of their own, so an agent loop run entirely inside one RunAsync is one step, with no tool steps.
• Inngest’s limits on the size of a step apply to the saved responses.