Python: [Breaking] Update runtime handling

python/samples/concepts/mcp/agent_with_mcp_sampling.py

@@ -52,6 +52,7 @@ async def main():
             "run",
             "mcp_server_with_sampling.py",
         ],
+        sampling_auto_approve=True,
     ) as plugin:
         agent = ChatCompletionAgent(
             service=OpenAIChatCompletion(),

python/semantic_kernel/connectors/ai/open_ai/services/_open_ai_realtime.py

@@ -480,8 +480,29 @@ async def _parse_function_call_arguments_done(
         )
 
         # Step 4: Invoke the function call
-        chat_history = ChatHistory()
+        # Fail closed: only invoke when a function choice behavior is available so the allowlist can be enforced.
+        # Without it, kernel.invoke_function_call would skip allowlist validation and execute any named function.
         function_behavior = self._current_settings.function_choice_behavior if self._current_settings else None
+        if function_behavior is None:
+            logger.warning(
+                "Skipping function call '%s-%s' because no function choice behavior is configured; "
+                "allowlist validation cannot be enforced.",
+                plugin_name,
+                function_name,
+            )
+            created_output = FunctionResultContent.from_function_call_content_and_result(
+                function_call_content=item,
+                result="Function call was not invoked because function choice behavior is not configured.",
+            )
+            result = RealtimeFunctionResultEvent(
+                service_type=SendEvents.CONVERSATION_ITEM_CREATE,
+                function_result=created_output,
+            )
+            await self.send(result)
+            await self.send(RealtimeEvent(service_type=SendEvents.RESPONSE_CREATE))
+            yield result
+            return
+        chat_history = ChatHistory()
         await self._kernel.invoke_function_call(item, chat_history, function_behavior=function_behavior)
         created_output: FunctionResultContent = chat_history.messages[-1].items[0]  # type: ignore
         # Step 5: Create the function result event
@@ -710,8 +731,7 @@ async def receive(
         if audio_output_callback:
             self.audio_output_callback = audio_output_callback
         while True:
-            event = await self._receive_buffer.get()
-            yield event
+            yield await self._receive_buffer.get()
 
     async def _send(self, event: RealtimeClientEvent) -> None:
         if not self.data_channel:

python/semantic_kernel/connectors/mcp.py

@@ -246,6 +246,7 @@ def __init__(
         kernel: Kernel | None = None,
         request_timeout: int | None = None,
         sampling_consent_callback: SamplingConsentCallback | None = None,
+        sampling_auto_approve: bool = False,
     ) -> None:
         """Initialize the MCP Plugin Base.
 
@@ -259,8 +260,11 @@ def __init__(
             request_timeout: The default timeout used for all requests.
             sampling_consent_callback: Optional callback for approving MCP sampling requests.
                 Receives the plugin name and MCP sampling request params. Return
-                False to deny the request. When omitted, sampling requests are
-                auto-approved and a warning is logged.
+                False to deny the request. Takes precedence over sampling_auto_approve.
+            sampling_auto_approve: Whether to auto-approve MCP sampling requests when no
+                sampling_consent_callback is configured. Defaults to False, meaning sampling
+                requests are denied unless a consent callback is provided or this flag is set
+                to True. Set to True only when connecting to a trusted MCP server.
         """
         self.name = name
         self.description = description
@@ -271,6 +275,7 @@ def __init__(
         self.kernel = kernel or None
         self.request_timeout = request_timeout
         self.sampling_consent_callback = sampling_consent_callback
+        self.sampling_auto_approve = sampling_auto_approve
         self._sampling_auto_approved_warning_logged = False
         self._mcp_reserved_attribute_names: set[str] | None = None
         self._current_task: asyncio.Task | None = None
@@ -383,13 +388,25 @@ async def sampling_callback(
 
         If a sampling consent callback is configured, it is called before forwarding the request to the configured
         chat completion service. Returning False denies the request. If no callback is configured, requests are
-        auto-approved and a warning is logged.
+        denied unless sampling_auto_approve is set to True, in which case they are auto-approved and a warning is
+        logged.
         """
         if self.sampling_consent_callback is None:
+            if not self.sampling_auto_approve:
+                logger.warning(
+                    "MCP sampling request for plugin '%s' was denied because no sampling consent callback was "
+                    "configured. Provide a sampling_consent_callback or set sampling_auto_approve=True to allow "
+                    "sampling requests.",
+                    self.name,
+                )
+                return types.ErrorData(
+                    code=types.INTERNAL_ERROR,
+                    message="Sampling denied: no consent callback configured.",
+                )
             if not self._sampling_auto_approved_warning_logged:
                 logger.warning(
-                    "MCP sampling request for plugin '%s' was auto-approved because no sampling consent callback "
-                    "was configured.",
+                    "MCP sampling request for plugin '%s' was auto-approved because sampling_auto_approve is "
+                    "enabled and no sampling consent callback was configured.",
                     self.name,
                 )
                 self._sampling_auto_approved_warning_logged = True
@@ -620,6 +637,7 @@ def __init__(
         encoding: str | None = None,
         kernel: Kernel | None = None,
         sampling_consent_callback: SamplingConsentCallback | None = None,
+        sampling_auto_approve: bool = False,
         **kwargs: Any,
     ) -> None:
         """Initialize the MCP stdio plugin.
@@ -643,8 +661,10 @@ def __init__(
             kernel: The kernel instance with one or more Chat Completion clients.
             sampling_consent_callback: Optional callback for approving MCP sampling requests.
                 Receives the plugin name and MCP sampling request params. Return
-                False to deny the request. When omitted, sampling requests are
-                auto-approved and a warning is logged.
+                False to deny the request. Takes precedence over sampling_auto_approve.
+            sampling_auto_approve: Whether to auto-approve MCP sampling requests when no
+                sampling_consent_callback is configured. Defaults to False (requests are denied).
+                Set to True only when connecting to a trusted MCP server.
             kwargs: Any extra arguments to pass to the stdio client.
 
         """
@@ -657,6 +677,7 @@ def __init__(
             load_prompts=load_prompts,
             request_timeout=request_timeout,
             sampling_consent_callback=sampling_consent_callback,
+            sampling_auto_approve=sampling_auto_approve,
         )
         self.command = command
         self.args = args or []
@@ -696,6 +717,7 @@ def __init__(
         sse_read_timeout: float | None = None,
         kernel: Kernel | None = None,
         sampling_consent_callback: SamplingConsentCallback | None = None,
+        sampling_auto_approve: bool = False,
         **kwargs: Any,
     ) -> None:
         """Initialize the MCP sse plugin.
@@ -720,8 +742,10 @@ def __init__(
             kernel: The kernel instance with one or more Chat Completion clients.
             sampling_consent_callback: Optional callback for approving MCP sampling requests.
                 Receives the plugin name and MCP sampling request params. Return
-                False to deny the request. When omitted, sampling requests are
-                auto-approved and a warning is logged.
+                False to deny the request. Takes precedence over sampling_auto_approve.
+            sampling_auto_approve: Whether to auto-approve MCP sampling requests when no
+                sampling_consent_callback is configured. Defaults to False (requests are denied).
+                Set to True only when connecting to a trusted MCP server.
             kwargs: Any extra arguments to pass to the sse client.
 
         """
@@ -734,6 +758,7 @@ def __init__(
             load_prompts=load_prompts,
             request_timeout=request_timeout,
             sampling_consent_callback=sampling_consent_callback,
+            sampling_auto_approve=sampling_auto_approve,
         )
         self.url = url
         self.headers = headers or {}
@@ -776,6 +801,7 @@ def __init__(
         terminate_on_close: bool | None = None,
         kernel: Kernel | None = None,
         sampling_consent_callback: SamplingConsentCallback | None = None,
+        sampling_auto_approve: bool = False,
         **kwargs: Any,
     ) -> None:
         """Initialize the MCP streamable http plugin.
@@ -801,8 +827,10 @@ def __init__(
             kernel: The kernel instance with one or more Chat Completion clients.
             sampling_consent_callback: Optional callback for approving MCP sampling requests.
                 Receives the plugin name and MCP sampling request params. Return
-                False to deny the request. When omitted, sampling requests are
-                auto-approved and a warning is logged.
+                False to deny the request. Takes precedence over sampling_auto_approve.
+            sampling_auto_approve: Whether to auto-approve MCP sampling requests when no
+                sampling_consent_callback is configured. Defaults to False (requests are denied).
+                Set to True only when connecting to a trusted MCP server.
             kwargs: Any extra arguments to pass to the sse client.
         """
         super().__init__(
@@ -814,6 +842,7 @@ def __init__(
             load_prompts=load_prompts,
             request_timeout=request_timeout,
             sampling_consent_callback=sampling_consent_callback,
+            sampling_auto_approve=sampling_auto_approve,
         )
         self.url = url
         self.headers = headers or {}
@@ -855,6 +884,7 @@ def __init__(
         description: str | None = None,
         kernel: Kernel | None = None,
         sampling_consent_callback: SamplingConsentCallback | None = None,
+        sampling_auto_approve: bool = False,
         **kwargs: Any,
     ) -> None:
         """Initialize the MCP websocket plugin.
@@ -876,8 +906,10 @@ def __init__(
             kernel: The kernel instance with one or more Chat Completion clients.
             sampling_consent_callback: Optional callback for approving MCP sampling requests.
                 Receives the plugin name and MCP sampling request params. Return
-                False to deny the request. When omitted, sampling requests are
-                auto-approved and a warning is logged.
+                False to deny the request. Takes precedence over sampling_auto_approve.
+            sampling_auto_approve: Whether to auto-approve MCP sampling requests when no
+                sampling_consent_callback is configured. Defaults to False (requests are denied).
+                Set to True only when connecting to a trusted MCP server.
             kwargs: Any extra arguments to pass to the websocket client.
 
         """
@@ -890,6 +922,7 @@ def __init__(
             load_prompts=load_prompts,
             request_timeout=request_timeout,
             sampling_consent_callback=sampling_consent_callback,
+            sampling_auto_approve=sampling_auto_approve,
         )
         self.url = url
         self._client_kwargs = kwargs

python/semantic_kernel/functions/kernel_function_extension.py

@@ -289,10 +289,21 @@ def get_function(self, plugin_name: str | None, function_name: str) -> "KernelFu
 
         """
         if plugin_name is None:
-            for plugin in self.plugins.values():
-                if function_name in plugin:
-                    return plugin[function_name]
-            raise KernelFunctionNotFoundError(f"Function '{function_name}' not found in any plugin.")
+            matches = [
+                (name, plugin[function_name]) for name, plugin in self.plugins.items() if function_name in plugin
+            ]
+            if not matches:
+                raise KernelFunctionNotFoundError(f"Function '{function_name}' not found in any plugin.")
+            if len(matches) > 1:
+                logger.warning(
+                    "Function '%s' is ambiguous: it exists in multiple plugins (%s). Resolving to '%s-%s' "
+                    "(first registered). Specify a plugin_name for security-relevant lookups to avoid shadowing.",
+                    function_name,
+                    ", ".join(name for name, _ in matches),
+                    matches[0][0],
+                    function_name,
+                )
+            return matches[0][1]
         if plugin_name not in self.plugins:
             raise KernelPluginNotFoundError(f"Plugin '{plugin_name}' not found")
         if function_name not in self.plugins[plugin_name]:

python/tests/unit/connectors/ai/open_ai/services/test_openai_realtime.py

@@ -636,6 +636,38 @@ async def test_parse_function_call_arguments_done_filters_block_unallowed(OpenAI
     assert "not part of the provided" in str(result_event.function_result.result)
 
 
+async def test_parse_function_call_arguments_done_no_settings_fails_closed(OpenAIWebsocket, kernel):
+    """When no settings/function choice behavior is present, the call must not be invoked (fail closed)."""
+    event = ResponseFunctionCallArgumentsDoneEvent(
+        call_id="call_id",
+        arguments='{"x": "result"}',
+        event_id="event_id",
+        output_index=0,
+        item_id="item_id",
+        name="plugin_name-function_name",
+        response_id="response_id",
+        type="response.function_call_arguments.done",
+    )
+    # No current settings -> function_behavior would be None -> allowlist cannot be enforced.
+    OpenAIWebsocket._current_settings = None
+    OpenAIWebsocket._call_id_to_function_map["call_id"] = "plugin_name-function_name"
+    func = kernel_function(name="function_name", description="function_description")(lambda x: x)
+    kernel.add_function(plugin_name="plugin_name", function_name="function_name", function=func)
+    OpenAIWebsocket._kernel = kernel
+
+    with (
+        patch.object(Kernel, "invoke_function_call") as mock_invoke,
+        patch.object(OpenAIWebsocket, "_send") as mock_send,
+    ):
+        events_received = [evt async for evt in OpenAIWebsocket._parse_function_call_arguments_done(event)]
+
+    # The function is never invoked, but a safe result is still sent back to the service.
+    mock_invoke.assert_not_awaited()
+    assert mock_send.await_count == 2
+    assert isinstance(events_received[-1], RealtimeFunctionResultEvent)
+    assert "not invoked" in str(events_received[-1].function_result.result)
+
+
 async def test_send_audio(OpenAIWebsocket):
     audio_event = RealtimeAudioEvent(
         audio=AudioContent(data=b"audio data", mime_type="audio/wav"),

python/tests/unit/connectors/mcp/test_mcp.py

@@ -132,7 +132,7 @@ async def test_mcp_sampling_consent_callback_error_denies_request(caplog):
     assert "MCP sampling consent callback failed" in caplog.text
 
 
-async def test_mcp_sampling_without_consent_callback_logs_auto_approve_warning(caplog):
+async def test_mcp_sampling_without_consent_callback_denies_by_default(caplog):
     plugin = MCPSsePlugin(name="TestMCPPlugin", url="http://localhost:8080/sse")
     params = types.CreateMessageRequestParams(
         messages=[types.SamplingMessage(role="user", content=types.TextContent(type="text", text="hello"))],
@@ -144,7 +144,28 @@ async def test_mcp_sampling_without_consent_callback_logs_auto_approve_warning(c
         result = await plugin.sampling_callback(MagicMock(), params)
 
     assert isinstance(result, types.ErrorData)
-    assert "auto-approved because no sampling consent callback was configured" in caplog.text
+    assert result.message == "Sampling denied: no consent callback configured."
+    assert "denied because no sampling consent callback was configured" in caplog.text
+
+
+async def test_mcp_sampling_auto_approve_logs_warning(caplog):
+    plugin = MCPSsePlugin(
+        name="TestMCPPlugin",
+        url="http://localhost:8080/sse",
+        sampling_auto_approve=True,
+    )
+    params = types.CreateMessageRequestParams(
+        messages=[types.SamplingMessage(role="user", content=types.TextContent(type="text", text="hello"))],
+        systemPrompt="server instructions",
+        maxTokens=100,
+    )
+
+    with caplog.at_level(logging.WARNING, logger="semantic_kernel.connectors.mcp"):
+        result = await plugin.sampling_callback(MagicMock(), params)
+
+    # No kernel configured, so the request is approved but then fails for lack of a chat service.
+    assert isinstance(result, types.ErrorData)
+    assert "auto-approved because sampling_auto_approve is enabled" in caplog.text
 
 
 async def test_mcp_tool_and_prompt_names_do_not_shadow_plugin_attributes():

python/tests/unit/kernel/test_kernel.py

@@ -973,6 +973,37 @@ def test_get_function_from_fqn_wo_plugin(kernel: Kernel, custom_plugin_class):
     assert func
 
 
+def test_get_function_bare_name_single_match(kernel: Kernel, custom_plugin_class):
+    kernel.add_plugin(custom_plugin_class(), "TestPlugin")
+    func = kernel.get_function(None, "getLightStatus")
+    assert func
+
+
+def test_get_function_bare_name_ambiguous_warns(kernel: Kernel, caplog):
+    import logging
+
+    class PluginA:
+        @kernel_function(name="check_permissions")
+        def check_permissions(self) -> str:
+            return "a"
+
+    class PluginB:
+        @kernel_function(name="check_permissions")
+        def check_permissions(self) -> str:
+            return "b"
+
+    kernel.add_plugin(PluginA(), "PluginA")
+    kernel.add_plugin(PluginB(), "PluginB")
+
+    with caplog.at_level(logging.WARNING, logger="semantic_kernel.functions.kernel_function_extension"):
+        func = kernel.get_function(None, "check_permissions")
+
+    # Warn-only: resolves to the first-registered match, but logs the ambiguity.
+    assert func is kernel.get_function("PluginA", "check_permissions")
+    assert "ambiguous" in caplog.text
+    assert "PluginA" in caplog.text and "PluginB" in caplog.text
+
+
 # endregion
 # region Services