.NET: Add streaming support to A2A agent handler (#5427)

* update a2a agent to the latest a2a sdk (#5257) * Move A2A samples from 04-hosting to 02-agents (#5267) Move the A2A sample projects (A2AAgent_AsFunctionTools and A2AAgent_PollingForTaskCompletion) from samples/04-hosting/A2A/ to samples/02-agents/A2A/ to better align with the sample directory structure. Update solution file and samples README accordingly. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * .NET: Fix stream reconnection for A2AAgent (#5275) * Add SSE stream reconnection support to A2AAgent Implement automatic reconnection for SSE streams that disconnect mid-task, using the Last-Event-ID header to resume from where the stream left off. Changes: - Add InvokeStreamingWithReconnectAsync method to A2AAgent with configurable max retries and delay between attempts - Add new log messages for reconnection events - Add A2AAgent_StreamReconnection sample demonstrating the feature - Update existing polling sample to use simplified SendMessageAsync API - Add unit tests for stream reconnection logic Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * address comments * Address PR review feedback - Dispose SSE enumerator before GetTaskAsync fallback to release HTTP connection - Wrap StreamWriter in using blocks with leaveOpen:true and explicit UTF-8 encoding - Print update.Text instead of update object in stream reconnection sample Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * .NET: Use IA2AClientFactory to create A2AClient (#5277) * Refactor A2A extensions to use IA2AClientFactory and add ProtocolSelection sample - Update A2AAgentCardExtensions to accept IA2AClientFactory instead of A2AClientOptions - Update A2ACardResolverExtensions to accept IA2AClientFactory - Update A2AClientExtensions to accept IA2AClientFactory - Update A2AAgent to use IA2AClientFactory for client creation - Add A2AAgent_ProtocolSelection sample demonstrating protocol selection - Add comprehensive unit tests for all changes - Update README files with new sample reference Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Reorder params: options before loggerFactory in A2A extensions Move A2AClientOptions parameter before ILoggerFactory in AsAIAgent and GetAIAgentAsync extension methods to follow the repo convention of keeping LoggerFactory and CancellationToken as the last parameters. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * .NET: Migrate A2A hosting to A2A SDK v1 (#5363) * .NET: Migrate A2A hosting to A2A SDK v1 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * remove unused agent card --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * .NET: Split A2A endpoint mapping into protocol-specific methods (#5413) * .NET: Refactor A2A hosting registration into A2AServerServiceCollectionExtensions - Rename A2AHostingOptions to A2AServerRegistrationOptions - Move server registration logic from A2AEndpointRouteBuilderExtensions and AIAgentExtensions into new A2AServerServiceCollectionExtensions - Remove A2AProtocolBinding and AIAgentExtensions (consolidated) - Update samples and tests to use the new registration API Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * address copilot comments --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Remove unnecessary using directive in AgentWebChat.AgentHost Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * restore AsyncEnumerable package version * address copilot initial feedback * address automated code review and formatting issues * fix formatting issues * Add streaming support to A2A agent handler Add HandleNewMessageStreamingAsync to A2AAgentHandler that routes StreamingResponse requests through RunStreamingAsync, enqueuing an A2A Message for each AgentResponseUpdate. Add MessageConverter.ToParts(AgentResponseUpdate) extension to convert streaming update contents to A2A Parts with unsupported-content filtering. Add CreateMessageFromUpdate to map AgentResponseUpdate to A2A Message. Add 16 new tests covering the streaming path and converter. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Add streaming edge-case tests for A2AAgentHandler Add two tests covering gaps in the streaming path: - ExecuteAsync_Streaming_WhenNoUpdates_EnqueuesNoMessagesAndSavesSessionAsync: Verifies that when RunStreamingAsync yields an empty async enumerable, no messages are enqueued and only SaveSessionAsync runs. - ExecuteAsync_Streaming_CancellationTokenIsPropagatedToRunStreamingAsyncAsync: Verifies that the CancellationToken from ExecuteAsync is propagated through to the inner agent's RunCoreStreamingAsync call. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * address copilot comments --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
2026-06-16 21:04:09 +08:00 · 2026-04-23 18:50:57 +00:00
parent 4d3e4f865f
commit 0dbcc9fe9d
4 changed files with 966 additions and 9 deletions
@@ -42,11 +42,19 @@ internal sealed class A2AAgentHandler : IAgentHandler
    /// <inheritdoc/>
    public Task ExecuteAsync(RequestContext context, AgentEventQueue eventQueue, CancellationToken cancellationToken)
    {
+        // Handle task updates
        if (context.IsContinuation)
        {
            return this.HandleTaskUpdateAsync(context, eventQueue, cancellationToken);
        }

+        // Handle messages received via streaming endpoint
+        if (context.StreamingResponse)
+        {
+            return this.HandleNewMessageStreamingAsync(context, eventQueue, cancellationToken);
+        }
+
+        // Handle new messages received via non-streaming endpoint
        return this.HandleNewMessageAsync(context, eventQueue, cancellationToken);
    }

@@ -80,13 +88,19 @@ internal sealed class A2AAgentHandler : IAgentHandler
            ? new AgentRunOptions { AllowBackgroundResponses = allowBackgroundResponses }
            : new AgentRunOptions { AllowBackgroundResponses = allowBackgroundResponses, AdditionalProperties = context.Metadata.ToAdditionalProperties() };

-        var response = await this._hostAgent.RunAsync(
-            chatMessages,
-            session: session,
-            options: options,
-            cancellationToken: cancellationToken).ConfigureAwait(false);
-
-        await this._hostAgent.SaveSessionAsync(contextId, session, cancellationToken).ConfigureAwait(false);
+        AgentResponse response;
+        try
+        {
+            response = await this._hostAgent.RunAsync(
+                chatMessages,
+                session: session,
+                options: options,
+                cancellationToken: cancellationToken).ConfigureAwait(false);
+        }
+        finally
+        {
+            await this._hostAgent.SaveSessionAsync(contextId, session, CancellationToken.None).ConfigureAwait(false);
+        }

        if (response.ContinuationToken is null)
        {
@@ -108,6 +122,39 @@ internal sealed class A2AAgentHandler : IAgentHandler
        }
    }

+    private async Task HandleNewMessageStreamingAsync(RequestContext context, AgentEventQueue eventQueue, CancellationToken cancellationToken)
+    {
+        var contextId = context.ContextId ?? Guid.NewGuid().ToString("N");
+        var session = await this._hostAgent.GetOrCreateSessionAsync(contextId, cancellationToken).ConfigureAwait(false);
+
+        // AIAgent does not support resuming from arbitrary prior tasks.
+        // Throw explicitly so the client gets a clear error rather than a response
+        // that silently ignores the referenced task context.
+        if (context.Message?.ReferenceTaskIds is { Count: > 0 })
+        {
+            throw new NotSupportedException("ReferenceTaskIds is not supported. AIAgent cannot resume from arbitrary prior task context.");
+        }
+
+        List<ChatMessage> chatMessages = context.Message is not null ? [context.Message.ToChatMessage()] : [];
+
+        var options = context.Metadata is { Count: > 0 }
+            ? new AgentRunOptions { AdditionalProperties = context.Metadata.ToAdditionalProperties() }
+            : null;
+
+        try
+        {
+            await foreach (var update in this._hostAgent.RunStreamingAsync(chatMessages, session, options, cancellationToken).ConfigureAwait(false))
+            {
+                var message = CreateMessageFromUpdate(contextId, update);
+                await eventQueue.EnqueueMessageAsync(message, cancellationToken).ConfigureAwait(false);
+            }
+        }
+        finally
+        {
+            await this._hostAgent.SaveSessionAsync(contextId, session, CancellationToken.None).ConfigureAwait(false);
+        }
+    }
+
    private async Task HandleTaskUpdateAsync(RequestContext context, AgentEventQueue eventQueue, CancellationToken cancellationToken)
    {
        var contextId = context.ContextId ?? Guid.NewGuid().ToString("N");
@@ -141,8 +188,10 @@ internal sealed class A2AAgentHandler : IAgentHandler
            await failUpdater.FailAsync(message: null, CancellationToken.None).ConfigureAwait(false);
            throw;
        }
-
-        await this._hostAgent.SaveSessionAsync(contextId, session, cancellationToken).ConfigureAwait(false);
+        finally
+        {
+            await this._hostAgent.SaveSessionAsync(contextId, session, CancellationToken.None).ConfigureAwait(false);
+        }

        if (response.ContinuationToken is null)
        {
@@ -174,6 +223,16 @@ internal sealed class A2AAgentHandler : IAgentHandler
            Metadata = response.AdditionalProperties?.ToA2AMetadata()
        };

+    private static Message CreateMessageFromUpdate(string contextId, AgentResponseUpdate update) =>
+        new()
+        {
+            MessageId = update.ResponseId ?? Guid.NewGuid().ToString("N"),
+            ContextId = contextId,
+            Role = Role.Agent,
+            Parts = update.ToParts(),
+            Metadata = update.AdditionalProperties?.ToA2AMetadata()
+        };
+
    private static List<ChatMessage> ExtractChatMessagesFromTaskHistory(AgentTask? agentTask)
    {
        if (agentTask?.History is not { Count: > 0 })
@@ -8,6 +8,26 @@ namespace Microsoft.Agents.AI.Hosting.A2A.Converters;

 internal static class MessageConverter
 {
+    public static List<Part> ToParts(this AgentResponseUpdate update)
+    {
+        if (update is null || update.Contents is not { Count: > 0 })
+        {
+            return [];
+        }
+
+        var parts = new List<Part>();
+        foreach (var content in update.Contents)
+        {
+            var part = content.ToPart();
+            if (part is not null)
+            {
+                parts.Add(part);
+            }
+        }
+
+        return parts;
+    }
+
    public static List<Part> ToParts(this IList<ChatMessage> chatMessages)
    {
        if (chatMessages is null || chatMessages.Count == 0)