MAF v1 — Agents in Workflows (Python + .NET)

Series note — Part of MAF v1: Python and .NET. Builds on Chapter 10 — Workflow Events and Builder by putting real LLM calls on the workflow’s event stream.

Repo — Full runnable code for this chapter is at https://github.com/nitin27may/e-commerce-agents/tree/main/tutorials/11-agents-in-workflows. Clone the repo, cd tutorials/11-agents-in-workflows, and follow the README.

Why this chapter

#

Chapters 09–10 taught raw executors that transform data. Chapter 11 replaces the interesting executors with agents: LLM-powered nodes that take a message in and produce another. Now one workflow can have deterministic steps (validation, enrichment, routing) and LLM steps (translation, summarization, classification) composed in the same graph — same scheduler, same event stream, same checkpoint boundaries.

The canonical example in the MAF docs is English → French → Spanish translation. Each arrow is a real LLM call; the workflow passes one agent’s output straight into the next agent’s input. Two round-trips to Azure OpenAI / OpenAI, one Pregel-style schedule.

What this chapter teaches. Two patterns for running an AIAgent inside a Workflow, with complete working code in both Python and .NET:

1. Manual adapter executors — you wrap each agent in a [MessageHandler]-decorated Executor (.NET) / subclass Executor and write @handler methods (Python). You get to emit custom events, manage sessions, mix agents with raw executors, short-circuit on errors. Verbose, maximum control.
2. Convenience builders — AgentWorkflowBuilder.BuildSequential(agents) (.NET) / feeding AIAgents straight into WorkflowBuilder (Python). One line; the framework wires the adapters for you. Terse, zero control knobs.

The rest of this series (Ch12 Sequential, Ch13 Concurrent, Ch14 Handoff, Ch16 Magentic) assumes you can reach for either.

Prerequisites

#

• Completed Chapter 10 — Workflow Events and Builder
• Python 3.12+ via uv; .NET 10 SDK
• A working .env at the repo root with either OPENAI_API_KEY (OpenAI) or AZURE_OPENAI_ENDPOINT + AZURE_OPENAI_DEPLOYMENT + AZURE_OPENAI_KEY (Azure). The examples cost two LLM calls per run (one French, one Spanish).

The concept

#

What happens on the wire

#

An agent inside a workflow is still an AIAgent — same .RunAsync() surface you’ve used since Ch01. The difference is that the workflow wants typed messages on edges, not raw agent I/O. MAF bridges this with two primitives:

• AgentExecutorRequest — carries a list of ChatMessages and a ShouldRespond flag. When a workflow dispatches to an agent-backed executor, it packs the inbound message into this request. ShouldRespond=false means “append to history, don’t call the LLM” — useful for pre-seeding context before another agent takes over.
• AgentExecutorResponse — carries the AgentResponse the LLM returned plus the full running conversation so downstream executors can see the complete history.

AgentExecutor (the built-in wrapper MAF provides) takes AgentExecutorRequest in and sends AgentExecutorResponse out. Your workflow input is a plain str (or a ChatMessage, or whatever your domain types are). Somewhere at the boundary you need to coerce. That’s the adapter.

Why adapters exist

#

A plain string is not a conversation. The agent wants a ChatMessage with a role, optional tool results, maybe metadata. Likewise, the consumer on the other side of the workflow wants the translated string, not a wrapped response with a conversation history attached. Adapter executors sit at the edges of the workflow to marshal between the domain type and the agent-protocol type.

In Python the adapters are explicit — you write them. In .NET the Workflows SDK accepts AIAgent anywhere it accepts an executor, and packs/unpacks the request for you. You still write adapters when you need custom behaviour; you skip them when you don’t.

The DAG we’re building

#

Two agents in a chain, with adapters on the ends:

Adapters sit at the workflow’s boundaries; agent-executors in the middle speak AgentExecutorRequest / AgentExecutorResponse. Each agent-executor triggers one round-trip to the LLM. The convenience builder (AgentWorkflowBuilder.BuildSequential in .NET, WorkflowBuilder accepting AIAgents directly in Python) emits exactly this graph — the adapters are still there, just not in your source file.

Jargon

#

