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/_workflows
_functional.py	343	14	95%	688, 713, 716–717, 725–726, 786–787, 793, 1073–1074, 1076, 1096, 1098
TOTAL	22151	2778	87%

Python Unit Test Overview

Tests	Skipped	Failures	Errors	Time
4624	244 💤	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

?

TaoChenOSU · 2026-02-25T22:01:06Z

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

+A ``@workflow``-decorated async function receives its input as the first
+positional argument.  If the function needs HITL (``request_info``), custom
+events, or key/value state, add a :class:`RunContext` parameter — otherwise it
+can be omitted.  Inside the function, plain ``async`` calls run normally.


Should this be:

Suggested change

can be omitted. Inside the function, plain ``async`` calls run normally.

can be omitted. Inside the workflow, plain ``async`` calls run normally.

?

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?

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?

TaoChenOSU · 2026-02-25T22:11:49Z

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

+
+        On first execution this suspends the workflow by raising an internal
+        ``WorkflowInterrupted`` signal (caught by the framework, never exposed
+        to user code).  The caller receives a ``WorkflowRunResult`` whose


The caller receives a WorkflowRunResult

What about streaming?

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?

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?

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.

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.

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.

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

		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

	can be omitted. Inside the function, plain ``async`` calls run normally.
	can be omitted. Inside the workflow, plain ``async`` calls run normally.



		# 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!

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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

markwallace-microsoft commented Feb 25, 2026 •

edited

Loading