.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 <[email protected]>

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

Co-authored-by: Roger Barreto <[email protected]>

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

Co-authored-by: Roger Barreto <[email protected]>

* address pr review feedback

* add clarification regarding background responses

---------

Co-authored-by: Roger Barreto <[email protected]>

Author: SergeyMenshykh

dotnet/agent-framework-dotnet.slnx

@@ -40,6 +40,7 @@
   <Folder Name="/Samples/GettingStarted/A2A/">
     <File Path="samples/GettingStarted/A2A/README.md" />
     <Project Path="samples/GettingStarted/A2A/A2AAgent_AsFunctionTools/A2AAgent_AsFunctionTools.csproj" />
+    <Project Path="samples/GettingStarted/A2A/A2AAgent_PollingForTaskCompletion/A2AAgent_PollingForTaskCompletion.csproj" />
   </Folder>
   <Folder Name="/Samples/GettingStarted/AgentProviders/">
     <File Path="samples/GettingStarted/AgentProviders/README.md" />
@@ -389,4 +390,4 @@
     <Project Path="tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests.csproj" />
     <Project Path="tests/Microsoft.Agents.AI.Workflows.UnitTests/Microsoft.Agents.AI.Workflows.UnitTests.csproj" />
   </Folder>
-</Solution>
+</Solution>

dotnet/samples/GettingStarted/A2A/A2AAgent_PollingForTaskCompletion/A2AAgent_PollingForTaskCompletion.csproj

@@ -0,0 +1,25 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net10.0</TargetFramework>
+
+    <Nullable>enable</Nullable>
+    <ImplicitUsings>enable</ImplicitUsings>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="A2A" />
+    <PackageReference Include="Azure.AI.OpenAI" />
+    <PackageReference Include="Azure.Identity" />
+    <PackageReference Include="Microsoft.Extensions.Hosting" />
+    <PackageReference Include="System.Net.ServerSentEvents" />
+    <PackageReference Include="Microsoft.Bcl.AsyncInterfaces" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\..\..\src\Microsoft.Agents.AI.A2A\Microsoft.Agents.AI.A2A.csproj" />
+    <ProjectReference Include="..\..\..\..\src\Microsoft.Agents.AI.OpenAI\Microsoft.Agents.AI.OpenAI.csproj" />
+  </ItemGroup>
+
+</Project>

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

@@ -0,0 +1,35 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+// This sample demonstrates how to poll for long-running task completion using continuation tokens with an A2A AI agent.
+
+using A2A;
+using Microsoft.Agents.AI;
+
+var a2aAgentHost = Environment.GetEnvironmentVariable("A2A_AGENT_HOST") ?? throw new InvalidOperationException("A2A_AGENT_HOST is not set.");
+
+// Initialize an A2ACardResolver to get an A2A agent card.
+A2ACardResolver agentCardResolver = new(new Uri(a2aAgentHost));
+
+// Get the agent card
+AgentCard agentCard = await agentCardResolver.GetAgentCardAsync();
+
+// Create an instance of the AIAgent for an existing A2A agent specified by the agent card.
+AIAgent agent = agentCard.GetAIAgent();
+
+AgentThread thread = agent.GetNewThread();
+
+// Start the initial run with a long-running task.
+AgentRunResponse response = await agent.RunAsync("Conduct a comprehensive analysis of quantum computing applications in cryptography, including recent breakthroughs, implementation challenges, and future roadmap. Please include diagrams and visual representations to illustrate complex concepts.", thread);
+
+// Poll until the response is complete.
+while (response.ContinuationToken is { } token)
+{
+    // Wait before polling again.
+    await Task.Delay(TimeSpan.FromSeconds(2));
+
+    // Continue with the token.
+    response = await agent.RunAsync(thread, options: new AgentRunOptions { ContinuationToken = token });
+}
+
+// Display the result
+Console.WriteLine(response);

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

@@ -0,0 +1,25 @@
+# Polling for A2A Agent Task Completion
+
+This sample demonstrates how to poll for long-running task completion using continuation tokens with an A2A AI agent, following the background responses pattern.
+
+The sample:
+
+- Connects to an A2A agent server specified in the `A2A_AGENT_HOST` environment variable
+- Sends a request to the agent that may take time to complete
+- Polls the agent at regular intervals using continuation tokens until a final response is received
+- Displays the final result
+
+This pattern is useful when an AI model cannot complete a complex task in a single response and needs multiple rounds of processing.
+
+# Prerequisites
+
+Before you begin, ensure you have the following prerequisites:
+
+- .NET 10.0 SDK or later
+- An A2A agent server running and accessible via HTTP
+
+Set the following environment variable:
+
+```powershell
+$env:A2A_AGENT_HOST="http://localhost:5000"  # Replace with your A2A agent server host
+```

dotnet/samples/GettingStarted/A2A/README.md

@@ -14,6 +14,7 @@ See the README.md for each sample for the prerequisites for that sample.
 |Sample|Description|
 |---|---|
 |[A2A Agent As Function Tools](./A2AAgent_AsFunctionTools/)|This sample demonstrates how to represent an A2A agent as a set of function tools, where each function tool corresponds to a skill of the A2A agent, and register these function tools with another AI agent so it can leverage the A2A agent's skills.|
+|[A2A Agent Polling For Task Completion](./A2AAgent_PollingForTaskCompletion/)|This sample demonstrates how to poll for long-running task completion using continuation tokens with an A2A agent.|
 
 ## Running the samples from the console
 

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,48 +75,63 @@ public override async Task<AgentRunResponse> RunAsync(IEnumerable<ChatMessage> m
     {
         _ = 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
             {
                 AgentId = this.Id,
                 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 @@ public override async IAsyncEnumerable<AgentRunResponseUpdate> RunStreamingAsync
     {
         _ = 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);
             }
+            else if (sseEvent.Data is AgentTask task)
+            {
+                contextId = task.ContextId;
+                taskId = task.Id;
 
-            UpdateThreadConversationId(typedThread, message.ContextId);
+                yield return this.ConvertToAgentResponseUpdate(task);
+            }
+            else if (sseEvent.Data is TaskUpdateEvent taskUpdateEvent)
+            {
+                contextId = taskUpdateEvent.ContextId;
+                taskId = taskUpdateEvent.TaskId;
 
-            yield return new AgentRunResponseUpdate
+                yield return this.ConvertToAgentResponseUpdate(taskUpdateEvent);
+            }
+            else
             {
-                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(),
-            };
+                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 @@ public override async IAsyncEnumerable<AgentRunResponseUpdate> RunStreamingAsync
     /// <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 @@ private static void UpdateThreadConversationId(A2AAgentThread? thread, string? c
 
         // 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;
     }
 }