Python: [BREAKING] Require building functional workflow instances

python/PACKAGE_STATUS.md

@@ -102,7 +102,8 @@ listed below.
 
 - `agent-framework-core`: functional workflow APIs from
   `agent_framework/_workflows/_functional.py`, including `RunContext`, `step`,
-  `FunctionalWorkflow`, `workflow`, and `FunctionalWorkflowAgent`
+  `FunctionalWorkflowDefinition`, `FunctionalWorkflow`, `workflow`, and
+  `FunctionalWorkflowAgent`
 
 #### `HARNESS`
 

python/packages/core/AGENTS.md

@@ -200,6 +200,12 @@ agent_framework/
   every output-capable executor not selected by `output_from`.
 - **`WorkflowRunResult`** - Non-streaming workflow result with Workflow Output `get_outputs()`
   and Intermediate Output `get_intermediate_outputs()` accessors
+- **Functional workflow definition/build lifecycle** - `@workflow` returns a stateless
+  `FunctionalWorkflowDefinition`. Call `build()` to create a stateful `FunctionalWorkflow` scoped to one logical
+  caller or session. The definition has no `run()` or `as_agent()` surface, so module-level decorated definitions
+  cannot accidentally retain caller state. Each built workflow and its `FunctionalWorkflowAgent` must remain scoped
+  to that caller/session. Pass a caller-scoped checkpoint storage to `build(checkpoint_storage=...)` when needed;
+  hosts remain responsible for authorizing and tenant-scoping access to any shared checkpoint adapter.
 - **Orchestrators**: `SequentialOrchestrator`, `ConcurrentOrchestrator`, `GroupChatOrchestrator`, `MagenticOrchestrator`, `HandoffOrchestrator`
 
 ## Built-in Providers

python/packages/core/agent_framework/__init__.py

