agent/agent-framework
2
0
Fork 0
mirror of https://github.com/microsoft/agent-framework.git synced 2026-06-16 21:04:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

.NET: [Feature Branch] Add Azure Functions hosting support for durable workflows (#3935)

Browse Source 
* Adding azure functions workflow support.

* - PR feedback fixes.
- Add example to demonstrate complex Object as payload.

* rename instanceId to runId.

* Use custom ITaskOrchestrator to run orchestrator function.
This commit is contained in:
Shyju Krishnankutty
2026-02-14 16:19:28 -08:00
committed by GitHub
Unverified
parent e8d0bd9051
commit b62b1f2191
26 changed files with 1769 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dotnet/Directory.Packages.props
+5 -5
View File
@@ -113,14 +113,14 @@
<PackageVersion Include="Microsoft.Agents.ObjectModel.PowerFx" Version="2026.1.2.3" />
<PackageVersion Include="Microsoft.PowerFx.Interpreter" Version="1.5.0-build.20251008-1002" />
<!-- Durable Task -->
<PackageVersion Include="Microsoft.DurableTask.Client" Version="1.18.0" />
<PackageVersion Include="Microsoft.DurableTask.Client.AzureManaged" Version="1.18.0" />
<PackageVersion Include="Microsoft.DurableTask.Worker" Version="1.18.0" />
<PackageVersion Include="Microsoft.DurableTask.Worker.AzureManaged" Version="1.18.0" />
<PackageVersion Include="Microsoft.DurableTask.Client" Version="1.19.1" />
<PackageVersion Include="Microsoft.DurableTask.Client.AzureManaged" Version="1.19.0" />
<PackageVersion Include="Microsoft.DurableTask.Worker" Version="1.19.0" />
<PackageVersion Include="Microsoft.DurableTask.Worker.AzureManaged" Version="1.19.0" />
<!-- Azure Functions -->
<PackageVersion Include="Microsoft.Azure.Functions.Worker" Version="2.50.0" />
<PackageVersion Include="Microsoft.Azure.Functions.Worker.ApplicationInsights" Version="2.50.0" />
<PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask" Version="1.11.0" />
<PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask" Version="1.13.1" />
<PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask.AzureManaged" Version="1.0.1" />
<PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.Http" Version="3.3.0" />
<PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.Http.AspNetCore" Version="2.1.0" />
dotnet/agent-framework-dotnet.slnx
+4
View File
@@ -53,6 +53,10 @@
<Project Path="samples/Durable/Workflow/ConsoleApps/03_ConditionalEdges/03_ConditionalEdges.csproj" />
<Project Path="samples/Durable/Workflow/ConsoleApps/04_WorkflowAndAgents/04_WorkflowAndAgents.csproj" />
</Folder>
<Folder Name="/Samples/Durable/Workflows/AzureFunctions/">
<Project Path="samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/01_SequentialWorkflow.csproj" />
<Project Path="samples/Durable/Workflow/AzureFunctions/02_ConcurrentWorkflow/02_ConcurrentWorkflow.csproj" />
</Folder>
<Folder Name="/Samples/GettingStarted/">
<File Path="samples/GettingStarted/README.md" />
</Folder>
dotnet/samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/01_SequentialWorkflow.csproj
+42
View File
@@ -0,0 +1,42 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFrameworks>net10.0</TargetFrameworks>
<AzureFunctionsVersion>v4</AzureFunctionsVersion>
<OutputType>Exe</OutputType>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
<!-- The Functions build tools don't like namespaces that start with a number -->
<AssemblyName>SingleAgent</AssemblyName>
<RootNamespace>SingleAgent</RootNamespace>
</PropertyGroup>
<ItemGroup>
<FrameworkReference Include="Microsoft.AspNetCore.App" />
</ItemGroup>
<!-- Azure Functions packages -->
<ItemGroup>
<PackageReference Include="Microsoft.Azure.Functions.Worker" />
<PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask" />
<PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask.AzureManaged" />
<PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.Http.AspNetCore" />
<PackageReference Include="Microsoft.Azure.Functions.Worker.Sdk" />
</ItemGroup>
<ItemGroup>
<PackageReference Include="Azure.AI.OpenAI" />
<PackageReference Include="Azure.Identity" />
</ItemGroup>
<!-- Local projects that should be switched to package references when using the sample outside of this MAF repo -->
<!--
<ItemGroup>
<PackageReference Include="Microsoft.Agents.AI.Hosting.AzureFunctions" />
<PackageReference Include="Microsoft.Agents.AI.OpenAI" />
</ItemGroup>
-->
<ItemGroup>
<ProjectReference Include="..\..\..\..\..\src\Microsoft.Agents.AI.Hosting.AzureFunctions\Microsoft.Agents.AI.Hosting.AzureFunctions.csproj" />
<ProjectReference Include="..\..\..\..\..\src\Microsoft.Agents.AI.OpenAI\Microsoft.Agents.AI.OpenAI.csproj" />
</ItemGroup>
</Project>
dotnet/samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/OrderCancelExecutors.cs
+215
View File
@@ -0,0 +1,215 @@
// Copyright (c) Microsoft. All rights reserved.
using Microsoft.Agents.AI.Workflows;
namespace SequentialWorkflow;
/// <summary>
/// Looks up an order by its ID and return an Order object.
/// </summary>
internal sealed class OrderLookup() : Executor<string, Order>("OrderLookup")
{
public override async ValueTask<Order> HandleAsync(
string message,
IWorkflowContext context,
CancellationToken cancellationToken = default)
{
Console.WriteLine();
Console.ForegroundColor = ConsoleColor.Magenta;
Console.WriteLine("┌─────────────────────────────────────────────────────────────────┐");
Console.WriteLine($"│ [Activity] OrderLookup: Starting lookup for order '{message}'");
Console.ResetColor();
// Simulate database lookup with delay
await Task.Delay(TimeSpan.FromMicroseconds(100), cancellationToken);
Order order = new(
Id: message,
OrderDate: DateTime.UtcNow.AddDays(-1),
IsCancelled: false,
Customer: new Customer(Name: "Jerry", Email: "jerry@example.com"));
Console.ForegroundColor = ConsoleColor.Magenta;
Console.WriteLine($"│ [Activity] OrderLookup: Found order '{message}' for customer '{order.Customer.Name}'");
Console.WriteLine("└─────────────────────────────────────────────────────────────────┘");
Console.ResetColor();
return order;
}
}
/// <summary>
/// Cancels an order.
/// </summary>
internal sealed class OrderCancel() : Executor<Order, Order>("OrderCancel")
{
public override async ValueTask<Order> HandleAsync(
Order message,
IWorkflowContext context,
CancellationToken cancellationToken = default)
{
Console.WriteLine();
Console.ForegroundColor = ConsoleColor.Yellow;
Console.WriteLine("┌─────────────────────────────────────────────────────────────────┐");
Console.WriteLine($"│ [Activity] OrderCancel: Starting cancellation for order '{message.Id}'");
Console.ResetColor();
// Simulate a slow cancellation process (e.g., calling external payment system)
for (int i = 1; i <= 3; i++)
{
await Task.Delay(TimeSpan.FromMilliseconds(100), cancellationToken);
Console.ForegroundColor = ConsoleColor.DarkYellow;
Console.WriteLine("│ [Activity] OrderCancel: Processing...");
Console.ResetColor();
}
Order cancelledOrder = message with { IsCancelled = true };
Console.ForegroundColor = ConsoleColor.Yellow;
Console.WriteLine($"│ [Activity] OrderCancel: ✓ Order '{cancelledOrder.Id}' has been cancelled");
Console.WriteLine("└─────────────────────────────────────────────────────────────────┘");
Console.ResetColor();
return cancelledOrder;
}
}
/// <summary>
/// Sends a cancellation confirmation email to the customer.
/// </summary>
internal sealed class SendEmail() : Executor<Order, string>("SendEmail")
{
public override ValueTask<string> HandleAsync(
Order message,
IWorkflowContext context,
CancellationToken cancellationToken = default)
{
Console.WriteLine();
Console.ForegroundColor = ConsoleColor.Cyan;
Console.WriteLine("┌─────────────────────────────────────────────────────────────────┐");
Console.WriteLine($"│ [Activity] SendEmail: Sending email to '{message.Customer.Email}'...");
Console.ResetColor();
string result = $"Cancellation email sent for order {message.Id} to {message.Customer.Email}.";
Console.ForegroundColor = ConsoleColor.Cyan;
Console.WriteLine("│ [Activity] SendEmail: ✓ Email sent successfully!");
Console.WriteLine("└─────────────────────────────────────────────────────────────────┘");
Console.ResetColor();
return ValueTask.FromResult(result);
}
}
internal sealed record Order(string Id, DateTime OrderDate, bool IsCancelled, Customer Customer);
internal sealed record Customer(string Name, string Email);
/// <summary>
/// Represents a batch cancellation request with multiple order IDs and a reason.
/// This demonstrates using a complex typed object as workflow input.
/// </summary>
#pragma warning disable CA1812 // Instantiated via JSON deserialization at runtime
internal sealed record BatchCancelRequest(string[] OrderIds, string Reason, bool NotifyCustomers);
#pragma warning restore CA1812
/// <summary>
/// Represents the result of processing a batch cancellation.
/// </summary>
internal sealed record BatchCancelResult(int TotalOrders, int CancelledCount, string Reason);
/// <summary>
/// Generates a status report for an order.
/// </summary>
internal sealed class StatusReport() : Executor<Order, string>("StatusReport")
{
public override ValueTask<string> HandleAsync(
Order message,
IWorkflowContext context,
CancellationToken cancellationToken = default)
{
Console.WriteLine();
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine("┌─────────────────────────────────────────────────────────────────┐");
Console.WriteLine($"│ [Activity] StatusReport: Generating report for order '{message.Id}'");
Console.ResetColor();
string status = message.IsCancelled ? "Cancelled" : "Active";
string result = $"Order {message.Id} for {message.Customer.Name}: Status={status}, Date={message.OrderDate:yyyy-MM-dd}";
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine($"│ [Activity] StatusReport: ✓ {result}");
Console.WriteLine("└─────────────────────────────────────────────────────────────────┘");
Console.ResetColor();
return ValueTask.FromResult(result);
}
}
/// <summary>
/// Processes a batch cancellation request. Accepts a complex <see cref="BatchCancelRequest"/> object
/// as input, demonstrating how workflows can receive structured JSON input.
/// </summary>
internal sealed class BatchCancelProcessor() : Executor<BatchCancelRequest, BatchCancelResult>("BatchCancelProcessor")
{
public override async ValueTask<BatchCancelResult> HandleAsync(
BatchCancelRequest message,
IWorkflowContext context,
CancellationToken cancellationToken = default)
{
Console.WriteLine();
Console.ForegroundColor = ConsoleColor.Yellow;
Console.WriteLine("┌─────────────────────────────────────────────────────────────────┐");
Console.WriteLine($"│ [Activity] BatchCancelProcessor: Processing {message.OrderIds.Length} orders");
Console.WriteLine($"│ [Activity] BatchCancelProcessor: Reason: {message.Reason}");
Console.WriteLine($"│ [Activity] BatchCancelProcessor: Notify customers: {message.NotifyCustomers}");
Console.ResetColor();
// Simulate processing each order
int cancelledCount = 0;
foreach (string orderId in message.OrderIds)
{
await Task.Delay(TimeSpan.FromMilliseconds(100), cancellationToken);
cancelledCount++;
Console.ForegroundColor = ConsoleColor.DarkYellow;
Console.WriteLine($"│ [Activity] BatchCancelProcessor: ✓ Cancelled order '{orderId}'");
Console.ResetColor();
}
BatchCancelResult result = new(message.OrderIds.Length, cancelledCount, message.Reason);
Console.ForegroundColor = ConsoleColor.Yellow;
Console.WriteLine($"│ [Activity] BatchCancelProcessor: ✓ Batch complete: {cancelledCount}/{message.OrderIds.Length} cancelled");
Console.WriteLine("└─────────────────────────────────────────────────────────────────┘");
Console.ResetColor();
return result;
}
}
/// <summary>
/// Generates a summary of the batch cancellation.
/// </summary>
internal sealed class BatchCancelSummary() : Executor<BatchCancelResult, string>("BatchCancelSummary")
{
public override ValueTask<string> HandleAsync(
BatchCancelResult message,
IWorkflowContext context,
CancellationToken cancellationToken = default)
{
Console.WriteLine();
Console.ForegroundColor = ConsoleColor.Cyan;
Console.WriteLine("┌─────────────────────────────────────────────────────────────────┐");
Console.WriteLine("│ [Activity] BatchCancelSummary: Generating summary");
Console.ResetColor();
string result = $"Batch cancellation complete: {message.CancelledCount}/{message.TotalOrders} orders cancelled. Reason: {message.Reason}";
Console.ForegroundColor = ConsoleColor.Cyan;
Console.WriteLine($"│ [Activity] BatchCancelSummary: ✓ {result}");
Console.WriteLine("└─────────────────────────────────────────────────────────────────┘");
Console.ResetColor();
return ValueTask.FromResult(result);
}
}
dotnet/samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/Program.cs
+52
View File
@@ -0,0 +1,52 @@
// Copyright (c) Microsoft. All rights reserved.
// This sample demonstrates three workflows that share executors.
// The CancelOrder workflow cancels an order and notifies the customer.
// The OrderStatus workflow looks up an order and generates a status report.
// The BatchCancelOrders workflow accepts a complex JSON input to cancel multiple orders.
// Both CancelOrder and OrderStatus reuse the same OrderLookup executor, demonstrating executor sharing.
using Microsoft.Agents.AI.Hosting.AzureFunctions;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Azure.Functions.Worker.Builder;
using Microsoft.Extensions.Hosting;
using SequentialWorkflow;
// Define executors for all workflows
OrderLookup orderLookup = new();
OrderCancel orderCancel = new();
SendEmail sendEmail = new();
StatusReport statusReport = new();
BatchCancelProcessor batchCancelProcessor = new();
BatchCancelSummary batchCancelSummary = new();
// Build the CancelOrder workflow: OrderLookup -> OrderCancel -> SendEmail
Workflow cancelOrder = new WorkflowBuilder(orderLookup)
.WithName("CancelOrder")
.WithDescription("Cancel an order and notify the customer")
.AddEdge(orderLookup, orderCancel)
.AddEdge(orderCancel, sendEmail)
.Build();
// Build the OrderStatus workflow: OrderLookup -> StatusReport
// This workflow shares the OrderLookup executor with the CancelOrder workflow.
Workflow orderStatus = new WorkflowBuilder(orderLookup)
.WithName("OrderStatus")
.WithDescription("Look up an order and generate a status report")
.AddEdge(orderLookup, statusReport)
.Build();
// Build the BatchCancelOrders workflow: BatchCancelProcessor -> BatchCancelSummary
// This workflow demonstrates using a complex JSON object as the workflow input.
Workflow batchCancelOrders = new WorkflowBuilder(batchCancelProcessor)
.WithName("BatchCancelOrders")
.WithDescription("Cancel multiple orders in a batch using a complex JSON input")
.AddEdge(batchCancelProcessor, batchCancelSummary)
.Build();
using IHost app = FunctionsApplication
.CreateBuilder(args)
.ConfigureFunctionsWebApplication()
.ConfigureDurableWorkflows(workflows => workflows.AddWorkflows(cancelOrder, orderStatus, batchCancelOrders))
.Build();
app.Run();
dotnet/samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/README.md
+100
View File
@@ -0,0 +1,100 @@
# Sequential Workflow Sample
This sample demonstrates how to use the Microsoft Agent Framework to create an Azure Functions app that hosts durable workflows with sequential executor chains. It showcases two workflows that share a common executor, demonstrating executor reuse across workflows.
## Key Concepts Demonstrated
- Defining workflows with sequential executor chains using `WorkflowBuilder`
- Sharing executors across multiple workflows (the `OrderLookup` executor is used by both workflows)
- Registering workflows with the Function app using `ConfigureDurableWorkflows`
- Durable orchestration ensuring workflows survive process restarts and failures
- Starting workflows via HTTP requests
- Viewing workflow execution history and status in the Durable Task Scheduler (DTS) dashboard
## Workflows
This sample defines two workflows:
1. **CancelOrder**: `OrderLookup` → `OrderCancel` → `SendEmail` — Looks up an order, cancels it, and sends a confirmation email.
2. **OrderStatus**: `OrderLookup` → `StatusReport` — Looks up an order and generates a status report.
Both workflows share the `OrderLookup` executor, which is registered only once by the framework.
## Environment Setup
See the [README.md](../../README.md) file in the parent directory for more information on how to configure the environment, including how to install and run common sample dependencies.
## Running the Sample
With the environment setup and function app running, you can test the sample by sending HTTP requests to the workflow endpoints.
You can use the `demo.http` file to trigger the workflows, or a command line tool like `curl` as shown below:
### Cancel an Order
Bash (Linux/macOS/WSL):
```bash
curl -X POST http://localhost:7071/api/workflows/CancelOrder/run \
-H "Content-Type: text/plain" \
-d "12345"
```
PowerShell:
```powershell
Invoke-RestMethod -Method Post `
-Uri http://localhost:7071/api/workflows/CancelOrder/run `
-ContentType text/plain `
-Body "12345"
```
The response will confirm the workflow orchestration has started:
```text
Workflow orchestration started for CancelOrder. Orchestration runId: abc123def456
```
> **Tip:** You can provide a custom run ID by appending a `runId` query parameter:
>
> ```bash
> curl -X POST "http://localhost:7071/api/workflows/CancelOrder/run?runId=my-order-123" \
> -H "Content-Type: text/plain" \
> -d "12345"
> ```
>
> If not provided, a unique run ID is auto-generated.
In the function app logs, you will see the sequential execution of each executor:
```text
│ [Activity] OrderLookup: Starting lookup for order '12345'
│ [Activity] OrderLookup: Found order '12345' for customer 'Jerry'
│ [Activity] OrderCancel: Starting cancellation for order '12345'
│ [Activity] OrderCancel: ✓ Order '12345' has been cancelled
│ [Activity] SendEmail: Sending email to 'jerry@example.com'...
│ [Activity] SendEmail: ✓ Email sent successfully!
```
### Get Order Status
```bash
curl -X POST http://localhost:7071/api/workflows/OrderStatus/run \
-H "Content-Type: text/plain" \
-d "12345"
```
The `OrderStatus` workflow reuses the same `OrderLookup` executor and then generates a status report:
```text
│ [Activity] OrderLookup: Starting lookup for order '12345'
│ [Activity] OrderLookup: Found order '12345' for customer 'Jerry'
│ [Activity] StatusReport: Generating report for order '12345'
│ [Activity] StatusReport: ✓ Order 12345 for Jerry: Status=Active, Date=2025-01-01
```
### Viewing Workflows in the DTS Dashboard
After running a workflow, you can navigate to the Durable Task Scheduler (DTS) dashboard to visualize the completed orchestration, inspect inputs/outputs for each step, and view execution history.
If you are using the DTS emulator, the dashboard is available at `http://localhost:8082`.
dotnet/samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/demo.http
+26
View File
@@ -0,0 +1,26 @@
# Default endpoint address for local testing
@authority=http://localhost:7071
### Cancel an order
POST {{authority}}/api/workflows/CancelOrder/run
Content-Type: text/plain
12345
### Cancel an order with a custom run ID
POST {{authority}}/api/workflows/CancelOrder/run?runId=my-custom-id-123
Content-Type: text/plain
99999
### Get order status (shares OrderLookup executor with CancelOrder)
POST {{authority}}/api/workflows/OrderStatus/run
Content-Type: text/plain
12345
### Batch cancel orders with a complex JSON input
POST {{authority}}/api/workflows/BatchCancelOrders/run
Content-Type: application/json
{"orderIds": ["1001", "1002", "1003"], "reason": "Customer requested cancellation", "notifyCustomers": true}
dotnet/samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/host.json
+20
View File
@@ -0,0 +1,20 @@
{
"version": "2.0",
"logging": {
"logLevel": {
"Microsoft.Agents.AI.DurableTask": "Information",
"Microsoft.Agents.AI.Hosting.AzureFunctions": "Information",
"DurableTask": "Information",
"Microsoft.DurableTask": "Information"
}
},
"extensions": {
"durableTask": {
"hubName": "default",
"storageProvider": {
"type": "AzureManaged",
"connectionStringName": "DURABLE_TASK_SCHEDULER_CONNECTION_STRING"
}
}
}
}
dotnet/samples/Durable/Workflow/AzureFunctions/02_ConcurrentWorkflow/02_ConcurrentWorkflow.csproj
+42
View File
@@ -0,0 +1,42 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFrameworks>net10.0</TargetFrameworks>
<AzureFunctionsVersion>v4</AzureFunctionsVersion>
<OutputType>Exe</OutputType>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
<!-- The Functions build tools don't like namespaces that start with a number -->
<AssemblyName>SingleAgent</AssemblyName>
<RootNamespace>SingleAgent</RootNamespace>
</PropertyGroup>
<ItemGroup>
<FrameworkReference Include="Microsoft.AspNetCore.App" />
</ItemGroup>
<!-- Azure Functions packages -->
<ItemGroup>
<PackageReference Include="Microsoft.Azure.Functions.Worker" />
<PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask" />
<PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask.AzureManaged" />
<PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.Http.AspNetCore" />
<PackageReference Include="Microsoft.Azure.Functions.Worker.Sdk" />
</ItemGroup>
<ItemGroup>
<PackageReference Include="Azure.AI.OpenAI" />
<PackageReference Include="Azure.Identity" />
</ItemGroup>
<!-- Local projects that should be switched to package references when using the sample outside of this MAF repo -->
<!--
<ItemGroup>
<PackageReference Include="Microsoft.Agents.AI.Hosting.AzureFunctions" />
<PackageReference Include="Microsoft.Agents.AI.OpenAI" />
</ItemGroup>
-->
<ItemGroup>
<ProjectReference Include="..\..\..\..\..\src\Microsoft.Agents.AI.Hosting.AzureFunctions\Microsoft.Agents.AI.Hosting.AzureFunctions.csproj" />
<ProjectReference Include="..\..\..\..\..\src\Microsoft.Agents.AI.OpenAI\Microsoft.Agents.AI.OpenAI.csproj" />
</ItemGroup>
</Project>
dotnet/samples/Durable/Workflow/AzureFunctions/02_ConcurrentWorkflow/ExpertExecutors.cs
+73
View File
@@ -0,0 +1,73 @@
// Copyright (c) Microsoft. All rights reserved.
using Microsoft.Agents.AI.Workflows;
namespace WorkflowConcurrency;
/// <summary>
/// Parses and validates the incoming question before sending to AI agents.
/// </summary>
internal sealed class ParseQuestionExecutor() : Executor<string, string>("ParseQuestion")
{
public override ValueTask<string> HandleAsync(
string message,
IWorkflowContext context,
CancellationToken cancellationToken = default)
{
Console.WriteLine();
Console.ForegroundColor = ConsoleColor.Magenta;
Console.WriteLine("┌─────────────────────────────────────────────────────────────────┐");
Console.WriteLine("│ [ParseQuestion] Preparing question for AI agents...");
string formattedQuestion = message.Trim();
if (!formattedQuestion.EndsWith('?'))
{
formattedQuestion += "?";
}
Console.WriteLine($"│ [ParseQuestion] Question: \"{formattedQuestion}\"");
Console.WriteLine("│ [ParseQuestion] → Sending to Physicist and Chemist in PARALLEL...");
Console.WriteLine("└─────────────────────────────────────────────────────────────────┘");
Console.ResetColor();
return ValueTask.FromResult(formattedQuestion);
}
}
/// <summary>
/// Aggregates responses from all AI agents into a comprehensive answer.
/// This is the Fan-in point where parallel results are collected.
/// </summary>
internal sealed class AggregatorExecutor() : Executor<string[], string>("Aggregator")
{
public override ValueTask<string> HandleAsync(
string[] message,
IWorkflowContext context,
CancellationToken cancellationToken = default)
{
Console.WriteLine();
Console.ForegroundColor = ConsoleColor.Cyan;
Console.WriteLine("┌─────────────────────────────────────────────────────────────────┐");
Console.WriteLine($"│ [Aggregator] 📋 Received {message.Length} AI agent responses");
Console.WriteLine("│ [Aggregator] Combining into comprehensive answer...");
Console.WriteLine("│ [Aggregator] ✓ Aggregation complete!");
Console.WriteLine("└─────────────────────────────────────────────────────────────────┘");
Console.ResetColor();
string aggregatedResult = "═══════════════════════════════════════════════════════════════\n" +
" AI EXPERT PANEL RESPONSES\n" +
"═══════════════════════════════════════════════════════════════\n\n";
for (int i = 0; i < message.Length; i++)
{
string expertLabel = i == 0 ? "⚛️ PHYSICIST" : "🧪 CHEMIST";
aggregatedResult += $"{expertLabel}:\n{message[i]}\n\n";
}
aggregatedResult += "═══════════════════════════════════════════════════════════════\n" +
$"Summary: Received perspectives from {message.Length} AI experts.\n" +
"═══════════════════════════════════════════════════════════════";
return ValueTask.FromResult(aggregatedResult);
}
}
dotnet/samples/Durable/Workflow/AzureFunctions/02_ConcurrentWorkflow/Program.cs
+45
View File
@@ -0,0 +1,45 @@
// Copyright (c) Microsoft. All rights reserved.
using Azure;
using Azure.AI.OpenAI;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.DurableTask;
using Microsoft.Agents.AI.Hosting.AzureFunctions;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Azure.Functions.Worker.Builder;
using Microsoft.Extensions.Hosting;
using OpenAI.Chat;
using WorkflowConcurrency;
string endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT")
?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set.");
string deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT")
?? throw new InvalidOperationException("AZURE_OPENAI_DEPLOYMENT is not set.");
string? azureOpenAiKey = Environment.GetEnvironmentVariable("AZURE_OPENAI_KEY");
// Create Azure OpenAI client
AzureOpenAIClient openAiClient = !string.IsNullOrEmpty(azureOpenAiKey)
? new AzureOpenAIClient(new Uri(endpoint), new AzureKeyCredential(azureOpenAiKey))
: new AzureOpenAIClient(new Uri(endpoint), new AzureCliCredential());
ChatClient chatClient = openAiClient.GetChatClient(deploymentName);
// Define the 4 executors for the workflow
ParseQuestionExecutor parseQuestion = new();
AIAgent physicist = chatClient.AsAIAgent("You are a physics expert. Be concise (2-3 sentences).", "Physicist");
AIAgent chemist = chatClient.AsAIAgent("You are a chemistry expert. Be concise (2-3 sentences).", "Chemist");
AggregatorExecutor aggregator = new();
// Build workflow: ParseQuestion -> [Physicist, Chemist] (parallel) -> Aggregator
Workflow workflow = new WorkflowBuilder(parseQuestion)
.WithName("ExpertReview")
.AddFanOutEdge(parseQuestion, [physicist, chemist])
.AddFanInEdge([physicist, chemist], aggregator)
.Build();
using IHost app = FunctionsApplication
.CreateBuilder(args)
.ConfigureFunctionsWebApplication()
.ConfigureDurableWorkflows(workflows => workflows.AddWorkflows(workflow))
.Build();
app.Run();
dotnet/samples/Durable/Workflow/AzureFunctions/02_ConcurrentWorkflow/README.md
+90
View File
@@ -0,0 +1,90 @@
# Concurrent Workflow Sample
This sample demonstrates how to use the Microsoft Agent Framework to create an Azure Functions app that orchestrates concurrent execution of multiple AI agents using the fan-out/fan-in pattern within a durable workflow.
## Key Concepts Demonstrated
- Defining workflows with fan-out/fan-in edges for parallel execution using `WorkflowBuilder`
- Mixing custom executors with AI agents in a single workflow
- Concurrent execution of multiple AI agents (physics and chemistry experts)
- Response aggregation from parallel branches into a unified result
- Durable orchestration with automatic checkpointing and resumption from failures
- Viewing workflow execution history and status in the Durable Task Scheduler (DTS) dashboard
## Workflow
This sample defines a single workflow:
**ExpertReview**: `ParseQuestion` → [`Physicist`, `Chemist`] (parallel) → `Aggregator`
1. **ParseQuestion** — A custom executor that validates and formats the incoming question.
2. **Physicist** and **Chemist** — AI agents that run concurrently, each providing an expert perspective.
3. **Aggregator** — A custom executor that combines the parallel responses into a comprehensive answer.
## Environment Setup
See the [README.md](../../README.md) file in the parent directory for more information on how to configure the environment, including how to install and run common sample dependencies.
This sample requires Azure OpenAI. Set the following environment variables:
- `AZURE_OPENAI_ENDPOINT` — Your Azure OpenAI endpoint URL.
- `AZURE_OPENAI_DEPLOYMENT` — The name of your chat model deployment.
- `AZURE_OPENAI_KEY` (optional) — Your Azure OpenAI API key. If not set, Azure CLI credentials are used.
## Running the Sample
With the environment setup and function app running, you can test the sample by sending an HTTP request with a science question to the workflow endpoint.
You can use the `demo.http` file to trigger the workflow, or a command line tool like `curl` as shown below:
Bash (Linux/macOS/WSL):
```bash
curl -X POST http://localhost:7071/api/workflows/ExpertReview/run \
-H "Content-Type: text/plain" \
-d "What is temperature?"
```
PowerShell:
```powershell
Invoke-RestMethod -Method Post `
-Uri http://localhost:7071/api/workflows/ExpertReview/run `
-ContentType text/plain `
-Body "What is temperature?"
```
The response will confirm the workflow orchestration has started:
```text
Workflow orchestration started for ExpertReview. Orchestration runId: abc123def456
```
> **Tip:** You can provide a custom run ID by appending a `runId` query parameter:
>
> ```bash
> curl -X POST "http://localhost:7071/api/workflows/ExpertReview/run?runId=my-review-123" \
> -H "Content-Type: text/plain" \
> -d "What is temperature?"
> ```
>
> If not provided, a unique run ID is auto-generated.
In the function app logs, you will see the fan-out/fan-in execution pattern:
```text
│ [ParseQuestion] Preparing question for AI agents...
│ [ParseQuestion] Question: "What is temperature?"
│ [ParseQuestion] → Sending to Physicist and Chemist in PARALLEL...
│ [Aggregator] 📋 Received 2 AI agent responses
│ [Aggregator] Combining into comprehensive answer...
│ [Aggregator] ✓ Aggregation complete!
```
The Physicist and Chemist AI agents execute concurrently, and the Aggregator combines their responses into a formatted expert panel result.
### Viewing Workflows in the DTS Dashboard
After running a workflow, you can navigate to the Durable Task Scheduler (DTS) dashboard to visualize the completed orchestration, inspect inputs/outputs for each step, and view execution history.
If you are using the DTS emulator, the dashboard is available at `http://localhost:8082`.
dotnet/samples/Durable/Workflow/AzureFunctions/02_ConcurrentWorkflow/demo.http
+14
View File
@@ -0,0 +1,14 @@
# Default endpoint address for local testing
@authority=http://localhost:7071
### Prompt the agent
POST {{authority}}/api/workflows/ExpertReview/run
Content-Type: text/plain
What is temperature?
### Start with a custom run ID
POST {{authority}}/api/workflows/ExpertReview/run?runId=my-review-123
Content-Type: text/plain
What is gravity?
dotnet/samples/Durable/Workflow/AzureFunctions/02_ConcurrentWorkflow/host.json
+20
View File
@@ -0,0 +1,20 @@
{
"version": "2.0",
"logging": {
"logLevel": {
"Microsoft.Agents.AI.DurableTask": "Information",
"Microsoft.Agents.AI.Hosting.AzureFunctions": "Information",
"DurableTask": "Information",
"Microsoft.DurableTask": "Information"
}
},
"extensions": {
"durableTask": {
"hubName": "default",
"storageProvider": {
"type": "AzureManaged",
"connectionStringName": "DURABLE_TASK_SCHEDULER_CONNECTION_STRING"
}
}
}
}
dotnet/src/Microsoft.Agents.AI.DurableTask/ServiceCollectionExtensions.cs
+11 -8
View File
@@ -197,24 +197,27 @@ public static class ServiceCollectionExtensions
// Configure Durable Task Worker - capture sharedOptions reference in closure.
// The options object is populated by all Configure* calls before the worker starts.
services.AddDurableTaskWorker(builder =>
{
workerBuilder?.Invoke(builder);
builder.AddTasks(registry => RegisterTasksFromOptions(registry, sharedOptions));
});
if (workerBuilder is not null)
{
services.AddDurableTaskWorker(builder =>
{
workerBuilder?.Invoke(builder);
builder.AddTasks(registry => RegisterTasksFromOptions(registry, sharedOptions));
});
}
// Configure Durable Task Client
if (clientBuilder is not null)
{
services.AddDurableTaskClient(clientBuilder);
services.TryAddSingleton<IWorkflowClient, DurableWorkflowClient>();
services.TryAddSingleton<IDurableAgentClient, DefaultDurableAgentClient>();
}
// Register workflow and agent services
services.TryAddSingleton<DurableWorkflowClient>();
services.TryAddSingleton<IWorkflowClient>(sp => sp.GetRequiredService<DurableWorkflowClient>());
services.TryAddSingleton<DataConverter, DurableDataConverter>();
services.TryAddSingleton<IDurableAgentClient, DefaultDurableAgentClient>();
// Register agent factories resolver - returns factories from the shared options
services.TryAddSingleton(
dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowRunner.cs
+11 -3
View File
@@ -75,7 +75,7 @@ internal sealed class DurableWorkflowRunner
/// Initializes a new instance of the <see cref="DurableWorkflowRunner"/> class.
/// </summary>
/// <param name="durableOptions">The durable options containing workflow configurations.</param>
internal DurableWorkflowRunner(DurableOptions durableOptions)
public DurableWorkflowRunner(DurableOptions durableOptions)
{
ArgumentNullException.ThrowIfNull(durableOptions);
@@ -144,8 +144,16 @@ internal sealed class DurableWorkflowRunner
{
SuperstepState state = new(workflow, edgeMap);
// Convert input to string for the message queue - serialize if not already a string
string inputString = initialInput is string s ? s : JsonSerializer.Serialize(initialInput);
// Convert input to string for the message queue.
// When DurableWorkflowInput<string> is deserialized as DurableWorkflowInput<object>,
// the Input property becomes a JsonElement instead of a string.
// We must extract the raw string value to avoid double-serialization.
string inputString = initialInput switch
{
string s => s,
JsonElement je when je.ValueKind == JsonValueKind.String => je.GetString() ?? string.Empty,
_ => JsonSerializer.Serialize(initialInput)
};
edgeMap.EnqueueInitialInput(inputString, state.MessageQueues);
dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctionExecutor.cs
+70 -5
View File
@@ -21,6 +21,15 @@ internal sealed class BuiltInFunctionExecutor : IFunctionExecutor
{
ArgumentNullException.ThrowIfNull(context);
// Orchestration triggers use a different input binding mechanism than other triggers.
// The encoded orchestrator state is retrieved via BindInputAsync on the orchestration trigger binding,
// not through IFunctionInputBindingFeature. Handle this case first to avoid unnecessary binding work.
if (context.FunctionDefinition.EntryPoint == BuiltInFunctions.RunWorkflowOrchestrationFunctionEntryPoint)
{
await ExecuteOrchestrationAsync(context);
return;
}
// Acquire the input binding feature (fail fast if missing rather than null-forgiving operator).
IFunctionInputBindingFeature? functionInputBindingFeature = context.Features.Get<IFunctionInputBindingFeature>() ??
throw new InvalidOperationException("Function input binding feature is not available on the current context.");
@@ -57,11 +66,39 @@ internal sealed class BuiltInFunctionExecutor : IFunctionExecutor
if (durableTaskClient is null)
{
// This is not expected to happen since all built-in functions are
// expected to have a Durable Task client binding.
// This is not expected to happen since all built-in functions (other than orchestration triggers)
// are expected to have a Durable Task client binding.
throw new InvalidOperationException($"Durable Task client binding is missing for the invocation {context.InvocationId}.");
}
if (context.FunctionDefinition.EntryPoint == BuiltInFunctions.RunWorkflowOrchestrationHttpFunctionEntryPoint)
{
if (httpRequestData == null)
{
throw new InvalidOperationException($"HTTP request data binding is missing for the invocation {context.InvocationId}.");
}
context.GetInvocationResult().Value = await BuiltInFunctions.RunWorkflowOrchestrationHttpTriggerAsync(
httpRequestData,
durableTaskClient,
context);
return;
}
if (context.FunctionDefinition.EntryPoint == BuiltInFunctions.InvokeWorkflowActivityFunctionEntryPoint)
{
if (encodedEntityRequest is null)
{
throw new InvalidOperationException($"Activity trigger input binding is missing for the invocation {context.InvocationId}.");
}
context.GetInvocationResult().Value = await BuiltInFunctions.InvokeWorkflowActivityAsync(
encodedEntityRequest,
durableTaskClient,
context);
return;
}
if (context.FunctionDefinition.EntryPoint == BuiltInFunctions.RunAgentHttpFunctionEntryPoint)
{
if (httpRequestData == null)
@@ -70,9 +107,9 @@ internal sealed class BuiltInFunctionExecutor : IFunctionExecutor
}
context.GetInvocationResult().Value = await BuiltInFunctions.RunAgentHttpAsync(
httpRequestData,
durableTaskClient,
context);
httpRequestData,
durableTaskClient,
context);
return;
}
@@ -104,4 +141,32 @@ internal sealed class BuiltInFunctionExecutor : IFunctionExecutor
throw new InvalidOperationException($"Unsupported function entry point '{context.FunctionDefinition.EntryPoint}' for invocation {context.InvocationId}.");
}
private static async ValueTask ExecuteOrchestrationAsync(FunctionContext context)
{
BindingMetadata? orchestrationBinding = null;
foreach (BindingMetadata binding in context.FunctionDefinition.InputBindings.Values)
{
if (string.Equals(binding.Type, "orchestrationTrigger", StringComparison.OrdinalIgnoreCase))
{
orchestrationBinding = binding;
break;
}
}
if (orchestrationBinding is null)
{
throw new InvalidOperationException($"Orchestration trigger binding is missing for the invocation {context.InvocationId}.");
}
InputBindingData<object> triggerInputData = await context.BindInputAsync<object>(orchestrationBinding);
if (triggerInputData?.Value is not string encodedOrchestratorState)
{
throw new InvalidOperationException($"Orchestration history state was either missing from the input or not a string value for invocation {context.InvocationId}.");
}
context.GetInvocationResult().Value = BuiltInFunctions.RunWorkflowOrchestration(
encodedOrchestratorState,
context);
}
}
dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctions.cs
+83
View File
@@ -3,9 +3,11 @@
using System.Net;
using System.Text.Json.Serialization;
using Microsoft.Agents.AI.DurableTask;
using Microsoft.Agents.AI.DurableTask.Workflows;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Azure.Functions.Worker.Extensions.Mcp;
using Microsoft.Azure.Functions.Worker.Http;
using Microsoft.DurableTask;
using Microsoft.DurableTask.Client;
using Microsoft.DurableTask.Worker.Grpc;
using Microsoft.Extensions.AI;
@@ -21,6 +23,87 @@ internal static class BuiltInFunctions
internal static readonly string RunAgentHttpFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(RunAgentHttpAsync)}";
internal static readonly string RunAgentEntityFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(InvokeAgentAsync)}";
internal static readonly string RunAgentMcpToolFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(RunMcpToolAsync)}";
internal static readonly string RunWorkflowOrchestrationHttpFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(RunWorkflowOrchestrationHttpTriggerAsync)}";
internal static readonly string RunWorkflowOrchestrationFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(RunWorkflowOrchestration)}";
internal static readonly string InvokeWorkflowActivityFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(InvokeWorkflowActivityAsync)}";
#pragma warning disable IL3000 // Avoid accessing Assembly file path when publishing as a single file - Azure Functions does not use single-file publishing
internal static readonly string ScriptFile = Path.GetFileName(typeof(BuiltInFunctions).Assembly.Location);
#pragma warning restore IL3000
/// <summary>
/// Starts a workflow orchestration in response to an HTTP request.
/// The workflow name is derived from the function name by stripping the <see cref="HttpPrefix"/>.
/// Callers can optionally provide a custom run ID via the <c>runId</c> query string parameter
/// (e.g., <c>/api/workflows/MyWorkflow/run?runId=my-id</c>). If not provided, one is auto-generated.
/// </summary>
public static async Task<HttpResponseData> RunWorkflowOrchestrationHttpTriggerAsync(
[HttpTrigger] HttpRequestData req,
[DurableClient] DurableTaskClient client,
FunctionContext context)
{
string workflowName = context.FunctionDefinition.Name.Replace(HttpPrefix, string.Empty);
string orchestrationFunctionName = WorkflowNamingHelper.ToOrchestrationFunctionName(workflowName);
string? inputMessage = await req.ReadAsStringAsync();
if (string.IsNullOrEmpty(inputMessage))
{
HttpResponseData errorResponse = req.CreateResponse(HttpStatusCode.BadRequest);
await errorResponse.WriteStringAsync("Workflow input cannot be empty.");
return errorResponse;
}
DurableWorkflowInput<string> orchestrationInput = new() { Input = inputMessage };
// Allow users to provide a custom run ID via query string; otherwise, auto-generate one.
string? instanceId = req.Query["runId"];
StartOrchestrationOptions? options = instanceId is not null ? new StartOrchestrationOptions(instanceId) : null;
string resolvedInstanceId = await client.ScheduleNewOrchestrationInstanceAsync(orchestrationFunctionName, orchestrationInput, options);
HttpResponseData response = req.CreateResponse(HttpStatusCode.Accepted);
await response.WriteStringAsync($"Workflow orchestration started for {workflowName}. Orchestration runId: {resolvedInstanceId}");
return response;
}
/// <summary>
/// Executes a workflow activity by looking up the registered executor and delegating to it.
/// The executor name is derived from the activity function name via <see cref="WorkflowNamingHelper"/>.
/// </summary>
public static Task<string> InvokeWorkflowActivityAsync(
[ActivityTrigger] string input,
[DurableClient] DurableTaskClient durableTaskClient,
FunctionContext functionContext)
{
ArgumentNullException.ThrowIfNull(input);
ArgumentNullException.ThrowIfNull(durableTaskClient);
ArgumentNullException.ThrowIfNull(functionContext);
string activityFunctionName = functionContext.FunctionDefinition.Name;
string executorName = WorkflowNamingHelper.ToWorkflowName(activityFunctionName);
DurableOptions durableOptions = functionContext.InstanceServices.GetRequiredService<DurableOptions>();
if (!durableOptions.Workflows.Executors.TryGetExecutor(executorName, out ExecutorRegistration? registration))
{
throw new InvalidOperationException($"Executor '{executorName}' not found in workflow options.");
}
return DurableActivityExecutor.ExecuteAsync(registration.Binding, input, functionContext.CancellationToken);
}
/// <summary>
/// Runs a workflow orchestration by delegating to <see cref="WorkflowOrchestrator"/>
/// via <see cref="GrpcOrchestrationRunner"/>.
/// </summary>
public static string RunWorkflowOrchestration(
string encodedOrchestratorRequest,
FunctionContext functionContext)
{
ArgumentNullException.ThrowIfNull(encodedOrchestratorRequest);
ArgumentNullException.ThrowIfNull(functionContext);
WorkflowOrchestrator orchestrator = new(functionContext.InstanceServices);
return GrpcOrchestrationRunner.LoadAndRun(encodedOrchestratorRequest, orchestrator, functionContext.InstanceServices);
}
// Exposed as an entity trigger via AgentFunctionsProvider
public static Task<string> InvokeAgentAsync(
dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableAgentFunctionMetadataTransformer.cs
+3 -41
View File
@@ -1,6 +1,5 @@
// Copyright (c) Microsoft. All rights reserved.
using Microsoft.Agents.AI.DurableTask;
using Microsoft.Azure.Functions.Worker.Core.FunctionMetadata;
using Microsoft.Extensions.Logging;
@@ -17,10 +16,6 @@ internal sealed class DurableAgentFunctionMetadataTransformer : IFunctionMetadat
private readonly IServiceProvider _serviceProvider;
private readonly IFunctionsAgentOptionsProvider _functionsAgentOptionsProvider;
#pragma warning disable IL3000 // Avoid accessing Assembly file path when publishing as a single file - Azure Functions does not use single-file publishing
private static readonly string s_builtInFunctionsScriptFile = Path.GetFileName(typeof(BuiltInFunctions).Assembly.Location);
#pragma warning restore IL3000
public DurableAgentFunctionMetadataTransformer(
IReadOnlyDictionary<string, Func<IServiceProvider, AIAgent>> agents,
ILogger<DurableAgentFunctionMetadataTransformer> logger,
@@ -45,14 +40,14 @@ internal sealed class DurableAgentFunctionMetadataTransformer : IFunctionMetadat
this._logger.LogRegisteringTriggerForAgent(agentName, "entity");
original.Add(CreateAgentTrigger(agentName));
original.Add(FunctionMetadataFactory.CreateEntityTrigger(agentName));
if (this._functionsAgentOptionsProvider.TryGet(agentName, out FunctionsAgentOptions? agentTriggerOptions))
{
if (agentTriggerOptions.HttpTrigger.IsEnabled)
{
this._logger.LogRegisteringTriggerForAgent(agentName, "http");
original.Add(CreateHttpTrigger(agentName, $"agents/{agentName}/run"));
original.Add(FunctionMetadataFactory.CreateHttpTrigger(agentName, $"agents/{agentName}/run", BuiltInFunctions.RunAgentHttpFunctionEntryPoint));
}
if (agentTriggerOptions.McpToolTrigger.IsEnabled)
@@ -65,39 +60,6 @@ internal sealed class DurableAgentFunctionMetadataTransformer : IFunctionMetadat
}
}
private static DefaultFunctionMetadata CreateAgentTrigger(string name)
{
return new DefaultFunctionMetadata()
{
Name = AgentSessionId.ToEntityName(name),
Language = "dotnet-isolated",
RawBindings =
[
"""{"name":"encodedEntityRequest","type":"entityTrigger","direction":"In"}""",
"""{"name":"client","type":"durableClient","direction":"In"}"""
],
EntryPoint = BuiltInFunctions.RunAgentEntityFunctionEntryPoint,
ScriptFile = s_builtInFunctionsScriptFile,
};
}
private static DefaultFunctionMetadata CreateHttpTrigger(string name, string route)
{
return new DefaultFunctionMetadata()
{
Name = $"{BuiltInFunctions.HttpPrefix}{name}",
Language = "dotnet-isolated",
RawBindings =
[
$"{{\"name\":\"req\",\"type\":\"httpTrigger\",\"direction\":\"In\",\"authLevel\":\"function\",\"methods\": [\"post\"],\"route\":\"{route}\"}}",
"{\"name\":\"$return\",\"type\":\"http\",\"direction\":\"Out\"}",
"{\"name\":\"client\",\"type\":\"durableClient\",\"direction\":\"In\"}"
],
EntryPoint = BuiltInFunctions.RunAgentHttpFunctionEntryPoint,
ScriptFile = s_builtInFunctionsScriptFile,
};
}
private static DefaultFunctionMetadata CreateMcpToolTrigger(string agentName, string? description)
{
return new DefaultFunctionMetadata
@@ -112,7 +74,7 @@ internal sealed class DurableAgentFunctionMetadataTransformer : IFunctionMetadat
"""{"name":"client","type":"durableClient","direction":"In"}"""
],
EntryPoint = BuiltInFunctions.RunAgentMcpToolFunctionEntryPoint,
ScriptFile = s_builtInFunctionsScriptFile,
ScriptFile = BuiltInFunctions.ScriptFile,
};
}
}
dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/FunctionMetadataFactory.cs
+100
View File
@@ -0,0 +1,100 @@
// Copyright (c) Microsoft. All rights reserved.
using Microsoft.Agents.AI.DurableTask;
using Microsoft.Azure.Functions.Worker.Core.FunctionMetadata;
namespace Microsoft.Agents.AI.Hosting.AzureFunctions;
/// <summary>
/// Provides factory methods for creating common <see cref="DefaultFunctionMetadata"/> instances
/// used by function metadata transformers.
/// </summary>
internal static class FunctionMetadataFactory
{
/// <summary>
/// Creates function metadata for an entity trigger function.
/// </summary>
/// <param name="name">The base name used to derive the entity function name.</param>
/// <returns>A <see cref="DefaultFunctionMetadata"/> configured for an entity trigger.</returns>
internal static DefaultFunctionMetadata CreateEntityTrigger(string name)
{
return new DefaultFunctionMetadata()
{
Name = AgentSessionId.ToEntityName(name),
Language = "dotnet-isolated",
RawBindings =
[
"""{"name":"encodedEntityRequest","type":"entityTrigger","direction":"In"}""",
"""{"name":"client","type":"durableClient","direction":"In"}"""
],
EntryPoint = BuiltInFunctions.RunAgentEntityFunctionEntryPoint,
ScriptFile = BuiltInFunctions.ScriptFile,
};
}
/// <summary>
/// Creates function metadata for an HTTP trigger function.
/// </summary>
/// <param name="name">The base name used to derive the HTTP function name.</param>
/// <param name="route">The HTTP route for the trigger.</param>
/// <param name="entryPoint">The entry point method for the HTTP trigger.</param>
/// <returns>A <see cref="DefaultFunctionMetadata"/> configured for an HTTP trigger.</returns>
internal static DefaultFunctionMetadata CreateHttpTrigger(string name, string route, string entryPoint)
{
return new DefaultFunctionMetadata()
{
Name = $"{BuiltInFunctions.HttpPrefix}{name}",
Language = "dotnet-isolated",
RawBindings =
[
$"{{\"name\":\"req\",\"type\":\"httpTrigger\",\"direction\":\"In\",\"authLevel\":\"function\",\"methods\": [\"post\"],\"route\":\"{route}\"}}",
"{\"name\":\"$return\",\"type\":\"http\",\"direction\":\"Out\"}",
"{\"name\":\"client\",\"type\":\"durableClient\",\"direction\":\"In\"}"
],
EntryPoint = entryPoint,
ScriptFile = BuiltInFunctions.ScriptFile,
};
}
/// <summary>
/// Creates function metadata for an activity trigger function.
/// </summary>
/// <param name="functionName">The name of the activity function.</param>
/// <returns>A <see cref="DefaultFunctionMetadata"/> configured for an activity trigger.</returns>
internal static DefaultFunctionMetadata CreateActivityTrigger(string functionName)
{
return new DefaultFunctionMetadata()
{
Name = functionName,
Language = "dotnet-isolated",
RawBindings =
[
"""{"name":"input","type":"activityTrigger","direction":"In","dataType":"String"}""",
"""{"name":"durableTaskClient","type":"durableClient","direction":"In"}"""
],
EntryPoint = BuiltInFunctions.InvokeWorkflowActivityFunctionEntryPoint,
ScriptFile = BuiltInFunctions.ScriptFile,
};
}
/// <summary>
/// Creates function metadata for an orchestration trigger function.
/// </summary>
/// <param name="functionName">The name of the orchestration function.</param>
/// <param name="entryPoint">The entry point method for the orchestration trigger.</param>
/// <returns>A <see cref="DefaultFunctionMetadata"/> configured for an orchestration trigger.</returns>
internal static DefaultFunctionMetadata CreateOrchestrationTrigger(string functionName, string entryPoint)
{
return new DefaultFunctionMetadata()
{
Name = functionName,
Language = "dotnet-isolated",
RawBindings =
[
"""{"name":"context","type":"orchestrationTrigger","direction":"In"}"""
],
EntryPoint = entryPoint,
ScriptFile = BuiltInFunctions.ScriptFile,
};
}
}
dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/FunctionsApplicationBuilderExtensions.cs
+84
View File
@@ -1,6 +1,7 @@
// Copyright (c) Microsoft. All rights reserved.
using Microsoft.Agents.AI.DurableTask;
using Microsoft.Agents.AI.DurableTask.Workflows;
using Microsoft.Azure.Functions.Worker.Builder;
using Microsoft.Azure.Functions.Worker.Core.FunctionMetadata;
using Microsoft.Extensions.DependencyInjection;
@@ -43,4 +44,87 @@ public static class FunctionsApplicationBuilderExtensions
return builder;
}
/// <summary>
/// Configures durable options for the functions application, allowing customization of Durable Task framework
/// settings.
/// </summary>
/// <remarks>This method ensures that a single shared <see cref="DurableOptions"/> instance is used across all
/// configuration calls. If any workflows have been added, it configures the necessary orchestrations and registers
/// required middleware.</remarks>
/// <param name="builder">The functions application builder to configure. Cannot be null.</param>
/// <param name="configure">An action that configures the <see cref="DurableOptions"/> instance. Cannot be null.</param>
/// <returns>The updated <see cref="FunctionsApplicationBuilder"/> instance, enabling method chaining.</returns>
public static FunctionsApplicationBuilder ConfigureDurableOptions(
this FunctionsApplicationBuilder builder,
Action<DurableOptions> configure)
{
ArgumentNullException.ThrowIfNull(builder);
ArgumentNullException.ThrowIfNull(configure);
builder.Services.ConfigureDurableOptions(configure);
// Read the shared options to check if workflows were added
DurableOptions sharedOptions = GetOrCreateSharedOptions(builder.Services);
if (sharedOptions.Workflows.Workflows.Count > 0)
{
builder.Services.TryAddEnumerable(ServiceDescriptor.Singleton<IFunctionMetadataTransformer, DurableWorkflowsFunctionMetadataTransformer>());
}
EnsureMiddlewareRegistered(builder);
return builder;
}
/// <summary>
/// Configures durable workflow support for the specified Azure Functions application builder.
/// </summary>
/// <param name="builder">The <see cref="FunctionsApplicationBuilder"/> instance to configure for durable workflows.</param>
/// <param name="configure">An action that configures the <see cref="DurableWorkflowOptions"/>, allowing customization of durable workflow behavior.</param>
/// <returns>The updated <see cref="FunctionsApplicationBuilder"/> instance, enabling method chaining.</returns>
public static FunctionsApplicationBuilder ConfigureDurableWorkflows(
this FunctionsApplicationBuilder builder,
Action<DurableWorkflowOptions> configure)
{
ArgumentNullException.ThrowIfNull(configure);
return builder.ConfigureDurableOptions(options => configure(options.Workflows));
}
private static void EnsureMiddlewareRegistered(FunctionsApplicationBuilder builder)
{
// Guard against registering the middleware filter multiple times in the pipeline.
if (builder.Services.Any(d => d.ServiceType == typeof(BuiltInFunctionExecutor)))
{
return;
}
builder.UseWhen<BuiltInFunctionExecutionMiddleware>(static context =>
string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunAgentHttpFunctionEntryPoint, StringComparison.Ordinal) ||
string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunAgentEntityFunctionEntryPoint, StringComparison.Ordinal) ||
string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunWorkflowOrchestrationHttpFunctionEntryPoint, StringComparison.Ordinal) ||
string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunWorkflowOrchestrationFunctionEntryPoint, StringComparison.Ordinal) ||
string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.InvokeWorkflowActivityFunctionEntryPoint, StringComparison.Ordinal)
);
builder.Services.TryAddSingleton<BuiltInFunctionExecutor>();
}
/// <summary>
/// Gets or creates a shared <see cref="DurableOptions"/> instance from the service collection.
/// </summary>
private static DurableOptions GetOrCreateSharedOptions(IServiceCollection services)
{
ServiceDescriptor? existingDescriptor = services.FirstOrDefault(
d => d.ServiceType == typeof(DurableOptions) && d.ImplementationInstance is not null);
if (existingDescriptor?.ImplementationInstance is DurableOptions existing)
{
return existing;
}
DurableOptions options = new();
services.AddSingleton(options);
return options;
}
}
dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/Logs.cs
+12
View File
@@ -17,4 +17,16 @@ internal static partial class Logs
Level = LogLevel.Information,
Message = "Registering {TriggerType} function for agent '{AgentName}'")]
public static partial void LogRegisteringTriggerForAgent(this ILogger logger, string agentName, string triggerType);
[LoggerMessage(
EventId = 102,
Level = LogLevel.Information,
Message = "Registering {TriggerType} trigger function '{FunctionName}' for workflow '{WorkflowKey}'")]
public static partial void LogRegisteringWorkflowTrigger(this ILogger logger, string workflowKey, string functionName, string triggerType);
[LoggerMessage(
EventId = 103,
Level = LogLevel.Information,
Message = "Function metadata transformation complete. Added {AddedCount} workflow function(s). Total function count: {TotalCount}")]
public static partial void LogTransformationComplete(this ILogger logger, int addedCount, int totalCount);
}
dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/Microsoft.Agents.AI.Hosting.AzureFunctions.csproj
+2 -1
View File
@@ -4,7 +4,8 @@
<TargetFrameworks>$(TargetFrameworksCore)</TargetFrameworks>
<ImplicitUsings>enable</ImplicitUsings>
<!-- CA2007: This rule should generally be suppressed in Durable Task libraries. Also, this is not library code. -->
<NoWarn>$(NoWarn);CA2007</NoWarn>
<!-- AD0001: Temporary workaround for Microsoft.DurableTask.Analyzers v0.2.0 bug (ArgumentNullException on 'node'). Remove when upgrading to a fixed analyzer version. -->
<NoWarn>$(NoWarn);CA2007;AD0001</NoWarn>
</PropertyGroup>
<Import Project="$(RepoRoot)/dotnet/nuget/nuget-package.props" />
dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/Workflows/DurableWorkflowsFunctionMetadataTransformer.cs
+119
View File
@@ -0,0 +1,119 @@
// Copyright (c) Microsoft. All rights reserved.
using Microsoft.Agents.AI.DurableTask;
using Microsoft.Agents.AI.DurableTask.Workflows;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Azure.Functions.Worker.Core.FunctionMetadata;
using Microsoft.Extensions.Logging;
namespace Microsoft.Agents.AI.Hosting.AzureFunctions;
/// <summary>
/// Transforms function metadata by dynamically registering Azure Functions triggers
/// for each configured durable workflow and its executors.
/// </summary>
/// <remarks>
/// For each workflow, this transformer registers:
/// <list type="bullet">
/// <item><description>An HTTP trigger function to start the workflow orchestration via HTTP.</description></item>
/// <item><description>An orchestration trigger function to run the workflow orchestration.</description></item>
/// <item><description>An activity trigger function for each non-agent executor in the workflow.</description></item>
/// <item><description>An entity trigger function for each AI agent executor in the workflow.</description></item>
/// </list>
/// When multiple workflows share the same executor, the corresponding function is registered only once.
/// </remarks>
internal sealed class DurableWorkflowsFunctionMetadataTransformer : IFunctionMetadataTransformer
{
private readonly ILogger<DurableWorkflowsFunctionMetadataTransformer> _logger;
private readonly DurableWorkflowOptions _options;
/// <summary>
/// Initializes a new instance of the <see cref="DurableWorkflowsFunctionMetadataTransformer"/> class.
/// </summary>
/// <param name="logger">The logger instance for diagnostic output.</param>
/// <param name="durableOptions">The durable options containing workflow configurations.</param>
public DurableWorkflowsFunctionMetadataTransformer(ILogger<DurableWorkflowsFunctionMetadataTransformer> logger, DurableOptions durableOptions)
{
this._logger = logger ?? throw new ArgumentNullException(nameof(logger));
ArgumentNullException.ThrowIfNull(durableOptions);
this._options = durableOptions.Workflows;
}
/// <inheritdoc />
public string Name => nameof(DurableWorkflowsFunctionMetadataTransformer);
/// <inheritdoc />
public void Transform(IList<IFunctionMetadata> original)
{
int initialCount = original.Count;
this._logger.LogTransformingFunctionMetadata(initialCount);
// Track registered function names to avoid duplicates when workflows share executors.
HashSet<string> registeredFunctions = [];
foreach (var workflow in this._options.Workflows)
{
string httpFunctionName = $"{BuiltInFunctions.HttpPrefix}{workflow.Key}";
if (this._logger.IsEnabled(LogLevel.Information))
{
this._logger.LogInformation("Registering durable workflow functions for workflow '{WorkflowKey}' with HTTP trigger function name '{HttpFunctionName}'", workflow.Key, httpFunctionName);
}
// Register an orchestration function for the workflow.
string orchestrationFunctionName = WorkflowNamingHelper.ToOrchestrationFunctionName(workflow.Key);
if (registeredFunctions.Add(orchestrationFunctionName))
{
this._logger.LogRegisteringWorkflowTrigger(workflow.Key, orchestrationFunctionName, "orchestration");
original.Add(FunctionMetadataFactory.CreateOrchestrationTrigger(
orchestrationFunctionName,
BuiltInFunctions.RunWorkflowOrchestrationFunctionEntryPoint));
}
// Register an HTTP trigger so users can start this workflow via HTTP.
if (registeredFunctions.Add(httpFunctionName))
{
this._logger.LogRegisteringWorkflowTrigger(workflow.Key, httpFunctionName, "http");
original.Add(FunctionMetadataFactory.CreateHttpTrigger(
workflow.Key,
$"workflows/{workflow.Key}/run",
BuiltInFunctions.RunWorkflowOrchestrationHttpFunctionEntryPoint));
}
// Register activity or entity functions for each executor in the workflow.
// ReflectExecutors() returns all executors across the graph; no need to manually traverse edges.
foreach (KeyValuePair<string, ExecutorBinding> entry in workflow.Value.ReflectExecutors())
{
// Sub-workflow bindings are handled as separate orchestrations, not activities.
if (entry.Value is SubworkflowBinding)
{
continue;
}
string executorName = WorkflowNamingHelper.GetExecutorName(entry.Key);
// AI agent executors are backed by durable entities; other executors use activity triggers.
if (entry.Value is AIAgentBinding)
{
string entityName = AgentSessionId.ToEntityName(executorName);
if (registeredFunctions.Add(entityName))
{
this._logger.LogRegisteringWorkflowTrigger(workflow.Key, entityName, "entity");
original.Add(FunctionMetadataFactory.CreateEntityTrigger(executorName));
}
}
else
{
string functionName = WorkflowNamingHelper.ToOrchestrationFunctionName(executorName);
if (registeredFunctions.Add(functionName))
{
this._logger.LogRegisteringWorkflowTrigger(workflow.Key, functionName, "activity");
original.Add(FunctionMetadataFactory.CreateActivityTrigger(functionName));
}
}
}
}
this._logger.LogTransformationComplete(original.Count - initialCount, original.Count);
}
}
dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/Workflows/WorkflowOrchestrator.cs
+51
View File
@@ -0,0 +1,51 @@
// Copyright (c) Microsoft. All rights reserved.
using Microsoft.Agents.AI.DurableTask.Workflows;
using Microsoft.DurableTask;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
namespace Microsoft.Agents.AI.Hosting.AzureFunctions;
/// <summary>
/// A custom <see cref="ITaskOrchestrator"/> implementation that delegates workflow orchestration
/// execution to the <see cref="DurableWorkflowRunner"/>.
/// </summary>
internal sealed class WorkflowOrchestrator : ITaskOrchestrator
{
private readonly IServiceProvider _serviceProvider;
/// <summary>
/// Initializes a new instance of the <see cref="WorkflowOrchestrator"/> class.
/// </summary>
/// <param name="serviceProvider">The service provider used to resolve workflow dependencies.</param>
public WorkflowOrchestrator(IServiceProvider serviceProvider)
{
this._serviceProvider = serviceProvider ?? throw new ArgumentNullException(nameof(serviceProvider));
}
/// <inheritdoc />
public Type InputType => typeof(DurableWorkflowInput<object>);
/// <inheritdoc />
public Type OutputType => typeof(string);
/// <inheritdoc />
public async Task<object?> RunAsync(TaskOrchestrationContext context, object? input)
{
ArgumentNullException.ThrowIfNull(context);
DurableWorkflowRunner runner = this._serviceProvider.GetRequiredService<DurableWorkflowRunner>();
ILogger logger = context.CreateReplaySafeLogger(context.Name);
DurableWorkflowInput<object> workflowInput = input switch
{
DurableWorkflowInput<object> existing => existing,
_ => new DurableWorkflowInput<object> { Input = input! }
};
// ConfigureAwait(true) is required to preserve the orchestration context
// across awaits, which the Durable Task framework uses for replay.
return await runner.RunWorkflowOrchestrationAsync(context, workflowInput, logger).ConfigureAwait(true);
}
}
dotnet/tests/Microsoft.Agents.AI.Hosting.AzureFunctions.IntegrationTests/WorkflowSamplesValidation.cs
+475
View File
@@ -0,0 +1,475 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Diagnostics;
using System.Reflection;
using System.Text;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging;
using Xunit.Abstractions;
namespace Microsoft.Agents.AI.Hosting.AzureFunctions.IntegrationTests;
/// <summary>
/// Integration tests for validating the durable workflow Azure Functions samples
/// located in samples/Durable/Workflow/AzureFunctions.
/// </summary>
[Collection("Samples")]
[Trait("Category", "SampleValidation")]
public sealed class WorkflowSamplesValidation(ITestOutputHelper outputHelper) : IAsyncLifetime
{
private const string AzureFunctionsPort = "7071";
private const string AzuritePort = "10000";
private const string DtsPort = "8080";
private static readonly string s_dotnetTargetFramework = GetTargetFramework();
private static readonly HttpClient s_sharedHttpClient = new();
private static readonly IConfiguration s_configuration =
new ConfigurationBuilder()
.AddUserSecrets(Assembly.GetExecutingAssembly())
.AddEnvironmentVariables()
.Build();
private static bool s_infrastructureStarted;
private static readonly TimeSpan s_orchestrationTimeout = TimeSpan.FromMinutes(1);
private static readonly string s_samplesPath = Path.GetFullPath(
Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "..", "..", "..", "..", "..", "samples", "Durable", "Workflow", "AzureFunctions"));
private readonly ITestOutputHelper _outputHelper = outputHelper;
async Task IAsyncLifetime.InitializeAsync()
{
if (!s_infrastructureStarted)
{
await this.StartSharedInfrastructureAsync();
s_infrastructureStarted = true;
}
}
async Task IAsyncLifetime.DisposeAsync()
{
await Task.CompletedTask;
}
[Fact]
public async Task SequentialWorkflowSampleValidationAsync()
{
string samplePath = Path.Combine(s_samplesPath, "01_SequentialWorkflow");
await this.RunSampleTestAsync(samplePath, requiresOpenAI: false, async (logs) =>
{
// Test the CancelOrder workflow
Uri cancelOrderUri = new($"http://localhost:{AzureFunctionsPort}/api/workflows/CancelOrder/run");
this._outputHelper.WriteLine($"Starting CancelOrder workflow via POST request to {cancelOrderUri}...");
using HttpContent cancelContent = new StringContent("12345", Encoding.UTF8, "text/plain");
using HttpResponseMessage cancelResponse = await s_sharedHttpClient.PostAsync(cancelOrderUri, cancelContent);
Assert.True(cancelResponse.IsSuccessStatusCode, $"CancelOrder request failed with status: {cancelResponse.StatusCode}");
string cancelResponseText = await cancelResponse.Content.ReadAsStringAsync();
Assert.Contains("CancelOrder", cancelResponseText);
this._outputHelper.WriteLine($"CancelOrder response: {cancelResponseText}");
// Wait for the CancelOrder workflow to complete by checking logs
await this.WaitForConditionAsync(
condition: () =>
{
lock (logs)
{
bool exists = logs.Any(log => log.Message.Contains("Workflow completed"));
return Task.FromResult(exists);
}
},
message: "CancelOrder workflow completed",
timeout: s_orchestrationTimeout);
// Verify the executor activities ran in sequence
lock (logs)
{
Assert.True(logs.Any(log => log.Message.Contains("[Activity] OrderLookup:")), "OrderLookup activity not found in logs.");
Assert.True(logs.Any(log => log.Message.Contains("[Activity] OrderCancel:")), "OrderCancel activity not found in logs.");
Assert.True(logs.Any(log => log.Message.Contains("[Activity] SendEmail:")), "SendEmail activity not found in logs.");
}
// Test the OrderStatus workflow (shares OrderLookup executor with CancelOrder)
Uri orderStatusUri = new($"http://localhost:{AzureFunctionsPort}/api/workflows/OrderStatus/run");
this._outputHelper.WriteLine($"Starting OrderStatus workflow via POST request to {orderStatusUri}...");
using HttpContent statusContent = new StringContent("67890", Encoding.UTF8, "text/plain");
using HttpResponseMessage statusResponse = await s_sharedHttpClient.PostAsync(orderStatusUri, statusContent);
Assert.True(statusResponse.IsSuccessStatusCode, $"OrderStatus request failed with status: {statusResponse.StatusCode}");
string statusResponseText = await statusResponse.Content.ReadAsStringAsync();
Assert.Contains("OrderStatus", statusResponseText);
this._outputHelper.WriteLine($"OrderStatus response: {statusResponseText}");
// Wait for the OrderStatus workflow to complete
await this.WaitForConditionAsync(
condition: () =>
{
lock (logs)
{
// Look for StatusReport activity which is unique to OrderStatus workflow
bool exists = logs.Any(log => log.Message.Contains("[Activity] StatusReport:"));
return Task.FromResult(exists);
}
},
message: "OrderStatus workflow completed",
timeout: s_orchestrationTimeout);
});
}
[Fact]
public async Task ConcurrentWorkflowSampleValidationAsync()
{
string samplePath = Path.Combine(s_samplesPath, "02_ConcurrentWorkflow");
await this.RunSampleTestAsync(samplePath, requiresOpenAI: true, async (logs) =>
{
// Start the ExpertReview workflow with a science question
const string RequestBody = "What is temperature?";
using HttpContent content = new StringContent(RequestBody, Encoding.UTF8, "text/plain");
Uri startUri = new($"http://localhost:{AzureFunctionsPort}/api/workflows/ExpertReview/run");
this._outputHelper.WriteLine($"Starting ExpertReview workflow via POST request to {startUri}...");
using HttpResponseMessage startResponse = await s_sharedHttpClient.PostAsync(startUri, content);
Assert.True(startResponse.IsSuccessStatusCode, $"ExpertReview request failed with status: {startResponse.StatusCode}");
string startResponseText = await startResponse.Content.ReadAsStringAsync();
Assert.Contains("ExpertReview", startResponseText);
this._outputHelper.WriteLine($"ExpertReview response: {startResponseText}");
// Wait for the ParseQuestion executor to run
await this.WaitForConditionAsync(
condition: () =>
{
lock (logs)
{
bool exists = logs.Any(log => log.Message.Contains("[ParseQuestion]"));
return Task.FromResult(exists);
}
},
message: "ParseQuestion executor ran",
timeout: s_orchestrationTimeout);
// Wait for the Aggregator to complete (indicates fan-in from parallel agents)
await this.WaitForConditionAsync(
condition: () =>
{
lock (logs)
{
bool exists = logs.Any(log => log.Message.Contains("Aggregation complete"));
return Task.FromResult(exists);
}
},
message: "Aggregator completed with parallel agent responses",
timeout: s_orchestrationTimeout);
// Verify the aggregator received responses from both AI agents
lock (logs)
{
Assert.True(
logs.Any(log => log.Message.Contains("AI agent responses")),
"Aggregator did not log receiving AI agent responses.");
}
});
}
private async Task StartSharedInfrastructureAsync()
{
// Start Azurite if it's not already running
if (!await this.IsAzuriteRunningAsync())
{
await this.StartDockerContainerAsync(
containerName: "azurite",
image: "mcr.microsoft.com/azure-storage/azurite",
ports: ["-p", "10000:10000", "-p", "10001:10001", "-p", "10002:10002"]);
await this.WaitForConditionAsync(this.IsAzuriteRunningAsync, "Azurite is running", TimeSpan.FromSeconds(30));
}
// Start DTS emulator if it's not already running
if (!await this.IsDtsEmulatorRunningAsync())
{
await this.StartDockerContainerAsync(
containerName: "dts-emulator",
image: "mcr.microsoft.com/dts/dts-emulator:latest",
ports: ["-p", "8080:8080", "-p", "8082:8082"]);
await this.WaitForConditionAsync(
condition: this.IsDtsEmulatorRunningAsync,
message: "DTS emulator is running",
timeout: TimeSpan.FromSeconds(30));
}
}
private async Task<bool> IsAzuriteRunningAsync()
{
this._outputHelper.WriteLine(
$"Checking if Azurite is running at http://localhost:{AzuritePort}/devstoreaccount1...");
try
{
using CancellationTokenSource timeoutCts = new(TimeSpan.FromSeconds(30));
using HttpResponseMessage response = await s_sharedHttpClient.GetAsync(
requestUri: new Uri($"http://localhost:{AzuritePort}/devstoreaccount1?comp=list"),
cancellationToken: timeoutCts.Token);
if (response.Headers.TryGetValues(
"Server",
out IEnumerable<string>? serverValues) && serverValues.Any(s => s.StartsWith("Azurite", StringComparison.OrdinalIgnoreCase)))
{
this._outputHelper.WriteLine($"Azurite is running, server: {string.Join(", ", serverValues)}");
return true;
}
this._outputHelper.WriteLine($"Azurite is not running. Status code: {response.StatusCode}");
return false;
}
catch (HttpRequestException ex)
{
this._outputHelper.WriteLine($"Azurite is not running: {ex.Message}");
return false;
}
}
private async Task<bool> IsDtsEmulatorRunningAsync()
{
this._outputHelper.WriteLine($"Checking if DTS emulator is running at http://localhost:{DtsPort}/healthz...");
using HttpClient http2Client = new()
{
DefaultRequestVersion = new Version(2, 0),
DefaultVersionPolicy = HttpVersionPolicy.RequestVersionExact
};
try
{
using CancellationTokenSource timeoutCts = new(TimeSpan.FromSeconds(30));
using HttpResponseMessage response = await http2Client.GetAsync(new Uri($"http://localhost:{DtsPort}/healthz"), timeoutCts.Token);
if (response.Content.Headers.ContentLength > 0)
{
string content = await response.Content.ReadAsStringAsync(timeoutCts.Token);
this._outputHelper.WriteLine($"DTS emulator health check response: {content}");
}
if (response.IsSuccessStatusCode)
{
this._outputHelper.WriteLine("DTS emulator is running");
return true;
}
this._outputHelper.WriteLine($"DTS emulator is not running. Status code: {response.StatusCode}");
return false;
}
catch (HttpRequestException ex)
{
this._outputHelper.WriteLine($"DTS emulator is not running: {ex.Message}");
return false;
}
}
private async Task StartDockerContainerAsync(string containerName, string image, string[] ports)
{
await this.RunCommandAsync("docker", ["stop", containerName]);
await this.RunCommandAsync("docker", ["rm", containerName]);
List<string> args = ["run", "-d", "--name", containerName];
args.AddRange(ports);
args.Add(image);
this._outputHelper.WriteLine(
$"Starting new container: {containerName} with image: {image} and ports: {string.Join(", ", ports)}");
await this.RunCommandAsync("docker", args.ToArray());
this._outputHelper.WriteLine($"Container started: {containerName}");
}
private async Task WaitForConditionAsync(Func<Task<bool>> condition, string message, TimeSpan timeout)
{
this._outputHelper.WriteLine($"Waiting for '{message}'...");
using CancellationTokenSource cancellationTokenSource = new(timeout);
while (true)
{
if (await condition())
{
return;
}
try
{
await Task.Delay(TimeSpan.FromSeconds(1), cancellationTokenSource.Token);
}
catch (OperationCanceledException) when (cancellationTokenSource.IsCancellationRequested)
{
throw new TimeoutException($"Timeout waiting for '{message}'");
}
}
}
private sealed record OutputLog(DateTime Timestamp, LogLevel Level, string Message);
private async Task RunSampleTestAsync(string samplePath, bool requiresOpenAI, Func<IReadOnlyList<OutputLog>, Task> testAction)
{
List<OutputLog> logsContainer = [];
using Process funcProcess = this.StartFunctionApp(samplePath, logsContainer, requiresOpenAI);
try
{
await this.WaitForAzureFunctionsAsync();
await testAction(logsContainer);
}
finally
{
await this.StopProcessAsync(funcProcess);
}
}
private Process StartFunctionApp(string samplePath, List<OutputLog> logs, bool requiresOpenAI)
{
ProcessStartInfo startInfo = new()
{
FileName = "dotnet",
Arguments = $"run -f {s_dotnetTargetFramework} --port {AzureFunctionsPort}",
WorkingDirectory = samplePath,
UseShellExecute = false,
RedirectStandardOutput = true,
RedirectStandardError = true,
};
if (requiresOpenAI)
{
string openAiEndpoint = s_configuration["AZURE_OPENAI_ENDPOINT"] ??
throw new InvalidOperationException("The required AZURE_OPENAI_ENDPOINT env variable is not set.");
string openAiDeployment = s_configuration["AZURE_OPENAI_CHAT_DEPLOYMENT_NAME"] ??
throw new InvalidOperationException("The required AZURE_OPENAI_CHAT_DEPLOYMENT_NAME env variable is not set.");
this._outputHelper.WriteLine($"Using Azure OpenAI endpoint: {openAiEndpoint}, deployment: {openAiDeployment}");
startInfo.EnvironmentVariables["AZURE_OPENAI_ENDPOINT"] = openAiEndpoint;
startInfo.EnvironmentVariables["AZURE_OPENAI_DEPLOYMENT"] = openAiDeployment;
}
startInfo.EnvironmentVariables["DURABLE_TASK_SCHEDULER_CONNECTION_STRING"] =
$"Endpoint=http://localhost:{DtsPort};TaskHub=default;Authentication=None";
startInfo.EnvironmentVariables["AzureWebJobsStorage"] = "UseDevelopmentStorage=true";
Process process = new() { StartInfo = startInfo };
process.ErrorDataReceived += (sender, e) =>
{
if (e.Data != null)
{
this._outputHelper.WriteLine($"[{startInfo.FileName}(err)]: {e.Data}");
lock (logs)
{
logs.Add(new OutputLog(DateTime.Now, LogLevel.Error, e.Data));
}
}
};
process.OutputDataReceived += (sender, e) =>
{
if (e.Data != null)
{
this._outputHelper.WriteLine($"[{startInfo.FileName}(out)]: {e.Data}");
lock (logs)
{
logs.Add(new OutputLog(DateTime.Now, LogLevel.Information, e.Data));
}
}
};
if (!process.Start())
{
throw new InvalidOperationException("Failed to start the function app");
}
process.BeginErrorReadLine();
process.BeginOutputReadLine();
return process;
}
private async Task WaitForAzureFunctionsAsync()
{
this._outputHelper.WriteLine(
$"Waiting for Azure Functions Core Tools to be ready at http://localhost:{AzureFunctionsPort}/...");
await this.WaitForConditionAsync(
condition: async () =>
{
try
{
using HttpRequestMessage request = new(HttpMethod.Head, $"http://localhost:{AzureFunctionsPort}/");
using HttpResponseMessage response = await s_sharedHttpClient.SendAsync(request);
this._outputHelper.WriteLine($"Azure Functions Core Tools response: {response.StatusCode}");
return response.IsSuccessStatusCode;
}
catch (HttpRequestException)
{
return false;
}
},
message: "Azure Functions Core Tools is ready",
timeout: TimeSpan.FromSeconds(60));
}
private async Task RunCommandAsync(string command, string[] args)
{
ProcessStartInfo startInfo = new()
{
FileName = command,
Arguments = string.Join(" ", args),
UseShellExecute = false,
RedirectStandardOutput = true,
RedirectStandardError = true,
CreateNoWindow = true
};
this._outputHelper.WriteLine($"Running command: {command} {string.Join(" ", args)}");
using Process process = new() { StartInfo = startInfo };
process.ErrorDataReceived += (sender, e) => this._outputHelper.WriteLine($"[{command}(err)]: {e.Data}");
process.OutputDataReceived += (sender, e) => this._outputHelper.WriteLine($"[{command}(out)]: {e.Data}");
if (!process.Start())
{
throw new InvalidOperationException("Failed to start the command");
}
process.BeginErrorReadLine();
process.BeginOutputReadLine();
using CancellationTokenSource cancellationTokenSource = new(TimeSpan.FromMinutes(1));
await process.WaitForExitAsync(cancellationTokenSource.Token);
this._outputHelper.WriteLine($"Command completed with exit code: {process.ExitCode}");
}
private async Task StopProcessAsync(Process process)
{
try
{
if (!process.HasExited)
{
this._outputHelper.WriteLine($"Killing process {process.ProcessName}#{process.Id}");
process.Kill(entireProcessTree: true);
using CancellationTokenSource timeoutCts = new(TimeSpan.FromSeconds(10));
await process.WaitForExitAsync(timeoutCts.Token);
this._outputHelper.WriteLine($"Process exited: {process.Id}");
}
}
catch (Exception ex)
{
this._outputHelper.WriteLine($"Failed to stop process: {ex.Message}");
}
}
private static string GetTargetFramework()
{
string filePath = new Uri(typeof(WorkflowSamplesValidation).Assembly.Location).LocalPath;
string directory = Path.GetDirectoryName(filePath)!;
string tfm = Path.GetFileName(directory);
if (tfm.StartsWith("net", StringComparison.OrdinalIgnoreCase))
{
return tfm;
}
throw new InvalidOperationException($"Unable to find target framework in path: {filePath}");
}
}