Events

The workflow event system provides observability into workflow execution. Events are emitted at key points during execution and can be consumed in real-time via streaming.

Built-in Event Types

// Workflow lifecycle events
WorkflowStartedEvent     // Workflow execution begins
WorkflowOutputEvent      // Workflow outputs data
WorkflowErrorEvent       // Workflow encounters an error
WorkflowWarningEvent     // Workflow encountered a warning

// Executor events
ExecutorInvokedEvent     // Executor starts processing
ExecutorCompletedEvent   // Executor finishes processing
ExecutorFailedEvent      // Executor encounters an error
AgentResponseEvent       // An agent run produces output
AgentResponseUpdateEvent // An agent run produces a streaming update

// Superstep events
SuperStepStartedEvent    // Superstep begins
SuperStepCompletedEvent  // Superstep completes

// Request events
RequestInfoEvent         // A request is issued

Note

When agents use approval-required tools, RequestInfoEvent typically carries a ToolApprovalRequestContent payload for tool calls that require human approval. See Human-in-the-Loop for details on handling these events.

# All events use the unified WorkflowEvent class with a type discriminator:

# Workflow lifecycle events
WorkflowEvent.type == "started"             # Workflow execution begins
WorkflowEvent.type == "status"              # Workflow state changed (use .state)
WorkflowEvent.type == "output"              # Workflow produces a terminal (final) output
WorkflowEvent.type == "intermediate"        # Workflow produces an intermediate (observational) output
WorkflowEvent.type == "failed"              # Workflow terminated with error (use .details)
WorkflowEvent.type == "error"               # Non-fatal error from user code
WorkflowEvent.type == "warning"             # Workflow encountered a warning

# Executor events
WorkflowEvent.type == "executor_invoked"    # Executor starts processing
WorkflowEvent.type == "executor_completed"  # Executor finishes processing
WorkflowEvent.type == "executor_failed"     # Executor encounters an error
WorkflowEvent.type == "data"                # Deprecated alias for "intermediate"

# Superstep events
WorkflowEvent.type == "superstep_started"   # Superstep begins
WorkflowEvent.type == "superstep_completed" # Superstep completes

# Request events
WorkflowEvent.type == "request_info"        # A request is issued

Note

When agents use approval-required tools, request_info events typically carry a Content payload with type == "function_approval_request" for tool calls that require human approval. See Human-in-the-Loop for details on handling these events.

Note

"output" and "intermediate" are the two output discriminators. An executor designated as a terminal output source emits "output" events (consumed by WorkflowRunResult.get_outputs()). One designated as an intermediate output source emits "intermediate" events (consumed by WorkflowRunResult.get_intermediate_outputs()). The "data" type is a deprecated alias for "intermediate" and will be removed in a future release; prefer filtering on "intermediate" in new code.

Go events implement workflow.Event and expose their payload through Data(). Use a type switch to inspect their typed fields:

Event type	Payload and purpose
workflow.StartedEvent	Starts an input-processing cycle. Message contains the input.
workflow.OutputEvent	A designated executor yields a value in Output. Tags identifies intermediate output.
workflow.ErrorEvent	Reports an error in Error. SubWorkflowID identifies a nested workflow when set.
workflow.WorkflowWarningEvent	Reports a warning in Message.
workflow.ExecutorInvokedEvent	An executor starts processing. ExecutorID identifies it, and Message contains its input.
workflow.ExecutorCompletedEvent	An executor completes processing. Result contains its result.
workflow.ExecutorFailedEvent	An executor fails. ExecutorID and Error identify the failure.
workflow.SuperStepStartedEvent	A superstep starts. StepNumber and StartInfo describe it.
workflow.SuperStepCompletedEvent	A superstep completes. CompletionInfo describes pending work, requests, and any checkpoint.
workflow.RequestInfoEvent	Request contains an external request that the host can answer.
workflow.RequestHaltEvent	An executor requests a halt. Result contains its payload, if any.

Go uses OutputEvent for both terminal and intermediate output. Call IsIntermediate() to distinguish them; it doesn't use Python's separate "output" and "intermediate" discriminators.

Hosted agents also use OutputEvent, with an *agent.ResponseUpdate or *agent.Response payload when the corresponding events are enabled. Go doesn't define separate AgentResponseEvent and AgentResponseUpdateEvent types. See Agent executors for configuration.

Consuming Events

using Microsoft.Agents.AI.Workflows;