• AgentExecutor — the built-in MAF Executor that wraps an AIAgent. Consumes AgentExecutorRequest, emits AgentExecutorResponse, runs one LLM turn per invocation.
• AgentExecutorRequest — typed envelope on the inbound edge of an agent-executor. Carries Messages (a list of ChatMessage) and ShouldRespond (bool). The agent only calls the LLM when ShouldRespond is true; otherwise it just appends to the conversation.
• AgentExecutorResponse — typed envelope on the outbound edge. Carries the AgentResponse the LLM produced plus the full running conversation.
• ShouldRespond — the knob that says “actually invoke the LLM” vs “just record this turn”. Lets you pre-seed context from multiple upstreams before the LLM fires.
• Adapter executor — a plain Executor at the workflow’s boundary whose only job is converting between the domain type (say, str) and the agent-protocol type (AgentExecutorRequest / AgentExecutorResponse). Hidden by the convenience builders; explicit when you need them.
• TurnToken (.NET) — a sentinel message you send into the workflow after the real input, which signals “agents, you may now fire”. .NET’s agent-wrapping executor caches inbound messages and waits for this token so fan-in patterns work. Python doesn’t need it — handlers fire on message arrival.
• Session sharing — passing the same AgentSession (.NET) / session kwarg (Python) to two agent-executors makes both agents see the same conversation. Skip it and each gets a fresh session per run.

Code walkthrough

#

using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Extensions.AI;

// Build two translation agents from one IChatClient.
AIAgent enToFr = chatClient.AsAIAgent(
    instructions: "Translate the user's message to French. Output ONLY the translation.",
    name: "en-to-fr");
AIAgent frToEs = chatClient.AsAIAgent(
    instructions: "Translate the user's message to Spanish. Output ONLY the translation.",
    name: "fr-to-es");

// One call. The framework wraps each agent as an AgentExecutor and wires
// the adapter plumbing internally.
Workflow workflow = AgentWorkflowBuilder.BuildSequential(enToFr, frToEs);

var messages = new List<ChatMessage> { new(ChatRole.User, "Hello, how are you?") };

await using StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, messages);

// Kick the pipeline. Agent-executors cache inbound messages and only fire
// when a TurnToken arrives — this matters for fan-in patterns where you
// want N upstreams to drop their messages before the agent runs.
await run.TrySendMessageAsync(new TurnToken(emitEvents: true));

await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    if (evt is AgentResponseUpdateEvent update && update.Data is AgentResponseUpdate chunk)
    {
        Console.Write(chunk.Text);  // streaming per-chunk text
    }
    else if (evt is WorkflowOutputEvent output && output.Data is List<ChatMessage> conversation)
    {
        var last = conversation.LastOrDefault(m => m.Role == ChatRole.Assistant);
        Console.WriteLine(last?.Text);  // "Hola, ¿cómo estás?"
    }
}

[SendsMessage(typeof(string))]
internal sealed partial class TranslationAgentExecutor : Executor
{
    private readonly AIAgent _agent;

    public TranslationAgentExecutor(string id, IChatClient chatClient, string targetLanguage)
        : base(id)
    {
        _agent = chatClient.AsAIAgent(
            instructions: $"Translate to {targetLanguage}. Output ONLY the translation.",
            name: id);
    }

    [MessageHandler]
    public async ValueTask HandleAsync(
        string message,
        IWorkflowContext context,
        CancellationToken cancellationToken = default)
    {
        // One LLM turn. We own the session story, the event story,
        // and the error story here.
        AgentResponse response = await _agent.RunAsync(message, cancellationToken: cancellationToken);

        await context.AddEventAsync(new TranslationCompletedEvent(Id, response.Text), cancellationToken);
        await context.SendMessageAsync(response.Text, cancellationToken);
    }
}

var inputAdapter = new InputAdapter();
var frenchExecutor = new TranslationAgentExecutor("en-to-fr", chatClient, "French");
var spanishExecutor = new TranslationAgentExecutor("fr-to-es", chatClient, "Spanish");
var outputAdapter = new OutputAdapter();

Workflow workflow = new WorkflowBuilder(inputAdapter)
    .AddEdge(inputAdapter, frenchExecutor)
    .AddEdge(frenchExecutor, spanishExecutor)
    .AddEdge(spanishExecutor, outputAdapter)
    .WithOutputFrom(outputAdapter)
    .Build();

