Create a Simple Sequential Workflow

This tutorial demonstrates how to create a simple sequential workflow using Agent Framework Workflows.

Sequential workflows are the foundation of building complex AI agent systems. This tutorial shows how to create a simple two-step workflow where each step processes data and passes it to the next step.

Overview

In this tutorial, you'll create a workflow with two executors:

Uppercase Executor - Converts input text to uppercase
Reverse Text Executor - Reverses the text and outputs the final result

The workflow demonstrates core concepts like:

Creating a custom executor with one handler
Creating a custom executor from a function
Using WorkflowBuilder to connect executors with edges
Processing data through sequential steps
Observing workflow execution through events

Concepts Covered

Prerequisites

.NET 8.0 SDK or later
No external AI services required for this basic example
A new console application

Step-by-Step Implementation

The following sections show how to build the sequential workflow step by step.

Step 1: Install NuGet packages

First, install the required packages for your .NET project:

dotnet add package Microsoft.Agents.AI.Workflows --prerelease

Step 2: Define the Uppercase Executor

Define an executor that converts text to uppercase:

using System;
using System.Linq;
using System.Threading.Tasks;
using Microsoft.Agents.AI.Workflows;

/// <summary>
/// First executor: converts input text to uppercase.
/// </summary>
Func<string, string> uppercaseFunc = s => s.ToUpperInvariant();
var uppercase = uppercaseFunc.BindExecutor("UppercaseExecutor");

Key Points:

Create a function that takes a string and returns the uppercase version
Use BindExecutor() to create an executor from the function

Step 3: Define the Reverse Text Executor

Define an executor that reverses the text:

/// <summary>
/// Second executor: reverses the input text and completes the workflow.
/// </summary>
internal sealed class ReverseTextExecutor() : Executor<string, string>("ReverseTextExecutor")
{
    public override ValueTask<string> HandleAsync(string input, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        // Reverse the input text
        return ValueTask.FromResult(new string(input.Reverse().ToArray()));
    }
}

ReverseTextExecutor reverse = new();

Key Points:

Create a class that inherits from Executor<TInput, TOutput>
Implement HandleAsync() to process the input and return the output

Step 4: Build and Connect the Workflow

Connect the executors using WorkflowBuilder:

// Build the workflow by connecting executors sequentially
WorkflowBuilder builder = new(uppercase);
builder.AddEdge(uppercase, reverse).WithOutputFrom(reverse);
var workflow = builder.Build();

Key Points:

WorkflowBuilder constructor takes the starting executor
AddEdge() creates a directed connection from uppercase to reverse
WithOutputFrom() specifies which executors produce workflow outputs
Build() creates the immutable workflow

Step 5: Execute the Workflow

Run the workflow and observe the results:

// Execute the workflow with input data
await using Run run = await InProcessExecution.RunAsync(workflow, "Hello, World!");
foreach (WorkflowEvent evt in run.NewEvents)
{
    switch (evt)
    {
        case ExecutorCompletedEvent executorComplete:
            Console.WriteLine($"{executorComplete.ExecutorId}: {executorComplete.Data}");
            break;
    }
}

Step 6: Understanding the Workflow Output

When you run the workflow, you'll see output like:

UppercaseExecutor: HELLO, WORLD!
ReverseTextExecutor: !DLROW ,OLLEH

The input "Hello, World!" is first converted to uppercase ("HELLO, WORLD!"), then reversed ("!DLROW ,OLLEH").

Key Concepts Explained

Executor Interface

Executors from functions:

Use BindExecutor() to create an executor from a function

Executors implement Executor<TInput, TOutput>:

TInput: The type of data this executor accepts
TOutput: The type of data this executor produces
HandleAsync: The method that processes the input and returns the output

.NET Workflow Builder Pattern

The WorkflowBuilder provides a fluent API for constructing workflows:

Constructor: Takes the starting executor
AddEdge(): Creates directed connections between executors
WithOutputFrom(): Specifies which executors produce workflow outputs
Build(): Creates the final immutable workflow

.NET Event Types

During execution, you can observe these event types:

ExecutorCompletedEvent - When an executor finishes processing

Complete .NET Example

For the complete, ready-to-run implementation, see the 01_ExecutorsAndEdges sample in the Agent Framework repository.

This sample includes:

Full implementation with all using statements and class structure
Additional comments explaining the workflow concepts
Complete project setup and configuration

Overview

In this tutorial, you'll create a workflow with two executors:

Upper Case Executor - Converts input text to uppercase
Reverse Text Executor - Reverses the text and outputs the final result

The workflow demonstrates core concepts like:

Using the @executor decorator to create workflow nodes
Connecting executors with WorkflowBuilder
Passing data between steps with ctx.send_message()
Yielding final output with ctx.yield_output()
Streaming events for real-time observability

Concepts Covered

Prerequisites

Python 3.10 or later
Agent Framework Core Python package installed: pip install agent-framework-core
No external AI services required for this basic example

Step-by-Step Implementation

