Python: (core): Add functional workflow API

[moonbox3]

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]

Python Test Coverage Report •

File	Stmts	Miss	Cover	Missing
packages/core/agent_framework/_workflows
_functional.py	341	14	95%	686, 716, 719–720, 728–729, 789–790, 796, 1076–1077, 1079, 1099, 1101
_workflow.py	267	19	92%	87, 268–270, 272–273, 291, 295, 429, 617, 638, 694, 706, 712, 717, 737–739, 752
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	22885	2794	87%

Python Unit Test Overview

Tests	Skipped	Failures	Errors	Time
4651	25 💤	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

[TaoChenOSU]

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?

[moonbox3]

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).

[TaoChenOSU]

How will a step let the workflow know that it needs human input?

[moonbox3]

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.

[TaoChenOSU]

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.

We can use the experimental flag now :)