Python: Forward function_invocation_kwargs through DevUI to agent.run…

python/packages/devui/agent_framework_devui/_executor.py

@@ -376,6 +376,17 @@ async def _execute_agent(
                 if session:
                     run_kwargs["session"] = session
 
+                # Forward request-scoped function_invocation_kwargs so tools can read them
+                # via FunctionInvocationContext. Accept them in extra_body (same channel as
+                # response_id/checkpoint_id) or as a top-level field (model_extra).
+                fik = None
+                if request.extra_body:
+                    fik = request.extra_body.get("function_invocation_kwargs")
+                if fik is None and request.model_extra:
+                    fik = request.model_extra.get("function_invocation_kwargs")
+                if isinstance(fik, dict):
+                    run_kwargs["function_invocation_kwargs"] = fik
+
                 stream = cast(Any, agent.run(user_message, **run_kwargs))
                 async for update in stream:
                     for trace_event in trace_collector.get_pending_events():

python/packages/devui/tests/devui/test_execution.py

@@ -596,6 +596,56 @@ def create_session(self, **kwargs):
     assert "Processed: hello" in text_events[0].delta
 
 
+@pytest.mark.parametrize(
+    "request_extras",
+    [
+        {"extra_body": {"function_invocation_kwargs": {"tenantId": "abc123"}}},  # documented channel
+        {"function_invocation_kwargs": {"tenantId": "abc123"}},  # top-level (model_extra), as issue #7344 tried
+    ],
+)
+async def test_agent_execution_forwards_function_invocation_kwargs(request_extras):
+    """DevUI forwards request function_invocation_kwargs into agent.run() (issue #7344)."""
+    from agent_framework import AgentResponseUpdate, AgentSession, Content
+
+    captured: dict[str, Any] = {}
+
+    class RecordingAgent:
+        """Agent that records the kwargs its run() is called with."""
+
+        id = "kwargs_test"
+        name = "Kwargs Test Agent"
+        description = "Records run() kwargs"
+
+        def run(self, messages=None, *, stream=False, session=None, **kwargs):
+            captured.update(kwargs)
+            return self._stream_impl(messages)
+
+        async def _stream_impl(self, messages):
+            yield AgentResponseUpdate(contents=[Content.from_text(text="ok")], role="assistant")
+
+        def create_session(self, **kwargs):
+            return AgentSession()
+
+    discovery = EntityDiscovery(None)
+    executor = AgentFrameworkExecutor(discovery, MessageMapper())
+
+    agent = RecordingAgent()
+    entity_info = await discovery.create_entity_info_from_object(agent, source="test")
+    discovery.register_entity(entity_info.id, entity_info, agent)
+
+    request = AgentFrameworkRequest(
+        metadata={"entity_id": entity_info.id},
+        input="hello",
+        stream=True,
+        **request_extras,
+    )
+
+    async for _event in executor.execute_streaming(request):
+        pass
+
+    assert captured.get("function_invocation_kwargs") == {"tenantId": "abc123"}
+
+
 # =============================================================================
 # Full Pipeline Tests for SequentialBuilder
 # =============================================================================