Note
Access to this page requires authorization. You can try signing in or changing directories.
Access to this page requires authorization. You can try changing directories.
Workflows let you chain multiple steps together — each step processes data and passes it to the next.
Define workflow steps (executors):
using Microsoft.Agents.AI.Workflows;
// Step 1: Convert text to uppercase
class UpperCase : Executor
{
[Handler]
public async Task ToUpperCase(string text, WorkflowContext<string> ctx)
{
await ctx.SendMessageAsync(text.ToUpper());
}
}
// Step 2: Reverse the string and yield output
[Executor(Id = "reverse_text")]
static async Task ReverseText(string text, WorkflowContext<Never, string> ctx)
{
var reversed = new string(text.Reverse().ToArray());
await ctx.YieldOutputAsync(reversed);
}
Build and run the workflow:
var upper = new UpperCase();
var workflow = new AgentWorkflowBuilder(startExecutor: upper)
.AddEdge(upper, ReverseText)
.Build();
var result = await workflow.RunAsync("hello world");
Console.WriteLine($"Output: {string.Join(", ", result.GetOutputs())}");
// Output: DLROW OLLEH
Tip
See the full sample for the complete runnable file.
Define workflow steps (executors) and connect them with edges:
# Step 1: A class-based executor that converts text to uppercase
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 input to uppercase and forward to the next node."""
await ctx.send_message(text.upper())
# Step 2: A function-based executor that reverses the string and yields output
@executor(id="reverse_text")
async def reverse_text(text: str, ctx: WorkflowContext[Never, str]) -> None:
"""Reverse the string and yield the final workflow output."""
await ctx.yield_output(text[::-1])
def create_workflow():
"""Build the workflow: UpperCase → reverse_text."""
upper = UpperCase(id="upper_case")
return (
WorkflowBuilder(start_executor=upper)
.add_edge(upper, reverse_text)
.build()
)
Build and run the workflow:
workflow = create_workflow()
events = await workflow.run("hello world")
print(f"Output: {events.get_outputs()}")
print(f"Final state: {events.get_final_state()}")
Tip
See the full sample for the complete runnable file.
Next steps
Go deeper:
- Workflows overview — understand workflow architecture
- Sequential workflows — linear step-by-step patterns
- Agents in workflows — using agents as workflow steps