@@ -319,6 +319,7 @@
     "._workflows._function_executor": ("FunctionExecutor", "executor"),
     "._workflows._functional": (
         "FunctionalWorkflow",
+        "FunctionalWorkflowDefinition",
         "FunctionalWorkflowAgent",
         "RunContext",
         "StepWrapper",
@@ -476,6 +477,7 @@
     "FunctionTool",
     "FunctionalWorkflow",
     "FunctionalWorkflowAgent",
+    "FunctionalWorkflowDefinition",
     "GeneratedEmbeddings",
     "GraphConnectivityError",
     "HistoryProvider",

python/packages/core/agent_framework/__init__.pyi

@@ -283,6 +283,7 @@ from ._workflows._function_executor import FunctionExecutor, executor
 from ._workflows._functional import (
     FunctionalWorkflow,
     FunctionalWorkflowAgent,
+    FunctionalWorkflowDefinition,
     RunContext,
     StepWrapper,
     get_run_context,
@@ -440,6 +441,7 @@ __all__ = [
     "FunctionTool",
     "FunctionalWorkflow",
     "FunctionalWorkflowAgent",
+    "FunctionalWorkflowDefinition",
     "GeneratedEmbeddings",
     "GraphConnectivityError",
     "HistoryProvider",

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

@@ -21,7 +21,10 @@
 
 Key public symbols:
 
-* :func:`workflow` / :class:`FunctionalWorkflow` — decorator and runtime.
+* :func:`workflow` / :class:`FunctionalWorkflowDefinition` — decorator and
+  stateless definition.
+* :class:`FunctionalWorkflow` — stateful runtime created by
+  :meth:`FunctionalWorkflowDefinition.build`.
 * :func:`step` / :class:`StepWrapper` — optional step decorator.
 * :class:`RunContext` — execution context injected into workflow and step
   functions.
@@ -628,6 +631,46 @@ def _decorator(fn: Callable[..., Awaitable[Any]]) -> StepWrapper[Any]:
     return _decorator
 
 
+# ---------------------------------------------------------------------------
+# FunctionalWorkflowDefinition
+# ---------------------------------------------------------------------------
+
+
+@experimental(feature_id=ExperimentalFeature.FUNCTIONAL_WORKFLOWS)
+class FunctionalWorkflowDefinition:
+    """Stateless definition produced by :func:`workflow`.
+
+    Call :meth:`build` to create a stateful :class:`FunctionalWorkflow`.
+    Each built workflow represents one logical caller or session.
+    """
+
+    def __init__(
+        self,
+        func: Callable[..., Awaitable[Any]],
+        *,
+        name: str | None = None,
+        description: str | None = None,
+    ) -> None:
+        FunctionalWorkflow._classify_signature(func)
+        self._func = func
+        self.name = name or func.__name__
+        self.description = description
+        functools.update_wrapper(self, func)  # type: ignore[arg-type]
+
+    def build(
+        self,
+        *,
+        checkpoint_storage: CheckpointStorage | None = None,
+    ) -> FunctionalWorkflow:
+        """Build a stateful workflow for one logical caller or session."""
+        return FunctionalWorkflow(
+            self._func,
+            name=self.name,
+            description=self.description,
+            checkpoint_storage=checkpoint_storage,
+        )
+
+
 # ---------------------------------------------------------------------------
 # FunctionalWorkflow
 # ---------------------------------------------------------------------------
@@ -637,15 +680,19 @@ def _decorator(fn: Callable[..., Awaitable[Any]]) -> StepWrapper[Any]:
 class FunctionalWorkflow:
     """A workflow backed by a user-defined async function.
 
-    Created by the :func:`workflow` decorator.  Exposes the same ``run()``
-    interface as graph-based :class:`Workflow` objects, returning a
+    Built from a :class:`FunctionalWorkflowDefinition`. Exposes the same
+    ``run()`` interface as graph-based :class:`Workflow` objects, returning a
     :class:`WorkflowRunResult` (or a :class:`ResponseStream` in streaming
     mode).
 
     The underlying function is executed directly — no graph compilation or
     edge wiring is involved.  Native Python control flow (``if``/``else``,
     ``for``, ``asyncio.gather``) is used for branching and parallelism.
 
+    Like graph-based :class:`Workflow`, each instance owns mutable execution
+    state across calls to :meth:`run`. Scope an instance to one logical
+    caller or session; build separate instances for independent callers.
+
     Args:
         func: The async function that implements the workflow logic.
         name: Display name for the workflow.  Defaults to ``func.__name__``.
@@ -664,7 +711,8 @@ async def my_pipeline(data: str) -> str:
                 return await to_upper(data)
 
 
-            result = await my_pipeline.run("hello")
+            pipeline = my_pipeline.build()
+            result = await pipeline.run("hello")
             print(result.get_outputs())  # ['HELLO']
     """
 
@@ -933,7 +981,7 @@ async def _run_core(
             if storage is None:
                 raise ValueError(
                     "Cannot restore from checkpoint without checkpoint_storage. "
-                    "Provide checkpoint_storage parameter or set it on the @workflow decorator."
+                    "Provide checkpoint_storage to build() or to this run."
                 )
             checkpoint = await storage.load(checkpoint_id)
             if checkpoint.graph_signature_hash != self.graph_signature_hash:
@@ -1258,16 +1306,15 @@ async def _run_cleanup(self) -> None:
 
 
 @overload
-def workflow(func: Callable[..., Awaitable[Any]]) -> FunctionalWorkflow: ...
+def workflow(func: Callable[..., Awaitable[Any]]) -> FunctionalWorkflowDefinition: ...
 
 
 @overload
 def workflow(
     *,
     name: str | None = None,
     description: str | None = None,
-    checkpoint_storage: CheckpointStorage | None = None,
-) -> Callable[[Callable[..., Awaitable[Any]]], FunctionalWorkflow]: ...
+) -> Callable[[Callable[..., Awaitable[Any]]], FunctionalWorkflowDefinition]: ...
 
 
 @experimental(feature_id=ExperimentalFeature.FUNCTIONAL_WORKFLOWS)
@@ -1276,29 +1323,26 @@ def workflow(
     *,
     name: str | None = None,
     description: str | None = None,
-    checkpoint_storage: CheckpointStorage | None = None,
-) -> FunctionalWorkflow | Callable[[Callable[..., Awaitable[Any]]], FunctionalWorkflow]:
-    """Decorator that converts an async function into a :class:`FunctionalWorkflow`.
+) -> FunctionalWorkflowDefinition | Callable[[Callable[..., Awaitable[Any]]], FunctionalWorkflowDefinition]:
+    """Decorator that creates a stateless :class:`FunctionalWorkflowDefinition`.
 
     Supports both bare ``@workflow`` and parameterized
     ``@workflow(name="my_wf")`` forms.
 
     The decorated function receives its input as the first positional argument
     and a :class:`RunContext` instance wherever a parameter is annotated with
-    that type.  The resulting :class:`FunctionalWorkflow` object exposes the
-    same ``run()`` interface as graph-based workflows.
+    that type. Call ``build()`` on the resulting definition to create a
+    stateful :class:`FunctionalWorkflow`.
 
     Args:
         func: The async function to decorate (when using the bare
             ``@workflow`` form).
         name: Display name for the workflow.  Defaults to ``func.__name__``.
         description: Optional human-readable description.
-        checkpoint_storage: Default :class:`CheckpointStorage` for
-            persisting step results and workflow state.
 
     Returns:
-        A :class:`FunctionalWorkflow` (bare form) or a decorator that
-        produces one (parameterized form).
+        A :class:`FunctionalWorkflowDefinition` (bare form) or a decorator
+        that produces one (parameterized form).
 
     Examples:
 
@@ -1311,14 +1355,17 @@ async def pipeline(data: str) -> str:
 
 
             # Parameterized form
-            @workflow(name="my_pipeline", checkpoint_storage=storage)
+            @workflow(name="my_pipeline")
             async def pipeline(data: str) -> str: ...
+
+
+            instance = pipeline.build(checkpoint_storage=storage)
     """
     if func is not None:
-        return FunctionalWorkflow(func, name=name, description=description, checkpoint_storage=checkpoint_storage)
+        return FunctionalWorkflowDefinition(func, name=name, description=description)
 
-    def _decorator(fn: Callable[..., Awaitable[Any]]) -> FunctionalWorkflow:
-        return FunctionalWorkflow(fn, name=name, description=description, checkpoint_storage=checkpoint_storage)
+    def _decorator(fn: Callable[..., Awaitable[Any]]) -> FunctionalWorkflowDefinition:
+        return FunctionalWorkflowDefinition(fn, name=name, description=description)
 
     return _decorator
 
@@ -1343,6 +1390,12 @@ class FunctionalWorkflowAgent:
     :class:`WorkflowAgent`), so HITL workflows are callable via this
     adapter.  Callers resume via ``responses=`` / ``checkpoint_id=``.
 
+    The wrapped workflow owns mutable execution state. Scope the workflow and
+    this adapter to one logical caller or session; create separate workflow
+    instances for independent or mutually untrusted callers. If those
+    instances use checkpoint storage, the host must also authorize and
+    tenant-scope access to that external store.
+
     Args:
         workflow: The :class:`FunctionalWorkflow` to wrap.
         name: Display name for the agent.  Defaults to the workflow name.