diff --git a/dotnet/Directory.Packages.props b/dotnet/Directory.Packages.props index 98c7376aaf..556d0a61d8 100644 --- a/dotnet/Directory.Packages.props +++ b/dotnet/Directory.Packages.props @@ -113,14 +113,14 @@ - - - - + + + + - + diff --git a/dotnet/agent-framework-dotnet.slnx b/dotnet/agent-framework-dotnet.slnx index 20552473b7..6c564d81b8 100644 --- a/dotnet/agent-framework-dotnet.slnx +++ b/dotnet/agent-framework-dotnet.slnx @@ -52,6 +52,13 @@ + + + + + + + @@ -403,7 +410,6 @@ - @@ -411,6 +417,7 @@ + @@ -422,8 +429,8 @@ - + @@ -433,8 +440,8 @@ - + @@ -448,13 +455,13 @@ - + diff --git a/dotnet/samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/01_SequentialWorkflow.csproj b/dotnet/samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/01_SequentialWorkflow.csproj new file mode 100644 index 0000000000..1c3ef2504a --- /dev/null +++ b/dotnet/samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/01_SequentialWorkflow.csproj @@ -0,0 +1,42 @@ + + + net10.0 + v4 + Exe + enable + enable + + SequentialWorkflowFunctionApp + SequentialWorkflowFunctionApp + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/dotnet/samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/Function.cs b/dotnet/samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/Function.cs new file mode 100644 index 0000000000..dd8de6e553 --- /dev/null +++ b/dotnet/samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/Function.cs @@ -0,0 +1,23 @@ +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Mvc; +using Microsoft.Azure.Functions.Worker; +using Microsoft.Extensions.Logging; + +namespace SequentialWorkflowFunctionApp; + +public class Function +{ + private readonly ILogger _logger; + + public Function(ILogger logger) + { + _logger = logger; + } + + [Function("Function")] + public IActionResult Run([HttpTrigger(AuthorizationLevel.Function, "get", "post")] HttpRequest req) + { + _logger.LogInformation("C# HTTP trigger function processed a request."); + return new OkObjectResult("Welcome to Azure Functions!"); + } +} \ No newline at end of file diff --git a/dotnet/samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/OrderCancelExecutors.cs b/dotnet/samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/OrderCancelExecutors.cs new file mode 100644 index 0000000000..39f5051a8c --- /dev/null +++ b/dotnet/samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/OrderCancelExecutors.cs @@ -0,0 +1,59 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Agents.AI.Workflows; + +namespace SequentialWorkflow; + +/// +/// Parses an Order ID from a string input and returns an Order object populated. +/// +internal sealed class OrderLookup() : Executor("OrderLookup") +{ + public override async ValueTask HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + // Populate Order information from OrderId. + return new Order(message, 100.0m); + } +} + +/// +/// Enriches an Order object with additional information. +/// +internal sealed class OrderEnrich() : Executor("EnrichOrder") +{ + public override async ValueTask HandleAsync(Order message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + if (message.Customer is null) + { + // populate customer information for the order from database. + message.Customer = new Customer(1, "Jerry"); + } + + return message; + } +} + +internal sealed class PaymentProcessor() : Executor("ProcessPayment") +{ + public override async ValueTask HandleAsync(Order message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + message.PaymentReferenceNumber = Guid.NewGuid().ToString()[^4..]; + + return message; + } +} + +internal sealed class Order +{ + public Order(string id, decimal amount) + { + this.Id = id; + this.Amount = amount; + } + public string Id { get; } + public decimal Amount { get; } + public Customer? Customer { get; set; } + public string? PaymentReferenceNumber { get; set; } +} + +public sealed record Customer(int Id, string Name); diff --git a/dotnet/samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/Program.cs b/dotnet/samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/Program.cs new file mode 100644 index 0000000000..8b6b457da5 --- /dev/null +++ b/dotnet/samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/Program.cs @@ -0,0 +1,31 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Agents.AI.Workflows; +using Microsoft.Azure.Functions.Worker.Builder; +using Microsoft.Agents.AI.Hosting.AzureFunctions; +using Microsoft.Extensions.Hosting; +using SequentialWorkflow; + +OrderLookup orderLookupExecutor = new(); +OrderEnrich orderEnricherExeecutor = new(); +PaymentProcessor paymentProcessorExecutor = new(); + +Workflow fulfillOrder = new WorkflowBuilder(orderLookupExecutor) + .WithName("FulfillOrder") + .WithDescription("Looks up an order by ID and run payment processing") + .AddEdge(orderLookupExecutor, orderEnricherExeecutor) + .AddEdge(orderEnricherExeecutor, paymentProcessorExecutor) + .Build(); + +// Configure the function app to host the AI agent. +// This will automatically generate HTTP API endpoints for the agent. +using IHost app = FunctionsApplication + .CreateBuilder(args) + .ConfigureFunctionsWebApplication() + .ConfigureDurableOptions(durableOption => + { + // Add a workflow. + durableOption.Workflows.AddWorkflow(fulfillOrder); + }) + .Build(); +app.Run(); diff --git a/dotnet/samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/demo.http b/dotnet/samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/demo.http new file mode 100644 index 0000000000..ec607e0f8d --- /dev/null +++ b/dotnet/samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/demo.http @@ -0,0 +1,8 @@ +# Default endpoint address for local testing +@authority=http://localhost:7071 + +### Prompt the agent +POST {{authority}}/api/workflows/FulfillOrder/run +Content-Type: text/plain + +987 \ No newline at end of file diff --git a/dotnet/samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/host.json b/dotnet/samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/host.json new file mode 100644 index 0000000000..9384a0a583 --- /dev/null +++ b/dotnet/samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/host.json @@ -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" + } + } + } +} diff --git a/dotnet/samples/Durable/Workflow/AzureFunctions/02_ConcurrentWorkflow/02_ConcurrentWorkflow.csproj b/dotnet/samples/Durable/Workflow/AzureFunctions/02_ConcurrentWorkflow/02_ConcurrentWorkflow.csproj new file mode 100644 index 0000000000..1c3ef2504a --- /dev/null +++ b/dotnet/samples/Durable/Workflow/AzureFunctions/02_ConcurrentWorkflow/02_ConcurrentWorkflow.csproj @@ -0,0 +1,42 @@ + + + net10.0 + v4 + Exe + enable + enable + + SequentialWorkflowFunctionApp + SequentialWorkflowFunctionApp + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/dotnet/samples/Durable/Workflow/AzureFunctions/02_ConcurrentWorkflow/ExpertExecutors.cs b/dotnet/samples/Durable/Workflow/AzureFunctions/02_ConcurrentWorkflow/ExpertExecutors.cs new file mode 100644 index 0000000000..40674126f6 --- /dev/null +++ b/dotnet/samples/Durable/Workflow/AzureFunctions/02_ConcurrentWorkflow/ExpertExecutors.cs @@ -0,0 +1,73 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Agents.AI.Workflows; + +namespace WorkflowConcurrency; + +/// +/// Parses and validates the incoming question before sending to AI agents. +/// +internal sealed class ParseQuestionExecutor() : Executor("ParseQuestion") +{ + public override ValueTask 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); + } +} + +/// +/// Aggregates responses from all AI agents into a comprehensive answer. +/// This is the Fan-in point where parallel results are collected. +/// +internal sealed class AggregatorExecutor() : Executor("Aggregator") +{ + public override ValueTask 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); + } +} diff --git a/dotnet/samples/Durable/Workflow/AzureFunctions/02_ConcurrentWorkflow/Program.cs b/dotnet/samples/Durable/Workflow/AzureFunctions/02_ConcurrentWorkflow/Program.cs new file mode 100644 index 0000000000..734ba1b46a --- /dev/null +++ b/dotnet/samples/Durable/Workflow/AzureFunctions/02_ConcurrentWorkflow/Program.cs @@ -0,0 +1,53 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Azure; +using Azure.AI.OpenAI; +using Azure.Identity; +using Microsoft.Agents.AI; +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; + +// Configuration +string dtsConnectionString = Environment.GetEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING") + ?? "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None"; +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(); + +// Configure the function app to host the AI agent. +// This will automatically generate HTTP API endpoints for the agent. +using IHost app = FunctionsApplication + .CreateBuilder(args) + .ConfigureFunctionsWebApplication() + .ConfigureDurableOptions(durableOption => + { + // Add a workflow. + durableOption.Workflows.AddWorkflow(workflow); + }) + .Build(); +app.Run(); diff --git a/dotnet/samples/Durable/Workflow/AzureFunctions/02_ConcurrentWorkflow/demo.http b/dotnet/samples/Durable/Workflow/AzureFunctions/02_ConcurrentWorkflow/demo.http new file mode 100644 index 0000000000..3129388a27 --- /dev/null +++ b/dotnet/samples/Durable/Workflow/AzureFunctions/02_ConcurrentWorkflow/demo.http @@ -0,0 +1,8 @@ +# 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 saturation? diff --git a/dotnet/samples/Durable/Workflow/AzureFunctions/02_ConcurrentWorkflow/host.json b/dotnet/samples/Durable/Workflow/AzureFunctions/02_ConcurrentWorkflow/host.json new file mode 100644 index 0000000000..9384a0a583 --- /dev/null +++ b/dotnet/samples/Durable/Workflow/AzureFunctions/02_ConcurrentWorkflow/host.json @@ -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" + } + } + } +} diff --git a/dotnet/samples/Durable/Workflow/AzureFunctions/03_ConditionalEdges/03_ConditionalEdges.csproj b/dotnet/samples/Durable/Workflow/AzureFunctions/03_ConditionalEdges/03_ConditionalEdges.csproj new file mode 100644 index 0000000000..1add0fdc50 --- /dev/null +++ b/dotnet/samples/Durable/Workflow/AzureFunctions/03_ConditionalEdges/03_ConditionalEdges.csproj @@ -0,0 +1,39 @@ + + + net10.0 + v4 + Exe + enable + enable + + ConditionalEdgesFunctionApp + ConditionalEdgesFunctionApp + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/dotnet/samples/Durable/Workflow/AzureFunctions/03_ConditionalEdges/OrderExecutors.cs b/dotnet/samples/Durable/Workflow/AzureFunctions/03_ConditionalEdges/OrderExecutors.cs new file mode 100644 index 0000000000..4d0518955a --- /dev/null +++ b/dotnet/samples/Durable/Workflow/AzureFunctions/03_ConditionalEdges/OrderExecutors.cs @@ -0,0 +1,118 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Agents.AI.Workflows; + +namespace ConditionalEdgesFunctionApp; + +/// +/// Represents an order with customer and payment details. +/// +internal sealed class Order +{ + public Order(string id, decimal amount) + { + this.Id = id; + this.Amount = amount; + } + + public string Id { get; } + public decimal Amount { get; } + public Customer? Customer { get; set; } + public string? PaymentReferenceNumber { get; set; } +} + +/// +/// Represents a customer associated with an order. +/// +public sealed record Customer(int Id, string Name, bool IsBlocked); + +/// +/// Parses the order ID and retrieves order details. +/// +internal sealed class OrderIdParser() : Executor("OrderIdParser") +{ + public override ValueTask HandleAsync( + string message, + IWorkflowContext context, + CancellationToken cancellationToken = default) + { + Console.WriteLine($"[OrderIdParser] Parsing order ID: {message}"); + Order order = new(message, 100.0m); + return ValueTask.FromResult(order); + } +} + +/// +/// Enriches the order with customer information. +/// Orders with IDs containing 'B' are associated with blocked customers. +/// +internal sealed class OrderEnrich() : Executor("EnrichOrder") +{ + public override ValueTask HandleAsync( + Order message, + IWorkflowContext context, + CancellationToken cancellationToken = default) + { + message.Customer = GetCustomerForOrder(message.Id); + Console.WriteLine($"[EnrichOrder] Customer: {message.Customer.Name}, IsBlocked: {message.Customer.IsBlocked}"); + return ValueTask.FromResult(message); + } + + private static Customer GetCustomerForOrder(string orderId) + { + if (orderId.Contains('B')) + { + return new Customer(101, "George", true); + } + + return new Customer(201, "Jerry", false); + } +} + +/// +/// Processes payment for valid (non-blocked) orders. +/// +internal sealed class PaymentProcessor() : Executor("PaymentProcessor") +{ + public override ValueTask HandleAsync( + Order message, + IWorkflowContext context, + CancellationToken cancellationToken = default) + { + message.PaymentReferenceNumber = Guid.NewGuid().ToString()[..4]; + Console.WriteLine($"[PaymentProcessor] Payment processed for order {message.Id}. Reference: {message.PaymentReferenceNumber}"); + return ValueTask.FromResult(message); + } +} + +/// +/// Notifies the fraud team when a blocked customer places an order. +/// +internal sealed class NotifyFraud() : Executor("NotifyFraud") +{ + public override ValueTask HandleAsync( + Order message, + IWorkflowContext context, + CancellationToken cancellationToken = default) + { + string result = $"Order {message.Id} flagged as fraudulent for customer {message.Customer?.Name}."; + Console.WriteLine($"[NotifyFraud] {result}"); + return ValueTask.FromResult(result); + } +} + +/// +/// Defines condition functions for routing orders based on customer status. +/// +internal static class OrderRouteConditions +{ + /// + /// Returns a condition that evaluates to true when the customer is blocked. + /// + internal static Func WhenBlocked() => order => order?.Customer?.IsBlocked == true; + + /// + /// Returns a condition that evaluates to true when the customer is not blocked. + /// + internal static Func WhenNotBlocked() => order => order?.Customer?.IsBlocked == false; +} diff --git a/dotnet/samples/Durable/Workflow/AzureFunctions/03_ConditionalEdges/Program.cs b/dotnet/samples/Durable/Workflow/AzureFunctions/03_ConditionalEdges/Program.cs new file mode 100644 index 0000000000..20bbe262a5 --- /dev/null +++ b/dotnet/samples/Durable/Workflow/AzureFunctions/03_ConditionalEdges/Program.cs @@ -0,0 +1,42 @@ +// Copyright (c) Microsoft. All rights reserved. + +// This sample demonstrates conditional edges in a workflow hosted as an Azure Function. +// Orders are routed to different executors based on customer status: +// - Blocked customers → NotifyFraud +// - Valid customers → PaymentProcessor + +using ConditionalEdgesFunctionApp; +using Microsoft.Agents.AI.Hosting.AzureFunctions; +using Microsoft.Agents.AI.Workflows; +using Microsoft.Azure.Functions.Worker.Builder; +using Microsoft.Extensions.Hosting; + +// Create executor instances +OrderIdParser orderParser = new(); +OrderEnrich orderEnrich = new(); +PaymentProcessor paymentProcessor = new(); +NotifyFraud notifyFraud = new(); + +// Build workflow with conditional edges. +// The condition functions evaluate the Order output from OrderEnrich +// to determine whether to route to NotifyFraud or PaymentProcessor. +Workflow auditOrder = new WorkflowBuilder(orderParser) + .WithName("AuditOrder") + .WithDescription("Audits an order and routes based on customer status") + .AddEdge(orderParser, orderEnrich) + .AddEdge(orderEnrich, notifyFraud, condition: OrderRouteConditions.WhenBlocked()) + .AddEdge(orderEnrich, paymentProcessor, condition: OrderRouteConditions.WhenNotBlocked()) + .Build(); + +// Configure the function app to host the workflow. +// This will automatically generate HTTP API endpoints for the workflow. +using IHost app = FunctionsApplication + .CreateBuilder(args) + .ConfigureFunctionsWebApplication() + .ConfigureDurableOptions(durableOption => + { + // Add the workflow. + durableOption.Workflows.AddWorkflow(auditOrder); + }) + .Build(); +app.Run(); diff --git a/dotnet/samples/Durable/Workflow/AzureFunctions/03_ConditionalEdges/README.md b/dotnet/samples/Durable/Workflow/AzureFunctions/03_ConditionalEdges/README.md new file mode 100644 index 0000000000..41eabf7b05 --- /dev/null +++ b/dotnet/samples/Durable/Workflow/AzureFunctions/03_ConditionalEdges/README.md @@ -0,0 +1,84 @@ +# Conditional Edges Workflow - Azure Functions Sample + +This sample demonstrates how to build a workflow with **conditional edges** hosted as an Azure Function. Orders are routed to different executors based on customer status. + +## Key Concepts Demonstrated + +- Building workflows with **conditional edges** using `AddEdge` with a `condition` parameter +- Hosting a conditional workflow as an Azure Function using `ConfigureDurableOptions` +- Defining reusable condition functions for routing logic +- Branching workflow execution based on data-driven decisions + +## Overview + +The workflow implements an order audit that routes orders differently based on whether the customer is blocked (flagged for fraud): + +``` +OrderIdParser --> OrderEnrich --[IsBlocked]--> NotifyFraud + | + +--[NotBlocked]--> PaymentProcessor +``` + +| Executor | Description | +|----------|-------------| +| OrderIdParser | Parses the order ID and retrieves order details | +| OrderEnrich | Enriches the order with customer information | +| PaymentProcessor | Processes payment for valid orders | +| NotifyFraud | Notifies the fraud team for blocked customers | + +## How Conditional Edges Work + +Conditional edges allow you to specify a condition function that determines whether the edge should be traversed: + +```csharp +builder + .AddEdge(orderParser, orderEnrich) + .AddEdge(orderEnrich, notifyFraud, condition: OrderRouteConditions.WhenBlocked()) + .AddEdge(orderEnrich, paymentProcessor, condition: OrderRouteConditions.WhenNotBlocked()); +``` + +The condition functions receive the output of the source executor and return a boolean: + +```csharp +internal static class OrderRouteConditions +{ + internal static Func WhenBlocked() => + order => order?.Customer?.IsBlocked == true; + + internal static Func WhenNotBlocked() => + order => order?.Customer?.IsBlocked == false; +} +``` + +### Routing Logic + +- Order IDs containing the letter **'B'** are associated with blocked customers → routed to `NotifyFraud` +- All other order IDs are associated with valid customers → routed to `PaymentProcessor` + +## Environment Setup + +This sample requires: + +- [.NET 10 SDK](https://dotnet.microsoft.com/download) +- [Azure Functions Core Tools v4](https://learn.microsoft.com/azure/azure-functions/functions-run-local) +- [Durable Task Scheduler](https://learn.microsoft.com/azure/azure-functions/durable/durable-functions-azure-managed-storage) running locally (default: `http://localhost:8080`) +- [Azurite](https://learn.microsoft.com/azure/storage/common/storage-use-azurite) for local Azure Storage emulation + +## Running the Sample + +```bash +cd dotnet/samples/Durable/Workflow/AzureFunctions/03_ConditionalEdges +func start +``` + +### Testing + +**Valid order (routes to PaymentProcessor):** +```bash +curl -X POST http://localhost:7071/api/workflows/AuditOrder/run -H "Content-Type: text/plain" -d "12345" +``` + +**Blocked order (routes to NotifyFraud):** +```bash +curl -X POST http://localhost:7071/api/workflows/AuditOrder/run -H "Content-Type: text/plain" -d "12345B" +``` diff --git a/dotnet/samples/Durable/Workflow/AzureFunctions/03_ConditionalEdges/demo.http b/dotnet/samples/Durable/Workflow/AzureFunctions/03_ConditionalEdges/demo.http new file mode 100644 index 0000000000..2516be9224 --- /dev/null +++ b/dotnet/samples/Durable/Workflow/AzureFunctions/03_ConditionalEdges/demo.http @@ -0,0 +1,14 @@ +# Default endpoint address for local testing +@authority=http://localhost:7071 + +### Valid order (routes to PaymentProcessor) +POST {{authority}}/api/workflows/AuditOrder/run +Content-Type: text/plain + +12345 + +### Blocked order (routes to NotifyFraud) - Order IDs containing 'B' are flagged +POST {{authority}}/api/workflows/AuditOrder/run +Content-Type: text/plain + +12345B diff --git a/dotnet/samples/Durable/Workflow/AzureFunctions/03_ConditionalEdges/host.json b/dotnet/samples/Durable/Workflow/AzureFunctions/03_ConditionalEdges/host.json new file mode 100644 index 0000000000..9384a0a583 --- /dev/null +++ b/dotnet/samples/Durable/Workflow/AzureFunctions/03_ConditionalEdges/host.json @@ -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" + } + } + } +} diff --git a/dotnet/samples/Durable/Workflow/AzureFunctions/04_NestedWorkflows/04_NestedWorkflows.csproj b/dotnet/samples/Durable/Workflow/AzureFunctions/04_NestedWorkflows/04_NestedWorkflows.csproj new file mode 100644 index 0000000000..22903f3fe2 --- /dev/null +++ b/dotnet/samples/Durable/Workflow/AzureFunctions/04_NestedWorkflows/04_NestedWorkflows.csproj @@ -0,0 +1,39 @@ + + + net10.0 + v4 + Exe + enable + enable + + NestedWorkflowsFunctionApp + NestedWorkflowsFunctionApp + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/dotnet/samples/Durable/Workflow/AzureFunctions/04_NestedWorkflows/OrderExecutors.cs b/dotnet/samples/Durable/Workflow/AzureFunctions/04_NestedWorkflows/OrderExecutors.cs new file mode 100644 index 0000000000..1fd697b434 --- /dev/null +++ b/dotnet/samples/Durable/Workflow/AzureFunctions/04_NestedWorkflows/OrderExecutors.cs @@ -0,0 +1,220 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Agents.AI.Workflows; + +namespace NestedWorkflowsFunctionApp; + +// ============================================ +// Order Processing Models +// ============================================ + +/// +/// Represents an order being processed through the workflow. +/// +internal sealed class OrderInfo +{ + public required string OrderId { get; set; } + public decimal Amount { get; set; } + public string? PaymentTransactionId { get; set; } + public string? InventoryReservationId { get; set; } + public string? TrackingNumber { get; set; } + public string? Carrier { get; set; } +} + +// ============================================ +// Main Workflow Executors +// ============================================ + +/// +/// Entry point executor that receives the order ID and creates an OrderInfo object. +/// +internal sealed class OrderReceived() : Executor("OrderReceived") +{ + public override ValueTask HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.WriteLine($"[OrderReceived] Processing order '{message}'"); + + OrderInfo order = new() + { + OrderId = message, + Amount = 99.99m + }; + + return ValueTask.FromResult(order); + } +} + +/// +/// Final executor that outputs the completed order summary. +/// +internal sealed class OrderCompleted() : Executor("OrderCompleted") +{ + public override ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.WriteLine($"[OrderCompleted] Order '{message.OrderId}' processed. Payment: {message.PaymentTransactionId}, Inventory: {message.InventoryReservationId}, Shipping: {message.Carrier} - {message.TrackingNumber}"); + + return ValueTask.FromResult($"Order {message.OrderId} completed. Tracking: {message.TrackingNumber}"); + } +} + +// ============================================ +// Payment Sub-Workflow Executors +// ============================================ + +/// +/// Validates payment information for an order. +/// +internal sealed class ValidatePayment() : Executor("ValidatePayment") +{ + public override async ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.WriteLine($"[Payment/ValidatePayment] Validating payment for order '{message.OrderId}'"); + + await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken); + + Console.WriteLine($"[Payment/ValidatePayment] Payment validated for ${message.Amount}"); + + return message; + } +} + +// ============================================ +// Fraud Check Sub-Sub-Workflow Executors (Level 2 nesting) +// ============================================ + +/// +/// Analyzes transaction patterns for potential fraud. +/// +internal sealed class AnalyzePatterns() : Executor("AnalyzePatterns") +{ + public override async ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.WriteLine($"[Payment/FraudCheck/AnalyzePatterns] Analyzing patterns for order '{message.OrderId}'"); + + await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken); + + Console.WriteLine("[Payment/FraudCheck/AnalyzePatterns] Pattern analysis complete"); + + return message; + } +} + +/// +/// Calculates a risk score for the transaction. +/// +internal sealed class CalculateRiskScore() : Executor("CalculateRiskScore") +{ + public override async ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.WriteLine($"[Payment/FraudCheck/CalculateRiskScore] Calculating risk score for order '{message.OrderId}'"); + + await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken); + + int riskScore = new Random().Next(1, 100); + + Console.WriteLine($"[Payment/FraudCheck/CalculateRiskScore] Risk score: {riskScore}/100 (Low risk)"); + + return message; + } +} + +/// +/// Charges the payment for an order. +/// +internal sealed class ChargePayment() : Executor("ChargePayment") +{ + public override async ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.WriteLine($"[Payment/ChargePayment] Charging ${message.Amount} for order '{message.OrderId}'"); + + await Task.Delay(TimeSpan.FromSeconds(2), cancellationToken); + + message.PaymentTransactionId = $"TXN-{Guid.NewGuid().ToString("N")[..8].ToUpperInvariant()}"; + + Console.WriteLine($"[Payment/ChargePayment] Payment processed: {message.PaymentTransactionId}"); + + return message; + } +} + +// ============================================ +// Inventory Sub-Workflow Executors +// ============================================ + +/// +/// Checks inventory availability for an order. +/// +internal sealed class CheckInventory() : Executor("CheckInventory") +{ + public override async ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.WriteLine($"[Inventory/CheckInventory] Checking inventory for order '{message.OrderId}'"); + + await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken); + + Console.WriteLine("[Inventory/CheckInventory] Items available in stock"); + + return message; + } +} + +/// +/// Reserves inventory for an order. +/// +internal sealed class ReserveInventory() : Executor("ReserveInventory") +{ + public override async ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.WriteLine($"[Inventory/ReserveInventory] Reserving items for order '{message.OrderId}'"); + + await Task.Delay(TimeSpan.FromSeconds(2), cancellationToken); + + message.InventoryReservationId = $"RES-{Guid.NewGuid().ToString("N")[..8].ToUpperInvariant()}"; + + Console.WriteLine($"[Inventory/ReserveInventory] Reserved: {message.InventoryReservationId}"); + + return message; + } +} + +// ============================================ +// Shipping Sub-Workflow Executors +// ============================================ + +/// +/// Selects a shipping carrier for an order. +/// +internal sealed class SelectCarrier() : Executor("SelectCarrier") +{ + public override async ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.WriteLine($"[Shipping/SelectCarrier] Selecting carrier for order '{message.OrderId}'"); + + await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken); + + message.Carrier = message.Amount > 50 ? "Express" : "Standard"; + + Console.WriteLine($"[Shipping/SelectCarrier] Selected carrier: {message.Carrier}"); + + return message; + } +} + +/// +/// Creates shipment and generates tracking number. +/// +internal sealed class CreateShipment() : Executor("CreateShipment") +{ + public override async ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.WriteLine($"[Shipping/CreateShipment] Creating shipment for order '{message.OrderId}'"); + + await Task.Delay(TimeSpan.FromSeconds(2), cancellationToken); + + message.TrackingNumber = $"TRACK-{Guid.NewGuid().ToString("N")[..10].ToUpperInvariant()}"; + + Console.WriteLine($"[Shipping/CreateShipment] Shipment created: {message.TrackingNumber}"); + + return message; + } +} diff --git a/dotnet/samples/Durable/Workflow/AzureFunctions/04_NestedWorkflows/Program.cs b/dotnet/samples/Durable/Workflow/AzureFunctions/04_NestedWorkflows/Program.cs new file mode 100644 index 0000000000..4817023868 --- /dev/null +++ b/dotnet/samples/Durable/Workflow/AzureFunctions/04_NestedWorkflows/Program.cs @@ -0,0 +1,122 @@ +// Copyright (c) Microsoft. All rights reserved. + +// This sample demonstrates nested workflows (sub-workflows) hosted as an Azure Function. +// One workflow can be used as an executor within another workflow, enabling modular, +// reusable workflow components with independent checkpointing and replay. +// +// Workflow structure: +// +// ┌─────────────────────────────────────────────────────────────────────────┐ +// │ OrderProcessing (Main Workflow) │ +// │ │ +// │ OrderReceived │ +// │ │ │ +// │ ▼ │ +// │ ┌─────────────────────────────────────────────────────────┐ │ +// │ │ Payment (Sub-Workflow) │ │ +// │ │ │ │ +// │ │ ValidatePayment │ │ +// │ │ │ │ │ +// │ │ ▼ │ │ +// │ │ ┌─────────────────────────────────────────┐ │ │ +// │ │ │ FraudCheck (Sub-Sub-Workflow) │ │ │ +// │ │ │ │ │ │ +// │ │ │ AnalyzePatterns ──► CalculateRiskScore │ │ │ +// │ │ └─────────────────────────────────────────┘ │ │ +// │ │ │ │ │ +// │ │ ▼ │ │ +// │ │ ChargePayment │ │ +// │ └─────────────────────────────────────────────────────────┘ │ +// │ │ │ +// │ ▼ │ +// │ ┌─────────────────────────────────────────────────────────┐ │ +// │ │ Inventory (Sub-Workflow) │ │ +// │ │ │ │ +// │ │ CheckInventory ──► ReserveInventory │ │ +// │ └─────────────────────────────────────────────────────────┘ │ +// │ │ │ +// │ ▼ │ +// │ ┌─────────────────────────────────────────────────────────┐ │ +// │ │ Shipping (Sub-Workflow) │ │ +// │ │ │ │ +// │ │ SelectCarrier ──► CreateShipment │ │ +// │ └─────────────────────────────────────────────────────────┘ │ +// │ │ │ +// │ ▼ │ +// │ OrderCompleted │ +// └─────────────────────────────────────────────────────────────────────────┘ + +using Microsoft.Agents.AI.Hosting.AzureFunctions; +using Microsoft.Agents.AI.Workflows; +using Microsoft.Azure.Functions.Worker.Builder; +using Microsoft.Extensions.Hosting; +using NestedWorkflowsFunctionApp; + +// Create executor instances for the Fraud Check sub-sub-workflow (Level 2 nesting) +AnalyzePatterns analyzePatterns = new(); +CalculateRiskScore calculateRiskScore = new(); + +Workflow fraudCheckWorkflow = new WorkflowBuilder(analyzePatterns) + .WithName("SubFraudCheck") + .WithDescription("Analyzes transaction patterns and calculates risk score") + .AddEdge(analyzePatterns, calculateRiskScore) + .Build(); + +// Create executor instances for the Payment Processing sub-workflow +ValidatePayment validatePayment = new(); +ExecutorBinding fraudCheckExecutor = fraudCheckWorkflow.BindAsExecutor("FraudCheck"); +ChargePayment chargePayment = new(); + +Workflow paymentWorkflow = new WorkflowBuilder(validatePayment) + .WithName("SubPaymentProcessing") + .WithDescription("Validates and processes payment for an order") + .AddEdge(validatePayment, fraudCheckExecutor) + .AddEdge(fraudCheckExecutor, chargePayment) + .Build(); + +// Create executor instances for the Inventory Management sub-workflow +CheckInventory checkInventory = new(); +ReserveInventory reserveInventory = new(); + +Workflow inventoryWorkflow = new WorkflowBuilder(checkInventory) + .WithName("SubInventoryManagement") + .WithDescription("Checks availability and reserves inventory") + .AddEdge(checkInventory, reserveInventory) + .Build(); + +// Create executor instances for the Shipping Arrangement sub-workflow +SelectCarrier selectCarrier = new(); +CreateShipment createShipment = new(); + +Workflow shippingWorkflow = new WorkflowBuilder(selectCarrier) + .WithName("SubShippingArrangement") + .WithDescription("Selects carrier and creates shipment") + .AddEdge(selectCarrier, createShipment) + .Build(); + +// Build the main Order Processing workflow using sub-workflows as executors +ExecutorBinding paymentExecutor = paymentWorkflow.BindAsExecutor("Payment"); +ExecutorBinding inventoryExecutor = inventoryWorkflow.BindAsExecutor("Inventory"); +ExecutorBinding shippingExecutor = shippingWorkflow.BindAsExecutor("Shipping"); + +OrderReceived orderReceived = new(); +OrderCompleted orderCompleted = new(); + +Workflow orderProcessingWorkflow = new WorkflowBuilder(orderReceived) + .WithName("OrderProcessing") + .WithDescription("Processes an order through payment, inventory, and shipping") + .AddEdge(orderReceived, paymentExecutor) + .AddEdge(paymentExecutor, inventoryExecutor) + .AddEdge(inventoryExecutor, shippingExecutor) + .AddEdge(shippingExecutor, orderCompleted) + .Build(); + +// Configure the function app to host the workflow. +// Sub-workflows are discovered and registered automatically. +using IHost app = FunctionsApplication + .CreateBuilder(args) + .ConfigureFunctionsWebApplication() + .ConfigureDurableOptions(durableOption => + durableOption.Workflows.AddWorkflow(orderProcessingWorkflow)) + .Build(); +app.Run(); diff --git a/dotnet/samples/Durable/Workflow/AzureFunctions/04_NestedWorkflows/README.md b/dotnet/samples/Durable/Workflow/AzureFunctions/04_NestedWorkflows/README.md new file mode 100644 index 0000000000..04973bd98f --- /dev/null +++ b/dotnet/samples/Durable/Workflow/AzureFunctions/04_NestedWorkflows/README.md @@ -0,0 +1,112 @@ +# Nested Workflows - Azure Functions Sample + +This sample demonstrates how to build **nested workflows** (sub-workflows) hosted as an Azure Function. One workflow can be used as an executor within another workflow, enabling modular, reusable workflow components. + +## Key Concepts Demonstrated + +- Building workflows with **sub-workflow executors** using `BindAsExecutor` +- **Multi-level nesting** — a sub-workflow contains its own sub-workflow (FraudCheck inside Payment) +- Hosting nested workflows as an Azure Function using `ConfigureDurableOptions` +- Automatic discovery and registration of sub-workflows + +## Overview + +The workflow implements an order processing pipeline where each stage is a separate sub-workflow: + +``` +OrderReceived + │ + ▼ +┌─────────────────────────────────┐ +│ Payment (Sub-Workflow) │ +│ ValidatePayment │ +│ ▼ │ +│ ┌────────────────────────────┐ │ +│ │ FraudCheck (Sub-Sub) │ │ +│ │ AnalyzePatterns │ │ +│ │ ▼ │ │ +│ │ CalculateRiskScore │ │ +│ └────────────────────────────┘ │ +│ ▼ │ +│ ChargePayment │ +└─────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────┐ +│ Inventory (Sub-Workflow) │ +│ CheckInventory ──► Reserve │ +└─────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────┐ +│ Shipping (Sub-Workflow) │ +│ SelectCarrier ──► CreateShip. │ +└─────────────────────────────────┘ + │ + ▼ +OrderCompleted +``` + +| Executor | Sub-Workflow | Description | +|----------|-------------|-------------| +| OrderReceived | Main | Receives order ID and creates OrderInfo | +| ValidatePayment | Payment | Validates payment information | +| AnalyzePatterns | FraudCheck (nested in Payment) | Analyzes transaction patterns | +| CalculateRiskScore | FraudCheck (nested in Payment) | Calculates fraud risk score | +| ChargePayment | Payment | Charges the payment | +| CheckInventory | Inventory | Checks item availability | +| ReserveInventory | Inventory | Reserves inventory items | +| SelectCarrier | Shipping | Selects shipping carrier | +| CreateShipment | Shipping | Creates shipment with tracking | +| OrderCompleted | Main | Outputs completed order summary | + +## How Nested Workflows Work + +Sub-workflows are created by binding a workflow as an executor: + +```csharp +// Build a sub-workflow +Workflow paymentWorkflow = new WorkflowBuilder(validatePayment) + .WithName("SubPaymentProcessing") + .AddEdge(validatePayment, fraudCheckExecutor) + .AddEdge(fraudCheckExecutor, chargePayment) + .Build(); + +// Bind it as an executor in the parent workflow +ExecutorBinding paymentExecutor = paymentWorkflow.BindAsExecutor("Payment"); + +// Use it like any other executor +Workflow mainWorkflow = new WorkflowBuilder(orderReceived) + .AddEdge(orderReceived, paymentExecutor) + .Build(); +``` + +Each sub-workflow runs as a separate orchestration instance, providing: +- **Modularity** — workflows can be composed from reusable sub-workflows +- **Independent checkpointing** — each sub-workflow has its own replay history +- **Hierarchical visualization** — view parent-child relationships in the DTS dashboard +- **Failure isolation** — sub-workflow failures don't corrupt parent state + +## Environment Setup + +This sample requires: + +- [.NET 10 SDK](https://dotnet.microsoft.com/download) +- [Azure Functions Core Tools v4](https://learn.microsoft.com/azure/azure-functions/functions-run-local) +- [Durable Task Scheduler](https://learn.microsoft.com/azure/azure-functions/durable/durable-functions-azure-managed-storage) running locally (default: `http://localhost:8080`) +- [Azurite](https://learn.microsoft.com/azure/storage/common/storage-use-azurite) for local Azure Storage emulation + +## Running the Sample + +```bash +cd dotnet/samples/Durable/Workflow/AzureFunctions/04_NestedWorkflows +func start +``` + +### Testing + +```bash +curl -X POST http://localhost:7071/api/workflows/OrderProcessing/run -H "Content-Type: text/plain" -d "ORD-2026-001" +``` + +Open the DTS dashboard at `http://localhost:8080` to see the parent-child orchestration hierarchy in the Timeline view. diff --git a/dotnet/samples/Durable/Workflow/AzureFunctions/04_NestedWorkflows/demo.http b/dotnet/samples/Durable/Workflow/AzureFunctions/04_NestedWorkflows/demo.http new file mode 100644 index 0000000000..6cb4223406 --- /dev/null +++ b/dotnet/samples/Durable/Workflow/AzureFunctions/04_NestedWorkflows/demo.http @@ -0,0 +1,14 @@ +# Default endpoint address for local testing +@authority=http://localhost:7071 + +### Process an order through nested workflows (Payment → Inventory → Shipping) +POST {{authority}}/api/workflows/OrderProcessing/run +Content-Type: text/plain + +ORD-2026-001 + +### Process another order +POST {{authority}}/api/workflows/OrderProcessing/run +Content-Type: text/plain + +ORD-2026-002 diff --git a/dotnet/samples/Durable/Workflow/AzureFunctions/04_NestedWorkflows/host.json b/dotnet/samples/Durable/Workflow/AzureFunctions/04_NestedWorkflows/host.json new file mode 100644 index 0000000000..9384a0a583 --- /dev/null +++ b/dotnet/samples/Durable/Workflow/AzureFunctions/04_NestedWorkflows/host.json @@ -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" + } + } + } +} diff --git a/dotnet/samples/Durable/Workflow/ConsoleApps/05_NestedWorkflows/05_NestedWorkflows.csproj b/dotnet/samples/Durable/Workflow/ConsoleApps/05_NestedWorkflows/05_NestedWorkflows.csproj new file mode 100644 index 0000000000..218b4dfe9a --- /dev/null +++ b/dotnet/samples/Durable/Workflow/ConsoleApps/05_NestedWorkflows/05_NestedWorkflows.csproj @@ -0,0 +1,29 @@ + + + net10.0 + Exe + enable + enable + NestedWorkflows + NestedWorkflows + + + + + + + + + + + + + + + + diff --git a/dotnet/samples/Durable/Workflow/ConsoleApps/05_NestedWorkflows/Executors.cs b/dotnet/samples/Durable/Workflow/ConsoleApps/05_NestedWorkflows/Executors.cs new file mode 100644 index 0000000000..e2c142a9dd --- /dev/null +++ b/dotnet/samples/Durable/Workflow/ConsoleApps/05_NestedWorkflows/Executors.cs @@ -0,0 +1,250 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Agents.AI.Workflows; + +namespace NestedWorkflows; + +// ============================================ +// Order Processing Models +// ============================================ + +/// +/// Represents an order being processed through the workflow. +/// +internal sealed class OrderInfo +{ + public required string OrderId { get; set; } + public decimal Amount { get; set; } + public string? PaymentTransactionId { get; set; } + public string? InventoryReservationId { get; set; } + public string? TrackingNumber { get; set; } + public string? Carrier { get; set; } +} + +// ============================================ +// Main Workflow Executors +// ============================================ + +/// +/// Entry point executor that receives the order ID and creates an OrderInfo object. +/// +internal sealed class OrderReceived() : Executor("OrderReceived") +{ + public override ValueTask HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.WriteLine(); + Console.ForegroundColor = ConsoleColor.Cyan; + Console.WriteLine($"[OrderReceived] Processing order '{message}'"); + Console.ResetColor(); + + OrderInfo order = new() + { + OrderId = message, + Amount = 99.99m + }; + + return ValueTask.FromResult(order); + } +} + +/// +/// Final executor that outputs the completed order summary. +/// +internal sealed class OrderCompleted() : Executor("OrderCompleted") +{ + public override ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.WriteLine(); + Console.ForegroundColor = ConsoleColor.Green; + Console.WriteLine("┌─────────────────────────────────────────────────────────────────┐"); + Console.WriteLine($"│ [OrderCompleted] Order '{message.OrderId}' successfully processed!"); + Console.WriteLine($"│ Payment: {message.PaymentTransactionId}"); + Console.WriteLine($"│ Inventory: {message.InventoryReservationId}"); + Console.WriteLine($"│ Shipping: {message.Carrier} - {message.TrackingNumber}"); + Console.WriteLine("└─────────────────────────────────────────────────────────────────┘"); + Console.ResetColor(); + + return ValueTask.FromResult($"Order {message.OrderId} completed. Tracking: {message.TrackingNumber}"); + } +} + +// ============================================ +// Payment Sub-Workflow Executors +// ============================================ + +/// +/// Validates payment information for an order. +/// +internal sealed class ValidatePayment() : Executor("ValidatePayment") +{ + public override async ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.WriteLine(); + Console.ForegroundColor = ConsoleColor.Yellow; + Console.WriteLine($" [Payment/ValidatePayment] Validating payment for order '{message.OrderId}'..."); + + await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken); + + Console.WriteLine($" [Payment/ValidatePayment] Payment validated for ${message.Amount}"); + Console.ResetColor(); + + return message; + } +} + +// ============================================ +// Fraud Check Sub-Sub-Workflow Executors (Level 2 nesting) +// ============================================ + +/// +/// Analyzes transaction patterns for potential fraud. +/// +internal sealed class AnalyzePatterns() : Executor("AnalyzePatterns") +{ + public override async ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.ForegroundColor = ConsoleColor.DarkYellow; + Console.WriteLine($" [Payment/FraudCheck/AnalyzePatterns] Analyzing patterns for order '{message.OrderId}'..."); + + await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken); + + Console.WriteLine(" [Payment/FraudCheck/AnalyzePatterns] ✓ Pattern analysis complete"); + Console.ResetColor(); + + return message; + } +} + +/// +/// Calculates a risk score for the transaction. +/// +internal sealed class CalculateRiskScore() : Executor("CalculateRiskScore") +{ + public override async ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.ForegroundColor = ConsoleColor.DarkYellow; + Console.WriteLine($" [Payment/FraudCheck/CalculateRiskScore] Calculating risk score for order '{message.OrderId}'..."); + + await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken); + + int riskScore = new Random().Next(1, 100); + + Console.WriteLine($" [Payment/FraudCheck/CalculateRiskScore] ✓ Risk score: {riskScore}/100 (Low risk)"); + Console.ResetColor(); + + return message; + } +} + +/// +/// Charges the payment for an order. +/// +internal sealed class ChargePayment() : Executor("ChargePayment") +{ + public override async ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.ForegroundColor = ConsoleColor.Yellow; + Console.WriteLine($" [Payment/ChargePayment] Charging ${message.Amount} for order '{message.OrderId}'..."); + + await Task.Delay(TimeSpan.FromSeconds(2), cancellationToken); + + message.PaymentTransactionId = $"TXN-{Guid.NewGuid().ToString("N")[..8].ToUpperInvariant()}"; + + Console.WriteLine($" [Payment/ChargePayment] ✓ Payment processed: {message.PaymentTransactionId}"); + Console.ResetColor(); + + return message; + } +} + +// ============================================ +// Inventory Sub-Workflow Executors +// ============================================ + +/// +/// Checks inventory availability for an order. +/// +internal sealed class CheckInventory() : Executor("CheckInventory") +{ + public override async ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.WriteLine(); + Console.ForegroundColor = ConsoleColor.Magenta; + Console.WriteLine($" [Inventory/CheckInventory] Checking inventory for order '{message.OrderId}'..."); + + await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken); + + Console.WriteLine(" [Inventory/CheckInventory] ✓ Items available in stock"); + Console.ResetColor(); + + return message; + } +} + +/// +/// Reserves inventory for an order. +/// +internal sealed class ReserveInventory() : Executor("ReserveInventory") +{ + public override async ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.ForegroundColor = ConsoleColor.Magenta; + Console.WriteLine($" [Inventory/ReserveInventory] Reserving items for order '{message.OrderId}'..."); + + await Task.Delay(TimeSpan.FromSeconds(2), cancellationToken); + + message.InventoryReservationId = $"RES-{Guid.NewGuid().ToString("N")[..8].ToUpperInvariant()}"; + + Console.WriteLine($" [Inventory/ReserveInventory] ✓ Reserved: {message.InventoryReservationId}"); + Console.ResetColor(); + + return message; + } +} + +// ============================================ +// Shipping Sub-Workflow Executors +// ============================================ + +/// +/// Selects a shipping carrier for an order. +/// +internal sealed class SelectCarrier() : Executor("SelectCarrier") +{ + public override async ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.WriteLine(); + Console.ForegroundColor = ConsoleColor.Blue; + Console.WriteLine($" [Shipping/SelectCarrier] Selecting carrier for order '{message.OrderId}'..."); + + await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken); + + message.Carrier = message.Amount > 50 ? "Express" : "Standard"; + + Console.WriteLine($" [Shipping/SelectCarrier] ✓ Selected carrier: {message.Carrier}"); + Console.ResetColor(); + + return message; + } +} + +/// +/// Creates shipment and generates tracking number. +/// +internal sealed class CreateShipment() : Executor("CreateShipment") +{ + public override async ValueTask HandleAsync(OrderInfo message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.ForegroundColor = ConsoleColor.Blue; + Console.WriteLine($" [Shipping/CreateShipment] Creating shipment for order '{message.OrderId}'..."); + + await Task.Delay(TimeSpan.FromSeconds(2), cancellationToken); + + message.TrackingNumber = $"TRACK-{Guid.NewGuid().ToString("N")[..10].ToUpperInvariant()}"; + + Console.WriteLine($" [Shipping/CreateShipment] ✓ Shipment created: {message.TrackingNumber}"); + Console.ResetColor(); + + return message; + } +} diff --git a/dotnet/samples/Durable/Workflow/ConsoleApps/05_NestedWorkflows/Program.cs b/dotnet/samples/Durable/Workflow/ConsoleApps/05_NestedWorkflows/Program.cs new file mode 100644 index 0000000000..14561a5bc5 --- /dev/null +++ b/dotnet/samples/Durable/Workflow/ConsoleApps/05_NestedWorkflows/Program.cs @@ -0,0 +1,228 @@ +// Copyright (c) Microsoft. All rights reserved. + +// This sample demonstrates nested workflows (sub-workflows) where one workflow +// can be used as an executor within another workflow. This enables modular, +// reusable workflow components with independent checkpointing and replay. +// +// Workflow structure: +// +// ┌─────────────────────────────────────────────────────────────────────────┐ +// │ OrderProcessing (Main Workflow) │ +// │ │ +// │ OrderReceived │ +// │ │ │ +// │ ▼ │ +// │ ┌─────────────────────────────────────────────────────────┐ │ +// │ │ Payment (Sub-Workflow) │ │ +// │ │ │ │ +// │ │ ValidatePayment │ │ +// │ │ │ │ │ +// │ │ ▼ │ │ +// │ │ ┌─────────────────────────────────────────┐ │ │ +// │ │ │ FraudCheck (Sub-Sub-Workflow) │ │ │ +// │ │ │ │ │ │ +// │ │ │ AnalyzePatterns ──► CalculateRiskScore │ │ │ +// │ │ └─────────────────────────────────────────┘ │ │ +// │ │ │ │ │ +// │ │ ▼ │ │ +// │ │ ChargePayment │ │ +// │ └─────────────────────────────────────────────────────────┘ │ +// │ │ │ +// │ ▼ │ +// │ ┌─────────────────────────────────────────────────────────┐ │ +// │ │ Inventory (Sub-Workflow) │ │ +// │ │ │ │ +// │ │ CheckInventory ──► ReserveInventory │ │ +// │ └─────────────────────────────────────────────────────────┘ │ +// │ │ │ +// │ ▼ │ +// │ ┌─────────────────────────────────────────────────────────┐ │ +// │ │ Shipping (Sub-Workflow) │ │ +// │ │ │ │ +// │ │ SelectCarrier ──► CreateShipment │ │ +// │ └─────────────────────────────────────────────────────────┘ │ +// │ │ │ +// │ ▼ │ +// │ OrderCompleted │ +// └─────────────────────────────────────────────────────────────────────────┘ +// +// Each sub-workflow runs as a separate orchestration instance, providing: +// - Modular, reusable workflow components +// - Independent checkpointing and replay +// - Hierarchical visualization in the DTS dashboard +// - Failure isolation between parent and child workflows + +using Microsoft.Agents.AI.DurableTask; +using Microsoft.Agents.AI.DurableTask.Workflows; +using Microsoft.Agents.AI.Workflows; +using Microsoft.DurableTask.Client.AzureManaged; +using Microsoft.DurableTask.Worker.AzureManaged; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using NestedWorkflows; + +string dtsConnectionString = Environment.GetEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING") + ?? "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None"; + +// ============================================ +// Step 1: Build the Fraud Check sub-sub-workflow (Level 2 nesting) +// ============================================ +AnalyzePatterns analyzePatterns = new(); +CalculateRiskScore calculateRiskScore = new(); + +Workflow fraudCheckWorkflow = new WorkflowBuilder(analyzePatterns) + .WithName("SubFraudCheck") + .WithDescription("Analyzes transaction patterns and calculates risk score") + .AddEdge(analyzePatterns, calculateRiskScore) + .Build(); + +// ============================================ +// Step 2: Build the Payment Processing sub-workflow (with nested FraudCheck) +// ============================================ +ValidatePayment validatePayment = new(); +ExecutorBinding fraudCheckExecutor = fraudCheckWorkflow.BindAsExecutor("FraudCheck"); +ChargePayment chargePayment = new(); + +Workflow paymentWorkflow = new WorkflowBuilder(validatePayment) + .WithName("SubPaymentProcessing") + .WithDescription("Validates and processes payment for an order") + .AddEdge(validatePayment, fraudCheckExecutor) + .AddEdge(fraudCheckExecutor, chargePayment) + .Build(); + +// ============================================ +// Step 3: Build the Inventory Management sub-workflow +// ============================================ +CheckInventory checkInventory = new(); +ReserveInventory reserveInventory = new(); + +Workflow inventoryWorkflow = new WorkflowBuilder(checkInventory) + .WithName("SubInventoryManagement") + .WithDescription("Checks availability and reserves inventory") + .AddEdge(checkInventory, reserveInventory) + .Build(); + +// ============================================ +// Step 4: Build the Shipping Arrangement sub-workflow +// ============================================ +SelectCarrier selectCarrier = new(); +CreateShipment createShipment = new(); + +Workflow shippingWorkflow = new WorkflowBuilder(selectCarrier) + .WithName("SubShippingArrangement") + .WithDescription("Selects carrier and creates shipment") + .AddEdge(selectCarrier, createShipment) + .Build(); + +// ============================================ +// Step 5: Build the Main Order Processing workflow using sub-workflows +// ============================================ +ExecutorBinding paymentExecutor = paymentWorkflow.BindAsExecutor("Payment"); +ExecutorBinding inventoryExecutor = inventoryWorkflow.BindAsExecutor("Inventory"); +ExecutorBinding shippingExecutor = shippingWorkflow.BindAsExecutor("Shipping"); + +OrderReceived orderReceived = new(); +OrderCompleted orderCompleted = new(); + +// Main workflow: OrderReceived -> Payment -> Inventory -> Shipping -> OrderCompleted +Workflow orderProcessingWorkflow = new WorkflowBuilder(orderReceived) + .WithName("OrderProcessing") + .WithDescription("Processes an order through payment, inventory, and shipping") + .AddEdge(orderReceived, paymentExecutor) + .AddEdge(paymentExecutor, inventoryExecutor) + .AddEdge(inventoryExecutor, shippingExecutor) + .AddEdge(shippingExecutor, orderCompleted) + .Build(); + +// ============================================ +// Step 6: Configure and start the host +// ============================================ +IHost host = Host.CreateDefaultBuilder(args) + .ConfigureLogging(logging => logging.SetMinimumLevel(LogLevel.Warning)) + .ConfigureServices(services => + { + // Register only the main workflow - sub-workflows are discovered automatically! + services.ConfigureDurableWorkflows( + workflowOptions => workflowOptions.AddWorkflow(orderProcessingWorkflow), + workerBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString), + clientBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString)); + }) + .Build(); + +await host.StartAsync(); + +IWorkflowClient workflowClient = host.Services.GetRequiredService(); + +Console.WriteLine("╔══════════════════════════════════════════════════════════════════╗"); +Console.WriteLine("║ Nested Workflows Sample ║"); +Console.WriteLine("╠══════════════════════════════════════════════════════════════════╣"); +Console.WriteLine("║ Main Workflow: OrderProcessing ║"); +Console.WriteLine("║ ├── Payment (sub-workflow) ║"); +Console.WriteLine("║ │ ├── ValidatePayment ║"); +Console.WriteLine("║ │ ├── FraudCheck (sub-sub-workflow) ← Level 2 nesting! ║"); +Console.WriteLine("║ │ │ ├── AnalyzePatterns ║"); +Console.WriteLine("║ │ │ └── CalculateRiskScore ║"); +Console.WriteLine("║ │ └── ChargePayment ║"); +Console.WriteLine("║ ├── Inventory (sub-workflow) ║"); +Console.WriteLine("║ │ ├── CheckInventory ║"); +Console.WriteLine("║ │ └── ReserveInventory ║"); +Console.WriteLine("║ └── Shipping (sub-workflow) ║"); +Console.WriteLine("║ ├── SelectCarrier ║"); +Console.WriteLine("║ └── CreateShipment ║"); +Console.WriteLine("╚══════════════════════════════════════════════════════════════════╝"); +Console.WriteLine(); +Console.WriteLine("Open the DTS dashboard at http://localhost:8080 to see the"); +Console.WriteLine("parent-child orchestration hierarchy in the Timeline view!"); +Console.WriteLine(); +Console.WriteLine("Enter an order ID (or 'exit'):"); + +while (true) +{ + Console.Write("> "); + string? input = Console.ReadLine(); + if (string.IsNullOrWhiteSpace(input) || input.Equals("exit", StringComparison.OrdinalIgnoreCase)) + { + break; + } + + try + { + await StartNewWorkflowAsync(input, orderProcessingWorkflow, workflowClient); + } + catch (Exception ex) + { + Console.WriteLine($"Error: {ex.Message}"); + } + + Console.WriteLine(); +} + +await host.StopAsync(); + +// Start a new workflow and wait for completion +static async Task StartNewWorkflowAsync(string orderId, Workflow workflow, IWorkflowClient client) +{ + Console.WriteLine($"\nStarting order processing for '{orderId}'..."); + + IAwaitableWorkflowRun run = (IAwaitableWorkflowRun)await client.RunAsync(workflow, orderId); + Console.WriteLine($"Run ID: {run.RunId}"); + Console.WriteLine("Check the DTS dashboard Timeline tab to see sub-orchestrations!"); + Console.WriteLine(); + + try + { + Console.WriteLine("Waiting for workflow to complete..."); + string? result = await run.WaitForCompletionAsync(); + + Console.ForegroundColor = ConsoleColor.Green; + Console.WriteLine($"✓ Order completed: {result}"); + Console.ResetColor(); + } + catch (InvalidOperationException ex) + { + Console.ForegroundColor = ConsoleColor.Red; + Console.WriteLine($"✗ Failed: {ex.Message}"); + Console.ResetColor(); + } +} diff --git a/dotnet/samples/Durable/Workflow/ConsoleApps/05_NestedWorkflows/README.md b/dotnet/samples/Durable/Workflow/ConsoleApps/05_NestedWorkflows/README.md new file mode 100644 index 0000000000..dda2e351d4 --- /dev/null +++ b/dotnet/samples/Durable/Workflow/ConsoleApps/05_NestedWorkflows/README.md @@ -0,0 +1,95 @@ +# Nested Workflows Sample + +This sample demonstrates how to use **nested workflows** (sub-workflows) where one workflow can be used as an executor within another workflow. This enables modular, reusable workflow components with independent checkpointing and replay. + +## Key Concepts Demonstrated + +- Using `Workflow.BindAsExecutor()` to embed a workflow as an executor in another workflow +- Multi-level nesting (sub-workflow within a sub-workflow) +- Automatic discovery of sub-workflows when registering the main workflow +- Independent orchestration instances for each sub-workflow + +## Overview + +The sample implements an order processing workflow with three sub-workflows and one sub-sub-workflow: + +``` +OrderProcessing (Main Workflow) +├── OrderReceived +├── Payment (sub-workflow) +│ ├── ValidatePayment +│ ├── FraudCheck (sub-sub-workflow) ← Level 2 nesting! +│ │ ├── AnalyzePatterns +│ │ └── CalculateRiskScore +│ └── ChargePayment +├── Inventory (sub-workflow) +│ ├── CheckInventory +│ └── ReserveInventory +├── Shipping (sub-workflow) +│ ├── SelectCarrier +│ └── CreateShipment +└── OrderCompleted +``` + +## How Nested Workflows Work + +Sub-workflows are created using `BindAsExecutor()`, which converts a `Workflow` into an `ExecutorBinding` that can be used in another workflow's graph: + +```csharp +// Build a sub-workflow +Workflow paymentWorkflow = new WorkflowBuilder(validatePayment) + .WithName("SubPaymentProcessing") + .AddEdge(validatePayment, fraudCheckExecutor) + .AddEdge(fraudCheckExecutor, chargePayment) + .Build(); + +// Bind as executor for use in the parent workflow +ExecutorBinding paymentExecutor = paymentWorkflow.BindAsExecutor("Payment"); + +// Use in the main workflow +Workflow mainWorkflow = new WorkflowBuilder(orderReceived) + .AddEdge(orderReceived, paymentExecutor) + .Build(); +``` + +Each sub-workflow runs as a separate orchestration instance in the Durable Task Scheduler, visible in the DTS dashboard Timeline view. + +## Environment Setup + +See the [README.md](../README.md) file in the parent directory for information on configuring the environment, including how to install and run the Durable Task Scheduler. + +## Running the Sample + +```bash +cd dotnet/samples/Durable/Workflow/ConsoleApps/05_NestedWorkflows +dotnet run --framework net10.0 +``` + +### Sample Output + +```text +╔══════════════════════════════════════════════════════════════════╗ +║ Nested Workflows Sample ║ +╠══════════════════════════════════════════════════════════════════╣ +║ Main Workflow: OrderProcessing ║ +║ ├── Payment (sub-workflow) ║ +║ │ ├── ValidatePayment ║ +║ │ ├── FraudCheck (sub-sub-workflow) ← Level 2 nesting! ║ +║ │ │ ├── AnalyzePatterns ║ +║ │ │ └── CalculateRiskScore ║ +║ │ └── ChargePayment ║ +║ ├── Inventory (sub-workflow) ║ +║ │ ├── CheckInventory ║ +║ │ └── ReserveInventory ║ +║ └── Shipping (sub-workflow) ║ +║ ├── SelectCarrier ║ +║ └── CreateShipment ║ +╚══════════════════════════════════════════════════════════════════╝ + +Enter an order ID (or 'exit'): +> ORD-001 +Starting order processing for 'ORD-001'... +Run ID: abc123... +Waiting for workflow to complete... +✓ Order completed: Order ORD-001 completed. Tracking: TRACK-1A2B3C4D5E +``` diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/ServiceCollectionExtensions.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/ServiceCollectionExtensions.cs index da763488c5..c7d8cc5a85 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/ServiceCollectionExtensions.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/ServiceCollectionExtensions.cs @@ -197,24 +197,33 @@ 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 => + + if (workerBuilder is not null) { - workerBuilder?.Invoke(builder); - - builder.AddTasks(registry => RegisterTasksFromOptions(registry, sharedOptions)); - }); + services.AddDurableTaskWorker(builder => + { + workerBuilder?.Invoke(builder); + builder.AddTasks(registry => RegisterTasksFromOptions(registry, sharedOptions)); + }); + } // Configure Durable Task Client + // Only register a client if explicitly configured. For Azure Functions, + // the Functions extension provides the DurableTaskClient automatically via bindings. if (clientBuilder is not null) { services.AddDurableTaskClient(clientBuilder); + + // These services depend on DurableTaskClient from DI, so only register them + // when we're registering a client. For Azure Functions, the client comes from + // bindings, not DI, so these won't work there. + services.TryAddSingleton(); + services.TryAddSingleton(sp => sp.GetRequiredService()); + services.TryAddSingleton(); } - // Register workflow and agent services - services.TryAddSingleton(); - services.TryAddSingleton(sp => sp.GetRequiredService()); + // Register workflow and agent services that don't depend on DurableTaskClient services.TryAddSingleton(); - services.TryAddSingleton(); // Register agent factories resolver - returns factories from the shared options services.TryAddSingleton( @@ -257,7 +266,11 @@ public static class ServiceCollectionExtensions ExecutorBinding binding = activity.Binding; registry.AddActivityFunc( activity.ActivityName, - (context, input) => DurableActivityExecutor.ExecuteAsync(binding, input)); + (context, input) => + { + // to do:a + return DurableActivityExecutor.ExecuteAsync(binding, input); + }); } } diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableExecutorDispatcher.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableExecutorDispatcher.cs index 903b2fb127..bd8e398ec3 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableExecutorDispatcher.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableExecutorDispatcher.cs @@ -14,19 +14,20 @@ using Microsoft.Extensions.Logging; namespace Microsoft.Agents.AI.DurableTask.Workflows; /// -/// Dispatches workflow executors to either activities or AI agents. +/// Dispatches workflow executors to activities, AI agents, or sub-workflow orchestrations. /// /// /// Called during the dispatch phase of each superstep by /// DurableWorkflowRunner.DispatchExecutorsInParallelAsync. For each executor that has /// pending input, this dispatcher determines whether the executor is an AI agent (stateful, -/// backed by Durable Entities) or a regular activity, and invokes the appropriate Durable Task API. +/// backed by Durable Entities), a sub-workflow (child orchestration), or a regular activity, +/// and invokes the appropriate Durable Task API. /// The serialised string result is returned to the runner for the routing phase. /// internal static class DurableExecutorDispatcher { /// - /// Dispatches an executor based on its type (activity or AI agent). + /// Dispatches an executor based on its type (activity, AI agent, or sub-workflow). /// /// The task orchestration context. /// Information about the executor to dispatch. @@ -46,6 +47,11 @@ internal static class DurableExecutorDispatcher return await ExecuteAgentAsync(context, executorInfo, logger, envelope.Message).ConfigureAwait(true); } + if (executorInfo.IsSubworkflowExecutor) + { + return await ExecuteSubWorkflowAsync(context, executorInfo, envelope.Message).ConfigureAwait(true); + } + return await ExecuteActivityAsync(context, executorInfo, envelope.Message, envelope.InputTypeName).ConfigureAwait(true); } @@ -96,4 +102,29 @@ internal static class DurableExecutorDispatcher return response.Text; } + + /// + /// Executes a sub-workflow as a child orchestration. + /// + /// + /// The input is passed as a raw JSON object (not a string) to avoid double-encoding. + /// + private static async Task ExecuteSubWorkflowAsync( + TaskOrchestrationContext context, + WorkflowExecutorInfo executorInfo, + string input) + { + string orchestrationName = WorkflowNamingHelper.ToOrchestrationFunctionName(executorInfo.SubWorkflow!.Name!); + + // Parse the input JSON to pass as an object, preventing double-encoding. + // The sub-workflow orchestrator receives DurableWorkflowInput where Input + // should be the deserialized object (not a JSON string). + using JsonDocument doc = JsonDocument.Parse(input); + object inputObj = doc.RootElement.Clone(); + DurableWorkflowInput workflowInput = new() { Input = inputObj }; + + return await context.CallSubOrchestratorAsync( + orchestrationName, + workflowInput).ConfigureAwait(true); + } } diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowRunner.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowRunner.cs index 8923b29dc1..fbb7469ae2 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowRunner.cs @@ -67,7 +67,7 @@ namespace Microsoft.Agents.AI.DurableTask.Workflows; /// /// Runs workflow orchestrations using message-driven superstep execution with Durable Task. /// -internal sealed class DurableWorkflowRunner +public class DurableWorkflowRunner { private const int MaxSupersteps = 100; @@ -75,7 +75,7 @@ internal sealed class DurableWorkflowRunner /// Initializes a new instance of the class. /// /// The durable options containing workflow configurations. - internal DurableWorkflowRunner(DurableOptions durableOptions) + public DurableWorkflowRunner(DurableOptions durableOptions) { ArgumentNullException.ThrowIfNull(durableOptions); diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctionExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctionExecutor.cs index fa0b9ef287..c740531abd 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctionExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctionExecutor.cs @@ -102,6 +102,43 @@ internal sealed class BuiltInFunctionExecutor : IFunctionExecutor return; } + if (context.FunctionDefinition.EntryPoint == BuiltInFunctions.RunWorkflowOrechstrtationHttpFunctionEntryPoint) + { + if (httpRequestData == null) + { + throw new InvalidOperationException($"HTTP request data binding is missing for the invocation {context.InvocationId}."); + } + context.GetInvocationResult().Value = await BuiltInFunctions.RunWorkflowOrechstrtationHttpTriggerAsync( + httpRequestData, + durableTaskClient, + context); + + return; + } + + if (context.FunctionDefinition.EntryPoint == BuiltInFunctions.InvokeWorkflowOrchestrationFunctionEntryPoint) + { + // The orchestration trigger binding provides an encoded orchestration request string. + // We use the same string binding that entities use (encodedEntityRequest is reused for orchestrations). + if (encodedEntityRequest is null) + { + throw new InvalidOperationException($"Orchestration trigger binding is missing for the invocation {context.InvocationId}."); + } + + // Execute the orchestration using the static method in BuiltInFunctions + context.GetInvocationResult().Value = BuiltInFunctions.InvokeWorkflowOrchestration( + encodedEntityRequest, + context); + return; + } + + if (context.FunctionDefinition.EntryPoint == BuiltInFunctions.InvokeWorkflowActivityFunctionEntryPoint) + { + // to do + context.GetInvocationResult().Value = await BuiltInFunctions.InvokeWorkflowActivityAsync(encodedEntityRequest!, durableTaskClient, context); + return; + } + throw new InvalidOperationException($"Unsupported function entry point '{context.FunctionDefinition.EntryPoint}' for invocation {context.InvocationId}."); } } diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctions.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctions.cs index 8573a80613..a67b92b454 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctions.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctions.cs @@ -3,13 +3,16 @@ 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; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; namespace Microsoft.Agents.AI.Hosting.AzureFunctions; @@ -22,6 +25,120 @@ internal static class BuiltInFunctions internal static readonly string RunAgentEntityFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(InvokeAgentAsync)}"; internal static readonly string RunAgentMcpToolFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(RunMcpToolAsync)}"; + internal static readonly string RunWorkflowOrechstrtationHttpFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(RunWorkflowOrechstrtationHttpTriggerAsync)}"; + internal static readonly string InvokeWorkflowOrchestrationFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(InvokeWorkflowOrchestration)}"; + + internal static readonly string InvokeWorkflowActivityFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(InvokeWorkflowActivityAsync)}"; + internal static readonly string RunWorkflowMcpToolFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(RunWorkflowMcpToolAsync)}"; + +#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 + + /// + /// Invokes a workflow orchestration using the encoded orchestration request. + /// + /// The base64-encoded protobuf payload for the orchestration. + /// The function context. + /// The encoded orchestration response. + internal static string InvokeWorkflowOrchestration( + string encodedOrchestratorRequest, + FunctionContext functionContext) + { + DurableOptions durableOptions = functionContext.InstanceServices.GetRequiredService(); + + return GrpcOrchestrationRunner.LoadAndRun, string>( + encodedOrchestratorRequest, + (orchestrationContext, input) => RunWorkflowOrchestrationCoreAsync(orchestrationContext, input, durableOptions)!, + functionContext.InstanceServices); + } + + private static async Task RunWorkflowOrchestrationCoreAsync( + TaskOrchestrationContext orchestrationContext, + DurableWorkflowInput? input, + DurableOptions durableOptions) + { + ILogger logger = orchestrationContext.CreateReplaySafeLogger("DurableWorkflow"); + DurableWorkflowRunner runner = new(durableOptions); + + // ConfigureAwait(true) is required in orchestration code for deterministic replay. + return await runner.RunWorkflowOrchestrationAsync( + orchestrationContext, + input ?? new DurableWorkflowInput { Input = string.Empty }, + logger).ConfigureAwait(true); + } + + /// + /// Invokes a workflow orchestration in response to an HTTP request. + /// + public static async Task RunWorkflowOrechstrtationHttpTriggerAsync( + [HttpTrigger] HttpRequestData req, + [DurableClient] DurableTaskClient client, + FunctionContext context) + { + var workflowName = context.FunctionDefinition.Name.Replace(HttpPrefix, string.Empty); + var orchestrationFunctionName = WorkflowNamingHelper.ToOrchestrationFunctionName(workflowName); + var inputMessage = await req.ReadAsStringAsync(); + + DurableWorkflowInput orchestrtionInput = new() { Input = inputMessage! }; + string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(orchestrationFunctionName, orchestrtionInput); + + HttpResponseData response = req.CreateResponse(HttpStatusCode.Accepted); + await response.WriteStringAsync($"InvokeWorkflowOrechstrtationAsync is invoked for {workflowName}. Orchestration instanceId: {instanceId}"); + return response; + } + + public static Task 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(); + 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); + } + + public static async Task RunWorkflowMcpToolAsync( + [McpToolTrigger("BuiltInWorkflowMcpTool")] ToolInvocationContext context, + [DurableClient] DurableTaskClient client, + FunctionContext functionContext) + { + if (context.Arguments is null) + { + throw new ArgumentException("MCP Tool invocation is missing required arguments."); + } + + if (!context.Arguments.TryGetValue("input", out object? inputObj) || inputObj is not string input) + { + throw new ArgumentException("MCP Tool invocation is missing required 'input' argument of type string."); + } + + // Extract workflow name from the MCP tool name (format: mcptool-workflow-{workflowName}) + string workflowName = context.Name; + string orchestrationFunctionName = WorkflowNamingHelper.ToOrchestrationFunctionName(workflowName); + + string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(orchestrationFunctionName, input); + + // Wait for the orchestration to complete and return the result + OrchestrationMetadata? metadata = await client.WaitForInstanceCompletionAsync( + instanceId, + getInputsAndOutputs: true, + cancellation: functionContext.CancellationToken); + + return metadata?.ReadOutputAs(); + } + // Exposed as an entity trigger via AgentFunctionsProvider public static Task InvokeAgentAsync( [DurableClient] DurableTaskClient client, diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/FunctionsApplicationBuilderExtensions.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/FunctionsApplicationBuilderExtensions.cs index e13c6008ea..eb549952db 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/FunctionsApplicationBuilderExtensions.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/FunctionsApplicationBuilderExtensions.cs @@ -1,11 +1,18 @@ // Copyright (c) Microsoft. All rights reserved. using Microsoft.Agents.AI.DurableTask; +using Microsoft.Agents.AI.DurableTask.Workflows; +using Microsoft.Agents.AI.Hosting.AzureFunctions.Workflows; +using Microsoft.Agents.AI.Workflows; +using Microsoft.Azure.Functions.Worker; using Microsoft.Azure.Functions.Worker.Builder; using Microsoft.Azure.Functions.Worker.Core.FunctionMetadata; +using Microsoft.DurableTask; +using Microsoft.DurableTask.Worker; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection.Extensions; using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; namespace Microsoft.Agents.AI.Hosting.AzureFunctions; @@ -14,6 +21,153 @@ namespace Microsoft.Agents.AI.Hosting.AzureFunctions; /// public static class FunctionsApplicationBuilderExtensions { + /// + /// Configures durable agents and workflows in a unified way. + /// + /// The Functions application builder. + /// A delegate to configure the durable options. + /// The Functions application builder for method chaining. + /// + /// This method provides a unified configuration point for both durable agents and workflows. + /// It automatically generates HTTP API endpoints for agents and workflows, and configures + /// the necessary middleware and services for durable execution. + /// + public static FunctionsApplicationBuilder ConfigureDurableOptions( + this FunctionsApplicationBuilder builder, + Action configure) + { + ArgumentNullException.ThrowIfNull(builder); + ArgumentNullException.ThrowIfNull(configure); + + DurableOptions options = new(); + configure(options); + + builder.Services.AddSingleton(options); + + if (options.Workflows.Workflows.Count > 0) + { + ConfigureWorkflowOrchestrations(builder, options.Workflows); + // Do things to enable workflow as orchestrator functions. + // Register the Workflow metadata transformer. + builder.ConfigureDurableWorkflows(durableWorkflwoOptions => + { + // what + }); + + builder.Services.AddSingleton(); + + builder.UseWhen(static context => + string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunAgentHttpFunctionEntryPoint, StringComparison.Ordinal) || + string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunAgentMcpToolFunctionEntryPoint, StringComparison.Ordinal) || + string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunAgentEntityFunctionEntryPoint, StringComparison.Ordinal) + || string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunWorkflowOrechstrtationHttpFunctionEntryPoint, StringComparison.Ordinal) + + || string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunWorkflowOrechstrtationHttpFunctionEntryPoint, StringComparison.Ordinal) + || string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.InvokeWorkflowOrchestrationFunctionEntryPoint, StringComparison.Ordinal) + || string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.InvokeWorkflowActivityFunctionEntryPoint, StringComparison.Ordinal) + ); + builder.Services.AddSingleton(); + + //builder.UseWhen(static context => + // string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunWorkflowOrechstrtationHttpFunctionEntryPoint, StringComparison.Ordinal) + // || string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.InvokeWorkflowOrchestrationFunctionEntryPoint, StringComparison.Ordinal) + // || string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.InvokeWorkflowActivityFunctionEntryPoint, StringComparison.Ordinal) + // || string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunAgentHttpFunctionEntryPoint, StringComparison.Ordinal) + // ); + //builder.Services.AddSingleton(); + } + + return builder; + } + + private static void ConfigureWorkflowOrchestrations(FunctionsApplicationBuilder builder, DurableWorkflowOptions workflows) + { + // Discover sub-workflows recursively and add them to the workflows dictionary + // so they are registered as separate orchestrations alongside the main workflows. + DiscoverSubWorkflows(workflows); + + builder.ConfigureDurableWorker().AddTasks(tasks => + { + foreach (string workflowName in workflows.Workflows.Select(kp => kp.Key)) + { + string orchestrationFunctionName = WorkflowNamingHelper.ToOrchestrationFunctionName(workflowName); + + tasks.AddOrchestratorFunc, string>( + orchestrationFunctionName, + async (orchestrationContext, orchInput) => + { + FunctionContext functionContext = orchestrationContext.GetFunctionContext() + ?? throw new InvalidOperationException("FunctionContext is not available in the orchestration context."); + + DurableWorkflowRunner runner = functionContext.InstanceServices.GetRequiredService(); + ILogger logger = orchestrationContext.CreateReplaySafeLogger(orchestrationFunctionName); + DurableWorkflowInput workflowInput = orchInput; + + return await runner.RunWorkflowOrchestrationAsync(orchestrationContext, workflowInput, logger).ConfigureAwait(true); + }); + } + }); + } + + private static void DiscoverSubWorkflows(DurableWorkflowOptions workflows) + { + HashSet visited = new(workflows.Workflows.Keys); + Queue queue = new(workflows.Workflows.Values); + + while (queue.Count > 0) + { + Workflow workflow = queue.Dequeue(); + + foreach (ExecutorBinding binding in workflow.ReflectExecutors().Values) + { + if (binding is SubworkflowBinding subworkflowBinding) + { + Workflow subWorkflow = subworkflowBinding.WorkflowInstance; + if (subWorkflow.Name is not null && visited.Add(subWorkflow.Name)) + { + workflows.AddWorkflow(subWorkflow); + queue.Enqueue(subWorkflow); + } + } + } + } + } + internal static FunctionsApplicationBuilder RegisterWorkflowServices(this FunctionsApplicationBuilder builder) + { + // Register FunctionsWorkflowRunner as a singleton + // builder.Services.TryAddSingleton(); + + // Also register it as DurableWorkflowRunner so orchestrations can resolve it by base type + //builder.Services.TryAddSingleton(sp => sp.GetRequiredService()); + + builder.Services.TryAddEnumerable(ServiceDescriptor.Singleton()); + + return builder; + } + + /// + /// Configures durable workflow services for the application and allows customization of durable workflow options. + /// + /// This method registers the services required for durable workflows using + /// Microsoft.DurableTask.Workflows. Call this method during application startup to enable durable workflows in your + /// Azure Functions app. + /// The application builder used to configure services and middleware for the Azure Functions app. + /// A delegate that is used to configure the durable workflow options. Cannot be null. + /// The same instance that this method was called on, to support method + /// chaining. + public static FunctionsApplicationBuilder ConfigureDurableWorkflows(this FunctionsApplicationBuilder builder, Action configure) + { + ArgumentNullException.ThrowIfNull(configure); + + //RegisterWorkflowServices(builder); + //builder.Services.AddSingleton(); + + // The main durable workflows services registration is done in Microsoft.DurableTask.Workflows. + builder.Services.ConfigureDurableWorkflows(configure); + + return builder; + } + /// /// Configures the application to use durable agents with a builder pattern. /// @@ -38,7 +192,13 @@ public static class FunctionsApplicationBuilderExtensions builder.UseWhen(static context => string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunAgentHttpFunctionEntryPoint, StringComparison.Ordinal) || string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunAgentMcpToolFunctionEntryPoint, StringComparison.Ordinal) || - string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunAgentEntityFunctionEntryPoint, StringComparison.Ordinal)); + string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunAgentEntityFunctionEntryPoint, StringComparison.Ordinal) + || string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunWorkflowOrechstrtationHttpFunctionEntryPoint, StringComparison.Ordinal) + + || string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunWorkflowOrechstrtationHttpFunctionEntryPoint, StringComparison.Ordinal) + || string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.InvokeWorkflowOrchestrationFunctionEntryPoint, StringComparison.Ordinal) + || string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.InvokeWorkflowActivityFunctionEntryPoint, StringComparison.Ordinal) + ); builder.Services.AddSingleton(); return builder; diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/Microsoft.Agents.AI.Hosting.AzureFunctions.csproj b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/Microsoft.Agents.AI.Hosting.AzureFunctions.csproj index ce67c9621e..ed99d264ff 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/Microsoft.Agents.AI.Hosting.AzureFunctions.csproj +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/Microsoft.Agents.AI.Hosting.AzureFunctions.csproj @@ -4,7 +4,10 @@ $(TargetFrameworksCore) enable - $(NoWarn);CA2007 + + $(NoWarn);CA2007;AD0001 + + false diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/Workflows/DurableWorkflowFunctionMetadataTransformer.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/Workflows/DurableWorkflowFunctionMetadataTransformer.cs new file mode 100644 index 0000000000..49a23d0584 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/Workflows/DurableWorkflowFunctionMetadataTransformer.cs @@ -0,0 +1,195 @@ +// 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.Workflows; + +internal sealed class DurableWorkflowFunctionMetadataTransformer : IFunctionMetadataTransformer +{ + private static readonly HashSet _registeredFunctionNames = new(); + private readonly ILogger _logger; + private readonly DurableWorkflowOptions _options; + + public DurableWorkflowFunctionMetadataTransformer(ILogger logger, DurableOptions durableOptions) + { + this._logger = logger ?? throw new ArgumentNullException(nameof(logger)); + ArgumentNullException.ThrowIfNull(durableOptions); + this._options = durableOptions.Workflows; + } + + public string Name => nameof(DurableWorkflowFunctionMetadataTransformer); + + public void Transform(IList original) + { + this._logger.LogTransformStart(original.Count); + + foreach (var workflow in this._options.Workflows) + { + this._logger.LogAddingWorkflowFunction(workflow.Key); + + // Currently due to how durable executor is registered, we are not able to bind TaskOrechestrationContext parameter properly + // because the InputBinding for TOC happens inside the DurableExecutor (rathen than in an input converter). + // So for now, we are going to use single orchestration function for all workflows. + //original.Add(CreateOrchestrationTrigger(workflow.Key)); + + // We also want to create an HTTP trigger for this orchestration so users can start it via HTTP. + this._logger.LogAddingHttpTrigger(workflow.Key); + original.Add(CreateHttpTrigger(workflow.Key, $"workflows/{workflow.Key}/run")); + + // Check if MCP tool trigger is enabled for this workflow + if (DurableWorkflowOptionsExtensions.TryGetWorkflowOptions(workflow.Key, out FunctionsWorkflowOptions? workflowOptions) && + workflowOptions?.McpToolTrigger.IsEnabled == true) + { + this._logger.LogAddingMcpToolTrigger(workflow.Key); + original.Add(CreateMcpToolTrigger(workflow.Key, workflow.Value.Description)); + } + + // Create activity/entity functions for each executor in the workflow based on their type + // Extract executor IDs from edges and start executor + HashSet executorIds = new() { workflow.Value.StartExecutorId }; + + var reflectedEdges = workflow.Value.ReflectEdges(); + foreach (var (sourceId, edgeSet) in reflectedEdges) + { + executorIds.Add(sourceId); + foreach (var edge in edgeSet) + { + foreach (var sinkId in edge.Connection.SinkIds) + { + executorIds.Add(sinkId); + } + } + } + + Dictionary executorBindings = workflow.Value.ReflectExecutors(); + + foreach (string executorId in executorIds) + { + if (executorBindings.TryGetValue(executorId, out ExecutorBinding? executorBinding)) + { + // Sub-workflow bindings are registered as separate orchestrations, not as activities + if (executorBinding is SubworkflowBinding) + { + continue; + } + + string executorName = WorkflowNamingHelper.GetExecutorName(executorId); + string functionName = WorkflowNamingHelper.ToOrchestrationFunctionName(executorName); + + // Skip if this function has already been registered by another workflow + if (!_registeredFunctionNames.Add(functionName)) + { + this._logger.LogSkippingDuplicateFunction(functionName, workflow.Key); + continue; + } + + // Check if the executor type is an agent-related type + if (executorBinding is AIAgentBinding) + { + this._logger.LogAddingAgentEntityFunction(executorId, executorBinding.ExecutorType.FullName ?? executorBinding.ExecutorType.Name, workflow.Key); + original.Add(CreateEntityTrigger(executorName)); + } + else + { + this._logger.LogAddingActivityFunction(executorId, executorBinding.ExecutorType.FullName ?? executorBinding.ExecutorType.Name, workflow.Key); + original.Add(CreateActivityTrigger(functionName)); + } + } + } + } + + this._logger.LogTransformFinished(original.Count); + } + + 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.RunWorkflowOrechstrtationHttpFunctionEntryPoint, + ScriptFile = BuiltInFunctions.ScriptFile + }; + } + + //private static DefaultFunctionMetadata CreateOrchestrationTrigger(string name) + //{ + // return new DefaultFunctionMetadata() + // { + // Name = AgentSessionId.ToEntityName(name), + // Language = "dotnet-isolated", + // RawBindings = + // [ + // // """{"name":"context","type":"orchestrationTrigger","direction":"In"}""", + // """{"name":"taskOrchestrationContext","type":"orchestrationTrigger","direction":"In"}""", + + // ], + // EntryPoint = BuiltInFunctions.RunWorkflowOrechstrtationFunctionEntryPoint, + // ScriptFile = BuiltInFunctions.ScriptFile, + // }; + //} + + //private static DefaultFunctionMetadata CreateOrchestrationFunction(string functionName) + //{ + // throw new NotImplementedException(); + //} + + private 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, + }; + } + + private static DefaultFunctionMetadata CreateEntityTrigger(string functionName) + { + return new DefaultFunctionMetadata() + { + Name = AgentSessionId.ToEntityName(functionName), + Language = "dotnet-isolated", + RawBindings = + [ + """{"name":"encodedEntityRequest","type":"entityTrigger","direction":"In"}""", + """{"name":"client","type":"durableClient","direction":"In"}""" + ], + EntryPoint = BuiltInFunctions.RunAgentEntityFunctionEntryPoint, + ScriptFile = BuiltInFunctions.ScriptFile, + }; + } + + private static DefaultFunctionMetadata CreateMcpToolTrigger(string workflowName, string? description) + { + return new DefaultFunctionMetadata + { + Name = $"{BuiltInFunctions.McpToolPrefix}{workflowName}", + Language = "dotnet-isolated", + RawBindings = + [ + $$"""{"name":"context","type":"mcpToolTrigger","direction":"In","toolName":"{{workflowName}}","description":"{{description ?? $"Run the {workflowName} workflow"}}","toolProperties":"[{\"propertyName\":\"input\",\"propertyType\":\"string\",\"description\":\"The input to the workflow.\",\"isRequired\":true,\"isArray\":false}]"}""", + """{"name":"input","type":"mcpToolProperty","direction":"In","propertyName":"input","description":"The input to the workflow","isRequired":true,"dataType":"String","propertyType":"string"}""", + """{"name":"client","type":"durableClient","direction":"In"}""" + ], + EntryPoint = BuiltInFunctions.RunWorkflowMcpToolFunctionEntryPoint, + ScriptFile = BuiltInFunctions.ScriptFile, + }; + } +} diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/Workflows/DurableWorkflowFunctionMetadataTransformerLogs.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/Workflows/DurableWorkflowFunctionMetadataTransformerLogs.cs new file mode 100644 index 0000000000..af91ddc00b --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/Workflows/DurableWorkflowFunctionMetadataTransformerLogs.cs @@ -0,0 +1,61 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Diagnostics.CodeAnalysis; +using Microsoft.Extensions.Logging; + +namespace Microsoft.Agents.AI.Hosting.AzureFunctions.Workflows; + +/// +/// Logging messages for . +/// +[ExcludeFromCodeCoverage] +internal static partial class DurableWorkflowFunctionMetadataTransformerLogs +{ + [LoggerMessage( + EventId = 200, + Level = LogLevel.Information, + Message = "Transforming function metadata to add durable workflow functions. Initial function count: {FunctionCount}")] + public static partial void LogTransformStart(this ILogger logger, int functionCount); + + [LoggerMessage( + EventId = 201, + Level = LogLevel.Information, + Message = "Adding durable workflow function for workflow: {WorkflowName}")] + public static partial void LogAddingWorkflowFunction(this ILogger logger, string workflowName); + + [LoggerMessage( + EventId = 202, + Level = LogLevel.Information, + Message = "Adding HTTP trigger function for workflow: {WorkflowName}")] + public static partial void LogAddingHttpTrigger(this ILogger logger, string workflowName); + + [LoggerMessage( + EventId = 203, + Level = LogLevel.Information, + Message = "Adding activity function for executor: {ExecutorId} (Type: {ExecutorType}) in workflow: {WorkflowName}")] + public static partial void LogAddingActivityFunction(this ILogger logger, string executorId, string executorType, string workflowName); + + [LoggerMessage( + EventId = 204, + Level = LogLevel.Information, + Message = "Adding agent entity function for executor: {ExecutorId} (Type: {ExecutorType}) in workflow: {WorkflowName}")] + public static partial void LogAddingAgentEntityFunction(this ILogger logger, string executorId, string executorType, string workflowName); + + [LoggerMessage( + EventId = 205, + Level = LogLevel.Information, + Message = "Adding MCP tool trigger function for workflow: {WorkflowName}")] + public static partial void LogAddingMcpToolTrigger(this ILogger logger, string workflowName); + + [LoggerMessage( + EventId = 206, + Level = LogLevel.Information, + Message = "Transform finished. Updated function count: {FunctionCount}")] + public static partial void LogTransformFinished(this ILogger logger, int functionCount); + + [LoggerMessage( + EventId = 207, + Level = LogLevel.Debug, + Message = "Skipping duplicate function registration: {FunctionName} (already registered by another workflow) in workflow: {WorkflowName}")] + public static partial void LogSkippingDuplicateFunction(this ILogger logger, string functionName, string workflowName); +} diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/Workflows/DurableWorkflowOptionsExtensions.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/Workflows/DurableWorkflowOptionsExtensions.cs new file mode 100644 index 0000000000..9a8496628f --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/Workflows/DurableWorkflowOptionsExtensions.cs @@ -0,0 +1,85 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Agents.AI.DurableTask.Workflows; +using Microsoft.Agents.AI.Workflows; + +namespace Microsoft.Agents.AI.Hosting.AzureFunctions.Workflows; + +/// +/// Provides extension methods for registering and configuring workflows in the context of the Azure Functions hosting environment. +/// +public static class DurableWorkflowOptionsExtensions +{ + // Registry of workflow options. + private static readonly Dictionary s_workflowOptions = new(StringComparer.OrdinalIgnoreCase); + + /// + /// Adds a workflow to the specified instance and optionally configures + /// workflow-specific options. + /// + /// The instance to which the workflow will be added. + /// The workflow to add. The workflow's Name property must not be null or empty. + /// An optional delegate to configure workflow-specific options. If null, default options are used. + /// The updated instance containing the added workflow. + /// Thrown when or is null. + /// Thrown when the workflow does not have a valid name. + public static DurableWorkflowOptions AddWorkflow( + this DurableWorkflowOptions options, + Workflow workflow, + Action? configure) + { + ArgumentNullException.ThrowIfNull(options); + ArgumentNullException.ThrowIfNull(workflow); + + if (string.IsNullOrEmpty(workflow.Name)) + { + throw new ArgumentException("Workflow must have a valid Name property.", nameof(workflow)); + } + + // Initialize with default behavior (MCP trigger disabled) + FunctionsWorkflowOptions workflowOptions = new(); + configure?.Invoke(workflowOptions); + + options.AddWorkflow(workflow); + s_workflowOptions[workflow.Name] = workflowOptions; + + return options; + } + + /// + /// Adds a workflow to the specified instance and configures + /// trigger support for MCP tool invocations. + /// + /// The instance to which the workflow will be added. + /// The workflow to add. The workflow's Name property must not be null or empty. + /// true to enable an MCP tool trigger for the workflow; otherwise, false. + /// The updated instance with the specified workflow and trigger configuration applied. + /// Thrown when or is null. + /// Thrown when the workflow does not have a valid name. + public static DurableWorkflowOptions AddWorkflow( + this DurableWorkflowOptions options, + Workflow workflow, + bool enableMcpToolTrigger) + { + return AddWorkflow(options, workflow, workflowOptions => workflowOptions.McpToolTrigger.IsEnabled = enableMcpToolTrigger); + } + + /// + /// Tries to get the for a workflow by name. + /// + /// The name of the workflow. + /// When this method returns, contains the workflow options if found; otherwise, null. + /// true if the workflow options were found; otherwise, false. + internal static bool TryGetWorkflowOptions(string workflowName, out FunctionsWorkflowOptions? workflowOptions) + { + return s_workflowOptions.TryGetValue(workflowName, out workflowOptions); + } + + /// + /// Builds the workflow options used for dependency injection (read-only copy). + /// + internal static IReadOnlyDictionary GetWorkflowOptionsSnapshot() + { + return new Dictionary(s_workflowOptions, StringComparer.OrdinalIgnoreCase); + } +} diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/Workflows/FunctionsWorkflowOptions.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/Workflows/FunctionsWorkflowOptions.cs new file mode 100644 index 0000000000..4f6b7e8248 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/Workflows/FunctionsWorkflowOptions.cs @@ -0,0 +1,17 @@ +// Copyright (c) Microsoft. All rights reserved. + +namespace Microsoft.Agents.AI.Hosting.AzureFunctions.Workflows; + +/// +/// Provides configuration options for enabling and customizing function triggers for a workflow. +/// +public sealed class FunctionsWorkflowOptions +{ + /// + /// Gets or sets the options used to configure the MCP tool trigger behavior. + /// + /// + /// By default, MCP tool trigger is disabled for workflows. + /// + public McpToolTriggerOptions McpToolTrigger { get; set; } = new(false); +} diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/Workflows/FunctionsWorkflowRunner.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/Workflows/FunctionsWorkflowRunner.cs new file mode 100644 index 0000000000..70cc023f9a --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/Workflows/FunctionsWorkflowRunner.cs @@ -0,0 +1,49 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Net; +using Microsoft.Agents.AI.DurableTask; +using Microsoft.Agents.AI.DurableTask.Workflows; +using Microsoft.Azure.Functions.Worker; +using Microsoft.Azure.Functions.Worker.Http; +using Microsoft.DurableTask.Client; + +namespace Microsoft.Agents.AI.Hosting.AzureFunctions.Workflows; + +/// +/// Provides functionality to invoke and manage workflow orchestrations in response to HTTP requests within an Azure +/// Functions environment. +/// +public sealed class FunctionsWorkflowRunner : DurableWorkflowRunner +{ + /// + /// Initializes a new instance of the FunctionsWorkflowRunner class using the specified DurableOptions. + /// + /// The DurableOptions that configure the behavior of the workflow runner. This parameter cannot be null. + public FunctionsWorkflowRunner(DurableOptions durableOptions) : base(durableOptions) + { + } + + /// + /// Invokes a workflow orchestration in response to an HTTP request. + /// + public static async Task RunWorkflowOrechstrtationHttpTriggerAsync( + [HttpTrigger] HttpRequestData req, + [DurableClient] DurableTaskClient client, + FunctionContext context) + { + var functionName = context.FunctionDefinition.Name; + var workflowName = functionName.Replace("-http", string.Empty); + var orchestrationFunctionName = WorkflowNamingHelper.ToOrchestrationFunctionName(workflowName); + var inputMessage = await req.ReadAsStringAsync(); + string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(orchestrationFunctionName, inputMessage); + + HttpResponseData response = req.CreateResponse(HttpStatusCode.Accepted); + await response.WriteStringAsync($"InvokeWorkflowOrechstrtationAsync is invoked for {workflowName}. Orchestration instanceId: {instanceId}"); + return response; + } + + internal async Task ExecuteActivityAsync(string activityFunctionName, string input, DurableTaskClient durableTaskClient, FunctionContext functionContext) + { + throw new NotImplementedException(); + } +}