The following sections show how to build the sequential workflow step by step.

Step 1: Import Required Modules

First, import the necessary modules from Agent Framework:

import asyncio
from typing_extensions import Never
from agent_framework import WorkflowBuilder, WorkflowContext, WorkflowOutputEvent, executor

Step 2: Create the First Executor

Create an executor that converts text to uppercase by implementing an executor with a handler method:

class UpperCase(Executor):
    def __init__(self, id: str):
        super().__init__(id=id)

    @handler
    async def to_upper_case(self, text: str, ctx: WorkflowContext[str]) -> None:
        """Convert the input to uppercase and forward it to the next node.

        Note: The WorkflowContext is parameterized with the type this handler will
        emit. Here WorkflowContext[str] means downstream nodes should expect str.
        """
        result = text.upper()

        # Send the result to the next executor in the workflow.
        await ctx.send_message(result)

Key Points:

The @executor decorator registers this function as a workflow node
WorkflowContext[str] indicates this executor sends a string downstream by specifying the first type parameter
ctx.send_message() passes data to the next step

Step 3: Create the Second Executor

Create an executor that reverses the text and yields the final output from a method decorated with @executor:

@executor(id="reverse_text_executor")
async def reverse_text(text: str, ctx: WorkflowContext[Never, str]) -> None:
    """Reverse the input and yield the workflow output."""
    result = text[::-1]

    # Yield the final output for this workflow run
    await ctx.yield_output(result)

Key Points:

WorkflowContext[Never, str] indicates this is a terminal executor that does not send any messages by specifying Never as the first type parameter but produce workflow outputs by specifying str as the second parameter
ctx.yield_output() provides the final workflow result
The workflow completes when it becomes idle

Step 4: Build the Workflow

Connect the executors using WorkflowBuilder:

upper_case = UpperCase(id="upper_case_executor")

workflow = (
    WorkflowBuilder()
    .add_edge(upper_case, reverse_text)
    .set_start_executor(upper_case)
    .build()
)

Key Points:

add_edge() creates directed connections between executors
set_start_executor() defines the entry point
build() finalizes the workflow

Step 5: Run the Workflow with Streaming

Execute the workflow and observe events in real-time:

async def main():
    # Run the workflow and stream events
    async for event in workflow.run_stream("hello world"):
        print(f"Event: {event}")
        if isinstance(event, WorkflowOutputEvent):
            print(f"Workflow completed with result: {event.data}")

if __name__ == "__main__":
    asyncio.run(main())

Step 6: Understanding the Output

When you run the workflow, you'll see events like:

Event: ExecutorInvokedEvent(executor_id=upper_case_executor)
Event: ExecutorCompletedEvent(executor_id=upper_case_executor)
Event: ExecutorInvokedEvent(executor_id=reverse_text_executor)
Event: ExecutorCompletedEvent(executor_id=reverse_text_executor)
Event: WorkflowOutputEvent(data='DLROW OLLEH', source_executor_id=reverse_text_executor)
Workflow completed with result: DLROW OLLEH

Key Concepts Explained

Workflow Context Types

The WorkflowContext generic type defines what data flows between executors:

WorkflowContext[str] - Sends a string to the next executor
WorkflowContext[Never, str] - Terminal executor that yields workflow output of type string

Event Types

During streaming execution, you'll observe these event types:

ExecutorInvokedEvent - When an executor starts processing
ExecutorCompletedEvent - When an executor finishes processing
WorkflowOutputEvent - Contains the final workflow result

Python Workflow Builder Pattern

The WorkflowBuilder provides a fluent API for constructing workflows:

add_edge(): Creates directed connections between executors
set_start_executor(): Defines the workflow entry point
build(): Finalizes and returns an immutable workflow object

Complete Example

For the complete, ready-to-run implementation, see the sample in the Agent Framework repository.

This sample includes:

Full implementation with all imports and documentation
Additional comments explaining the workflow concepts
Sample output showing the expected results

Next Steps

Learn about creating a simple concurrent workflow

Feedback

Was this page helpful?

Last updated on 2025-11-21

Share via

Create a Simple Sequential Workflow

Overview

Concepts Covered

Prerequisites

Step-by-Step Implementation

Step 1: Install NuGet packages

Step 2: Define the Uppercase Executor

Step 3: Define the Reverse Text Executor

Step 4: Build and Connect the Workflow

Step 5: Execute the Workflow

Step 6: Understanding the Workflow Output

Key Concepts Explained

Executor Interface

.NET Workflow Builder Pattern

.NET Event Types

Complete .NET Example

Overview

Concepts Covered

Prerequisites

Step-by-Step Implementation

Step 1: Import Required Modules

Step 2: Create the First Executor

Step 3: Create the Second Executor

Step 4: Build the Workflow

Step 5: Run the Workflow with Streaming

Step 6: Understanding the Output

Key Concepts Explained

Workflow Context Types

Event Types

Python Workflow Builder Pattern

Complete Example

Next Steps

Feedback

Additional resources