Stream the tool calling rounds of the Anthropic messages API

app/MindWork AI Studio/Provider/Anthropic/AnthropicContentBlockBuilder.cs

@@ -0,0 +1,227 @@
+using System.Buffers;
+using System.Text;
+using System.Text.Json;
+
+namespace AIStudio.Provider.Anthropic;
+
+/// <summary>
+/// Puts one streamed content block back together.
+/// </summary>
+/// <remarks>
+/// A block opens with a seed, grows through fragments, and has to end up as the very block the
+/// provider would have sent had we not streamed: it goes back on the next request, and Anthropic
+/// checks what it gets. A thinking block is the sharp edge here -- its signature has to return
+/// byte for byte with the text it was made for, or the next round is refused with a 400.<br/><br/>
+/// This is a pure function over bytes: no HTTP, no state beyond the block itself. That is what
+/// makes it the piece worth testing against recorded streams.
+/// </remarks>
+public sealed class AnthropicContentBlockBuilder
+{
+    private const string TYPE_TEXT = "text";
+    private const string TYPE_TOOL_USE = "tool_use";
+    private const string TYPE_THINKING = "thinking";
+    
+    private const string DELTA_TEXT = "text_delta";
+    private const string DELTA_INPUT_JSON = "input_json_delta";
+    private const string DELTA_THINKING = "thinking_delta";
+    private const string DELTA_SIGNATURE = "signature_delta";
+    
+    private const string EMPTY_OBJECT = "{}";
+    
+    private readonly JsonElement seed;
+    private readonly StringBuilder text = new();
+    private readonly StringBuilder toolArguments = new();
+    private readonly StringBuilder thinking = new();
+    private string signature;
+    
+    /// <summary>
+    /// Opens a block from the seed the provider sent for it.
+    /// </summary>
+    /// <param name="contentBlock">The block as it opened.</param>
+    public AnthropicContentBlockBuilder(JsonElement contentBlock)
+    {
+        //
+        // The seed is cloned because the document it was read from is gone by the time this block
+        // is built, and an element which outlives its document reads memory that is no longer
+        // there.
+        //
+        this.seed = contentBlock.ValueKind is JsonValueKind.Object ? contentBlock.Clone() : default;
+        this.BlockType = ReadString(this.seed, "type");
+        
+        //
+        // Anthropic seeds a block with what it already has, which is usually nothing. When it is
+        // not nothing, it belongs in front of everything that follows.
+        //
+        this.text.Append(ReadString(this.seed, TYPE_TEXT));
+        this.thinking.Append(ReadString(this.seed, TYPE_THINKING));
+        this.signature = ReadString(this.seed, "signature");
+    }
+    
+    /// <summary>
+    /// What kind of block this is: text, a tool use, thinking, or something we do not know.
+    /// </summary>
+    public string BlockType { get; }
+    
+    /// <summary>
+    /// The ID of the tool use, for a tool use block.
+    /// </summary>
+    public string ToolUseId => ReadString(this.seed, "id");
+    
+    /// <summary>
+    /// The tool arguments as they came off the wire, set only when they never parsed into an object.
+    /// </summary>
+    /// <remarks>
+    /// The block itself carries an empty object then, because that is what may go back to the
+    /// provider. The call still has to be rejected rather than run with no arguments at all,
+    /// which is what this text is for.
+    /// </remarks>
+    public string? UnparsableToolArguments { get; private set; }
+    
+    /// <summary>
+    /// Adds the next piece of this block.
+    /// </summary>
+    /// <param name="delta">The piece as it arrived.</param>
+    /// <returns>The text to show, empty for every piece which is not text.</returns>
+    public string Append(AnthropicStreamDelta delta)
+    {
+        switch (delta.Type)
+        {
+            case DELTA_TEXT when delta.Text is not null:
+                this.text.Append(delta.Text);
+                return delta.Text;
+            
+            case DELTA_INPUT_JSON when delta.PartialJson is not null:
+                this.toolArguments.Append(delta.PartialJson);
+                return string.Empty;
+            
+            case DELTA_THINKING when delta.Thinking is not null:
+                this.thinking.Append(delta.Thinking);
+                return string.Empty;
+            
+            case DELTA_SIGNATURE when delta.Signature is not null:
+                this.signature = delta.Signature;
+                return string.Empty;
+            
+            default:
+                return string.Empty;
+        }
+    }
+    
+    /// <summary>
+    /// Builds the finished block, in the shape a non-streamed call would have returned it.
+    /// </summary>
+    public JsonElement Build()
+    {
+        switch (this.BlockType)
+        {
+            case TYPE_TEXT:
+                return this.BuildFromSeed(new()
+                {
+                    ["type"] = JsonSerializer.Serialize(TYPE_TEXT),
+                    ["text"] = JsonSerializer.Serialize(this.text.ToString()),
+                });
+            
+            case TYPE_THINKING:
+                //
+                // The signature travels with the thinking it belongs to. Anthropic refuses the
+                // next round without it, so it is written even when it stayed empty: a missing
+                // field and an empty one fail the same way, and the empty one says where to look.
+                //
+                return this.BuildFromSeed(new()
+                {
+                    ["type"] = JsonSerializer.Serialize(TYPE_THINKING),
+                    ["thinking"] = JsonSerializer.Serialize(this.thinking.ToString()),
+                    ["signature"] = JsonSerializer.Serialize(this.signature),
+                });
+            
+            case TYPE_TOOL_USE:
+                return this.BuildFromSeed(new()
+                {
+                    ["input"] = this.BuildToolInput(),
+                });
+            
+            default:
+                //
+                // Redacted thinking and anything we have not seen before go back untouched. We
+                // cannot read them, which is precisely why we must not rewrite them either.
+                //
+                return this.seed;
+        }
+    }
+    
+    /// <summary>
+    /// The tool arguments as the JSON object they have to be.
+    /// </summary>
+    /// <remarks>
+    /// A tool without arguments gets no fragment at all, so an empty buffer is an empty object.
+    /// A buffer which is not an object is kept aside instead: the block needs something the
+    /// provider accepts, while the call needs the text that made it invalid.
+    /// </remarks>
+    private string BuildToolInput()
+    {
+        var arguments = this.toolArguments.ToString();
+        if (string.IsNullOrWhiteSpace(arguments))
+            return EMPTY_OBJECT;
+        
+        try
+        {
+            using var document = JsonDocument.Parse(arguments);
+            if (document.RootElement.ValueKind is JsonValueKind.Object)
+                return arguments;
+        }
+        catch (JsonException)
+        {
+            // Falls through to the same place a well-formed non-object does:
+        }
+        
+        this.UnparsableToolArguments = arguments;
+        return EMPTY_OBJECT;
+    }
+    
+    /// <summary>
+    /// Writes the given properties over a copy of the seed.
+    /// </summary>
+    /// <remarks>
+    /// Copying rather than rebuilding keeps whatever the provider sent along that we do not know
+    /// about. The values are JSON text, so that a string is escaped exactly once.
+    /// </remarks>
+    /// <param name="overrides">The properties to write, as property name to JSON text.</param>
+    private JsonElement BuildFromSeed(Dictionary<string, string> overrides)
+    {
+        var buffer = new ArrayBufferWriter<byte>();
+        using (var writer = new Utf8JsonWriter(buffer))
+        {
+            writer.WriteStartObject();
+            if (this.seed.ValueKind is JsonValueKind.Object)
+                foreach (var property in this.seed.EnumerateObject())
+                {
+                    if (overrides.ContainsKey(property.Name))
+                        continue;
+                    
+                    property.WriteTo(writer);
+                }
+            
+            foreach (var (propertyName, json) in overrides)
+            {
+                writer.WritePropertyName(propertyName);
+                using var value = JsonDocument.Parse(json);
+                value.RootElement.WriteTo(writer);
+            }
+            
+            writer.WriteEndObject();
+        }
+        
+        using var document = JsonDocument.Parse(buffer.WrittenMemory);
+        return document.RootElement.Clone();
+    }
+    
+    private static string ReadString(JsonElement item, string propertyName)
+    {
+        if (item.ValueKind is not JsonValueKind.Object ||
+            !item.TryGetProperty(propertyName, out var property) ||
+            property.ValueKind is not JsonValueKind.String)
+            return string.Empty;
+
+        return property.GetString() ?? string.Empty;
+    }
+}

