microsoft/semantic-kernel · error · RuntimeError

{self!r} is bound to a different event loop

Error message

{self!r} is bound to a different event loop

What it means

_LoopBoundMixin binds a queue (or similar async primitive) to the first event loop it is used on. On subsequent uses, if asyncio.get_running_loop() returns a different loop object, RuntimeError is raised. This prevents cross-loop corruption of internal async state.

Source

Thrown at python/semantic_kernel/agents/runtime/in_process/queue.py:28

from typing import Generic, TypeVar

from semantic_kernel.utils.feature_stage_decorator import experimental

_global_lock = threading.Lock()


class _LoopBoundMixin:
    _loop = None

    def _get_loop(self) -> asyncio.AbstractEventLoop:
        loop = asyncio.get_running_loop()

        if self._loop is None:
            with _global_lock:
                if self._loop is None:
                    self._loop = loop
        if loop is not self._loop:
            raise RuntimeError(f"{self!r} is bound to a different event loop")
        return loop


@experimental
class QueueShutDown(Exception):
    """Raised when putting on to or getting from a shut-down Queue."""

    pass


T = TypeVar("T")


@experimental
class Queue(_LoopBoundMixin, Generic[T]):
    """A queue class that supports async operations."""

    def __init__(self, maxsize: int = 0):

View on GitHub (pinned to c028a0c7dc)

Solutions

1. Ensure the runtime is created, started, used, and stopped all within the same event loop.
2. In pytest, use a consistent event_loop fixture scope (e.g. function-scoped for both runtime and tests).
3. Do not reuse a runtime across asyncio.run() calls — create a new instance each time.
4. If using threads, ensure all runtime access happens on the loop's owning thread.

Example fix

# before
async def setup():
    runtime.start()

asyncio.run(setup())  # binds to loop A
async def use():
    await runtime.send_message(...)  # different loop → raises
asyncio.run(use())

# after
async def main():
    runtime.start()
    await runtime.send_message(...)
    await runtime.stop_when_idle()

asyncio.run(main())  # all on same loop

import asyncio

# Ensure all runtime access happens on the same loop
loop = asyncio.get_running_loop()
# Pass this loop reference around; before using the runtime from a new context:
current_loop = asyncio.get_running_loop()
if current_loop is not loop:
    raise RuntimeError('Runtime bound to a different event loop')

null

try:
    await runtime.send_message(msg, recipient)
except RuntimeError as e:
    if 'bound to a different event loop' in str(e):
        # Recreate runtime on the current loop
        runtime = InProcessRuntime()
        runtime.start()

When it happens

Trigger: Using the same runtime (and its internal Queue) across two different event loops — e.g. starting the runtime in one asyncio loop, then calling runtime methods from a newly created loop. Common with pytest-asyncio when session-scoped fixtures run on different event loops than test-scoped code.

Common situations: pytest-asyncio with different event_loop policies per test scope. Running the runtime in one thread/loop and accessing it from another. Jupyter notebooks that create new event loops per cell. asyncio.run() called multiple times in sequence while holding a reference to the same runtime.

Related errors

• task_done() called too many times
• Agent with name {agentId.Type} not found.
• receive_task must be an instance of asyncio.Task or None
• The invocation was canceled before it could complete.
• AgentInstantiationContext cannot be instantiated. It is a st

AI-assisted analysis of microsoft/semantic-kernel@c028a0c7dc (2026-08-13). Data as JSON: /api/errors/d295a0e1110189b8. Report an issue: GitHub.