MAF v1 — Declarative Workflows (Python + .NET)

Series note — Part of MAF v1: Python and .NET. Third of four advanced chapters. New material — no counterpart in the original Python-only series.

Repo — Full runnable code for this chapter is at https://github.com/nitin27may/e-commerce-agents/tree/main/tutorials/19-declarative-workflows. Clone the repo, cd tutorials/19-declarative-workflows, and follow the per-language instructions below.

When you’d reach for this — A workflow’s shape is now a configurable concern, not a code concern. Examples: ops needs to insert a “validate compliance” step before the existing “issue refund” step without an engineering deploy; a customer-success lead wants to A/B-test two different orderings of a return flow; a compliance review needs the workflow graph to live in a YAML the auditor can read without opening a C# project. If any of those sound familiar, declarative belongs on the table. If your team owns the graph end-to-end and changes it via PR-then-deploy like every other code change, the imperative builders from Ch09–Ch18 are a better fit.

Why this chapter

#

Ch09 through Ch18 built every workflow in code. That’s right for engineers who own the graph — strong typing, IDE completion, refactors that fail loudly at build time. It stops being right the moment a non-engineer needs to tweak ordering, a support team wants to swap in a different retry policy during an incident, or operations wants to ship a pipeline change without a deploy.

Declarative — the opposite of imperative; you describe what the pipeline looks like (a YAML document), not how to build it (a sequence of WorkflowBuilder.add_edge(...) calls). A loader reads the document and emits the same Workflow object code would have.

Three concrete reasons to reach for a declarative loader:

• GitOps — a workflow edit shows up in a pull request as a YAML diff, not a C#/Python diff. Reviewers who don’t read code can still approve it. “Does the graph now run Validate before Log?” is answerable from the diff alone.
• Hot reload — restart the service (or a config-watcher inside it) and a new graph takes effect. No rebuild, no image push.
• Authoring by non-engineers — ops, compliance, customer-success. Anyone who can edit a YAML file in the admin UI can reorder steps or flip a config value.

MAF ships an officially supported declarative schema via Microsoft.Agents.AI.Workflows.Declarative (.NET) and agent_framework.declarative (Python). That schema is much richer than what we’ll build here — it supports kind: Workflow, SetVariable, If/ConditionGroup/Foreach, PowerFx expressions, agent invocations, and checkpointing. We’ll point at it at the end of the chapter.

This chapter rolls its own minimal loader instead, because the shape of the pattern — parse YAML → look up an op by name → instantiate executors → wire edges → return a Workflow — is worth seeing end-to-end in ~90 lines. Once you’ve built one, the production-grade one reads the same way, just with more dials.

Prerequisites

#

• Completed Chapter 09 — Workflow Executors and Edges — we reuse the same three-step text pipeline from that chapter.
• Python 3.12+ via uv; .NET 10 SDK.
• No LLM key required — workflows are deterministic coordinators and our example is pure text manipulation.
• A passing familiarity with YAML (keys, lists, strings) is enough.

The concept

#

A workflow spec names three things: executors, edges, and a start node. Every executor entry carries an id, an op (operation-by-name — a string key the loader looks up in an op registry to decide what behaviour the executor has), and optional per-op config. Every edge entry names two executor ids.

# workflow.yaml
name: text-pipeline
start: uppercase
executors:
  - id: uppercase
    op: upper
  - id: validate
    op: non_empty         # emits a terminal output when input is blank
  - id: log
    op: prefix
    prefix: "LOGGED: "    # free-form op config
edges:
  - from: uppercase
    to: validate
  - from: validate
    to: log

Jargon used above, defined once:

• Declarative — the workflow is described in data, not built by code. The loader is the only imperative step.
• Op (operation-by-name) — a string in the YAML that selects behaviour. op: upper means “the uppercase transform”; op: prefix means “wrap the message with a prefix”. Ops are a minimal domain-specific language; each one is a pure function registered at startup.
• Op registry — the dict/dictionary that maps op names to factory functions. The single extension point: adding a new op means adding one entry. No new YAML syntax.
• WorkflowFactory — the name MAF gives to the officially supported declarative loader (agent_framework.declarative.WorkflowFactory in Python; DeclarativeWorkflowBuilder.Build<TInput>(...) in .NET). Our chapter calls its equivalent load_workflow / DeclarativeWorkflowLoader.Load to make the pedagogical intent obvious.
• GitOps — workflow definition lives in git; deploy is “change the file and push”; review is a diff on the YAML; rollback is a revert. Shorthand coined by Weaveworks in 2017.
• Lazy instantiation — only build the executor when the loader hits its YAML entry, not upfront for every op in the registry. Our loader does this implicitly: the registry holds factories, not instances.