app/MindWork AI Studio/Provider/Anthropic/AnthropicMessageStreamAccumulator.cs

@@ -0,0 +1,151 @@
+using System.Text.Json;
+
+namespace AIStudio.Provider.Anthropic;
+
+/// <summary>
+/// Reads a streamed Anthropic messages call back into the answer the tool calling loop works with.
+/// </summary>
+/// <remarks>
+/// Anthropic streams a message as a set of content blocks which open, grow, and close, correlated
+/// by their index and interleaved with one another. This type keeps one builder per index and
+/// hands out text as it arrives; everything else is bookkeeping until the message ends.<br/><br/>
+/// No HTTP, no dependency injection, no provider: what happens here are decisions about bytes,
+/// and those are the decisions worth having a test for.
+/// </remarks>
+public sealed class AnthropicMessageStreamAccumulator
+{
+    private const string EVENT_BLOCK_START = "content_block_start";
+    private const string EVENT_BLOCK_DELTA = "content_block_delta";
+    private const string EVENT_BLOCK_STOP = "content_block_stop";
+    private const string EVENT_MESSAGE_DELTA = "message_delta";
+    private const string EVENT_MESSAGE_STOP = "message_stop";
+    
+    private const string DELTA_TEXT = "text_delta";
+    
+    private readonly Dictionary<int, AnthropicContentBlockBuilder> openBlocks = [];
+    private readonly SortedDictionary<int, JsonElement> finishedBlocks = [];
+    private readonly Dictionary<string, string> unparsableToolArguments = [];
+    private string stopReason = string.Empty;
+    private bool messageEnded;
+    
+    /// <summary>
+    /// Takes the next event of the stream and returns what it has to show.
+    /// </summary>
+    /// <param name="serverSentEvent">The event to read.</param>
+    /// <returns>The text of this event, empty when it carried none.</returns>
+    public AnthropicStreamPart Process(ServerSentEvent serverSentEvent)
+    {
+        if (serverSentEvent.Data.Length is 0)
+            return AnthropicStreamPart.Nothing;
+
+        AnthropicStreamLine line;
+        try
+        {
+            line = JsonSerializer.Deserialize<AnthropicStreamLine>(serverSentEvent.Data, ProviderJsonOptions.OPTIONS);
+        }
+        catch (JsonException)
+        {
+            // A line we cannot read is a line we skip, exactly as the plain text path does:
+            return AnthropicStreamPart.Nothing;
+        }
+
+        switch (line.Type)
+        {
+            case EVENT_BLOCK_START:
+                this.openBlocks[line.Index] = new AnthropicContentBlockBuilder(line.ContentBlock);
+                return AnthropicStreamPart.Nothing;
+            
+            case EVENT_BLOCK_DELTA:
+                if (!this.openBlocks.TryGetValue(line.Index, out var openBlock))
+                {
+                    //
+                    // A delta for a block which never opened. Only text can be salvaged from
+                    // that: a tool use without its ID and name is unanswerable, and thinking
+                    // without its signature would have the next round refused. Text is kept as a
+                    // block of its own so that what the user reads is what the model is told it
+                    // said.
+                    //
+                    if (line.Delta.Type is not DELTA_TEXT)
+                        return AnthropicStreamPart.Nothing;
+
+                    openBlock = new AnthropicContentBlockBuilder(EmptyTextBlock());
+                    this.openBlocks[line.Index] = openBlock;
+                }
+
+                return new AnthropicStreamPart(openBlock.Append(line.Delta));
+            
+            case EVENT_BLOCK_STOP:
+                if (this.openBlocks.Remove(line.Index, out var finishedBlock))
+                    this.Finish(line.Index, finishedBlock);
+                
+                return AnthropicStreamPart.Nothing;
+            
+            case EVENT_MESSAGE_DELTA:
+                //
+                // The stop reason ends the message as surely as the closing event does. Taking
+                // both means a gateway which sends only one of them still gets a round out.
+                //
+                if (!string.IsNullOrWhiteSpace(line.Delta.StopReason))
+                {
+                    this.stopReason = line.Delta.StopReason;
+                    this.messageEnded = true;
+                }
+                
+                return AnthropicStreamPart.Nothing;
+            
+            case EVENT_MESSAGE_STOP:
+                this.messageEnded = true;
+                this.MaterializeOpenBlocks();
+                return AnthropicStreamPart.Nothing;
+            
+            default:
+                return AnthropicStreamPart.Nothing;
+        }
+    }
+    
+    /// <summary>
+    /// Builds the answer of the round from everything the stream said.
+    /// </summary>
+    /// <returns>
+    /// The answer, or null when the stream ended before the message did. Null is how a failed
+    /// request and a stream cut off mid-sentence look from here, and both end the round.
+    /// </returns>
+    public AnthropicResponse? Build()
+    {
+        if (!this.messageEnded)
+            return null;
+
+        // Blocks whose closing event never came are finished here rather than dropped:
+        this.MaterializeOpenBlocks();
+        
+        return new AnthropicResponse
+        {
+            StopReason = this.stopReason,
+            Content = [..this.finishedBlocks.Values],
+            UnparsableToolInputs = this.unparsableToolArguments,
+        };
+    }
+    
+    private void MaterializeOpenBlocks()
+    {
+        foreach (var (index, builder) in this.openBlocks)
+            this.Finish(index, builder);
+        
+        this.openBlocks.Clear();
+    }
+    
+    private void Finish(int index, AnthropicContentBlockBuilder builder)
+    {
+        this.finishedBlocks[index] = builder.Build();
+        
+        // Read after the block was built, because that is when the arguments are parsed:
+        if (builder.UnparsableToolArguments is not null && !string.IsNullOrWhiteSpace(builder.ToolUseId))
+            this.unparsableToolArguments[builder.ToolUseId] = builder.UnparsableToolArguments;
+    }
+    
+    private static JsonElement EmptyTextBlock() => JsonSerializer.SerializeToElement(new
+    {
+        type = "text",
+        text = string.Empty,
+    });
+}