await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    switch (evt)
    {
        case ExecutorInvokedEvent invoke:
            Console.WriteLine($"Starting {invoke.ExecutorId}");
            break;

        case ExecutorCompletedEvent complete:
            Console.WriteLine($"Completed {complete.ExecutorId}: {complete.Data}");
            break;

        case WorkflowOutputEvent output:
            Console.WriteLine($"Workflow output: {output.Data}");
            return;

        case WorkflowErrorEvent error:
            Console.WriteLine($"Workflow error: {error.Exception}");
            return;
    }
}

from agent_framework import WorkflowEvent

async for event in workflow.run(input_message, stream=True):
    if event.type == "executor_invoked":
        print(f"Starting {event.executor_id}")
    elif event.type == "executor_completed":
        print(f"Completed {event.executor_id}: {event.data}")
    elif event.type == "intermediate":
        print(f"Intermediate output from {event.executor_id}: {event.data}")
    elif event.type == "output":
        print(f"Terminal output: {event.data}")
        return
    elif event.type == "error":
        print(f"Workflow error: {event.data}")
        return

Use a type switch for Go workflow events. Handle iterator errors and failure events separately: receiving an output doesn't guarantee that the rest of the workflow succeeds.

Observe events

inproc.Default.Run advances to the next halt and accumulates events. The run can pause for external input; a halt isn't necessarily completion.

The following function prints executor activity and output, returns failures, and closes the run:

import (
    "context"
    "errors"
    "fmt"

    "github.com/microsoft/agent-framework-go/workflow"
    "github.com/microsoft/agent-framework-go/workflow/inproc"
)

func observeEvents(
    ctx context.Context, wf *workflow.Workflow, input any,
) (err error) {
    run, err := inproc.Default.Run(ctx, wf, input)
    if err != nil {
        return err
    }
    defer func() {
        err = errors.Join(err, run.Close(context.WithoutCancel(ctx)))
    }()

    for event := range run.NewEvents() {
        switch event := event.(type) {
        case workflow.ExecutorInvokedEvent:
            fmt.Printf("Starting %s\n", event.ExecutorID)
        case workflow.ExecutorCompletedEvent:
            fmt.Printf("Completed %s: %v\n",
                event.ExecutorID, event.Result)
        case workflow.OutputEvent:
            kind := "Terminal"
            if event.IsIntermediate() {
                kind = "Intermediate"
            }
            fmt.Printf("%s output from %s: %v\n",
                kind, event.ExecutorID, event.Output)
        case workflow.ErrorEvent:
            return event.Error
        case workflow.ExecutorFailedEvent:
            return fmt.Errorf("executor %q failed: %w",
                event.ExecutorID, event.Error)
        }
    }
    return nil
}

NewEvents() returns the events accumulated since its previous call. Its bookmark advances when you call it, even if you stop iterating early. Use OutgoingEvents() to replay all accumulated events instead of calling NewEvents() again.

Streaming events

Use RunStreaming to observe events as they arrive. This function observes a workflow that doesn't require external input:

import (
    "context"
    "errors"
    "fmt"

    "github.com/microsoft/agent-framework-go/workflow"
    "github.com/microsoft/agent-framework-go/workflow/inproc"
)

func streamEvents(
    ctx context.Context, wf *workflow.Workflow, input any,
) (err error) {
    run, err := inproc.Default.RunStreaming(ctx, wf, input)
    if err != nil {
        return err
    }
    defer func() {
        err = errors.Join(err, run.Close(context.WithoutCancel(ctx)))
    }()

    for event, streamErr := range run.WatchStream(ctx) {
        if streamErr != nil {
            return streamErr
        }
        switch event := event.(type) {
        case workflow.OutputEvent:
            kind := "Terminal"
            if event.IsIntermediate() {
                kind = "Intermediate"
            }
            fmt.Printf("%s output from %s: %v\n",
                kind, event.ExecutorID, event.Output)
        case workflow.ErrorEvent:
            return event.Error
        case workflow.ExecutorFailedEvent:
            return fmt.Errorf("executor %q failed: %w",
                event.ExecutorID, event.Error)
        }
    }
    return nil
}

WatchStream keeps waiting when a workflow has pending external requests. Handle RequestInfoEvent and send matching responses through SendResponse when you use it for request-driven workflows.

Use WatchUntilHalt instead when the consumer must return at a pause without waiting for a response. Only one event iterator can consume a streaming run at a time. See Human-in-the-loop workflows for request handling and resumption.

Custom Events

Custom events let executors emit domain-specific signals during workflow execution tailored to your application's needs. Some example use cases include:

• Track progress — report intermediate steps so callers can show status updates.
• Emit diagnostics — surface warnings, metrics, or debug information without changing the workflow output.
• Relay domain data — push structured payloads (e.g., database writes, tool calls) to listeners in real time.

Defining Custom Events

Define a custom event by subclassing WorkflowEvent. The base constructor accepts an optional object? data payload that is exposed through the Data property.

using Microsoft.Agents.AI.Workflows;

// Simple event with a string payload
internal sealed class ProgressEvent(string step) : WorkflowEvent(step) { }

// Event with a structured payload
internal sealed class MetricsEvent(MetricsData metrics) : WorkflowEvent(metrics) { }

In Python, create custom events using the WorkflowEvent class directly with a custom type discriminator string. The type and data parameters carry all the information.

from agent_framework import WorkflowEvent

# Create a custom event with a custom type string and payload
event = WorkflowEvent(type="progress", data="Step 1 complete")

# Custom event with a structured payload
event = WorkflowEvent(type="metrics", data={"latency_ms": 42, "tokens": 128})

Note

The event types "started", "status", and "failed" are reserved for framework lifecycle notifications. If an executor attempts to emit one of these types, the event is ignored and a warning is logged.

Define a custom event by creating a type that implements the workflow.Event interface. The Data method returns the event payload.

import "github.com/microsoft/agent-framework-go/workflow"

type ProgressEvent struct {
    Step string
}

func (e ProgressEvent) Data() any {
    return e.Step
}

var _ workflow.Event = ProgressEvent{}

Emitting Custom Events

Emit custom events from an executor's message handler by calling AddEventAsync on the IWorkflowContext:

using Microsoft.Agents.AI.Workflows;

internal sealed class ProgressEvent(string step) : WorkflowEvent(step) { }

internal sealed partial class CustomExecutor() : Executor("CustomExecutor")
{
    [MessageHandler]
    private async ValueTask HandleAsync(string message, IWorkflowContext context)
    {
        await context.AddEventAsync(new ProgressEvent("Validating input"));

        // Executor logic...

        await context.AddEventAsync(new ProgressEvent("Processing complete"));
    }
}

Emit custom events from a handler by calling add_event on the WorkflowContext:

from agent_framework import (
    handler,
    Executor,
    WorkflowContext,
    WorkflowEvent,
)

class CustomExecutor(Executor):

    @handler
    async def handle(self, message: str, ctx: WorkflowContext[str]) -> None:
        await ctx.add_event(WorkflowEvent(type="progress", data="Validating input"))

        # Executor logic...

        await ctx.add_event(WorkflowEvent(type="progress", data="Processing complete"))

Emit custom events from an executor handler by calling AddEvent on the workflow.Context:

import "github.com/microsoft/agent-framework-go/workflow"

customExecutor := workflow.NewExecutor("CustomExecutor", func(ctx *workflow.Context, message string) error {
    if err := ctx.AddEvent(ProgressEvent{Step: "Validating input"}); err != nil {
        return err
    }

    // Executor logic...

    return ctx.AddEvent(ProgressEvent{Step: "Processing complete"})
}).Bind()

Custom events don't require WithOutputFrom: AddEvent emits them directly. Use YieldOutput and designate an output source when you also need ordinary workflow output.

Consuming Custom Events

Use pattern matching to filter for your custom event type in the event stream:

await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    switch (evt)
    {
        case ProgressEvent progress:
            Console.WriteLine($"Progress: {progress.Data}");
            break;

        case WorkflowOutputEvent output:
            Console.WriteLine($"Done: {output.Data}");
            return;
    }
}

Filter on the custom type discriminator string:

async for event in workflow.run(input_message, stream=True):
    if event.type == "progress":
        print(f"Progress: {event.data}")
    elif event.type == "output":
        print(f"Done: {event.data}")
        return

Use a type switch or type assertion to filter for your custom event type in the event stream:

for evt, err := range run.WatchStream(ctx) {
    if err != nil {
        return err
    }

    switch e := evt.(type) {
    case ProgressEvent:
        fmt.Printf("Progress: %v\n", e.Data())
    case workflow.OutputEvent:
        fmt.Printf("Done: %v\n", e.Output)
    case workflow.ErrorEvent:
        return e.Error
    case workflow.ExecutorFailedEvent:
        return fmt.Errorf("executor %q failed: %w", e.ExecutorID, e.Error)
    }
}

Next steps

Workflow Builder & Execution

Related topics:

• Agents in Workflows
• State Management
• Checkpoints & Resuming
• Observability