Loader flow

#

The loader is a small pipeline in its own right — a spec on disk becomes a Workflow through one parse, one registry lookup per entry, and one builder pass.

The bytes on disk and the Workflow object are separated by two lookups — a YAML parse and a registry hit per executor. Unknown ops are caught at load time, not at run time. The output is the same Workflow type Ch09’s hand-wired code produced; nothing downstream (streaming, checkpoints, visualisation) knows the graph came from YAML.

Tradeoff — when declarative pays, when it costs

#

Every declarative system is a small language of its own. You trade off Python’s or C#’s expressive power for a smaller, reviewable surface.

Two hard limits specifically on the rolled-our-own pattern in this chapter:

• Ops must be pre-registered. register_op("my_op", factory) runs at startup. A YAML file that references an op nobody registered fails at load time. That’s a feature — unreviewed code can’t execute — but it means the YAML is not a full programming language. You cannot inline arbitrary logic.
• No runtime-state access from the spec. The YAML has no way to say “if the user is an admin” or “if the previous step’s output was empty”. Conditional edges would need to be modelled as ops that inspect the message, or by extending the schema (Microsoft’s built-in loader does the latter via PowerFx).

Both constraints vanish if you reach for Microsoft.Agents.AI.Workflows.Declarative — its schema includes If, ConditionGroup, Foreach, SetVariable, and PowerFx expressions, at the cost of ~7× the schema surface. Pick the smallest loader that solves your problem.

Code walkthrough

#

// dotnet/Program.cs (excerpt)
using Microsoft.Agents.AI.Workflows;
using YamlDotNet.Serialization;
using YamlDotNet.Serialization.NamingConventions;

// ── Executor ──────────────────────────────────────────────
[SendsMessage(typeof(string))]
[YieldsOutput(typeof(string))]
internal sealed partial class DeclarativeExecutor : Executor
{
    private readonly OpFunction _op;

    public DeclarativeExecutor(string id, OpFunction op) : base(id) => _op = op;

    [MessageHandler]
    public async ValueTask HandleAsync(
        string message, IWorkflowContext ctx, CancellationToken ct = default)
    {
        (string? forward, string? terminal) = _op(message);
        if (terminal is not null) { await ctx.YieldOutputAsync(terminal, ct); return; }
        if (forward  is not null) { await ctx.SendMessageAsync(forward, ct); }
    }
}

// ── Op registry ───────────────────────────────────────────
public delegate (string? Forward, string? Terminal) OpFunction(string input);

public static class OpRegistry
{
    private static readonly Dictionary<string, Func<ExecutorSpec, OpFunction>> _ops = new()
    {
        ["upper"]     = _    => s => (s.ToUpperInvariant(), null),
        ["non_empty"] = spec => s => string.IsNullOrWhiteSpace(s)
                                      ? (null, spec.EmptyOutput ?? "[skipped: empty input]")
                                      : (s, null),
        ["prefix"]    = spec => s => (null, (spec.Prefix ?? "") + s),
        // ... lower, strip, reverse, passthrough
    };

    public static void Register(string name, Func<ExecutorSpec, OpFunction> factory)
        => _ops[name] = factory;

    public static OpFunction Build(ExecutorSpec spec) => _ops[spec.Op](spec);
}

// ── Loader ────────────────────────────────────────────────
public static Workflow Load(string specPath)
{
    IDeserializer yaml = new DeserializerBuilder()
        .WithNamingConvention(UnderscoredNamingConvention.Instance)
        .IgnoreUnmatchedProperties().Build();
    WorkflowSpec spec = yaml.Deserialize<WorkflowSpec>(File.ReadAllText(specPath));

    var execs = spec.Executors.ToDictionary(
        e => e.Id, e => new DeclarativeExecutor(e.Id, OpRegistry.Build(e)));

    WorkflowBuilder builder = new(execs[spec.Start]);
    foreach (var exec in execs.Values) builder = builder.WithOutputFrom(exec);
    foreach (var edge in spec.Edges)   builder = builder.AddEdge(execs[edge.From], execs[edge.To]);
    return builder.Build();
}

cd tutorials/19-declarative-workflows/dotnet
dotnet run -- "hello world"
# spec:   workflow.yaml
# input:  'hello world'
# output: 'LOGGED: HELLO WORLD'

