Python: (core): Add functional workflow API by moonbox3 · Pull Request #4238 · microsoft/agent-framework

moonbox3 · 2026-02-25T08:40:58Z

Motivation and Context

The functional API is a stepping stone between single-agent use and the full graph API. Users write workflows as plain async functions -- no executor classes, no edges, no builder patterns.

Add @workflow and @step decorators for writing workflows as plain async functions
Native Python control flow (if/else, loops, asyncio.gather) replaces graph concepts
@step is opt-in: plain functions work inside @workflow without it. Use @step on expensive operations (agent calls, API requests) to save their results and skip re-execution on
HITL resume or crash recovery
Streaming support via run(stream=True)
HITL support via ctx.request_info() with replay
.as_agent() wraps a functional workflow as an agent-compatible object

A very basic example of the functional workflow API:

@workflow
async def pipeline(data: str) -> str:
    upper = await to_upper(data)
    return await reverse(upper)

result = await pipeline.run("hello")
print(result.get_outputs())  # ['OLLEH']

Note: @step is opt-in for functions where per-step checkpointing matters (for example, agent calls). Without @step, workflows still support HITL and checkpointing — functions just re-execute on resume.

@step
async def call_agent(prompt: str) -> str:
    return (await agent.run(prompt)).text

@workflow
async def pipeline(data: str, ctx: RunContext) -> str:
    result = await call_agent(data)        # saved by @step
    validated = await validate(result)      # plain function, re-runs on resume
    feedback = await ctx.request_info(...)  # HITL pause
    return await finalize(result, feedback)

ctx: RunContext is only needed when you use HITL (request_info), custom events (add_event), or state (get_state/set_state). Otherwise, omit it for a cleaner signature.

Description

Closes Python: Add functional workflow API #3537

Contribution Checklist

The code builds clean without any errors or warnings
The PR follows the Contribution Guidelines
All unit tests pass, and I have added new tests where possible
Is this a breaking change? If yes, add "[BREAKING]" prefix to the title of the PR.

markwallace-microsoft · 2026-02-25T08:43:36Z

Python Test Coverage Report •

File	Stmts	Miss	Cover	Missing
packages/core/agent_framework
_agents.py	342	43	87%	435, 439, 491, 856, 892, 908, 991–994, 1055–1057, 1178, 1194, 1196, 1209, 1215, 1251, 1253, 1262–1267, 1272, 1274, 1280–1281, 1288, 1290–1291, 1299–1300, 1303–1305, 1313–1314, 1316, 1321, 1323
packages/core/agent_framework/_workflows
_functional.py	341	14	95%	686, 711, 714–715, 723–724, 784–785, 791, 1071–1072, 1074, 1094, 1096
packages/orchestrations/agent_framework_orchestrations
_handoff.py	385	58	84%	105–106, 108, 163–173, 175, 177, 179, 184, 278, 284, 316, 339, 361, 417, 442, 500, 532, 590–591, 623, 631, 635–636, 674–676, 681–683, 801, 804, 817, 879, 884, 891, 901, 903, 922, 924, 1006–1007, 1039–1040, 1122, 1129, 1201–1202, 1204
TOTAL	22563	2772	87%

Python Unit Test Overview

Tests	Skipped	Failures	Errors	Time
4780	247 💤	0 ❌	0 🔥	1m 18s ⏱️

Copilot

Pull request overview

This PR introduces a functional workflow API as an alternative to the existing graph-based workflow API. The functional approach allows users to write workflows as plain async functions decorated with @workflow, using native Python control flow (if/else, loops, asyncio.gather) instead of explicit graph construction with executors and edges. The @step decorator is optional and provides per-step checkpointing, caching, and observability.

Changes:

Added core implementation (_functional.py) with @workflow, @step decorators, RunContext, FunctionalWorkflow, and FunctionalWorkflowAgent classes
Added comprehensive test suite (40+ test cases covering basic execution, HITL, checkpointing, streaming, error handling, edge cases)
Added 6 sample files demonstrating functional workflows (basic pipeline, streaming, parallel execution, checkpointing, HITL, agent integration)
Restructured getting-started samples to introduce functional workflows before graph workflows
Updated exports in __init__.py to expose new functional API symbols

Reviewed changes

Copilot reviewed 13 out of 14 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
`python/packages/core/agent_framework/_workflows/_functional.py`	Core implementation of functional workflow API with RunContext, StepWrapper, FunctionalWorkflow, and FunctionalWorkflowAgent classes (1105 lines)
`python/packages/core/agent_framework/__init__.py`	Added exports for FunctionalWorkflow, FunctionalWorkflowAgent, RunContext, StepWrapper, step, and workflow
`python/packages/core/tests/workflow/test_functional_workflow.py`	Comprehensive test suite covering basic execution, events, parallelism, HITL, errors, streaming, state, checkpointing, control flow, and edge cases (1031 lines)
`python/samples/01-get-started/05_first_functional_workflow.py`	Getting started sample demonstrating basic functional workflow with plain async functions
`python/samples/01-get-started/06_first_graph_workflow.py`	Renamed and updated graph workflow sample (previously 05_first_workflow.py)
`python/samples/01-get-started/07_host_your_agent.py`	Renamed agent hosting sample (previously 06_host_your_agent.py)
`python/samples/01-get-started/README.md`	Updated sample listing to include both functional and graph workflow samples
`python/samples/03-workflows/functional/basic_pipeline.py`	Sample showing simplest sequential pipeline with `@workflow` decorator
`python/samples/03-workflows/functional/basic_streaming_pipeline.py`	Sample demonstrating streaming workflow events with run(stream=True)
`python/samples/03-workflows/functional/parallel_pipeline.py`	Sample showing fan-out/fan-in with asyncio.gather
`python/samples/03-workflows/functional/steps_and_checkpointing.py`	Sample explaining `@step` decorator for per-step checkpointing and observability
`python/samples/03-workflows/functional/hitl_review.py`	Sample demonstrating HITL with ctx.request_info() and resume
`python/samples/03-workflows/functional/agent_integration.py`	Sample showing agent calls inside workflows and .as_agent() wrapper
`python/samples/03-workflows/README.md`	Added functional workflow section to samples overview

python/samples/01-get-started/README.md

python/packages/core/agent_framework/_workflows/_functional.py

eavanvalkenburg · 2026-02-25T10:11:16Z

python/samples/03-workflows/functional/basic_pipeline.py

+@workflow
+async def data_pipeline(url: str) -> str:
+    """A simple sequential data pipeline."""
+    raw = await fetch_data(url)


I think it would be usefull to also demonstrate that because this is just a function, you do not have to wrap everything in steps, you can do some of the manipulation just as simple code between steps, making it a lot simpler

eavanvalkenburg · 2026-02-25T10:12:42Z

python/samples/03-workflows/functional/hitl_review.py

+    return f"Draft document about '{topic}': Lorem ipsum dolor sit amet..."
+
+
+@step


how this does step compare to handler is there a lot of overlap and could we reuse steps in graphs, or handler here?

I want them to be conceptually different things. A @handler handles a message routed to it by the graph: it's reactive, tied to the executor contract. A @step marks a function call in a sequential flow: it's proactive, just "I called this function as step N." Different mental models, different names.

eavanvalkenburg · 2026-02-25T10:13:44Z

python/samples/03-workflows/functional/hitl_review.py

+    print(f"State: {result1.get_final_state()}")
+    assert result1.get_final_state() == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS
+
+    requests = result1.get_request_info_events()


what would happen here if state != WorkflowRunState.IDLE_WITH_PENDING_REQUESTS?

If request_info() was never reached (an early return), the workflow completes normally with state IDLE and get_request_info_events() returns an empty list. Added a comment in the sample clarifying this.

so requests would be None or would the get... call raise?