await using StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, "Hello, how are you?");
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    if (evt is TranslationCompletedEvent t)
        Console.WriteLine($"  [{t.ExecutorId}] {t.Text}");
    else if (evt is WorkflowOutputEvent o && o.Data is string final)
        Console.WriteLine($"output: {final}");
}

cd tutorials/11-agents-in-workflows/dotnet
dotnet build

# Convenience — one-liner builder.
dotnet run -- --sequential "Hello, how are you?"
# mode:   Sequential
# input:  'Hello, how are you?'
#   [en-to-fr] Bonjour, comment ça va ?
#   [fr-to-es] Hola, ¿cómo estás?
# output: 'Hola, ¿cómo estás?'

# Manual — custom [MessageHandler] executors.
dotnet run -- --manual "Good night"
# mode:   Manual
# input:  'Good night'
#   [en-to-fr] Bonne nuit
#   [fr-to-es] Buenas noches
# output: 'Buenas noches'

# Tests (5 xUnit tests — deterministic, no LLM).
cd tests && dotnet test

from agent_framework import Agent, Message
from agent_framework._workflows._agent_executor import (
    AgentExecutor, AgentExecutorRequest, AgentExecutorResponse,
)
from agent_framework._workflows._executor import Executor, handler
from agent_framework._workflows._workflow_builder import WorkflowBuilder
from agent_framework._workflows._workflow_context import WorkflowContext


def translator(target_language: str, name: str) -> Agent:
    return Agent(
        _default_client(),
        instructions=(
            f"You are a translator. Translate to {target_language}. "
            "Output ONLY the translation — no preamble."
        ),
        name=name,
    )


class InputAdapter(Executor):
    def __init__(self) -> None:
        super().__init__(id="input-adapter")

    @handler
    async def run(self, message: str, ctx: WorkflowContext[AgentExecutorRequest]) -> None:
        await ctx.send_message(AgentExecutorRequest(
            messages=[Message(role="user", contents=[message])],
            should_respond=True,  # actually call the LLM (vs. pre-seed history)
        ))


class OutputAdapter(Executor):
    def __init__(self) -> None:
        super().__init__(id="output-adapter")

    @handler
    async def run(self, response: AgentExecutorResponse, ctx: WorkflowContext[None, str]) -> None:
        await ctx.yield_output(response.agent_response.text)


def build_workflow():
    input_adapter = InputAdapter()
    en_to_fr = AgentExecutor(translator("French", name="en-to-fr"), id="en-to-fr")
    fr_to_es = AgentExecutor(translator("Spanish", name="fr-to-es"), id="fr-to-es")
    output_adapter = OutputAdapter()

    return (
        WorkflowBuilder(start_executor=input_adapter)
        .add_edge(input_adapter, en_to_fr)
        .add_edge(en_to_fr, fr_to_es)
        .add_edge(fr_to_es, output_adapter)
        .build()
    )

uv run python tutorials/11-agents-in-workflows/python/main.py "Hello, how are you?"
# English input: Hello, how are you?
# Spanish output: Hola, ¿cómo estás?

Which pattern when

#

The default is convenience. The production capstone ends up with manual adapters everywhere because domain events are load-bearing (see “How this shows up in the capstone” below).

Side-by-side differences

#

The mental model is identical. Python’s adapters are explicit where .NET’s are hidden behind BuildSequential; both hop on the same runtime and emit the same event shapes. Python skips TurnToken because @handler fires on message arrival; .NET needs it because the auto-generated AgentExecutor wrapper intentionally batches inbound messages until the token drops.

Gotchas

#

