Edit

Share via

Executors

Executors are the fundamental building blocks that process messages in a workflow. They are autonomous processing units that receive typed messages, perform operations, and can produce output messages or events.

Overview

Each executor has a unique identifier and can handle specific message types. Executors can be:

  • Custom logic components — process data, call APIs, or transform messages
  • AI agents — use LLMs to generate responses (see Agents in Workflows)

Important

The recommended way to define executor message handlers in C# is to use the [MessageHandler] attribute on methods within a partial class that derives from Executor. This uses compile-time source generation for handler registration, providing better performance, compile-time validation, and Native AOT compatibility.

Basic Executor Structure

Executors derive from the Executor base class and use the [MessageHandler] attribute to declare handler methods. The class must be marked partial to enable source generation.

using Microsoft.Agents.AI.Workflows;

internal sealed partial class UppercaseExecutor() : Executor("UppercaseExecutor")
{
    [MessageHandler]
    private ValueTask<string> HandleAsync(string message, IWorkflowContext context)
    {
        string result = message.ToUpperInvariant();
        return ValueTask.FromResult(result); // Return value is automatically sent to connected executors
    }
}

You can also send messages manually without returning a value:

internal sealed partial class UppercaseExecutor() : Executor("UppercaseExecutor")
{
    [MessageHandler]
    private async ValueTask HandleAsync(string message, IWorkflowContext context)
    {
        string result = message.ToUpperInvariant();
        await context.SendMessageAsync(result); // Manually send messages to connected executors
    }
}

Multiple Input Types

Handle multiple input types by defining multiple [MessageHandler] methods:

internal sealed partial class SampleExecutor() : Executor("SampleExecutor")
{
    [MessageHandler]
    private ValueTask<string> HandleStringAsync(string message, IWorkflowContext context)
    {
        return ValueTask.FromResult(message.ToUpperInvariant());
    }

    [MessageHandler]
    private ValueTask<int> HandleIntAsync(int message, IWorkflowContext context)
    {
        return ValueTask.FromResult(message * 2);
    }
}

Function-Based Executors

Create an executor from a function using the BindExecutor extension method:

Func<string, string> uppercaseFunc = s => s.ToUpperInvariant();
var uppercase = uppercaseFunc.BindExecutor("UppercaseExecutor");

Basic Executor Structure

Executors inherit from the Executor base class. Each executor uses methods decorated with the @handler decorator. Handlers must have proper type annotations to specify the message types they process.

from agent_framework import (
    Executor,
    WorkflowContext,
    handler,
)

class UpperCase(Executor):

    @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."""
        await ctx.send_message(text.upper())

Function-Based Executors

Create an executor from a function using the @executor decorator:

from agent_framework import (
    WorkflowContext,
    executor,
)

@executor(id="upper_case_executor")
async def upper_case(text: str, ctx: WorkflowContext[str]) -> None:
    """Convert the input to uppercase and forward it to the next node."""
    await ctx.send_message(text.upper())

Multiple Input Types

Handle multiple input types by defining multiple handlers:

class SampleExecutor(Executor):

    @handler
    async def to_upper_case(self, text: str, ctx: WorkflowContext[str]) -> None:
        await ctx.send_message(text.upper())

    @handler
    async def double_integer(self, number: int, ctx: WorkflowContext[int]) -> None:
        await ctx.send_message(number * 2)

Explicit Type Parameters

As an alternative to type annotations, you can specify types explicitly via decorator parameters:

Important

When using explicit type parameters, you must specify all types via the decorator — you cannot mix explicit parameters with type annotations. The input parameter is required; output and workflow_output are optional.

class ExplicitTypesExecutor(Executor):

    @handler(input=str, output=str)
    async def to_upper_case(self, text, ctx) -> None:
        await ctx.send_message(text.upper())

    @handler(input=str | int, output=str)
    async def handle_mixed(self, message, ctx) -> None:
        await ctx.send_message(str(message).upper())

    @handler(input=str, output=int, workflow_output=bool)
    async def process_with_workflow_output(self, message, ctx) -> None:
        await ctx.send_message(len(message))
        await ctx.yield_output(True)

The WorkflowContext Object

The WorkflowContext provides methods for interacting with the workflow during execution:

  • send_message — send messages to connected executors
  • yield_output — produce workflow outputs returned/streamed to the caller
class OutputExecutor(Executor):

    @handler
    async def handle(self, message: str, ctx: WorkflowContext[Never, str]) -> None:
        await ctx.yield_output("Hello, World!")

If a handler neither sends messages nor yields outputs, no type parameter is needed:

class LogExecutor(Executor):

    @handler
    async def handle(self, message: str, ctx: WorkflowContext) -> None:
        print("Doing some work...")

Next steps

Edges

  • Last updated on