app/MindWork AI Studio/Provider/Anthropic/AnthropicResponse.cs

@@ -11,6 +11,15 @@ public sealed record AnthropicResponse
 
     public IList<JsonElement> Content { get; init; } = [];
 
+    /// <summary>
+    /// The argument text of those tool uses whose arguments never parsed, by tool use ID.
+    /// </summary>
+    /// <remarks>
+    /// Empty for a non-streamed answer, where the arguments either arrived as an object or did
+    /// not arrive at all.
+    /// </remarks>
+    public IReadOnlyDictionary<string, string> UnparsableToolInputs { get; init; } = new Dictionary<string, string>();
+
     /// <summary>
     /// The tool calls the model asked for.
     /// </summary>
@@ -25,6 +34,7 @@ public sealed record AnthropicResponse
             Id = ReadString(x, "id"),
             Name = ReadString(x, "name"),
             Input = x.TryGetProperty("input", out var input) ? input : default,
+            UnparsableArguments = this.UnparsableToolInputs.GetValueOrDefault(ReadString(x, "id")),
         })
         .Where(x => !string.IsNullOrWhiteSpace(x.Id) && !string.IsNullOrWhiteSpace(x.Name))
         .ToList();

app/MindWork AI Studio/Provider/Anthropic/AnthropicStreamDelta.cs

