From 1be0a6bfc3353d97556c331cc230776d2a561e6b Mon Sep 17 00:00:00 2001 From: Thorsten Sommer Date: Sun, 20 Sep 2026 10:03:20 +0200 Subject: [PATCH] Stream the tool calling rounds of the Anthropic messages API --- .../Anthropic/AnthropicContentBlockBuilder.cs | 227 ++++++++++++++++++ .../AnthropicMessageStreamAccumulator.cs | 151 ++++++++++++ .../Provider/Anthropic/AnthropicResponse.cs | 10 + .../Anthropic/AnthropicStreamDelta.cs | 18 ++ .../Provider/Anthropic/AnthropicStreamLine.cs | 12 + .../Provider/Anthropic/AnthropicStreamPart.cs | 22 ++ .../Anthropic/AnthropicToolCallingAdapter.cs | 32 +-- .../Provider/Anthropic/AnthropicToolUse.cs | 13 +- .../Provider/Anthropic/ProviderAnthropic.cs | 35 ++- 9 files changed, 485 insertions(+), 35 deletions(-) create mode 100644 app/MindWork AI Studio/Provider/Anthropic/AnthropicContentBlockBuilder.cs create mode 100644 app/MindWork AI Studio/Provider/Anthropic/AnthropicMessageStreamAccumulator.cs create mode 100644 app/MindWork AI Studio/Provider/Anthropic/AnthropicStreamDelta.cs create mode 100644 app/MindWork AI Studio/Provider/Anthropic/AnthropicStreamLine.cs create mode 100644 app/MindWork AI Studio/Provider/Anthropic/AnthropicStreamPart.cs diff --git a/app/MindWork AI Studio/Provider/Anthropic/AnthropicContentBlockBuilder.cs b/app/MindWork AI Studio/Provider/Anthropic/AnthropicContentBlockBuilder.cs new file mode 100644 index 00000000..857219a6 --- /dev/null +++ b/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; + +/// +/// Puts one streamed content block back together. +/// +/// +/// 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.

+/// 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. +/// +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; + + /// + /// Opens a block from the seed the provider sent for it. + /// + /// The block as it opened. + 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"); + } + + /// + /// What kind of block this is: text, a tool use, thinking, or something we do not know. + /// + public string BlockType { get; } + + /// + /// The ID of the tool use, for a tool use block. + /// + public string ToolUseId => ReadString(this.seed, "id"); + + /// + /// The tool arguments as they came off the wire, set only when they never parsed into an object. + /// + /// + /// 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. + /// + public string? UnparsableToolArguments { get; private set; } + + /// + /// Adds the next piece of this block. + /// + /// The piece as it arrived. + /// The text to show, empty for every piece which is not text. + 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; + } + } + + /// + /// Builds the finished block, in the shape a non-streamed call would have returned it. + /// + 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; + } + } + + /// + /// The tool arguments as the JSON object they have to be. + /// + /// + /// 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. + /// + 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; + } + + /// + /// Writes the given properties over a copy of the seed. + /// + /// + /// 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. + /// + /// The properties to write, as property name to JSON text. + private JsonElement BuildFromSeed(Dictionary overrides) + { + var buffer = new ArrayBufferWriter(); + 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; + } +} \ No newline at end of file diff --git a/app/MindWork AI Studio/Provider/Anthropic/AnthropicMessageStreamAccumulator.cs b/app/MindWork AI Studio/Provider/Anthropic/AnthropicMessageStreamAccumulator.cs new file mode 100644 index 00000000..075a1978 --- /dev/null +++ b/app/MindWork AI Studio/Provider/Anthropic/AnthropicMessageStreamAccumulator.cs @@ -0,0 +1,151 @@ +using System.Text.Json; + +namespace AIStudio.Provider.Anthropic; + +/// +/// Reads a streamed Anthropic messages call back into the answer the tool calling loop works with. +/// +/// +/// 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.