dotnet run -- ""
# output: '[skipped: empty input]'

OpRegistry.Register("check_stock", spec =>
{
    int threshold = int.Parse(spec.Prefix ?? "1"); // reuse Prefix for config, or add a field
    return productId =>
    {
        // call DB, return (forward, terminal)
        return (productId, null);
    };
});

# python/main.py (excerpt)
import yaml
from agent_framework._workflows._executor import Executor, handler
from agent_framework._workflows._workflow_builder import WorkflowBuilder
from agent_framework._workflows._workflow_context import WorkflowContext


# ── Op registry ───────────────────────────────────────────
def _build_op(op: str, config: dict):
    if op == "upper":     return lambda s: (s.upper(), None)
    if op == "lower":     return lambda s: (s.lower(), None)
    if op == "strip":     return lambda s: (s.strip(), None)
    if op == "reverse":   return lambda s: (s[::-1], None)
    if op == "non_empty":
        def _impl(s):
            return (s, None) if s.strip() else (None, "[skipped: empty input]")
        return _impl
    if op == "prefix":
        prefix = config.get("prefix", "")
        return lambda s: (None, f"{prefix}{s}")
    raise ValueError(f"unknown op: {op!r}")


# ── Executor ──────────────────────────────────────────────
class DeclarativeExecutor(Executor):
    """One Executor subclass. Behaviour comes from an op, not a method body."""

    def __init__(self, executor_id: str, op: str, config: dict) -> None:
        super().__init__(id=executor_id)
        self._op = _build_op(op, config)

    @handler
    async def run(self, message: str, ctx: WorkflowContext[str, str]) -> None:
        forward, terminal = self._op(message)
        if terminal is not None:
            await ctx.yield_output(terminal)   # short-circuit
        elif forward is not None:
            await ctx.send_message(forward)    # forward downstream


# ── Loader ────────────────────────────────────────────────
def load_workflow(spec_path):
    spec = yaml.safe_load(spec_path.read_text())
    execs = {
        e["id"]: DeclarativeExecutor(e["id"], e["op"], e)
        for e in spec["executors"]
    }
    builder = WorkflowBuilder(start_executor=execs[spec["start"]], name=spec["name"])
    for edge in spec["edges"]:
        builder = builder.add_edge(execs[edge["from"]], execs[edge["to"]])
    return builder.build()

cd tutorials/19-declarative-workflows/python
uv run python main.py "hello world"
# spec: workflow.yaml
# input: 'hello world'
# output: 'LOGGED: HELLO WORLD'

uv run python main.py ""
# input: ''
# output: '[skipped: empty input]'

from shared.workflow_loader import register_op

def _op_check_stock(config: dict):
    threshold = int(config.get("threshold", 1))

    def _impl(product_id: str):
        # call DB, return (forward, terminal)
        ...

    return _impl

register_op("check_stock", _op_check_stock)

- id: stock-gate
  op: check_stock
  threshold: 5

Side-by-side differences

#

Structural parity, as usual. The two real differences:

• .NET makes outbound types explicit via attributes, Python via generics. Same information, different place.
• .NET requires a per-executor WithOutputFrom(...) when every executor can yield. Python’s runtime surfaces every yield_output without an opt-in list.

Gotchas

#

• Schema freedom is a tax. Every declarative system is a small language you’ll end up maintaining. Validate aggressively — our loader rejects unknown ops, missing id/op fields, duplicate ids, and undeclared edge endpoints at load time. Production loaders should go further (jsonschema on the YAML, a CI check that every committed spec loads clean).
• Executors aren’t free. Each YAML entry builds a real DeclarativeExecutor instance at load time. Our loader uses lazy instantiation at the op level — registry entries are factories, not instances — but the executor itself is eager. For 50+ entry specs, consider building executors on first edge traversal.
• Edge validation is your job. A from: typo that doesn’t match any declared id is caught by the loader here. If you loosen that check and let the WorkflowBuilder catch it, the error surfaces deeper in the stack and is harder to read.
• Custom ops must live somewhere. Our example embeds them in the loader module. Production systems register ops by name from a plugin folder, a DI container, or an explicit bootstrap call. Either way, ops must be pre-registered before the loader runs — the YAML cannot inline behaviour.
• YAML’s boolean/string coercion. prefix: "" is fine, but prefix: off becomes Python’s False / .NET’s bool. Always quote string values if there’s any doubt ("off").
• WithOutputFrom per-executor in .NET. Miss an executor that calls YieldOutputAsync and the output never surfaces. The Python runtime doesn’t have this trap.
• Don’t confuse our loader with the built-in one. Ours is load_workflow / DeclarativeWorkflowLoader.Load on a minimal op+edge schema. MAF’s built-in is WorkflowFactory / DeclarativeWorkflowBuilder on a much richer kind: Workflow schema with PowerFx, agents, and control-flow actions. Don’t mix the two in one project — pick the schema that fits, stay consistent.
• Hot-reload is a race condition. If you watch the YAML file and rebuild the Workflow on change, be sure you don’t swap the reference mid-run. Either drain in-flight runs first, or version the Workflow and keep old ones alive until their runs complete.