• Don’t send raw strings between agent-executors. When you use AgentExecutor directly (Python’s explicit form), it expects AgentExecutorRequest on the inbound edge and emits AgentExecutorResponse on the outbound edge. Piping a str into it silently routes nothing. If you’re mixing raw and agent nodes, either adapt at each boundary or use the convenience builder that handles it for you.
• ShouldRespond=False skips the LLM but appends to history. Useful for pre-seeding context from two upstreams into a third agent — the first two emit AgentExecutorRequest(..., should_respond=False), the third agent sees both turns in its history before firing. Misuse and the agent never calls the LLM; it stays silent while the workflow waits for an output that never comes.
• TurnToken is easy to forget. AgentWorkflowBuilder.BuildSequential and direct-agent AddEdge both produce a workflow whose first event is a no-op until TurnToken arrives. Symptom: RunStreamingAsync returns, WatchStreamAsync yields the WorkflowStartedEvent, and then nothing happens. Fix: await run.TrySendMessageAsync(new TurnToken(emitEvents: true)) right after RunStreamingAsync.
• Unique executor ids are still required. Two AgentExecutors on the same agent can’t share an id. BuildSequential generates ids like en_to_fr_<guid> to avoid the problem — if you want readable ids, use the manual pattern.
• List<ChatMessage> vs string for workflow input. BuildSequential’s adapters expect List<ChatMessage>; the manual pattern from this chapter uses string. Wire a mismatched input type and the workflow will throw at runtime with “no handler for type X” because the adapter’s [MessageHandler] signature is specific.
• Session sharing crosses conversation boundaries. Two agent-executors sharing the same AgentSession will both see every message — including the other’s system prompt turn history. That’s usually what you want for classify-then-answer patterns; it’s usually not what you want for parallel specialists that should each start with a fresh context.
• Don’t mix [SendsMessage(typeof(AgentExecutorResponse))] with convenience builders. The convenience path’s auto-generated AgentExecutor doesn’t expose its AgentExecutorResponse on outbound edges — that shape exists only between internal adapters. If you want to inspect the response, switch to the manual pattern or subscribe to AgentResponseEvent on the event stream.

Tests

#

# Python — 1 wiring test + 3 real-LLM integration tests (skipped if no creds).
source agents/.venv/bin/activate
python -m pytest tutorials/11-agents-in-workflows/python/tests/ -v
# 4 passed (3 hit real Azure OpenAI / OpenAI)

# .NET — 5 xUnit tests, all deterministic (fake IChatClient, no network).
cd tutorials/11-agents-in-workflows/dotnet/tests
dotnet test --nologo
# Passed! - Failed: 0, Passed: 5, Skipped: 0, Total: 5, Duration: <100ms

The fake IChatClient in the .NET test file is the pattern to reuse for every workflow-chapter test from here on. Real LLM tests go in smoke/nightly suites; the deterministic fake catches the wiring bugs that would otherwise fire on every PR.

How this shows up in the capstone

#

• Phase 7 — plans/refactor/08-pre-purchase-concurrent.md. The pre-purchase workflow wraps each specialist (ProductDiscovery, PricingPromotions, InventoryFulfillment) as a manual agent-adapter executor so the executor can emit OrderProgressEvent / ReviewSummaryEvent payloads on the event stream. Those payloads travel verbatim to the frontend as SSE chunks. The refactor uses AgentWorkflowBuilder.BuildConcurrent for the fan-out layer and manual adapters for the conditional EstimateShipping step.
• Phase 8 — plans/refactor/10-orchestrator-to-handoff.md. The top-level orchestrator moves from a single-agent call-router to a HandoffBuilder mesh. Each mesh node is an AIAgent the builder wraps as an AgentExecutor under the hood — the same primitive from this chapter, stitched into a graph with handoff tool-call edges.
• .NET port — plans/dotnet-port/02-orchestrator.md. The .NET orchestrator port uses the manual pattern for every node because TreatWarningsAsErrors=true and the custom event subclasses ride through the WorkflowEvent type hierarchy cleanly.

In other words: the convenience path is great for learning and for straight chains; production work in both languages gravitates to manual adapters because domain events and custom sessions matter.

Further reading

#

• MAF docs — Agents in Workflows
• MAF docs — Executors
• Sample — 03_AgentWorkflowPatterns (Sequential / Concurrent / Handoff / GroupChat via AgentWorkflowBuilder)
• Sample — CustomAgentExecutors (manual adapters with structured output)

What’s next

#

• Next chapter: Chapter 12 — Sequential Orchestration takes the sequential pattern seriously — turn limits, content filters, handoff conditions. Builds straight on AgentWorkflowBuilder.BuildSequential from this chapter.
• Full source: tutorials/11-agents-in-workflows