get_request_info_events() returns an empty list. It's a filter over the event stream, so no requests means []. No exception raised.

python/samples/03-workflows/functional/agent_integration.py

python/packages/core/agent_framework/_workflows/_functional.py

python/samples/01-get-started/05_first_functional_workflow.py

TaoChenOSU · 2026-02-25T22:00:22Z

python/packages/core/agent_framework/_workflows/_functional.py

+``asyncio.gather``) instead of a graph-based topology.
+
+A ``@workflow``-decorated async function receives its input as the first
+positional argument.  If the function needs HITL (``request_info``), custom


Should it be:

Suggested change

positional argument. If the function needs HITL (``request_info``), custom

positional argument. If a step needs HITL (``request_info``), custom

?

The docstring is describing the @workflow-decorated function, not a @step. The workflow function is what receives RunContext and calls request_info(). A @step doesn't receive context — so "If the function needs HITL" is referring to the workflow function.

python/packages/core/agent_framework/_workflows/_functional.py

TaoChenOSU · 2026-02-25T22:05:14Z

python/packages/core/agent_framework/_workflows/_functional.py

+class RunContext:
+    """Execution context injected into ``@workflow`` functions.
+
+    Every ``@workflow`` invocation receives a ``RunContext`` instance that


I am a bit confused on which entity receives the context. Is it the workflow or the steps or both?

Only the @workflow function receives RunContext. Steps don't — they're plain async functions with caching. If a step needs HITL input, the workflow calls request_info() and passes the result to the step as an argument.

TaoChenOSU · 2026-02-25T22:08:45Z

python/packages/core/agent_framework/_workflows/_functional.py

+            @workflow
+            async def hitl_pipeline(data: str, ctx: RunContext) -> str:
+                feedback = await ctx.request_info({"draft": data}, response_type=str)
+                return feedback


My brain maps @workflow to the graph-based Workflow and @step to Executor. I can see the benefit of allowing request_info at the workflow level. It's kind of like an executor whose sole purpose is to get user feedback. But should we also allow request_info inside a @step?

By design, request_info lives at the workflow level — the workflow is the orchestrator that decides when to pause for input. Steps are meant to be self-contained units of work. If a step needs human input, the workflow calls request_info() first and passes the result to the step. This keeps steps simple and testable in isolation (no framework dependency).

python/packages/core/agent_framework/_workflows/_functional.py

TaoChenOSU · 2026-02-25T22:24:26Z

python/packages/core/agent_framework/_workflows/_functional.py

+    def __init__(self, func: Callable[..., Awaitable[R]], *, name: str | None = None) -> None:
+        if not inspect.iscoroutinefunction(func):
+            raise TypeError(
+                f"@step can only decorate async functions, but '{func.__name__}' is not a coroutine function."


This is probably not super important but should we also support not async methods?

We could, but it adds complexity for limited benefit — the workflow itself is async, so wrapping a sync function in asyncio.to_thread() or similar is straightforward for users. For now, @step requires async. We can revisit if users ask for it.

TaoChenOSU · 2026-02-25T22:28:14Z

python/packages/core/agent_framework/_workflows/_functional.py

+        functools.update_wrapper(self, func)  # type: ignore[arg-type]
+
+    # ------------------------------------------------------------------
+    # run() — same overloaded interface as graph Workflow


Should we extract these methods and make both workflow types get them for free?

Agree this is worth considering. The run() overload pattern and _finalize_events logic are similar between FunctionalWorkflow and graph Workflow. Worth a follow-up refactoring PR to extract a shared base, but out of scope for this one.

TaoChenOSU · 2026-02-25T22:32:40Z

python/packages/core/agent_framework/_workflows/_functional.py

+                self._last_step_cache = dict(ctx._step_cache)
+
+                # Yield collected events
+                for event in ctx._get_events():


Question: Is this true streaming?

It looks like all events have been produced at this point.

Currently events are collected during execution and yielded after. Within a single function execution there's no natural suspension point to stream intermediate events (unlike graph workflows where each superstep boundary is a yield point). However, @step events are collected incrementally, and the async generator yields them as they're produced. For true mid-function streaming, users can use ctx.add_event() — though these are still batched by the current implementation. This is a known limitation we can improve in a follow-up.

TaoChenOSU · 2026-02-25T22:35:59Z

python/packages/core/agent_framework/_workflows/_functional.py

+        cache_key = ctx._get_step_cache_key(self.name)
+        found, cached = ctx._get_cached_result(cache_key)
+        invocation_data = deepcopy({"args": args, "kwargs": kwargs}) if args or kwargs else None
+        if found:


Should we also check if the input arguments have the same values?

There could be scenarios where some steps are checkpointed and some are not (the ones without the decorator). If a checkpointed step depends on the output of a non-checkpointed step and its output changes, it may lead to incorrect results.

The cache key is (step_name, call_index) and relies on the documented determinism requirement — if the workflow function is deterministic w.r.t. step results, the arguments will always match on replay. Adding argument comparison would be defensive but expensive (deep equality on arbitrary objects) and could give false negatives for objects that don't implement __eq__.

This is by design and documented: workflow functions must be deterministic w.r.t. step results. If a non-@step function is nondeterministic and a cached @step depends on it, results can diverge. The fix is to use @step on any function whose output affects downstream cached steps. This tradeoff keeps the simple case simple.

TaoChenOSU · 2026-02-25T22:39:46Z

python/samples/01-get-started/05_first_functional_workflow.py

+
+
+# Plain async functions — no decorators needed
+async def to_upper_case(text: str) -> str:


nit: this sample will become even simpler if we can remove the asyncs.

Same as the sync support discussion — supporting sync functions would simplify the getting-started sample, but @workflow is inherently async (it runs an async event loop internally). We can consider sync support as a follow-up.

moonbox3 · 2026-02-27T04:10:04Z

Btw, @eavanvalkenburg and @TaoChenOSU I think it would be best to stick this functional API in to its own package. We want to get some more signal around the APIs and use of it before we deem it "GA worthy," IMO.

moonbox3 added 3 commits February 25, 2026 13:19

Add functional workflow api

e886798

cleanup

15c7561

More cleanup

e9bdaa3

Copilot AI review requested due to automatic review settings February 25, 2026 08:40

markwallace-microsoft added documentation Improvements or additions to documentation python labels Feb 25, 2026

Copilot started reviewing on behalf of moonbox3 February 25, 2026 08:41 View session

Copilot AI reviewed Feb 25, 2026

View reviewed changes

python/samples/01-get-started/README.md Outdated Show resolved Hide resolved

python/packages/core/agent_framework/_workflows/_functional.py Outdated Show resolved Hide resolved

address copilot feedback

8087814

eavanvalkenburg reviewed Feb 25, 2026

View reviewed changes

moonbox3 added 2 commits February 25, 2026 20:10

Address PR feedbacK

27dbbef

updates

503df78

TaoChenOSU reviewed Feb 25, 2026

View reviewed changes

moonbox3 added 2 commits February 27, 2026 12:53

Merge main

629f36d

PR feedback

c459af5

moonbox3 requested review from TaoChenOSU and eavanvalkenburg February 27, 2026 04:08

		return f"Draft document about '{topic}': Lorem ipsum dolor sit amet..."


		@step

	positional argument. If the function needs HITL (``request_info``), custom
	positional argument. If a step needs HITL (``request_info``), custom



		# Plain async functions — no decorators needed
		async def to_upper_case(text: str) -> str:

Conversation

moonbox3 commented Feb 25, 2026

Motivation and Context

Description

Contribution Checklist

Uh oh!

markwallace-microsoft commented Feb 25, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Python Unit Test Overview

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

moonbox3 commented Feb 27, 2026

Uh oh!

Reviewers

Assignees

Labels

markwallace-microsoft commented Feb 25, 2026 •

edited

Loading