Tests

#

# Python — 10 unit tests: every built-in op, unknown-op error, loader wiring,
# happy path + short-circuit path through the YAML-built workflow.
python -m pytest tutorials/19-declarative-workflows/python/tests/ -v
# 10 passed

# .NET — same workflow, happy path + short-circuit path via dotnet run.
cd tutorials/19-declarative-workflows/dotnet
dotnet build
dotnet run -- "hello world"
dotnet run -- ""

The Python suite splits by concern: op unit tests (each op is a pure function — assert op("hello") == ("HELLO", None)), registry tests (unknown op raises cleanly), and loader end-to-end tests (load the YAML, run it, assert terminal output). The same split works for any extension — when you register check_stock, write unit tests against the op function first, then one integration test that runs it from a YAML file.

How this shows up in the capstone

#

The capstone extends the exact pattern in this chapter:

• agents/python/shared/workflow_loader.py (load_workflow, lines 158–227) is the full production version. Same op-registry shape. Same YAML schema. The delta vs this chapter is error handling, bulk load_workflows_directory for agents/config/workflows/*.yaml, and a description field piped through to WorkflowBuilder.
• Built-in ops — passthrough, upper, lower, strip, reverse, non_empty, prefix — match this chapter one-for-one.
• register_op(name, factory) is the extension point. Phase 7’s pre-purchase rewrite (plans/refactor/08-pre-purchase-concurrent.md) registers domain ops (check_stock, fetch_reviews, get_price_history) and drops them into agents/config/workflows/pre-purchase.yaml. The workflow file becomes reviewable by product/support without diving into Python.
• DeclarativeExecutor (lines 129–148) is the one-for-one twin of this chapter’s class — same handler signature, same op call, same dual-output (forward vs terminal) contract.
• .NET parity port lives at agents/dotnet/src/ECommerceAgents.Shared/Workflows/DeclarativeWorkflow.cs. (That port predates the MAF-native refactor and currently executes ops with a hand-rolled Run() loop rather than via WorkflowBuilder; plans/dotnet-port/05-declarative.md tracks the migration to the shape shown in this chapter.)

The pattern the capstone avoids: a separate schema per workflow type. One YAML schema, one op registry, one loader — the workflows differ only in which ops and edges they wire. That’s also why the capstone ships a scripts/visualize_workflows.py (Ch20) that renders any registered spec to Mermaid without per-workflow code.

Further reading

#

This chapter

• Source on GitHub: tutorials/19-declarative-workflows
• Previous: Chapter 18 — State and Checkpoints · Next: Chapter 20 — Visualization

Microsoft Agent Framework docs

• Declarative Workflows — overview — the officially supported schema (kind: Workflow, PowerFx, control-flow actions).
• Workflows — Workflow Builder — the code-first API we’re wrapping.
• agent_framework.declarative.WorkflowFactory (Python) · Microsoft.Agents.AI.Workflows.Declarative.DeclarativeWorkflowBuilder (NuGet, prerelease).

Where it lives in the capstone

• Production loader: agents/python/shared/workflow_loader.py:1-244 (Python), agents/dotnet/src/ECommerceAgents.Shared/Workflows/DeclarativeWorkflow.cs (.NET).
• Workflow specs: agents/python/config/workflows/*.yaml.
• Migration plan: plans/refactor/12-declarative-workflows.md, plans/dotnet-port/05-declarative.md.

What’s next

#

Chapter 20 — Visualization shows how to render any Workflow — code-built or declarative — to Mermaid and Graphviz DOT in one line. Combined with this chapter, you can commit the YAML spec and its rendered diagram to git, so reviewers see the graph diff without a Mermaid preview extension.