Testing how Open Api based plugins can use HttpResponseContentReader delegate and IAutoFunctionInvocationFilter and SignalR

[radrad]

I am testing how Open Api based plugins can use HttpResponseContentReader delegate and IAutoFunctionInvocationFilter and if these two can play nicely together.
If I want to intercept the function results (which I want to interpret as Stream instead of string when MediaType == "application/json")
and using SignalR I want to send a collection of json objects to react.js chat client to be displayed as soon as available.
I am getting error here (see the image):
result = await chatCompletionService.GetChatMessageContentAsync( _chatHistory, executionSettings: _openAIPromptExecutionSettings, kernel: _kernel );
This is how I import Open Api plugin
kernel.ImportPluginFromOpenApiAsync( pluginName: "listRepairs", // Ensure this API is running and accessible when the main app starts uri: new Uri("https://localhost:7194/swagger/v1/swagger.json"), executionParameters: new OpenApiFunctionExecutionParameters() { EnablePayloadNamespacing = true, HttpResponseContentReader = CustomHttpContentReader.ReadHttpResponseContentAsync } ).GetAwaiter().GetResult();

This is my custom delegate:
`
using Microsoft.SemanticKernel.Plugins.OpenApi;

namespace SemanticKernelDemoApi.Services;

public static class CustomHttpContentReader
{
///

/// A custom HTTP content reader to change the default behavior of reading HTTP content.
///

/// The HTTP response content reader context.
/// The cancellation token.
/// The HTTP response content.
public static async Task<object?> ReadHttpResponseContentAsync(HttpResponseContentReaderContext context, CancellationToken cancellationToken)
{
// Read JSON content as a stream rather than as a string, which is the default behavior
if (context.Response.Content.Headers.ContentType?.MediaType == "application/json")
{
return await context.Response.Content.ReadAsStreamAsync(cancellationToken);
}

    // HTTP request and response properties can be used to decide how to read the content.
    // The 'if' operator below is not relevant to the current example and is just for demonstration purposes.
    if (context.Request.Headers.Contains("x-stream"))
    {
        return await context.Response.Content.ReadAsStreamAsync(cancellationToken);
    }

    // Return null to indicate that any other HTTP content not handled above should be read by the default reader.
    return null;
}

}
`

Here is how I simulate Repair objects retrieval:
`
namespace ProductsApi;

internal record Repair(int Id, string Title, string Description, string AssignedTo, string Date, string Image);

internal class ProductsService
{
private readonly List repairs = new List
{
new Repair(1, "Engine Repair", "Fix engine issues", "Tech Magician", "2025-04-10", ""),
new Repair(2, "Battery Replacement", "Replace battery", "Tech Magician", "2025-04-11", ""),
};

public IEnumerable<Repair> Get() => repairs;   

`

Here is my custom IAutoFunctionInvocationFilter implemenation.
Noticed I played with context.Result = new FunctionResult..... numereous times but no solution.
If I comment out that code, I am receiving 2 calls. One with a stream as reponse but immediately next one with String
(since I serialized fullResponse into String) and I am receiving the error you see in the picture.

`
using Microsoft.AspNetCore.SignalR;
using Microsoft.SemanticKernel;
using System.Globalization;
using System.Text.Json;

namespace SemanticKernelDemoApi.Services;

internal record Repair(int Id, string Title, string Description, string AssignedTo, string Date, string Image);

public class ChatAutoFunctionInvocationFilter : IAutoFunctionInvocationFilter
{
private readonly IHubContext _hubContext;

public ChatAutoFunctionInvocationFilter(IHubContext<RepairsHub> hubContext)
{
    _hubContext = hubContext;
}

public async Task OnAutoFunctionInvocationAsync(AutoFunctionInvocationContext context, Func<AutoFunctionInvocationContext, Task> next)
{
    string functionName = context.Function.Name;

    if (context.Function.PluginName == "listRepairs" && (functionName == "get_all_repairs" || functionName == "stream_repairs"))
    {
        var fullResponse = new List<Repair>();
        var batch = new List<Repair>();
        const int batchSize = 10;

        await next(context);

        Console.WriteLine(context.Result.GetType().FullName);

        
        if (context.Result.GetValue<RestApiOperationResponse>() is not RestApiOperationResponse listResult)
        {
            throw new InvalidOperationException("Unexpected result type.");
        }

        if (listResult.Content is Stream originalStream)
        {

            //using var stream = (Stream)listResult.Content!;

            using var memoryStream = new MemoryStream();
            await originalStream.CopyToAsync(memoryStream);
            memoryStream.Position = 0;

            using var reader = new StreamReader(memoryStream);

            var buffer = new char[1024];
            int bytesRead;

            while ((bytesRead = await reader.ReadAsync(buffer, 0, buffer.Length)) > 0)
            {
                var jsonChunk = new string(buffer, 0, bytesRead);
                //var repairs = JsonSerializer.Deserialize<List<Repair>>(jsonChunk);

                var repairs = JsonSerializer.Deserialize<List<Repair>>(jsonChunk, new JsonSerializerOptions
                {
                    PropertyNameCaseInsensitive = true // Enables case-insensitive property matching
                });

                if (repairs != null)
                {
                    foreach (var repair in repairs)
                    {
                        fullResponse.Add(repair);
                        batch.Add(repair);

                        if (batch.Count >= batchSize)
                        {
                            await _hubContext.Clients.All.SendAsync("ReceiveRepairBatch", batch);
                            batch.Clear();
                        }
                    }
                }
            }

            if (batch.Count > 0)
            {
                await _hubContext.Clients.All.SendAsync("ReceiveRepairBatch", batch);
            }

        }
        /*
        string response = JsonSerializer.Serialize(fullResponse);

        context.Result = new FunctionResult(
            context.Function,
            response,
            metadata: null,
            culture: CultureInfo.InvariantCulture
        );
        */
    }
    else
    {
        await next(context);
    }
}

}
`

[sophialagerkranspandey]

Tagging @SergeyMenshykh

[SergeyMenshykh]

Hi @radrad, I tried to reproduce the issue but was not able to.

Here's the sample that imports the repair service API as a plugin and uses LLM to call it:

[Fact]
public async Task GetRepairs()
{
    IKernelBuilder builder = Kernel.CreateBuilder();

    builder.AddOpenAIChatCompletion(
        modelId: TestConfiguration.OpenAI.ChatModelId,
        apiKey: TestConfiguration.OpenAI.ApiKey);

    Kernel kernel = builder.Build();
    kernel.AutoFunctionInvocationFilters.Add(new OpenApiFunctionInvocationFilter());

    await kernel.ImportPluginFromOpenApiAsync(
        "listRepairs",
        new Uri("https://localhost:7261/swagger/v1/swagger.json"),
        new OpenApiFunctionExecutionParameters()
        {
            EnablePayloadNamespacing = true,
            HttpResponseContentReader = ReadHttpResponseContentAsync
        });

    IChatCompletionService chatCompletionService = kernel.Services.GetRequiredService<IChatCompletionService>();

    OpenAIPromptExecutionSettings settings = new() { FunctionChoiceBehavior = FunctionChoiceBehavior.Auto() };

    ChatHistory chatHistory = [];
    chatHistory.AddUserMessage("List all repairs");

    ChatMessageContent content = await chatCompletionService.GetChatMessageContentAsync(chatHistory, settings, kernel);

    Console.WriteLine(content.ToString());
}

the content reader and the filter are identical to yours:

private static async Task<object?> ReadHttpResponseContentAsync(HttpResponseContentReaderContext context, CancellationToken cancellationToken)
{
    // Read JSON content as a stream rather than as a string, which is the default behavior
    if (context.Response.Content.Headers.ContentType?.MediaType == "application/json")
    {
        return await context.Response.Content.ReadAsStreamAsync(cancellationToken);
    }

    // Return null to indicate that any other HTTP content not handled above should be read by the default reader.
    return null;
}

internal class OpenApiFunctionInvocationFilter : IAutoFunctionInvocationFilter
{
    public async Task OnAutoFunctionInvocationAsync(AutoFunctionInvocationContext context, Func<AutoFunctionInvocationContext, Task> next)
    {
        string functionName = context.Function.Name;

        if (context.Function.PluginName == "listRepairs")
        {
            var fullResponse = new List<Repair>();
            var batch = new List<Repair>();
            const int batchSize = 10;

            await next(context);

            Console.WriteLine(context.Result.GetType().FullName);

            if (context.Result.GetValue<RestApiOperationResponse>() is not RestApiOperationResponse listResult)
            {
                throw new InvalidOperationException("Unexpected result type.");
            }

            if (listResult.Content is Stream originalStream)
            {
                //using var stream = (Stream)listResult.Content!;

                using var memoryStream = new MemoryStream();
                await originalStream.CopyToAsync(memoryStream);
                memoryStream.Position = 0;

                using var reader = new StreamReader(memoryStream);

                var buffer = new char[1024];
                int bytesRead;

                while ((bytesRead = await reader.ReadAsync(buffer, 0, buffer.Length)) > 0)
                {
                    var jsonChunk = new string(buffer, 0, bytesRead);
                    //var repairs = JsonSerializer.Deserialize<List<Repair>>(jsonChunk);

                    var repairs = JsonSerializer.Deserialize<List<Repair>>(jsonChunk, new JsonSerializerOptions
                    {
                        PropertyNameCaseInsensitive = true // Enables case-insensitive property matching
                    });

                    if (repairs != null)
                    {
                        foreach (var repair in repairs)
                        {
                            fullResponse.Add(repair);
                            batch.Add(repair);

                            if (batch.Count >= batchSize)
                            {
                                //await _hubContext.Clients.All.SendAsync("ReceiveRepairBatch", batch);
                                batch.Clear();
                            }
                        }
                    }
                }

                if (batch.Count > 0)
                {
                    //await _hubContext.Clients.All.SendAsync("ReceiveRepairBatch", batch);
                }
            }

            string response = JsonSerializer.Serialize(fullResponse);

            context.Result = new FunctionResult(
                context.Function,
                response,
                metadata: null,
                culture: CultureInfo.InvariantCulture
            );
        }
        else
        {
            await next(context);
        }
    }
}

Here's the result and chat history content:

As a next step, could you copy the code and run it in a console app to see if it's still reproducible?