Add workflow.in_workflow() utility

temporalio/workflow.py

@@ -844,6 +844,11 @@ def instance() -> Any:
     return _Runtime.current().workflow_instance()
 
 
+def in_workflow() -> bool:
+    """Whether the code is currently running in a workflow."""
+    return _Runtime.maybe_current() is not None
+
+
 def memo() -> Mapping[str, Any]:
     """Current workflow's memo values, converted without type hints.
 

tests/worker/test_workflow.py

@@ -6981,6 +6981,27 @@ async def test_update_handler_semaphore_acquisition_respects_timeout(
     )
 
 
+def check_in_workflow() -> str:
+    return "in workflow" if workflow.in_workflow() else "not in workflow"
+
+
+@workflow.defn
+class InWorkflowUtilWorkflow:
+    @workflow.run
+    async def run(self) -> str:
+        return check_in_workflow()
+
+
+async def test_in_workflow_util(client: Client):
+    assert check_in_workflow() == "not in workflow"
+    async with new_worker(client, InWorkflowUtilWorkflow) as worker:
+        assert "in workflow" == await client.execute_workflow(
+            InWorkflowUtilWorkflow.run,
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+        )
+
+
 deadlock_interruptible_completed = 0