+/// No HTTP, no dependency injection, no provider: what happens here are decisions about bytes, +/// and those are the decisions worth having a test for. +/// +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 openBlocks = []; + private readonly SortedDictionary finishedBlocks = []; + private readonly Dictionary unparsableToolArguments = []; + private string stopReason = string.Empty; + private bool messageEnded; + + /// + /// Takes the next event of the stream and returns what it has to show. + /// + /// The event to read. + /// The text of this event, empty when it carried none. + public AnthropicStreamPart Process(ServerSentEvent serverSentEvent) + { + if (serverSentEvent.Data.Length is 0) + return AnthropicStreamPart.Nothing; + + AnthropicStreamLine line; + try + { + line = JsonSerializer.Deserialize(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; + } + } + + /// + /// Builds the answer of the round from everything the stream said. + /// + /// + /// 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. + /// + 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, + }); +} \ No newline at end of file diff --git a/app/MindWork AI Studio/Provider/Anthropic/AnthropicResponse.cs b/app/MindWork AI Studio/Provider/Anthropic/AnthropicResponse.cs index b94d23cb..99071fc9 100644 --- a/app/MindWork AI Studio/Provider/Anthropic/AnthropicResponse.cs +++ b/app/MindWork AI Studio/Provider/Anthropic/AnthropicResponse.cs @@ -11,6 +11,15 @@ public sealed record AnthropicResponse public IList Content { get; init; } = []; + /// + /// The argument text of those tool uses whose arguments never parsed, by tool use ID. + /// + /// + /// Empty for a non-streamed answer, where the arguments either arrived as an object or did + /// not arrive at all. + /// + public IReadOnlyDictionary UnparsableToolInputs { get; init; } = new Dictionary(); + /// /// The tool calls the model asked for. /// @@ -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(); diff --git a/app/MindWork AI Studio/Provider/Anthropic/AnthropicStreamDelta.cs b/app/MindWork AI Studio/Provider/Anthropic/AnthropicStreamDelta.cs new file mode 100644 index 00000000..573538d3 --- /dev/null +++ b/app/MindWork AI Studio/Provider/Anthropic/AnthropicStreamDelta.cs @@ -0,0 +1,18 @@ +namespace AIStudio.Provider.Anthropic; + +/// +/// One piece of a streamed content block. +/// +/// +/// 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. +/// +/// What kind of piece this is. +/// The piece of text, for a text delta. +/// The fragment of the tool arguments, for an input JSON delta. +/// The piece of thinking, for a thinking delta. +/// The signature of a thinking block, for a signature delta. +/// Why the model stopped, for the message delta. +public readonly record struct AnthropicStreamDelta(string? Type, string? Text, string? PartialJson, string? Thinking, string? Signature, string? StopReason); \ No newline at end of file diff --git a/app/MindWork AI Studio/Provider/Anthropic/AnthropicStreamLine.cs b/app/MindWork AI Studio/Provider/Anthropic/AnthropicStreamLine.cs new file mode 100644 index 00000000..e58ffff1 --- /dev/null +++ b/app/MindWork AI Studio/Provider/Anthropic/AnthropicStreamLine.cs @@ -0,0 +1,12 @@ +using System.Text.Json; + +namespace AIStudio.Provider.Anthropic; + +/// +/// One line of a streamed Anthropic messages call. +/// +/// The kind of event this line reports. +/// Which content block the event belongs to; blocks are correlated by it. +/// The block as it opens, for a content block start. +/// The piece this event adds, for a content block delta or a message delta. +public readonly record struct AnthropicStreamLine(string? Type, int Index, JsonElement ContentBlock, AnthropicStreamDelta Delta); \ No newline at end of file diff --git a/app/MindWork AI Studio/Provider/Anthropic/AnthropicStreamPart.cs b/app/MindWork AI Studio/Provider/Anthropic/AnthropicStreamPart.cs new file mode 100644 index 00000000..bc8f1bd1 --- /dev/null +++ b/app/MindWork AI Studio/Provider/Anthropic/AnthropicStreamPart.cs @@ -0,0 +1,22 @@ +namespace AIStudio.Provider.Anthropic; + +/// +/// What one line of a streamed Anthropic messages call has to show to the user. +/// +/// +/// 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. +/// +/// The text this line carried, empty when it carried none. +public readonly record struct AnthropicStreamPart(string TextDelta) +{ + /// + /// The part of a line that says nothing to the user, such as an opening or closing block. + /// + public static AnthropicStreamPart Nothing => new(string.Empty); + + /// + /// Whether this part has anything to show at all. + /// + public bool HasContent => this.TextDelta.Length > 0; +} \ No newline at end of file diff --git a/app/MindWork AI Studio/Provider/Anthropic/AnthropicToolCallingAdapter.cs b/app/MindWork AI Studio/Provider/Anthropic/AnthropicToolCallingAdapter.cs index 210c4fc4..cee4d160 100644 --- a/app/MindWork AI Studio/Provider/Anthropic/AnthropicToolCallingAdapter.cs +++ b/app/MindWork AI Studio/Provider/Anthropic/AnthropicToolCallingAdapter.cs @@ -17,7 +17,7 @@ namespace AIStudio.Provider.Anthropic; /// public sealed class AnthropicToolCallingAdapter(Model chatModel, IList baseMessages, string systemPrompt, int maxTokens, IDictionary apiParameters, IReadOnlyList<(ToolDefinition Definition, IToolImplementation Implementation)> runnableTools, - Func> executeRequestAsync) : IToolCallingProviderAdapter + Func> streamRequestAsync) : IToolCallingProviderAdapter { private readonly List internalMessages = []; private readonly List pendingToolResults = []; @@ -41,7 +41,7 @@ public sealed class AnthropicToolCallingAdapter(Model chatModel, IList 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, diff --git a/app/MindWork AI Studio/Provider/Anthropic/AnthropicToolUse.cs b/app/MindWork AI Studio/Provider/Anthropic/AnthropicToolUse.cs index bd42c326..52f3a62a 100644 --- a/app/MindWork AI Studio/Provider/Anthropic/AnthropicToolUse.cs +++ b/app/MindWork AI Studio/Provider/Anthropic/AnthropicToolUse.cs @@ -10,8 +10,19 @@ public sealed record AnthropicToolUse public JsonElement Input { get; init; } + /// + /// The arguments as they came off the wire, set only when they never parsed into an object. + /// + /// + /// 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. + /// + public string? UnparsableArguments { get; init; } + /// /// The arguments as JSON text, which is what the tool executor works with. /// - 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()); } \ No newline at end of file diff --git a/app/MindWork AI Studio/Provider/Anthropic/ProviderAnthropic.cs b/app/MindWork AI Studio/Provider/Anthropic/ProviderAnthropic.cs index 17ad8309..21b32049 100644 --- a/app/MindWork AI Studio/Provider/Anthropic/ProviderAnthropic.cs +++ b/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(); var toolExecutor = Program.SERVICE_PROVIDER.GetService(); @@ -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(); var loopContext = new ToolCallingLoopContext @@ -151,30 +151,25 @@ public sealed class ProviderAnthropic() : BaseProvider(LLMProviders.ANTHROPIC, n } /// - /// Runs one non-streamed messages request, as the tool rounds need it. + /// Runs one round of a tool calling conversation against the messages API. /// /// - /// 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. /// - /// The answer, or null when the request failed and the user was already told. - private async Task ExecuteMessagesRequest(ChatRequest requestDto, RequestedSecret requestedSecret, CancellationToken token) + private IAsyncEnumerable StreamMessagesRequest(ChatRequest requestDto, RequestedSecret requestedSecret, CancellationToken token) { - using 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) + async Task RequestBuilder() { - 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; + 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"); + return request; } - return await response.Content.ReadFromJsonAsync(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