@@ -0,0 +1,18 @@
+namespace AIStudio.Provider.Anthropic;
+
+/// <summary>
+/// One piece of a streamed content block.
+/// </summary>
+/// <remarks>
+/// Which of the fields is set depends on what the block is made of: text arrives as text, tool
+/// arguments as fragments of JSON, and a thinking block brings its signature in one piece at the
+/// end. The stop reason belongs to the message rather than to a block, and shares this shape
+/// because the API sends it in a delta of its own.
+/// </remarks>
+/// <param name="Type">What kind of piece this is.</param>
+/// <param name="Text">The piece of text, for a text delta.</param>
+/// <param name="PartialJson">The fragment of the tool arguments, for an input JSON delta.</param>
+/// <param name="Thinking">The piece of thinking, for a thinking delta.</param>
+/// <param name="Signature">The signature of a thinking block, for a signature delta.</param>
+/// <param name="StopReason">Why the model stopped, for the message delta.</param>
+public readonly record struct AnthropicStreamDelta(string? Type, string? Text, string? PartialJson, string? Thinking, string? Signature, string? StopReason);

app/MindWork AI Studio/Provider/Anthropic/AnthropicStreamLine.cs

@@ -0,0 +1,12 @@
+using System.Text.Json;
+
+namespace AIStudio.Provider.Anthropic;
+
+/// <summary>
+/// One line of a streamed Anthropic messages call.
+/// </summary>
+/// <param name="Type">The kind of event this line reports.</param>
+/// <param name="Index">Which content block the event belongs to; blocks are correlated by it.</param>
+/// <param name="ContentBlock">The block as it opens, for a content block start.</param>
+/// <param name="Delta">The piece this event adds, for a content block delta or a message delta.</param>
+public readonly record struct AnthropicStreamLine(string? Type, int Index, JsonElement ContentBlock, AnthropicStreamDelta Delta);

app/MindWork AI Studio/Provider/Anthropic/AnthropicStreamPart.cs

