Python: fix(declarative): keep internal workflow-routing kwargs out of the Agent.run splat

python/packages/declarative/agent_framework_declarative/_workflows/_executors_agents.py

@@ -806,8 +806,10 @@ async def _invoke_agent_and_store_results(
             # Merge caller-provided options to avoid duplicate keyword argument
             options = dict(run_kwargs.get("options") or {})
             options["additional_function_arguments"] = run_kwargs
-            # Exclude 'options' from splat to avoid TypeError on duplicate keyword
-            run_kwargs = {k: v for k, v in run_kwargs.items() if k != "options"}
+            # Exclude 'options' from splat to avoid TypeError on duplicate keyword,
+            # and keep internal workflow-routing copies (stored under underscore
+            # keys for nested executors) out of the public Agent.run signature
+            run_kwargs = {k: v for k, v in run_kwargs.items() if k != "options" and not k.startswith("_")}
 
         # Use run() method to get properly structured messages (including tool calls and results)
         # This is critical for multi-turn conversations where tool calls must be followed

python/packages/declarative/tests/test_graph_executors.py

@@ -427,7 +427,6 @@ def mock_set(key, value):
     async def test_invoke_agent_not_found(self, mock_context, mock_state):
         """Test InvokeAzureAgentExecutor raises error when agent not found."""
         from agent_framework.exceptions import AgentInvalidRequestException
-
         from agent_framework_declarative._workflows import (
             InvokeAzureAgentExecutor,
         )
@@ -449,6 +448,80 @@ async def test_invoke_agent_not_found(self, mock_context, mock_state):
         assert "non_existent_agent" in str(exc_info.value)
         assert "not found in registry" in str(exc_info.value)
 
+    @pytest.mark.asyncio
+    async def test_invoke_agent_skips_internal_workflow_kwargs(self, mock_context, mock_state):
+        """Internal underscore keys in the run-kwargs bag must not reach Agent.run.
+
+        Regression for #8413: the workflow state bag carries internal routing
+        copies (e.g. `_raw_function_invocation_kwargs`) next to the public
+        kwargs, and the declarative agent step used to splat the whole bag into
+        `Agent.run`, which takes no `**kwargs` and raised TypeError.
+        """
+        from types import SimpleNamespace
+
+        from agent_framework_declarative._workflows import InvokeAzureAgentExecutor
+
+        state = DeclarativeWorkflowState(mock_state)
+        state.initialize()
+
+        class _StrictAgent:
+            """Same public surface as Agent.run: no **kwargs catch-all."""
+
+            def __init__(self) -> None:
+                self.received_options: dict[str, Any] | None = None
+                self.received_public_kwargs: dict[str, Any] = {}
+
+            async def run(
+                self,
+                messages,
+                options=None,
+                function_invocation_kwargs=None,
+                client_kwargs=None,
+                session=None,
+                tools=None,
+                stream=False,
+            ):
+                self.received_options = options
+                self.received_public_kwargs = {
+                    "function_invocation_kwargs": function_invocation_kwargs,
+                    "client_kwargs": client_kwargs,
+                }
+                return SimpleNamespace(text="ok", messages=[], tool_calls=[])
+
+        agent = _StrictAgent()
+        run_kwargs_bag = {
+            "function_invocation_kwargs": {"forwarded_props": {"a": 1}},
+            "_raw_function_invocation_kwargs": {"forwarded_props": {"a": 1}},
+        }
+        mock_context.get_state = MagicMock(
+            side_effect=lambda key, default=None: run_kwargs_bag if key == "_workflow_run_kwargs" else default
+        )
+
+        executor = InvokeAzureAgentExecutor(
+            {"kind": "InvokeAzureAgent", "agent": "StubAgent", "input": "hello"},
+            agents={"StubAgent": agent},
+        )
+
+        # Before the fix this raises TypeError: Agent.run() got an unexpected
+        # keyword argument '_raw_function_invocation_kwargs'.
+        await executor._invoke_agent_and_store_results(
+            agent,
+            "StubAgent",
+            "hello",
+            state,
+            mock_context,
+            messages_var=None,
+            response_obj_var=None,
+            result_property=None,
+            auto_send=False,
+        )
+
+        assert agent.received_options is not None
+        # The full bag still reaches tool forwarding, internal copies included.
+        forwarded = agent.received_options["additional_function_arguments"]
+        assert forwarded["_raw_function_invocation_kwargs"] == {"forwarded_props": {"a": 1}}
+        assert forwarded["function_invocation_kwargs"] == {"forwarded_props": {"a": 1}}
+
 
 class TestHumanInputExecutors:
     """Tests for human input executors."""
@@ -1824,7 +1897,6 @@ async def test_workflow_run_kwargs_survive_file_checkpoint_restore(
             RESOLVED_WORKFLOW_RUN_KWARGS_KEY,
             WORKFLOW_RUN_KWARGS_KEY,
         )
-
         from agent_framework_declarative._workflows._executors_external_input import ExternalInputResponse
 
         storage = FileCheckpointStorage(
@@ -1911,7 +1983,9 @@ def build_workflow() -> tuple[Any, Any]:
             )
             expected_run_kwargs = {kwargs_channel: invocation_kwargs, raw_key: invocation_kwargs}
             assert call_kwargs[kwargs_channel] == invocation_kwargs
-            assert call_kwargs[raw_key] == invocation_kwargs
+            # The internal raw copy stays routing state for nested executors;
+            # it is not a public Agent.run parameter (#8413).
+            assert raw_key not in call_kwargs
         else:
             expected_run_kwargs = {kwargs_channel: {"shared": "G", "specific": "S"}}
             assert call_kwargs[kwargs_channel] == {"shared": "G", "specific": "S"}
@@ -1924,7 +1998,6 @@ async def test_invoke_agent_forwards_kwargs(self):
         """InvokeAzureAgentExecutor should forward run_kwargs to agent.run()."""
         from agent_framework._workflows._const import WORKFLOW_RUN_KWARGS_KEY
         from agent_framework._workflows._state import State
-
         from agent_framework_declarative._workflows._executors_agents import (
             InvokeAzureAgentExecutor,
         )
@@ -1996,7 +2069,6 @@ async def test_invoke_agent_merges_caller_options(self):
         """Caller-provided options in run_kwargs should be merged, not cause TypeError."""
         from agent_framework._workflows._const import WORKFLOW_RUN_KWARGS_KEY
         from agent_framework._workflows._state import State
-
         from agent_framework_declarative._workflows._executors_agents import (
             InvokeAzureAgentExecutor,
         )