diff --git a/python/PACKAGE_STATUS.md b/python/PACKAGE_STATUS.md
index f1dfb5bdc..d14af5d06 100644
--- a/python/PACKAGE_STATUS.md
+++ b/python/PACKAGE_STATUS.md
@@ -115,7 +115,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`
 
diff --git a/python/packages/core/AGENTS.md b/python/packages/core/AGENTS.md
index 298537336..593aa1e4b 100644
--- a/python/packages/core/AGENTS.md
+++ b/python/packages/core/AGENTS.md
@@ -201,6 +201,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
diff --git a/python/packages/core/agent_framework/__init__.py b/python/packages/core/agent_framework/__init__.py
index d7ab91687..46e0d9285 100644
--- a/python/packages/core/agent_framework/__init__.py
+++ b/python/packages/core/agent_framework/__init__.py
@@ -321,6 +321,7 @@ _LAZY_MODULE_EXPORTS: Final[Mapping[str, tuple[str, ...]]] = {
     "._workflows._function_executor": ("FunctionExecutor", "executor"),
     "._workflows._functional": (
         "FunctionalWorkflow",
+        "FunctionalWorkflowDefinition",
         "FunctionalWorkflowAgent",
         "RunContext",
         "StepWrapper",
@@ -478,6 +479,7 @@ __all__ = [
     "FunctionTool",
     "FunctionalWorkflow",
     "FunctionalWorkflowAgent",
+    "FunctionalWorkflowDefinition",
     "GeneratedEmbeddings",
     "GraphConnectivityError",
     "HistoryProvider",
diff --git a/python/packages/core/agent_framework/__init__.pyi b/python/packages/core/agent_framework/__init__.pyi
index f72269c02..4f4f10fb0 100644
--- a/python/packages/core/agent_framework/__init__.pyi
+++ b/python/packages/core/agent_framework/__init__.pyi
@@ -285,6 +285,7 @@ from ._workflows._function_executor import FunctionExecutor, executor
 from ._workflows._functional import (
     FunctionalWorkflow,
     FunctionalWorkflowAgent,
+    FunctionalWorkflowDefinition,
     RunContext,
     StepWrapper,
     get_run_context,
@@ -442,6 +443,7 @@ __all__ = [
     "FunctionTool",
     "FunctionalWorkflow",
     "FunctionalWorkflowAgent",
+    "FunctionalWorkflowDefinition",
     "GeneratedEmbeddings",
     "GraphConnectivityError",
     "HistoryProvider",
diff --git a/python/packages/core/agent_framework/_workflows/_functional.py b/python/packages/core/agent_framework/_workflows/_functional.py
index 2ffe99807..9e4deb340 100644
--- a/python/packages/core/agent_framework/_workflows/_functional.py
+++ b/python/packages/core/agent_framework/_workflows/_functional.py
@@ -21,7 +21,10 @@ parameter to access HITL and state APIs directly.
 
 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 step(
     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,8 +680,8 @@ def step(
 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).
 
@@ -646,6 +689,10 @@ class FunctionalWorkflow:
     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 @@ class FunctionalWorkflow:
                 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 @@ class FunctionalWorkflow:
             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,7 +1306,7 @@ class FunctionalWorkflow:
 
 
 @overload
-def workflow(func: Callable[..., Awaitable[Any]]) -> FunctionalWorkflow: ...
+def workflow(func: Callable[..., Awaitable[Any]]) -> FunctionalWorkflowDefinition: ...
 
 
 @overload
@@ -1266,8 +1314,7 @@ 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 @@ def workflow(
 
 
             # 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.
diff --git a/python/packages/core/tests/workflow/test_functional_workflow.py b/python/packages/core/tests/workflow/test_functional_workflow.py
index c4313e4f4..96b7e6edf 100644
--- a/python/packages/core/tests/workflow/test_functional_workflow.py
+++ b/python/packages/core/tests/workflow/test_functional_workflow.py
@@ -7,17 +7,20 @@ from __future__ import annotations
 import asyncio
 import json
 import logging
-from collections.abc import Iterator
+from collections.abc import Awaitable, Callable, Iterator
 from contextlib import contextmanager
 from dataclasses import dataclass
+from typing import Any, overload
 
 import pytest
 
 from agent_framework import (
     AgentResponseUpdate,
+    CheckpointStorage,
     ExperimentalFeature,
     FunctionalWorkflow,
     FunctionalWorkflowAgent,
+    FunctionalWorkflowDefinition,
     InMemoryCheckpointStorage,
     RunContext,
     StepWrapper,
@@ -37,6 +40,34 @@ from agent_framework._workflows._functional import (
 # ---------------------------------------------------------------------------
 
 
+@overload
+def built_workflow(func: Callable[..., Awaitable[Any]]) -> FunctionalWorkflow: ...
+
+
+@overload
+def built_workflow(
+    *,
+    name: str | None = None,
+    description: str | None = None,
+    checkpoint_storage: CheckpointStorage | None = None,
+) -> Callable[[Callable[..., Awaitable[Any]]], FunctionalWorkflow]: ...
+
+
+def built_workflow(
+    func: Callable[..., Awaitable[Any]] | None = None,
+    *,
+    name: str | None = None,
+    description: str | None = None,
+    checkpoint_storage: CheckpointStorage | None = None,
+) -> FunctionalWorkflow | Callable[[Callable[..., Awaitable[Any]]], FunctionalWorkflow]:
+    """Build a fresh executable workflow for behavior-focused tests."""
+
+    def decorate(fn: Callable[..., Awaitable[Any]]) -> FunctionalWorkflow:
+        return workflow(name=name, description=description)(fn).build(checkpoint_storage=checkpoint_storage)
+
+    return decorate(func) if func is not None else decorate
+
+
 @step
 async def add_one(x: int) -> int:
     return x + 1
@@ -69,7 +100,7 @@ async def failing_step(x: int) -> int:
 
 class TestBasicExecution:
     async def test_simple_sequential_pipeline(self):
-        @workflow
+        @built_workflow
         async def pipeline(x: int) -> int:
             a = await add_one(x)
             return await double(a)
@@ -80,7 +111,7 @@ class TestBasicExecution:
         assert outputs == [12]  # (5+1)*2
 
     async def test_workflow_with_string_data(self):
-        @workflow
+        @built_workflow
         async def upper_pipeline(text: str) -> str:
             return await to_upper(text)
 
@@ -88,7 +119,7 @@ class TestBasicExecution:
         assert result.get_outputs() == ["HELLO"]
 
     async def test_workflow_returns_result(self):
-        @workflow
+        @built_workflow
         async def simple(x: int) -> int:
             return await add_one(x)
 
@@ -96,14 +127,14 @@ class TestBasicExecution:
         assert result.get_outputs() == [11]
 
     async def test_workflow_name_defaults_to_function_name(self):
-        @workflow
+        @built_workflow
         async def my_pipeline(x: int) -> int:
             return x
 
         assert my_pipeline.name == "my_pipeline"
 
     async def test_workflow_custom_name(self):
-        @workflow(name="custom_wf", description="A test workflow")
+        @built_workflow(name="custom_wf", description="A test workflow")
         async def wf(x: int) -> int:
             return x
 
@@ -118,7 +149,7 @@ class TestBasicExecution:
 
 class TestEventEmission:
     async def test_step_events_emitted(self):
-        @workflow
+        @built_workflow
         async def pipeline(x: int) -> int:
             return await add_one(x)
 
@@ -129,7 +160,7 @@ class TestEventEmission:
         assert "output" in event_types
 
     async def test_step_events_carry_executor_id(self):
-        @workflow
+        @built_workflow
         async def pipeline(x: int) -> int:
             return await add_one(x)
 
@@ -144,7 +175,7 @@ class TestEventEmission:
         assert completed_events[0].data == 6
 
     async def test_status_events_in_timeline(self):
-        @workflow
+        @built_workflow
         async def pipeline(x: int) -> int:
             return x
 
@@ -154,7 +185,7 @@ class TestEventEmission:
         assert WorkflowRunState.IDLE in states
 
     async def test_final_state_is_idle(self):
-        @workflow
+        @built_workflow
         async def pipeline(x: int) -> int:
             return x
 
@@ -164,7 +195,7 @@ class TestEventEmission:
     async def test_custom_event(self):
         from agent_framework import WorkflowEvent
 
-        @workflow
+        @built_workflow
         async def pipeline(x: int, ctx: RunContext) -> int:
             await ctx.add_event(WorkflowEvent("intermediate", executor_id="pipeline", data="custom_data"))
             return x
@@ -192,7 +223,7 @@ class TestParallelExecution:
             await asyncio.sleep(0.01)
             return x * 2
 
-        @workflow
+        @built_workflow
         async def parallel_wf(x: int) -> list[int]:
             a, b = await asyncio.gather(slow_add(x), slow_double(x))
             return [a, b]
@@ -210,7 +241,7 @@ class TestParallelExecution:
         async def task_b(x: int) -> int:
             return x * 2
 
-        @workflow
+        @built_workflow
         async def par_wf(x: int) -> tuple[int, int]:
             a, b = await asyncio.gather(task_a(x), task_b(x))
             return (a, b)
@@ -228,8 +259,50 @@ class TestParallelExecution:
 
 
 class TestHITL:
-    async def test_request_info_interrupts(self):
+    async def test_workflow_definition_builds_isolated_pending_continuations(self):
         @workflow
+        async def review_wf(doc: str, ctx: RunContext) -> str:
+            feedback = await ctx.request_info(doc, response_type=str)
+            return f"{doc}:{feedback}"
+
+        assert not hasattr(review_wf, "run")
+
+        caller_a = review_wf.build()
+        caller_b = review_wf.build()
+
+        caller_a_paused = await caller_a.run("caller-a")
+        request_id = caller_a_paused.get_request_info_events()[0].request_id
+
+        with pytest.raises(ValueError, match="no pending request_info events"):
+            await caller_b.run(responses={request_id: "caller-b-response"})
+
+        caller_a_completed = await caller_a.run(responses={request_id: "caller-a-response"})
+        assert caller_a_completed.get_outputs() == ["caller-a:caller-a-response"]
+
+    async def test_build_does_not_inherit_checkpoint_storage_by_default(self):
+        @workflow
+        async def review_wf(doc: str) -> str:
+            return doc
+
+        caller = review_wf.build()
+
+        with pytest.raises(ValueError, match="checkpoint_storage"):
+            await caller.run(checkpoint_id="missing")
+
+    async def test_build_accepts_tenant_scoped_checkpoint_storage(self):
+        caller_storage = InMemoryCheckpointStorage()
+
+        @workflow
+        async def review_wf(doc: str, ctx: RunContext) -> str:
+            return await ctx.request_info(doc, response_type=str)
+
+        caller = review_wf.build(checkpoint_storage=caller_storage)
+        await caller.run("caller")
+
+        assert len(await caller_storage.list_checkpoints(workflow_name="review_wf")) == 1
+
+    async def test_request_info_interrupts(self):
+        @built_workflow
         async def review_wf(doc: str, ctx: RunContext) -> str:
             feedback = await ctx.request_info({"draft": doc}, response_type=str, request_id="req1")
             return f"Final: {feedback}"
@@ -242,7 +315,7 @@ class TestHITL:
         assert request_events[0].request_id == "req1"
 
     async def test_request_info_resume(self):
-        @workflow
+        @built_workflow
         async def review_wf(doc: str, ctx: RunContext) -> str:
             feedback = await ctx.request_info({"draft": doc}, response_type=str, request_id="req1")
             return f"Final: {feedback}"
@@ -260,7 +333,7 @@ class TestHITL:
     async def test_fresh_message_while_pending_requests_warns(self, caplog: pytest.LogCaptureFixture) -> None:
         """A fresh message while request_info events are pending is allowed but logs a warning."""
 
-        @workflow
+        @built_workflow
         async def review_wf(doc: str, ctx: RunContext) -> str:
             feedback = await ctx.request_info({"draft": doc}, response_type=str, request_id="req1")
             return f"Final: {feedback}"
@@ -279,7 +352,7 @@ class TestHITL:
     async def test_responses_while_pending_requests_does_not_warn(self, caplog: pytest.LogCaptureFixture) -> None:
         """Delivering responses is the normal completion path and must not warn."""
 
-        @workflow
+        @built_workflow
         async def review_wf(doc: str, ctx: RunContext) -> str:
             feedback = await ctx.request_info({"draft": doc}, response_type=str, request_id="req1")
             return f"Final: {feedback}"
@@ -297,7 +370,7 @@ class TestHITL:
     async def test_untyped_ctx_parameter(self):
         """ctx is injected by parameter name even without a RunContext annotation."""
 
-        @workflow  # pyright: ignore[reportUnknownArgumentType]
+        @built_workflow  # pyright: ignore[reportUnknownArgumentType]
         async def review_wf(doc: str, ctx) -> str:  # pyright: ignore[reportMissingParameterType, reportUnknownParameterType]
             feedback: str = await ctx.request_info({"draft": doc}, response_type=str, request_id="req1")  # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]
             return f"Final: {feedback}"
@@ -309,7 +382,7 @@ class TestHITL:
         assert result2.get_outputs() == ["Final: LGTM"]
 
     async def test_multiple_sequential_interrupts(self):
-        @workflow
+        @built_workflow
         async def multi_hitl(data: str, ctx: RunContext) -> str:
             r1 = await ctx.request_info("step1", response_type=str, request_id="r1")
             r2 = await ctx.request_info("step2", response_type=str, request_id="r2")
@@ -330,7 +403,7 @@ class TestHITL:
         assert result3.get_outputs() == ["A+B"]
 
     async def test_request_info_auto_generates_id(self):
-        @workflow
+        @built_workflow
         async def auto_id_wf(x: int, ctx: RunContext) -> None:
             await ctx.request_info("need data", response_type=str)
 
@@ -347,7 +420,7 @@ class TestHITL:
 
 class TestErrorHandling:
     async def test_step_failure_propagates(self):
-        @workflow
+        @built_workflow
         async def failing_wf(x: int) -> None:
             await failing_step(x)
 
@@ -355,7 +428,7 @@ class TestErrorHandling:
             await failing_wf.run(42)
 
     async def test_step_failure_emits_executor_failed(self):
-        @workflow
+        @built_workflow
         async def failing_wf(x: int) -> None:
             await failing_step(x)
 
@@ -371,7 +444,7 @@ class TestErrorHandling:
         assert failed_events[0].executor_id == "failing_step"
 
     async def test_workflow_failure_emits_failed_status(self):
-        @workflow
+        @built_workflow
         async def bad_wf(x: int) -> None:
             raise RuntimeError("workflow broke")
 
@@ -387,7 +460,7 @@ class TestErrorHandling:
         assert any(e.state == WorkflowRunState.FAILED for e in status_events)
 
     async def test_invalid_params_message_and_responses(self):
-        @workflow
+        @built_workflow
         async def wf(x: int) -> None:
             pass
 
@@ -395,7 +468,7 @@ class TestErrorHandling:
             await wf.run("hello", responses={"r1": "val"})
 
     async def test_invalid_params_message_and_checkpoint(self):
-        @workflow
+        @built_workflow
         async def wf(x: int) -> None:
             pass
 
@@ -403,7 +476,7 @@ class TestErrorHandling:
             await wf.run("hello", checkpoint_id="abc")
 
     async def test_invalid_params_nothing(self):
-        @workflow
+        @built_workflow
         async def wf(x: int) -> None:
             pass
 
@@ -418,7 +491,7 @@ class TestErrorHandling:
 
 class TestStreaming:
     async def test_streaming_yields_events(self):
-        @workflow
+        @built_workflow
         async def pipeline(x: int) -> int:
             return await add_one(x)
 
@@ -434,7 +507,7 @@ class TestStreaming:
         assert "output" in event_types
 
     async def test_streaming_final_response(self):
-        @workflow
+        @built_workflow
         async def pipeline(x: int) -> int:
             return await add_one(x)
 
@@ -446,7 +519,7 @@ class TestStreaming:
     async def test_streaming_context_reports_streaming(self):
         streaming_flag = None
 
-        @workflow
+        @built_workflow
         async def wf(x: int, ctx: RunContext) -> int:
             nonlocal streaming_flag
             streaming_flag = ctx.is_streaming()  # type: ignore[assignment]
@@ -491,7 +564,7 @@ class TestStepPassthrough:
 
 class TestStateManagement:
     async def test_get_set_state(self):
-        @workflow
+        @built_workflow
         async def stateful_wf(x: int, ctx: RunContext) -> int:
             ctx.set_state("counter", x)
             return ctx.get_state("counter")
@@ -500,7 +573,7 @@ class TestStateManagement:
         assert result.get_outputs() == [42]
 
     async def test_get_state_default(self):
-        @workflow
+        @built_workflow
         async def wf(x: int, ctx: RunContext) -> str:
             return ctx.get_state("missing", "default_val")
 
@@ -521,7 +594,7 @@ class TestCheckpointing:
         async def expensive(x: int) -> int:
             return x * 100
 
-        @workflow(checkpoint_storage=storage)
+        @built_workflow(checkpoint_storage=storage)
         async def ckpt_wf(x: int) -> int:
             return await expensive(x)
 
@@ -539,7 +612,7 @@ class TestCheckpointing:
         async def compute(x: int) -> int:
             return x + 1
 
-        @workflow
+        @built_workflow
         async def wf(x: int) -> int:
             return await compute(x)
 
@@ -559,7 +632,7 @@ class TestCheckpointing:
             call_count += 1
             return x + 1
 
-        @workflow(checkpoint_storage=storage)
+        @built_workflow(checkpoint_storage=storage)
         async def wf(x: int) -> int:
             return await counting_task(x)
 
@@ -580,7 +653,7 @@ class TestCheckpointing:
     async def test_checkpoint_hitl_resume(self):
         storage = InMemoryCheckpointStorage()
 
-        @workflow(checkpoint_storage=storage)
+        @built_workflow(checkpoint_storage=storage)
         async def hitl_wf(doc: str, ctx: RunContext) -> str:
             feedback = await ctx.request_info({"draft": doc}, response_type=str, request_id="req1")
             return f"Done: {feedback}"
@@ -598,7 +671,7 @@ class TestCheckpointing:
         assert result2.get_outputs() == ["Done: Approved!"]
 
     async def test_checkpoint_without_storage_raises(self):
-        @workflow
+        @built_workflow
         async def wf(x: int) -> int:
             return x
 
@@ -608,7 +681,7 @@ class TestCheckpointing:
     async def test_checkpoint_preserves_state(self):
         storage = InMemoryCheckpointStorage()
 
-        @workflow(checkpoint_storage=storage)
+        @built_workflow(checkpoint_storage=storage)
         async def stateful_wf(x: int, ctx: RunContext) -> str:
             ctx.set_state("key", "value")
             feedback = await ctx.request_info("need info", response_type=str, request_id="r1")
@@ -648,7 +721,7 @@ class TestCheckpointing:
                 raise RuntimeError("simulated crash")
             return x * 2
 
-        @workflow(checkpoint_storage=storage)
+        @built_workflow(checkpoint_storage=storage)
         async def crash_wf(x: int) -> int:
             a = await slow_step1(x)
             return await crashing_step2(a)
@@ -687,7 +760,7 @@ class TestCheckpointing:
         async def s3(x: int) -> int:
             return x + 3
 
-        @workflow(checkpoint_storage=storage)
+        @built_workflow(checkpoint_storage=storage)
         async def multi_step_wf(x: int) -> int:
             a = await s1(x)
             b = await s2(a)
@@ -708,7 +781,7 @@ class TestCheckpointing:
         async def compute(x: int) -> int:
             return x + 1
 
-        @workflow(checkpoint_storage=storage)
+        @built_workflow(checkpoint_storage=storage)
         async def wf(x: int) -> int:
             return await compute(x)
 
@@ -748,7 +821,7 @@ class TestControlFlow:
         async def quarantine(text: str) -> str:
             return f"quarantined: {text}"
 
-        @workflow
+        @built_workflow
         async def email_pipeline(email: str) -> str:
             cl = await classify(email)
             if cl.is_spam:
@@ -775,7 +848,7 @@ class TestNestedWorkflows:
         async def step_a(x: int) -> int:
             return x + 1
 
-        @workflow
+        @built_workflow
         async def inner_wf(x: int) -> int:
             return await step_a(x)
 
@@ -784,7 +857,7 @@ class TestNestedWorkflows:
             result = await inner_wf.run(x)
             return result.get_outputs()[0]
 
-        @workflow
+        @built_workflow
         async def outer_wf(x: int) -> int:
             return await call_inner(x)
 
@@ -799,7 +872,7 @@ class TestNestedWorkflows:
 
 class TestAsAgent:
     async def test_as_agent_returns_agent(self):
-        @workflow
+        @built_workflow
         async def wf(x: int) -> str:
             return f"result: {x}"
 
@@ -807,7 +880,7 @@ class TestAsAgent:
         assert agent.name == "wf"
 
     async def test_as_agent_custom_name(self):
-        @workflow
+        @built_workflow
         async def wf(x: int) -> int:
             return x
 
@@ -815,7 +888,7 @@ class TestAsAgent:
         assert agent.name == "my_agent"
 
     async def test_as_agent_run(self):
-        @workflow
+        @built_workflow
         async def wf(x: int) -> int:
             return await add_one(x)
 
@@ -824,7 +897,7 @@ class TestAsAgent:
         assert response.text == "11"
 
     async def test_as_agent_run_streaming(self):
-        @workflow
+        @built_workflow
         async def wf(x: int) -> str:
             return f"result: {x}"
 
@@ -840,7 +913,7 @@ class TestAsAgent:
         assert len(response.messages) >= 1
 
     async def test_as_agent_has_id_and_description(self):
-        @workflow(description="A test workflow")
+        @built_workflow(description="A test workflow")
         async def wf(x: int) -> int:
             return x
 
@@ -856,7 +929,7 @@ class TestAsAgent:
 
 class TestConcurrencyGuard:
     async def test_concurrent_run_raises(self):
-        @workflow
+        @built_workflow
         async def slow_wf(x: int) -> int:
             await asyncio.sleep(0.1)
             return x
@@ -872,7 +945,7 @@ class TestConcurrencyGuard:
         await stream.get_final_response()
 
     async def test_run_after_completion(self):
-        @workflow
+        @built_workflow
         async def wf(x: int) -> int:
             return x
 
@@ -911,7 +984,7 @@ class TestDecoratorForms:
         async def my_wf(x: int) -> None:
             pass
 
-        assert isinstance(my_wf, FunctionalWorkflow)
+        assert isinstance(my_wf, FunctionalWorkflowDefinition)
         assert my_wf.name == "my_wf"
 
     def test_workflow_with_params(self):
@@ -919,7 +992,7 @@ class TestDecoratorForms:
         async def my_wf(x: int) -> None:
             pass
 
-        assert isinstance(my_wf, FunctionalWorkflow)
+        assert isinstance(my_wf, FunctionalWorkflowDefinition)
         assert my_wf.name == "custom"
         assert my_wf.description == "desc"
 
@@ -931,7 +1004,7 @@ class TestDecoratorForms:
 
 class TestIncludeStatusEvents:
     async def test_status_events_excluded_by_default(self):
-        @workflow
+        @built_workflow
         async def wf(x: int) -> int:
             return x
 
@@ -940,7 +1013,7 @@ class TestIncludeStatusEvents:
         assert len(status_in_list) == 0
 
     async def test_status_events_included_when_requested(self):
-        @workflow
+        @built_workflow
         async def wf(x: int) -> int:
             return x
 
@@ -956,7 +1029,7 @@ class TestIncludeStatusEvents:
 
 class TestEdgeCases:
     async def test_workflow_with_no_tasks(self):
-        @workflow
+        @built_workflow
         async def no_tasks(x: int) -> int:
             return x * 2
 
@@ -964,7 +1037,7 @@ class TestEdgeCases:
         assert result.get_outputs() == [10]
 
     async def test_workflow_with_no_output(self):
-        @workflow
+        @built_workflow
         async def silent_wf(x: int) -> None:
             pass  # returns None — no output emitted
 
@@ -974,7 +1047,7 @@ class TestEdgeCases:
     async def test_return_value_auto_yields_output(self):
         """Returning a non-None value automatically emits it as an output."""
 
-        @workflow
+        @built_workflow
         async def wf(x: int) -> int:
             return x * 3
 
@@ -982,7 +1055,7 @@ class TestEdgeCases:
         assert result.get_outputs() == [15]
 
     async def test_step_called_multiple_times(self):
-        @workflow
+        @built_workflow
         async def wf(x: int) -> int:
             a = await add_one(x)
             b = await add_one(a)
@@ -1005,7 +1078,7 @@ class TestEdgeCases:
 
 class TestRecoveryAfterErrors:
     async def test_run_after_failure_is_allowed(self):
-        @workflow
+        @built_workflow
         async def wf(x: int) -> int:
             if x == 1:
                 raise RuntimeError("boom")
@@ -1036,7 +1109,7 @@ class TestWorkflowInterruptedIsBaseException:
         """User code with ``except Exception`` should not catch WorkflowInterrupted."""
         caught = False
 
-        @workflow
+        @built_workflow
         async def wf(x: int, ctx: RunContext) -> str:
             nonlocal caught
             try:
@@ -1064,7 +1137,7 @@ class TestCheckpointValidation:
 
         storage = InMemoryCheckpointStorage()
 
-        @workflow(name="my_wf", checkpoint_storage=storage)
+        @built_workflow(name="my_wf", checkpoint_storage=storage)
         async def wf(x: int) -> int:
             return x
 
@@ -1108,7 +1181,7 @@ class TestExecutorBypassed:
             call_count += 1
             return x + 1
 
-        @workflow(checkpoint_storage=storage)
+        @built_workflow(checkpoint_storage=storage)
         async def wf(x: int) -> int:
             return await tracked(x)
 
@@ -1141,7 +1214,7 @@ class TestExecutorBypassed:
         async def compute(x: int) -> int:
             return x * 10
 
-        @workflow(checkpoint_storage=storage)
+        @built_workflow(checkpoint_storage=storage)
         async def wf(x: int) -> int:
             return await compute(x)
 
@@ -1169,7 +1242,7 @@ class TestRequestInfoInStep:
             feedback = await ctx.request_info({"draft": doc}, response_type=str, request_id="s1")
             return f"reviewed: {feedback}"
 
-        @workflow
+        @built_workflow
         async def wf(doc: str) -> str:
             return await review_step(doc)
 
@@ -1205,7 +1278,7 @@ class TestRequestInfoInStep:
             ctx.set_state("seen", data)
             return f"{data}:{ctx.get_state('seen')}"
 
-        @workflow
+        @built_workflow
         async def wf(data: str) -> str:
             return await needs_ctx_first(data)
 
@@ -1225,7 +1298,7 @@ class TestRequestInfoInStep:
             captured_ctx = get_run_context()  # type: ignore[assignment]
             return x
 
-        @workflow
+        @built_workflow
         async def wf(x: int) -> int:
             return await capture_ctx(x)
 
@@ -1249,7 +1322,7 @@ class TestNoneResponseHandling:
     async def test_none_response_logs_warning(self):
         """Providing None as a response value should log a warning."""
 
-        @workflow
+        @built_workflow
         async def wf(doc: str, ctx: RunContext) -> str:
             val = await ctx.request_info("need input", response_type=str, request_id="r1")
             return f"got: {val}"
@@ -1267,7 +1340,7 @@ class TestNoneResponseHandling:
     async def test_none_response_is_returned(self):
         """None is a valid (if discouraged) response value."""
 
-        @workflow
+        @built_workflow
         async def wf(x: int, ctx: RunContext) -> str:
             val = await ctx.request_info("need data", response_type=str, request_id="r1")
             return f"value={val}"
@@ -1322,7 +1395,7 @@ class TestHITLInStepWithCaching:
             feedback = await ctx.request_info({"val": val}, response_type=str, request_id="r1")
             return f"{val}:{feedback}"
 
-        @workflow
+        @built_workflow
         async def wf(x: int) -> str:
             a = await step_a(x)
             return await step_b(a)
@@ -1353,7 +1426,7 @@ class TestHITLInStepWithCaching:
             feedback = await ctx.request_info({"val": val}, response_type=str, request_id="rev")
             return f"reviewed({val}):{feedback}"
 
-        @workflow(checkpoint_storage=storage)
+        @built_workflow(checkpoint_storage=storage)
         async def wf(x: int) -> str:
             v = await compute(x)
             return await review(v)
@@ -1384,7 +1457,7 @@ class TestHITLInStepWithCaching:
             val = await ctx.request_info({"doc": doc}, response_type=str, request_id="r1")
             return f"got:{val}"
 
-        @workflow
+        @built_workflow
         async def wf(doc: str) -> str:
             return await needs_feedback(doc)
 
@@ -1403,7 +1476,7 @@ class TestHITLInStepWithCaching:
         async def hitl_step(x: int, ctx: RunContext) -> str:
             return await ctx.request_info("need data", response_type=str, request_id="r1")
 
-        @workflow
+        @built_workflow
         async def wf(x: int) -> str:
             return await hitl_step(x)
 
@@ -1422,7 +1495,7 @@ class TestDeterministicAutoRequestId:
     """Regression for bug_001: auto-generated request_info ids must be stable across replay."""
 
     async def test_auto_request_id_roundtrips_on_resume(self):
-        @workflow
+        @built_workflow
         async def wf(x: int, ctx: RunContext) -> str:
             # No request_id — framework must generate a deterministic one
             val = await ctx.request_info("need data", response_type=str)
@@ -1441,7 +1514,7 @@ class TestDeterministicAutoRequestId:
         assert result2.get_outputs() == ["got:hello"]
 
     async def test_multiple_auto_ids_are_distinct_and_stable(self):
-        @workflow
+        @built_workflow
         async def wf(x: int, ctx: RunContext) -> str:
             a = await ctx.request_info("first", response_type=str)
             b = await ctx.request_info("second", response_type=str)
@@ -1468,7 +1541,7 @@ class TestDeterministicAutoRequestId:
         async def second_review(value: int, ctx: RunContext) -> str:
             return await ctx.request_info({"step": "second", "value": value}, response_type=str)
 
-        @workflow
+        @built_workflow
         async def wf(value: int) -> str:
             first = await first_review(value)
             second = await second_review(value)
@@ -1495,7 +1568,7 @@ class TestPendingRequestsPruned:
     async def test_final_checkpoint_no_longer_claims_resolved_requests_pending(self):
         storage = InMemoryCheckpointStorage()
 
-        @workflow(checkpoint_storage=storage)
+        @built_workflow(checkpoint_storage=storage)
         async def wf(x: int, ctx: RunContext) -> str:
             a = await ctx.request_info("q1", response_type=str, request_id="r1")
             b = await ctx.request_info("q2", response_type=str, request_id="r2")
@@ -1523,7 +1596,7 @@ class TestArityValidation:
                 return f"{a}+{b}"
 
     async def test_ctx_only_workflow_with_message_raises_clear_error(self):
-        @workflow
+        @built_workflow
         async def wf(ctx: RunContext) -> str:
             return "no message used"
 
@@ -1535,7 +1608,7 @@ class TestArityValidation:
         # message-receiving parameter.  (Running it without a message still
         # requires providing responses or a checkpoint_id — that's
         # _validate_run_params's job, not ours.)
-        @workflow
+        @built_workflow
         async def wf(ctx: RunContext) -> str:
             return "ok"
 
@@ -1546,7 +1619,7 @@ class TestStaleResponsesRejected:
     """Regression for bug_014: stale responses after clean completion must be rejected."""
 
     async def test_responses_after_clean_completion_raise(self):
-        @workflow
+        @built_workflow
         async def wf(x: int) -> int:
             return x * 2
 
@@ -1555,7 +1628,7 @@ class TestStaleResponsesRejected:
             await wf.run(responses={"stale": "x"})
 
     async def test_responses_mismatched_key_raises(self):
-        @workflow
+        @built_workflow
         async def wf(x: int, ctx: RunContext) -> str:
             return await ctx.request_info("q", response_type=str, request_id="r1")
 
@@ -1568,7 +1641,7 @@ class TestReservedStateKeys:
     """Regression for bug_017: set_state must reject underscore-prefixed keys."""
 
     async def test_underscore_key_rejected(self):
-        @workflow
+        @built_workflow
         async def wf(x: int, ctx: RunContext) -> int:
             ctx.set_state("_private", "user value")
             return x
@@ -1577,7 +1650,7 @@ class TestReservedStateKeys:
             await wf.run(1)
 
     async def test_normal_key_still_works(self):
-        @workflow
+        @built_workflow
         async def wf(x: int, ctx: RunContext) -> int:
             ctx.set_state("normal_key", "v")
             assert ctx.get_state("normal_key") == "v"
@@ -1597,7 +1670,7 @@ class TestDeepcopyOnCacheHit:
         async def takes_lock(lock: threading.Lock, n: int) -> int:
             return n + 1
 
-        @workflow
+        @built_workflow
         async def wf(x: int) -> int:
             lock = threading.Lock()
             return await takes_lock(lock, x)
@@ -1613,11 +1686,11 @@ class TestStepDiscoveryAttributeAccess:
     """Regression for bug_008: checkpoint hash must differ when function body changes."""
 
     async def test_signature_hash_changes_when_function_body_changes(self):
-        @workflow
+        @built_workflow
         async def wf_a(x: int) -> int:
             return x + 1
 
-        @workflow(name="wf_b")
+        @built_workflow(name="wf_b")
         async def wf_b(x: int) -> int:
             return x * 100
 
@@ -1630,7 +1703,7 @@ class TestAsAgentSignatureParity:
     """Regression for bug_015: as_agent signature must accept description/context_providers."""
 
     async def test_as_agent_accepts_description_override(self):
-        @workflow(description="workflow level")
+        @built_workflow(description="workflow level")
         async def wf(x: str) -> str:
             return x.upper()
 
@@ -1638,7 +1711,7 @@ class TestAsAgentSignatureParity:
         assert agent.description == "agent level"
 
     async def test_as_agent_accepts_context_providers_kwarg(self):
-        @workflow
+        @built_workflow
         async def wf(x: str) -> str:
             return x
 
@@ -1647,7 +1720,7 @@ class TestAsAgentSignatureParity:
         assert list(agent.context_providers or []) == providers
 
     async def test_as_agent_description_defaults_to_workflow_description(self):
-        @workflow(description="from workflow")
+        @built_workflow(description="from workflow")
         async def wf(x: str) -> str:
             return x
 
@@ -1659,7 +1732,7 @@ class TestFunctionalWorkflowAgentHITL:
     """Regression for bug_013: .as_agent() must surface request_info events."""
 
     async def test_request_info_surfaces_as_function_approval_request(self):
-        @workflow
+        @built_workflow
         async def wf(x: str, ctx: RunContext) -> str:
             answer = await ctx.request_info({"need": x}, response_type=str, request_id="rid-1")
             return f"got:{answer}"
@@ -1686,7 +1759,7 @@ class TestFunctionalWorkflowAgentHITL:
             target_agent: str
             reason: str
 
-        @workflow
+        @built_workflow
         async def wf(x: str, ctx: RunContext) -> str:
             answer = await ctx.request_info(
                 HandoffRequest(target_agent=x, reason="overflow"),
@@ -1712,7 +1785,7 @@ class TestFunctionalWorkflowAgentHITL:
         assert json.loads(json.dumps(function_call_arguments)) == function_call_arguments
 
     async def test_resume_via_agent_responses_kwarg(self):
-        @workflow
+        @built_workflow
         async def wf(x: str, ctx: RunContext) -> str:
             answer = await ctx.request_info(x, response_type=str, request_id="rid-1")
             return f"got:{answer}"
@@ -1751,6 +1824,7 @@ class TestFunctionalWorkflowExperimentalStage:
             StepWrapper,
             step,
             FunctionalWorkflow,
+            FunctionalWorkflowDefinition,
             workflow,
             FunctionalWorkflowAgent,
         ]
diff --git a/python/samples/01-get-started/05_functional_workflow_with_agents.py b/python/samples/01-get-started/05_functional_workflow_with_agents.py
index 9aff97826..e77c1bc74 100644
--- a/python/samples/01-get-started/05_functional_workflow_with_agents.py
+++ b/python/samples/01-get-started/05_functional_workflow_with_agents.py
@@ -47,7 +47,8 @@ async def poem_workflow(topic: str) -> str:
 
 
 async def main() -> None:
-    result = await poem_workflow.run("a cat learning to code")
+    workflow_instance = poem_workflow.build()
+    result = await workflow_instance.run("a cat learning to code")
     print(result.get_outputs()[0])
 
 
diff --git a/python/samples/01-get-started/06_functional_workflow_basics.py b/python/samples/01-get-started/06_functional_workflow_basics.py
index a679e70a9..ec6fd79f5 100644
--- a/python/samples/01-get-started/06_functional_workflow_basics.py
+++ b/python/samples/01-get-started/06_functional_workflow_basics.py
@@ -43,7 +43,8 @@ async def text_workflow(text: str) -> str:
 
 async def main() -> None:
     # <run_workflow>
-    result = await text_workflow.run("hello world")
+    workflow_instance = text_workflow.build()
+    result = await workflow_instance.run("hello world")
     print(f"Output: {result.get_outputs()}")
     print(f"Final state: {result.get_final_state()}")
     # </run_workflow>
diff --git a/python/samples/03-workflows/functional/agent_integration.py b/python/samples/03-workflows/functional/agent_integration.py
index b5911cb69..23699e271 100644
--- a/python/samples/03-workflows/functional/agent_integration.py
+++ b/python/samples/03-workflows/functional/agent_integration.py
@@ -94,12 +94,15 @@ async def cached_pipeline(document: str) -> str:
 
 
 async def main():
+    simple_workflow = simple_pipeline.build()
+    cached_workflow = cached_pipeline.build()
+
     # Simple version — agents called inline
-    result = await simple_pipeline.run("This is a technical document about machine learning...")
+    result = await simple_workflow.run("This is a technical document about machine learning...")
     print(result.get_outputs()[0])
 
     # Cached version — same result, but steps won't re-execute on resume
-    result = await cached_pipeline.run("This is a technical document about machine learning...")
+    result = await cached_workflow.run("This is a technical document about machine learning...")
     print(f"\nCached: {result.get_outputs()[0]}")
 
 
diff --git a/python/samples/03-workflows/functional/basic_pipeline.py b/python/samples/03-workflows/functional/basic_pipeline.py
index 81514da53..eb32b54c6 100644
--- a/python/samples/03-workflows/functional/basic_pipeline.py
+++ b/python/samples/03-workflows/functional/basic_pipeline.py
@@ -23,8 +23,8 @@ async def transform_data(data: dict[str, str | int]) -> str:
     return f"[{data['status']}] {data['content']}"
 
 
-# @workflow turns this async function into a FunctionalWorkflow object.
-# Without it, this is just a normal async function. With it, you get:
+# @workflow turns this async function into a stateless workflow definition.
+# Call .build() to create a stateful FunctionalWorkflow with:
 #   - .run() that returns a WorkflowRunResult with events and outputs
 #   - .run(stream=True) for streaming events in real time
 #   - .as_agent() to use this workflow anywhere an agent is expected
@@ -48,8 +48,8 @@ async def data_pipeline(url: str) -> str:
 
 
 async def main():
-    # .run() is provided by @workflow — a plain async function wouldn't have it
-    result = await data_pipeline.run("https://example.com/api/data")
+    workflow_instance = data_pipeline.build()
+    result = await workflow_instance.run("https://example.com/api/data")
     print("Output:", result.get_outputs()[0])
     print("State:", result.get_final_state())
 
diff --git a/python/samples/03-workflows/functional/basic_streaming_pipeline.py b/python/samples/03-workflows/functional/basic_streaming_pipeline.py
index 4ee61da60..a986b97ee 100644
--- a/python/samples/03-workflows/functional/basic_streaming_pipeline.py
+++ b/python/samples/03-workflows/functional/basic_streaming_pipeline.py
@@ -26,9 +26,9 @@ async def validate_result(summary: str) -> bool:
     return len(summary) > 0 and "[200]" in summary
 
 
-# @workflow enables .run(stream=True), which returns a ResponseStream
-# you can iterate over with `async for`. Without @workflow, you'd just
-# have a normal async function with no streaming capability.
+# @workflow creates a definition. Its built workflow enables
+# .run(stream=True), which returns a ResponseStream you can iterate over with
+# `async for`.
 @workflow
 async def data_pipeline(url: str) -> str:
     """A simple sequential data pipeline."""
@@ -40,10 +40,12 @@ async def data_pipeline(url: str) -> str:
 
 
 async def main():
+    workflow_instance = data_pipeline.build()
+
     # run(stream=True) returns a ResponseStream that yields events as they
     # are produced. The raw stream includes lifecycle events (started, status)
     # alongside application events — filter by event.type to find what you need.
-    stream = data_pipeline.run("https://example.com/api/data", stream=True)
+    stream = workflow_instance.run("https://example.com/api/data", stream=True)
     async for event in stream:
         if event.type == "output":
             print(f"Output: {event.data}")
diff --git a/python/samples/03-workflows/functional/hitl_review.py b/python/samples/03-workflows/functional/hitl_review.py
index 39f2dae88..a9343978a 100644
--- a/python/samples/03-workflows/functional/hitl_review.py
+++ b/python/samples/03-workflows/functional/hitl_review.py
@@ -59,14 +59,17 @@ async def review_pipeline(topic: str, ctx: RunContext) -> str:
 
 
 async def main():
+    workflow_instance = review_pipeline.build()
+
     # Phase 1: Run until the workflow pauses for human input
     print("=== Phase 1: Initial run ===")
-    result1 = await review_pipeline.run("AI Safety")
+    result1 = await workflow_instance.run("AI Safety")
 
     # If request_info() was reached, the state is IDLE_WITH_PENDING_REQUESTS.
     # If the workflow completed without hitting request_info(), it would be IDLE.
     print(f"State: {(final_state := result1.get_final_state())}")
-    assert final_state == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS
+    if final_state != WorkflowRunState.IDLE_WITH_PENDING_REQUESTS:
+        raise RuntimeError(f"Expected pending review input, but workflow entered {final_state}.")
 
     requests = result1.get_request_info_events()
     print(f"Pending request: {requests[0].request_id}")
@@ -74,7 +77,7 @@ async def main():
     # Phase 2: Resume with the human's response
     print("\n=== Phase 2: Resume with feedback ===")
     print("(write_draft should NOT execute again — saved by @step)")
-    result2 = await review_pipeline.run(responses={"review_request": "Add more details about alignment research"})
+    result2 = await workflow_instance.run(responses={"review_request": "Add more details about alignment research"})
 
     print(f"State: {result2.get_final_state()}")
     print(f"Output: {result2.get_outputs()[0]}")
diff --git a/python/samples/03-workflows/functional/naive_group_chat.py b/python/samples/03-workflows/functional/naive_group_chat.py
index 45a326604..255b185fb 100644
--- a/python/samples/03-workflows/functional/naive_group_chat.py
+++ b/python/samples/03-workflows/functional/naive_group_chat.py
@@ -74,7 +74,8 @@ async def group_chat(question: str) -> str:
 
 
 async def main():
-    result = await group_chat.run("What's the difference between a list and a tuple in Python?")
+    workflow_instance = group_chat.build()
+    result = await workflow_instance.run("What's the difference between a list and a tuple in Python?")
     print(result.get_outputs()[0])
 
 
diff --git a/python/samples/03-workflows/functional/parallel_pipeline.py b/python/samples/03-workflows/functional/parallel_pipeline.py
index 21fd8dcea..87a695bf8 100644
--- a/python/samples/03-workflows/functional/parallel_pipeline.py
+++ b/python/samples/03-workflows/functional/parallel_pipeline.py
@@ -36,9 +36,9 @@ async def synthesize(sources: list[str]) -> str:
     return "Research Summary:\n" + "\n".join(f"  - {s}" for s in sources)
 
 
-# @workflow wraps the orchestration logic so you get .run(), streaming,
-# and events. The functions it calls are plain Python — no decorators
-# needed just because they're inside a workflow.
+# @workflow defines the orchestration logic. Build the definition to get a
+# stateful workflow with .run(), streaming, and events. The functions it calls
+# are plain Python — no decorators needed just because they're inside a workflow.
 @workflow
 async def research_pipeline(topic: str) -> str:
     """Fan-out to three research tasks, then synthesize results."""
@@ -58,7 +58,8 @@ async def research_pipeline(topic: str) -> str:
 
 
 async def main():
-    result = await research_pipeline.run("AI agents")
+    workflow_instance = research_pipeline.build()
+    result = await workflow_instance.run("AI agents")
     print(result.get_outputs()[0])
 
 
diff --git a/python/samples/03-workflows/functional/steps_and_checkpointing.py b/python/samples/03-workflows/functional/steps_and_checkpointing.py
index 93a9423df..a41649184 100644
--- a/python/samples/03-workflows/functional/steps_and_checkpointing.py
+++ b/python/samples/03-workflows/functional/steps_and_checkpointing.py
@@ -55,9 +55,9 @@ async def validate_result(summary: str) -> bool:
 storage = InMemoryCheckpointStorage()
 
 
-# checkpoint_storage tells @workflow where to persist step results.
+# Build with checkpoint_storage to persist step results.
 # Each @step saves a checkpoint after it completes.
-@workflow(checkpoint_storage=storage)
+@workflow
 async def data_pipeline(url: str) -> str:
     """Mix of @step functions and plain functions."""
     raw = await fetch_data(url)
@@ -68,9 +68,11 @@ async def data_pipeline(url: str) -> str:
 
 
 async def main():
+    workflow_instance = data_pipeline.build(checkpoint_storage=storage)
+
     # --- Run 1: Everything executes normally ---
     print("=== Run 1: Fresh execution ===")
-    result = await data_pipeline.run("https://example.com/api/data")
+    result = await workflow_instance.run("https://example.com/api/data")
     print(f"Output: {result.get_outputs()[0]}")
     print(f"fetch_calls={fetch_calls}, transform_calls={transform_calls}")
 
@@ -85,9 +87,10 @@ async def main():
     # Only validate_result() (no @step) actually runs again.
     print("\n=== Run 2: Restored from checkpoint ===")
     latest = await storage.get_latest(workflow_name="data_pipeline")
-    assert latest is not None
+    if latest is None:
+        raise RuntimeError("Expected a checkpoint from the first run.")
 
-    result2 = await data_pipeline.run(checkpoint_id=latest.checkpoint_id)
+    result2 = await workflow_instance.run(checkpoint_id=latest.checkpoint_id)
     print(f"Output: {result2.get_outputs()[0]}")
     print(f"fetch_calls={fetch_calls}, transform_calls={transform_calls}")
     print("(call counts unchanged — @step results were restored from checkpoint)")