Python: Safely clone plugins

python/semantic_kernel/kernel.py

@@ -543,9 +543,23 @@ def clone(self) -> "Kernel":
         New lists of plugins and filters are created. It will not affect the original lists when the new instance
         is mutated. A new `ai_service_selector` is created. It will not affect the original instance when the new
         instance is mutated.
+
+        Important: Plugins are cloned without deep-copying their underlying callable methods. This avoids attempting
+        to pickle/clone unpickleable objects (e.g., async generators), which can be present when plugins wrap async
+        context managers such as MCP client sessions. Function metadata is deep-copied while callables are shared.
         """
+        # Safely clone plugins by copying function metadata while retaining callable references.
+        # This avoids deepcopying bound methods that may reference unpickleable async components.
+        new_plugins: dict[str, KernelPlugin] = {}
+        for plugin_name, plugin in self.plugins.items():
+            cloned_plugin = KernelPlugin(name=plugin.name, description=plugin.description)
+            # Using KernelPlugin.add will copy functions via KernelFunction.function_copy(),
+            # which deep-copies metadata but keeps callables shallow.
+            cloned_plugin.add(plugin.functions)
+            new_plugins[plugin_name] = cloned_plugin
+
         return Kernel(
-            plugins=deepcopy(self.plugins),
+            plugins=new_plugins,
             # Shallow copy of the services, as they are not serializable
             services={k: v for k, v in self.services.items()},
             ai_service_selector=deepcopy(self.ai_service_selector),

python/tests/unit/kernel/test_kernel.py

@@ -3,6 +3,7 @@
 import os
 import tempfile
 from collections.abc import Callable
+from copy import deepcopy
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Union
@@ -361,6 +362,50 @@ async def test_kernel_invoke_deep_copy_preserves_previous_state():
     assert snapshot2[-1] == {"id": 4, "name": "Desk lamp", "is_on": True}
 
 
+# region Clone safety with unpickleable async context
+
+
+class _AsyncGenPlugin:
+    """Plugin holding an async generator to emulate MCP-style async internals.
+
+    Deep-copying objects that reference async generators typically fails with
+    `TypeError: cannot pickle 'async_generator' object`.
+    """
+
+    def __init__(self):
+        async def _agen():
+            yield "tick"
+
+        # Store an async generator object on the instance to make it unpickleable.
+        self._unpickleable_async_gen = _agen()
+
+    @kernel_function(name="do", description="Return OK to validate plugin wiring")
+    async def do(self) -> str:
+        return "ok"
+
+
+@pytest.mark.asyncio
+async def test_kernel_clone_with_unpickleable_plugin_does_not_raise():
+    kernel = Kernel()
+    plugin_instance = _AsyncGenPlugin()
+    kernel.add_plugin(plugin_instance)
+
+    # Sanity: naive deepcopy of plugins should raise due to async generator state
+    with pytest.raises(TypeError):
+        deepcopy(kernel.plugins)
+
+    # Clone should succeed and preserve function usability
+    cloned = kernel.clone()
+
+    func = cloned.get_function(plugin_instance.__class__.__name__, "do")
+    result = await func.invoke(cloned)
+    assert result is not None
+    assert result.value == "ok"
+
+
+# endregion
+
+
 async def test_invoke_function_call_throws_during_invoke(kernel: Kernel, get_tool_call_mock):
     tool_call_mock = get_tool_call_mock
     result_mock = MagicMock(spec=ChatMessageContent)