.NET: Add support for background responses to A2A agent (#2381)

* add support for baackground responses to a2a agent

* fix line endings

* address pr review comments

* address pr review comments

* update sample to net10.0

* Update dotnet/samples/GettingStarted/A2A/A2AAgent_PollingForTaskCompletion/README.md

Co-authored-by: Roger Barreto <19890735+rogerbarreto@users.noreply.github.com>

* Update dotnet/src/Microsoft.Agents.AI.A2A/Extensions/A2AAgentTaskExtensions.cs

Co-authored-by: Roger Barreto <19890735+rogerbarreto@users.noreply.github.com>

* Update dotnet/samples/GettingStarted/A2A/A2AAgent_PollingForTaskCompletion/Program.cs

Co-authored-by: Roger Barreto <19890735+rogerbarreto@users.noreply.github.com>

* address pr review feedback

* add clarification regarding background responses

---------

Co-authored-by: Roger Barreto <19890735+rogerbarreto@users.noreply.github.com>

dotnet/src/Microsoft.Agents.AI.A2A/A2AAgent.cs

@@ -3,6 +3,7 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Net.ServerSentEvents;
 using System.Runtime.CompilerServices;
 using System.Text.Json;
 using System.Threading;
@@ -74,26 +75,32 @@ internal sealed class A2AAgent : AIAgent
     {
         _ = Throw.IfNull(messages);
 
-        var a2aMessage = messages.ToA2AMessage();
-
         thread ??= this.GetNewThread();
         if (thread is not A2AAgentThread typedThread)
         {
             throw new InvalidOperationException("The provided thread is not compatible with the agent. Only threads created by the agent can be used.");
         }
 
-        // Linking the message to the existing conversation, if any.
-        a2aMessage.ContextId = typedThread.ContextId;
-
         this._logger.LogA2AAgentInvokingAgent(nameof(RunAsync), this.Id, this.Name);
 
-        var a2aResponse = await this._a2aClient.SendMessageAsync(new MessageSendParams { Message = a2aMessage }, cancellationToken).ConfigureAwait(false);
+        A2AResponse? a2aResponse = null;
+
+        if (GetContinuationToken(messages, options) is { } token)
+        {
+            a2aResponse = await this._a2aClient.GetTaskAsync(token.TaskId, cancellationToken).ConfigureAwait(false);
+        }
+        else
+        {
+            var a2aMessage = CreateA2AMessage(typedThread, messages);
+
+            a2aResponse = await this._a2aClient.SendMessageAsync(new MessageSendParams { Message = a2aMessage }, cancellationToken).ConfigureAwait(false);
+        }
 
         this._logger.LogAgentChatClientInvokedAgent(nameof(RunAsync), this.Id, this.Name);
 
         if (a2aResponse is AgentMessage message)
         {
-            UpdateThreadConversationId(typedThread, message.ContextId);
+            UpdateThread(typedThread, message.ContextId);
 
             return new AgentRunResponse
             {
@@ -101,21 +108,30 @@ internal sealed class A2AAgent : AIAgent
                 ResponseId = message.MessageId,
                 RawRepresentation = message,
                 Messages = [message.ToChatMessage()],
-                AdditionalProperties = message.Metadata.ToAdditionalProperties(),
+                AdditionalProperties = message.Metadata?.ToAdditionalProperties(),
             };
         }
+
         if (a2aResponse is AgentTask agentTask)
         {
-            UpdateThreadConversationId(typedThread, agentTask.ContextId);
+            UpdateThread(typedThread, agentTask.ContextId, agentTask.Id);
 
-            return new AgentRunResponse
+            var response = new AgentRunResponse
             {
                 AgentId = this.Id,
                 ResponseId = agentTask.Id,
                 RawRepresentation = agentTask,
-                Messages = agentTask.ToChatMessages(),
-                AdditionalProperties = agentTask.Metadata.ToAdditionalProperties(),
+                Messages = agentTask.ToChatMessages() ?? [],
+                ContinuationToken = CreateContinuationToken(agentTask.Id, agentTask.Status.State),
+                AdditionalProperties = agentTask.Metadata?.ToAdditionalProperties(),
             };
+
+            if (agentTask.ToChatMessages() is { Count: > 0 } taskMessages)
+            {
+                response.Messages = taskMessages;
+            }
+
+            return response;
         }
 
         throw new NotSupportedException($"Only Message and AgentTask responses are supported from A2A agents. Received: {a2aResponse.GetType().FullName ?? "null"}");
@@ -126,43 +142,67 @@ internal sealed class A2AAgent : AIAgent
     {
         _ = Throw.IfNull(messages);
 
-        var a2aMessage = messages.ToA2AMessage();
-
         thread ??= this.GetNewThread();
         if (thread is not A2AAgentThread typedThread)
         {
             throw new InvalidOperationException("The provided thread is not compatible with the agent. Only threads created by the agent can be used.");
         }
 
-        // Linking the message to the existing conversation, if any.
-        a2aMessage.ContextId = typedThread.ContextId;
-
         this._logger.LogA2AAgentInvokingAgent(nameof(RunStreamingAsync), this.Id, this.Name);
 
-        var a2aSseEvents = this._a2aClient.SendMessageStreamingAsync(new MessageSendParams { Message = a2aMessage }, cancellationToken).ConfigureAwait(false);
+        ConfiguredCancelableAsyncEnumerable<SseItem<A2AEvent>> a2aSseEvents;
+
+        if (options?.ContinuationToken is not null)
+        {
+            // Task stream resumption is not well defined in the A2A v2.* specification, leaving it to the agent implementations.  
+            // The v3.0 specification improves this by defining task stream reconnection that allows obtaining the same stream  
+            // from the beginning, but it does not define stream resumption from a specific point in the stream.  
+            // Therefore, the code should be updated once the A2A .NET library supports the A2A v3.0 specification,  
+            // and AF has the necessary model to allow consumers to know whether they need to resume the stream and add new updates to  
+            // the existing ones or reconnect the stream and obtain all updates again.  
+            // For more details, see the following issue: https://github.com/microsoft/agent-framework/issues/1764  
+            throw new InvalidOperationException("Reconnecting to task streams using continuation tokens is not supported yet.");
+            // a2aSseEvents = this._a2aClient.SubscribeToTaskAsync(token.TaskId, cancellationToken).ConfigureAwait(false);  
+        }
+
+        var a2aMessage = CreateA2AMessage(typedThread, messages);
+
+        a2aSseEvents = this._a2aClient.SendMessageStreamingAsync(new MessageSendParams { Message = a2aMessage }, cancellationToken).ConfigureAwait(false);
 
         this._logger.LogAgentChatClientInvokedAgent(nameof(RunStreamingAsync), this.Id, this.Name);
 
+        string? contextId = null;
+        string? taskId = null;
+
         await foreach (var sseEvent in a2aSseEvents)
         {
-            if (sseEvent.Data is not AgentMessage message)
+            if (sseEvent.Data is AgentMessage message)
             {
-                throw new NotSupportedException($"Only message responses are supported from A2A agents. Received: {sseEvent.Data?.GetType().FullName ?? "null"}");
+                contextId = message.ContextId;
+
+                yield return this.ConvertToAgentResponseUpdate(message);
             }
-
-            UpdateThreadConversationId(typedThread, message.ContextId);
-
-            yield return new AgentRunResponseUpdate
+            else if (sseEvent.Data is AgentTask task)
             {
-                AgentId = this.Id,
-                ResponseId = message.MessageId,
-                RawRepresentation = message,
-                Role = ChatRole.Assistant,
-                MessageId = message.MessageId,
-                Contents = [.. message.Parts.Select(part => part.ToAIContent()).OfType<AIContent>()],
-                AdditionalProperties = message.Metadata.ToAdditionalProperties(),
-            };
+                contextId = task.ContextId;
+                taskId = task.Id;
+
+                yield return this.ConvertToAgentResponseUpdate(task);
+            }
+            else if (sseEvent.Data is TaskUpdateEvent taskUpdateEvent)
+            {
+                contextId = taskUpdateEvent.ContextId;
+                taskId = taskUpdateEvent.TaskId;
+
+                yield return this.ConvertToAgentResponseUpdate(taskUpdateEvent);
+            }
+            else
+            {
+                throw new NotSupportedException($"Only message, task, task update events are supported from A2A agents. Received: {sseEvent.Data.GetType().FullName ?? "null"}");
+            }
         }
+
+        UpdateThread(typedThread, contextId, taskId);
     }
 
     /// <inheritdoc/>
@@ -177,7 +217,7 @@ internal sealed class A2AAgent : AIAgent
     /// <inheritdoc/>
     public override string? Description => this._description ?? base.Description;
 
-    private static void UpdateThreadConversationId(A2AAgentThread? thread, string? contextId)
+    private static void UpdateThread(A2AAgentThread? thread, string? contextId, string? taskId = null)
     {
         if (thread is null)
         {
@@ -194,5 +234,93 @@ internal sealed class A2AAgent : AIAgent
 
         // Assign a server-generated context Id to the thread if it's not already set.
         thread.ContextId ??= contextId;
+        thread.TaskId = taskId;
+    }
+
+    private static AgentMessage CreateA2AMessage(A2AAgentThread typedThread, IEnumerable<ChatMessage> messages)
+    {
+        var a2aMessage = messages.ToA2AMessage();
+
+        // Linking the message to the existing conversation, if any.
+        // See: https://github.com/a2aproject/A2A/blob/main/docs/topics/life-of-a-task.md#group-related-interactions
+        a2aMessage.ContextId = typedThread.ContextId;
+
+        // Link the message as a follow-up to an existing task, if any.
+        // See: https://github.com/a2aproject/A2A/blob/main/docs/topics/life-of-a-task.md#task-refinements
+        a2aMessage.ReferenceTaskIds = typedThread.TaskId is null ? null : [typedThread.TaskId];
+
+        return a2aMessage;
+    }
+
+    private static A2AContinuationToken? GetContinuationToken(IEnumerable<ChatMessage> messages, AgentRunOptions? options = null)
+    {
+        if (options?.ContinuationToken is ResponseContinuationToken token)
+        {
+            if (messages.Any())
+            {
+                throw new InvalidOperationException("Messages are not allowed when continuing a background response using a continuation token.");
+            }
+
+            return A2AContinuationToken.FromToken(token);
+        }
+
+        return null;
+    }
+
+    private static A2AContinuationToken? CreateContinuationToken(string taskId, TaskState state)
+    {
+        if (state == TaskState.Submitted || state == TaskState.Working)
+        {
+            return new A2AContinuationToken(taskId);
+        }
+
+        return null;
+    }
+
+    private AgentRunResponseUpdate ConvertToAgentResponseUpdate(AgentMessage message)
+    {
+        return new AgentRunResponseUpdate
+        {
+            AgentId = this.Id,
+            ResponseId = message.MessageId,
+            RawRepresentation = message,
+            Role = ChatRole.Assistant,
+            MessageId = message.MessageId,
+            Contents = message.Parts.ConvertAll(part => part.ToAIContent()),
+            AdditionalProperties = message.Metadata?.ToAdditionalProperties(),
+        };
+    }
+
+    private AgentRunResponseUpdate ConvertToAgentResponseUpdate(AgentTask task)
+    {
+        return new AgentRunResponseUpdate
+        {
+            AgentId = this.Id,
+            ResponseId = task.Id,
+            RawRepresentation = task,
+            Role = ChatRole.Assistant,
+            Contents = task.ToAIContents(),
+            AdditionalProperties = task.Metadata?.ToAdditionalProperties(),
+        };
+    }
+
+    private AgentRunResponseUpdate ConvertToAgentResponseUpdate(TaskUpdateEvent taskUpdateEvent)
+    {
+        AgentRunResponseUpdate responseUpdate = new()
+        {
+            AgentId = this.Id,
+            ResponseId = taskUpdateEvent.TaskId,
+            RawRepresentation = taskUpdateEvent,
+            Role = ChatRole.Assistant,
+            AdditionalProperties = taskUpdateEvent.Metadata?.ToAdditionalProperties() ?? [],
+        };
+
+        if (taskUpdateEvent is TaskArtifactUpdateEvent artifactUpdateEvent)
+        {
+            responseUpdate.Contents = artifactUpdateEvent.Artifact.ToAIContents();
+            responseUpdate.RawRepresentation = artifactUpdateEvent;
+        }
+
+        return responseUpdate;
     }
 }

dotnet/src/Microsoft.Agents.AI.A2A/A2AAgentThread.cs

@@ -1,5 +1,6 @@
 // Copyright (c) Microsoft. All rights reserved.
 
+using System;
 using System.Text.Json;
 
 namespace Microsoft.Agents.AI.A2A;
@@ -7,22 +8,59 @@ namespace Microsoft.Agents.AI.A2A;
 /// <summary>
 /// Thread for A2A based agents.
 /// </summary>
-public sealed class A2AAgentThread : ServiceIdAgentThread
+public sealed class A2AAgentThread : AgentThread
 {
     internal A2AAgentThread()
     {
     }
 
-    internal A2AAgentThread(JsonElement serializedThreadState, JsonSerializerOptions? jsonSerializerOptions = null) : base(serializedThreadState, jsonSerializerOptions)
+    internal A2AAgentThread(JsonElement serializedThreadState, JsonSerializerOptions? jsonSerializerOptions = null)
     {
+        if (serializedThreadState.ValueKind != JsonValueKind.Object)
+        {
+            throw new ArgumentException("The serialized thread state must be a JSON object.", nameof(serializedThreadState));
+        }
+
+        var state = serializedThreadState.Deserialize(
+            A2AJsonUtilities.DefaultOptions.GetTypeInfo(typeof(A2AAgentThreadState))) as A2AAgentThreadState;
+
+        if (state?.ContextId is string contextId)
+        {
+            this.ContextId = contextId;
+        }
+
+        if (state?.TaskId is string taskId)
+        {
+            this.TaskId = taskId;
+        }
     }
 
     /// <summary>
     /// Gets the ID for the current conversation with the A2A agent.
     /// </summary>
-    public string? ContextId
+    public string? ContextId { get; internal set; }
+
+    /// <summary>
+    /// Gets the ID for the task the agent is currently working on.
+    /// </summary>
+    public string? TaskId { get; internal set; }
+
+    /// <inheritdoc/>
+    public override JsonElement Serialize(JsonSerializerOptions? jsonSerializerOptions = null)
     {
-        get { return this.ServiceThreadId; }
-        internal set { this.ServiceThreadId = value; }
+        var state = new A2AAgentThreadState
+        {
+            ContextId = this.ContextId,
+            TaskId = this.TaskId
+        };
+
+        return JsonSerializer.SerializeToElement(state, A2AJsonUtilities.DefaultOptions.GetTypeInfo(typeof(A2AAgentThreadState)));
+    }
+
+    internal sealed class A2AAgentThreadState
+    {
+        public string? ContextId { get; set; }
+
+        public string? TaskId { get; set; }
     }
 }

dotnet/src/Microsoft.Agents.AI.A2A/A2AContinuationToken.cs

@@ -0,0 +1,81 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System;
+using System.IO;
+using System.Text.Json;
+using Microsoft.Extensions.AI;
+using Microsoft.Shared.Diagnostics;
+
+namespace Microsoft.Agents.AI.A2A;
+#pragma warning disable MEAI001 // Type is for evaluation purposes only and is subject to change or removal in future updates. Suppress this diagnostic to proceed.
+internal class A2AContinuationToken : ResponseContinuationToken
+{
+    internal A2AContinuationToken(string taskId)
+    {
+        _ = Throw.IfNullOrEmpty(taskId);
+
+        this.TaskId = taskId;
+    }
+
+    internal string TaskId { get; }
+
+    internal static A2AContinuationToken FromToken(ResponseContinuationToken token)
+    {
+        if (token is A2AContinuationToken longRunContinuationToken)
+        {
+            return longRunContinuationToken;
+        }
+
+        ReadOnlyMemory<byte> data = token.ToBytes();
+
+        if (data.Length == 0)
+        {
+            Throw.ArgumentException(nameof(token), "Failed to create A2AContinuationToken from provided token because it does not contain any data.");
+        }
+
+        Utf8JsonReader reader = new(data.Span);
+
+        string taskId = null!;
+
+        reader.Read();
+
+        while (reader.Read())
+        {
+            if (reader.TokenType == JsonTokenType.EndObject)
+            {
+                break;
+            }
+
+            string propertyName = reader.GetString() ?? throw new JsonException("Failed to read property name from continuation token.");
+
+            switch (propertyName)
+            {
+                case "taskId":
+                    reader.Read();
+                    taskId = reader.GetString()!;
+                    break;
+                default:
+                    throw new JsonException($"Unrecognized property '{propertyName}'.");
+            }
+        }
+
+        return new(taskId);
+    }
+
+    public override ReadOnlyMemory<byte> ToBytes()
+    {
+        using MemoryStream stream = new();
+        using Utf8JsonWriter writer = new(stream);
+
+        writer.WriteStartObject();
+
+        writer.WriteString("taskId", this.TaskId);
+
+        writer.WriteEndObject();
+
+        writer.Flush();
+        stream.Position = 0;
+
+        return stream.ToArray();
+    }
+}

dotnet/src/Microsoft.Agents.AI.A2A/A2AJsonUtilities.cs

@@ -0,0 +1,80 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Diagnostics.CodeAnalysis;
+using System.Text.Encodings.Web;
+using System.Text.Json;
+using System.Text.Json.Serialization;
+using Microsoft.Agents.AI.A2A;
+
+namespace Microsoft.Agents.AI;
+
+/// <summary>
+/// Provides utility methods and configurations for JSON serialization operations for A2A agent types.
+/// </summary>
+public static partial class A2AJsonUtilities
+{
+    /// <summary>
+    /// Gets the default <see cref="JsonSerializerOptions"/> instance used for JSON serialization operations of A2A agent types.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// For Native AOT or applications disabling <see cref="JsonSerializer.IsReflectionEnabledByDefault"/>, this instance
+    /// includes source generated contracts for A2A agent types.
+    /// </para>
+    /// <para>
+    /// It additionally turns on the following settings:
+    /// <list type="number">
+    /// <item><description>Enables <see cref="JsonSerializerDefaults.Web"/> defaults.</description></item>
+    /// <item><description>Enables <see cref="JsonIgnoreCondition.WhenWritingNull"/> as the default ignore condition for properties.</description></item>
+    /// <item><description>Enables <see cref="JsonNumberHandling.AllowReadingFromString"/> as the default number handling for number types.</description></item>
+    /// <item><description>
+    /// Enables <see cref="JavaScriptEncoder.UnsafeRelaxedJsonEscaping"/> when escaping JSON strings.
+    /// Consuming applications must ensure that JSON outputs are adequately escaped before embedding in other document formats, such as HTML and XML.
+    /// </description></item>
+    /// </list>
+    /// </para>
+    /// </remarks>
+    public static JsonSerializerOptions DefaultOptions { get; } = CreateDefaultOptions();
+
+    /// <summary>
+    /// Creates and configures the default JSON serialization options for agent abstraction types.
+    /// </summary>
+    /// <returns>The configured options.</returns>
+    [UnconditionalSuppressMessage("ReflectionAnalysis", "IL3050:RequiresDynamicCode", Justification = "Converter is guarded by IsReflectionEnabledByDefault check.")]
+    [UnconditionalSuppressMessage("Trimming", "IL2026:Members annotated with 'RequiresUnreferencedCodeAttribute' require dynamic access", Justification = "Converter is guarded by IsReflectionEnabledByDefault check.")]
+    private static JsonSerializerOptions CreateDefaultOptions()
+    {
+        // Copy the configuration from the source generated context.
+        JsonSerializerOptions options = new(JsonContext.Default.Options)
+        {
+            Encoder = JavaScriptEncoder.UnsafeRelaxedJsonEscaping, // same as AIJsonUtilities
+        };
+
+        // Chain in the resolvers from both AIJsonUtilities and our source generated context.
+        // We want AIJsonUtilities first to ensure any M.E.AI types are handled via its resolver.
+        options.TypeInfoResolverChain.Clear();
+        options.TypeInfoResolverChain.Add(AgentAbstractionsJsonUtilities.DefaultOptions.TypeInfoResolver!);
+
+        // If reflection-based serialization is enabled by default, this includes
+        // the default type info resolver that utilizes reflection, but we need to manually
+        // apply the same converter AIJsonUtilities adds for string-based enum serialization,
+        // as that's not propagated as part of the resolver.
+        if (JsonSerializer.IsReflectionEnabledByDefault)
+        {
+            options.Converters.Add(new JsonStringEnumConverter());
+        }
+
+        options.MakeReadOnly();
+        return options;
+    }
+
+    [JsonSourceGenerationOptions(JsonSerializerDefaults.Web,
+        UseStringEnumConverter = true,
+        DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull,
+        NumberHandling = JsonNumberHandling.AllowReadingFromString)]
+
+    // A2A agent types
+    [JsonSerializable(typeof(A2AAgentThread.A2AAgentThreadState))]
+    [ExcludeFromCodeCoverage]
+    private sealed partial class JsonContext : JsonSerializerContext;
+}

dotnet/src/Microsoft.Agents.AI.A2A/Extensions/A2AAgentTaskExtensions.cs

@@ -11,20 +11,37 @@ namespace A2A;
 /// </summary>
 internal static class A2AAgentTaskExtensions
 {
-    internal static IList<ChatMessage> ToChatMessages(this AgentTask agentTask)
+    internal static IList<ChatMessage>? ToChatMessages(this AgentTask agentTask)
     {
         _ = Throw.IfNull(agentTask);
 
-        List<ChatMessage> messages = [];
+        List<ChatMessage>? messages = null;
 
-        if (agentTask.Artifacts is not null)
+        if (agentTask?.Artifacts is { Count: > 0 })
         {
             foreach (var artifact in agentTask.Artifacts)
             {
-                messages.Add(artifact.ToChatMessage());
+                (messages ??= []).Add(artifact.ToChatMessage());
             }
         }
 
         return messages;
     }
+
+    internal static IList<AIContent>? ToAIContents(this AgentTask agentTask)
+    {
+        _ = Throw.IfNull(agentTask);
+
+        List<AIContent>? aiContents = null;
+
+        if (agentTask.Artifacts is not null)
+        {
+            foreach (var artifact in agentTask.Artifacts)
+            {
+                (aiContents ??= []).AddRange(artifact.ToAIContents());
+            }
+        }
+
+        return aiContents;
+    }
 }

dotnet/src/Microsoft.Agents.AI.A2A/Extensions/A2AArtifactExtensions.cs

@@ -12,21 +12,15 @@ internal static class A2AArtifactExtensions
 {
     internal static ChatMessage ToChatMessage(this Artifact artifact)
     {
-        List<AIContent>? aiContents = null;
-
-        foreach (var part in artifact.Parts)
-        {
-            var content = part.ToAIContent();
-            if (content is not null)
-            {
-                (aiContents ??= []).Add(content);
-            }
-        }
-
-        return new ChatMessage(ChatRole.Assistant, aiContents)
+        return new ChatMessage(ChatRole.Assistant, artifact.ToAIContents())
         {
             AdditionalProperties = artifact.Metadata.ToAdditionalProperties(),
             RawRepresentation = artifact,
         };
     }
+
+    internal static List<AIContent> ToAIContents(this Artifact artifact)
+    {
+        return artifact.Parts.ConvertAll(part => part.ToAIContent());
+    }
 }

dotnet/src/Microsoft.Agents.AI.A2A/Microsoft.Agents.AI.A2A.csproj

@@ -2,12 +2,14 @@
 
   <PropertyGroup>
     <VersionSuffix>preview</VersionSuffix>
+    <NoWarn>$(NoWarn);MEAI001</NoWarn>
   </PropertyGroup>
 
   <Import Project="$(RepoRoot)/dotnet/nuget/nuget-package.props" />
 
   <PropertyGroup>
     <InjectSharedThrow>true</InjectSharedThrow>
+    <InjectTrimAttributesOnLegacy>true</InjectTrimAttributesOnLegacy>
   </PropertyGroup>
 
   <ItemGroup>