@@ -0,0 +1,22 @@
+namespace AIStudio.Provider.Anthropic;
+
+/// <summary>
+/// What one line of a streamed Anthropic messages call has to show to the user.
+/// </summary>
+/// <remarks>
+/// Only text ever shows. Thinking does not: neither of the two paths has ever put it on screen,
+/// and doing so would be a feature of its own rather than a side effect of streaming.
+/// </remarks>
+/// <param name="TextDelta">The text this line carried, empty when it carried none.</param>
+public readonly record struct AnthropicStreamPart(string TextDelta)
+{
+    /// <summary>
+    /// The part of a line that says nothing to the user, such as an opening or closing block.
+    /// </summary>
+    public static AnthropicStreamPart Nothing => new(string.Empty);
+    
+    /// <summary>
+    /// Whether this part has anything to show at all.
+    /// </summary>
+    public bool HasContent => this.TextDelta.Length > 0;
+}

app/MindWork AI Studio/Provider/Anthropic/AnthropicToolCallingAdapter.cs

@@ -17,7 +17,7 @@ namespace AIStudio.Provider.Anthropic;
 /// </remarks>
 public sealed class AnthropicToolCallingAdapter(Model chatModel, IList<IMessageBase> baseMessages, string systemPrompt, int maxTokens,
     IDictionary<string, object> apiParameters, IReadOnlyList<(ToolDefinition Definition, IToolImplementation Implementation)> runnableTools,
-    Func<ChatRequest, CancellationToken, Task<AnthropicResponse?>> executeRequestAsync) : IToolCallingProviderAdapter
+    Func<ChatRequest, CancellationToken, IAsyncEnumerable<ServerSentEvent>> streamRequestAsync) : IToolCallingProviderAdapter
 {
     private readonly List<IMessageBase> internalMessages = [];
     private readonly List<AnthropicToolResultContent> pendingToolResults = [];
@@ -41,7 +41,7 @@ public sealed class AnthropicToolCallingAdapter(Model chatModel, IList<IMessageB
             this.pendingToolResults.Clear();
         }
 
-        var response = await executeRequestAsync(new ChatRequest
+        var request = new ChatRequest
         {
             Model = chatModel.Id,
             Messages = [..baseMessages, ..this.internalMessages],
@@ -50,26 +50,30 @@ public sealed class AnthropicToolCallingAdapter(Model chatModel, IList<IMessageB
                 : $"{systemPrompt}{Environment.NewLine}{Environment.NewLine}{finalResponseInstruction}",
 
             MaxTokens = maxTokens,
-            Stream = false,
+            Stream = true,
             Tools = includeTools && this.tools.Count > 0 ? this.tools : null,
             AdditionalApiParameters = apiParameters,
-        }, token);
+        };
 
+        //
+        // The text goes out while it is being written; the blocks are put back together behind
+        // it, because they have to return to the provider exactly as they arrived.
+        //
+        var accumulator = new AnthropicMessageStreamAccumulator();
+        await foreach (var serverSentEvent in streamRequestAsync(request, token))
+        {
+            var part = accumulator.Process(serverSentEvent);
+            if (part.HasContent)
+                yield return ToolCallingStreamEvent.TextDelta(part.TextDelta);
+        }
+
+        var response = accumulator.Build();
         if (response is null)
             yield break;
 
         this.lastResponse = response;
-        
-        //
-        // The whole round arrives at once for now, so its text goes out as one delta. What the
-        // loop and the UI see is already the streaming shape; only the pieces are still large.
-        //
-        var textOutput = response.GetTextOutput();
-        if (!string.IsNullOrEmpty(textOutput))
-            yield return ToolCallingStreamEvent.TextDelta(textOutput);
-
         yield return ToolCallingStreamEvent.RoundCompleted(new ToolCallingRound(
-            textOutput,
+            response.GetTextOutput(),
             response.GetToolUses()
                 .Select(toolUse => new ToolCallingRequestedCall(
                     toolUse.Id,

app/MindWork AI Studio/Provider/Anthropic/AnthropicToolUse.cs

@@ -10,8 +10,19 @@ public sealed record AnthropicToolUse
 
     public JsonElement Input { get; init; }
 
+    /// <summary>
+    /// The arguments as they came off the wire, set only when they never parsed into an object.
+    /// </summary>
+    /// <remarks>
+    /// Only a streamed round can have these: the arguments arrive in fragments there, and a
+    /// stream which ends mid-fragment leaves text which is not an object. The block carries an
+    /// empty object in that case, because that is what may go back to the provider -- while the
+    /// call itself has to be rejected rather than run without the arguments it asked for.
+    /// </remarks>
+    public string? UnparsableArguments { get; init; }
+
     /// <summary>
     /// The arguments as JSON text, which is what the tool executor works with.
     /// </summary>
-    public string Arguments => this.Input.ValueKind is JsonValueKind.Undefined ? "{}" : this.Input.GetRawText();
+    public string Arguments => this.UnparsableArguments ?? (this.Input.ValueKind is JsonValueKind.Undefined ? "{}" : this.Input.GetRawText());
 }

app/MindWork AI Studio/Provider/Anthropic/ProviderAnthropic.cs

@@ -75,8 +75,8 @@ public sealed class ProviderAnthropic() : BaseProvider(LLMProviders.ANTHROPIC, n
 
         //
         // Prepare the tools we want to use. When the model may call one, the conversation runs
-        // through the harness instead of being streamed straight away: tool rounds are not
-        // streamed, only the final answer is.
+        // through the harness instead of going straight to the streaming path below. It streams
+        // there as well, round by round -- what the harness adds is the tools in between.
         //
         var toolRegistry = Program.SERVICE_PROVIDER.GetService<ToolRegistry>();
         var toolExecutor = Program.SERVICE_PROVIDER.GetService<ToolExecutor>();
@@ -93,7 +93,7 @@ public sealed class ProviderAnthropic() : BaseProvider(LLMProviders.ANTHROPIC, n
         if (toolExecutor is not null && runnableTools.Count > 0)
         {
             var adapter = new AnthropicToolCallingAdapter(chatModel, [..messages], systemPrompt, maxTokens, apiParameters, runnableTools,
-                (requestDto, requestToken) => this.ExecuteMessagesRequest(requestDto, requestedSecret, requestToken));
+                (requestDto, requestToken) => this.StreamMessagesRequest(requestDto, requestedSecret, requestToken));
 
             var loop = Program.SERVICE_PROVIDER.GetRequiredService<IToolCallingLoop>();
             var loopContext = new ToolCallingLoopContext
@@ -151,30 +151,25 @@ public sealed class ProviderAnthropic() : BaseProvider(LLMProviders.ANTHROPIC, n
     }
 
     /// <summary>
-    /// Runs one non-streamed messages request, as the tool rounds need it.
+    /// Runs one round of a tool calling conversation against the messages API.
     /// </summary>
     /// <remarks>
-    /// Tool rounds are not streamed: the whole answer has to be there before its tool calls can
-    /// be executed. Only the final answer reaches the user through the streaming path.
+    /// Nothing but the HTTP request is done here. The retries, the timeouts, and the error
+    /// classification come from the shared stream reader, which the tool rounds used to go
+    /// without; reading the events is the adapter's business.
     /// </remarks>
-    /// <returns>The answer, or null when the request failed and the user was already told.</returns>
-    private async Task<AnthropicResponse?> ExecuteMessagesRequest(ChatRequest requestDto, RequestedSecret requestedSecret, CancellationToken token)
+    private IAsyncEnumerable<ServerSentEvent> StreamMessagesRequest(ChatRequest requestDto, RequestedSecret requestedSecret, CancellationToken token)
     {
-        using var request = new HttpRequestMessage(HttpMethod.Post, "messages");
+        async Task<HttpRequestMessage> RequestBuilder()
+        {
+            var request = new HttpRequestMessage(HttpMethod.Post, "messages");
             request.Headers.Add("x-api-key", await requestedSecret.Secret.Decrypt(Program.ENCRYPTION));
             request.Headers.Add("anthropic-version", "2023-06-01");
             request.Content = new StringContent(JsonSerializer.Serialize(requestDto, JSON_SERIALIZER_OPTIONS), Encoding.UTF8, "application/json");
-
-        using var response = await this.HttpClient.SendAsync(request, token);
-        if (!response.IsSuccessStatusCode)
-        {
-            var responseBody = await response.Content.ReadAsStringAsync(token);
-            LOGGER.LogError("Tool calling messages request failed with status code {ResponseStatusCode} and body: '{ResponseBody}'.", response.StatusCode, responseBody);
-            await ToolCallingMessages.SendToolCallingRequestFailedAsync((int)response.StatusCode);
-            return null;
+            return request;
         }
 
-        return await response.Content.ReadFromJsonAsync<AnthropicResponse>(JSON_SERIALIZER_OPTIONS, token);
+        return this.ReadServerSentEventsAsync("Anthropic", "messages call", RequestBuilder, token);
     }
 
     #pragma warning disable CS1998 // Async method lacks 'await' operators and will run synchronously