Python: fix(workflows): support keyword-only RunContext in functional workflows

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

@@ -2294,22 +2294,29 @@ async def _execute(self, ctx: RunContext, message: Any) -> Any:
             #   async def my_workflow(data: str, ctx: RunContext) -> str:
             #   async def my_workflow(data: str, ctx) -> str:
             call_args: list[Any] = []
+            call_kwargs: dict[str, Any] = {}
             message_injected = False
 
             for param in params:
                 resolved = hints.get(param.name, param.annotation)
                 if resolved is RunContext or param.name == "ctx":
-                    call_args.append(ctx)
+                    if param.kind == inspect.Parameter.KEYWORD_ONLY:
+                        call_kwargs[param.name] = ctx
+                    else:
+                        call_args.append(ctx)
                 elif not message_injected:
                     # First non-ctx param gets the message
-                    call_args.append(message)
+                    if param.kind == inspect.Parameter.KEYWORD_ONLY:
+                        call_kwargs[param.name] = message
+                    else:
+                        call_args.append(message)
                     message_injected = True
 
             root_task_ready = asyncio.Event()
 
             async def _invoke_workflow() -> Any:
                 await root_task_ready.wait()
-                return await self._func(*call_args)
+                return await self._func(*call_args, **call_kwargs)
 
             workflow_task = asyncio.create_task(_invoke_workflow())
             ctx._root_task = workflow_task

python/packages/core/tests/workflow/test_functional_workflow.py

@@ -4393,6 +4393,57 @@ async def wf(ctx: RunContext) -> str:
         assert wf is not None
 
 
+class TestKeywordOnlyRunContext:
+    """Regression tests for Issue #9324: keyword-only RunContext parameter injection."""
+
+    async def test_workflow_with_keyword_only_run_context(self):
+        """A @workflow function with a keyword-only RunContext parameter injects properly."""
+
+        @built_workflow
+        async def wf(text: str, *, ctx: RunContext) -> str:
+            ctx.set_state("text", text)
+            return ctx.get_state("text")
+
+        result = await wf.run("hello")
+        assert result.get_outputs() == ["hello"]
+
+    async def test_workflow_with_both_keyword_only_parameters(self):
+        """A @workflow function with both message and context keyword-only injects properly."""
+
+        @built_workflow
+        async def wf(*, text: str, ctx: RunContext) -> str:
+            ctx.set_state("text", text)
+            return ctx.get_state("text")
+
+        result = await wf.run("hello")
+        assert result.get_outputs() == ["hello"]
+
+    async def test_workflow_with_keyword_only_ctx_named_param(self):
+        """An unannotated keyword-only ctx parameter is injected properly."""
+
+        @built_workflow
+        async def wf(text: str, *, ctx) -> str:
+            ctx.set_state("val", text)
+            return ctx.get_state("val")
+
+        result = await wf.run("world")
+        assert result.get_outputs() == ["world"]
+
+    async def test_workflow_streaming_with_keyword_only_run_context(self):
+        """Streaming mode properly passes keyword-only RunContext parameter."""
+
+        @built_workflow
+        async def wf(text: str, *, ctx: RunContext) -> str:
+            ctx.set_state("text", text)
+            return ctx.get_state("text")
+
+        stream = wf.run("hello", stream=True)
+        async for _ in stream:
+            pass
+        result = await stream.get_final_response()
+        assert result.get_outputs() == ["hello"]
+
+
 class TestStaleResponsesRejected:
     """Regression for bug_014: stale responses after clean completion must be rejected."""