Skip to content

Calling Async Code from Synchronous Code Safely

The new client library is async-only, and the code that needs it is not: a Flask view, a Celery task, a CLI command, a pytest fixture written years ago. The first attempt is asyncio.run(client.fetch()) inside the function, which works in a script, crashes with RuntimeError: asyncio.run() cannot be called from a running event loop inside Jupyter or an async framework, and in a threaded web server quietly creates and destroys an event loop — plus the client's connection pool — on every request. The second attempt is nest_asyncio, which patches the loop to be re-entrant and trades the error for subtle ordering bugs. There are three correct shapes, and which one fits depends only on where the synchronous caller runs: at a program entry point, repeatedly on one thread, or concurrently on many threads.

Prerequisites

Where does the synchronous caller run? A decision on Who is calling the async code with 3 outcomes. Where does the synchronous caller run? Who is calling the async code? program entry point asyncio.run() one loop, then exit one thread, many calls asyncio.Runner reuse the same loop many threads at once shared loop thread run_coroutine_threadsafe If a loop is already running in the caller, the answer is await, not a new loop.

1. Use asyncio.run() only at a true entry point

asyncio.run() creates a loop, runs one coroutine to completion, cancels anything left over, shuts down async generators and the default executor, and closes the loop. That is exactly right for main() of a script or CLI command, where the process does one async job and exits.

import asyncio


async def fetch(i: int) -> int:
    await asyncio.sleep(0.01)                # stand-in for an async client call
    return i * 2


def cli_main() -> None:
    result = asyncio.run(fetch(21))          # one loop for the whole command
    print(result)


async def called_from_async_code() -> None:
    try:
        asyncio.run(fetch(1))                # wrong: a loop is already running here
    except RuntimeError as exc:
        print(exc)                           # asyncio.run() cannot be called from a running event loop


if __name__ == "__main__":
    cli_main()

The error in the second function is a feature, not an obstacle. It means the caller is already async and should simply await fetch(1); if it cannot because it sits in a synchronous helper called from async code, the fix is to make that helper async, not to start a loop inside a loop. Note that the refused call still created the fetch(1) coroutine, so Python also warns that it was never awaited.

Verify: the CLI prints 42. Anywhere you see the running-loop error, trace the call stack upward to the nearest async def — that is where an await belongs.

2. Reuse one loop with asyncio.Runner on a single thread

A synchronous worker that calls async code many times — a batch job looping over records, a Celery task using an async client — should not pay for a new loop per call. Worse, async clients bind their connection pools to the loop that created them, so a client created in one asyncio.run() fails with "attached to a different loop" or "Event loop is closed" in the next. asyncio.Runner keeps one loop alive across calls on the same thread.

import asyncio


class AsyncService:
    """Long-lived async client used from synchronous code on one thread."""

    def __init__(self) -> None:
        self._runner = asyncio.Runner()
        self._client = self._runner.run(self._connect())   # created on the runner's loop

    async def _connect(self) -> dict:
        await asyncio.sleep(0)                             # e.g. httpx.AsyncClient()
        return {"pool": "open"}

    def fetch(self, i: int) -> int:
        return self._runner.run(fetch(i))                  # same loop every call

    def close(self) -> None:
        self._runner.close()                               # cancels leftovers, closes the loop


service = AsyncService()
try:
    print([service.fetch(i) for i in range(5)])
finally:
    service.close()

A Runner is tied to the thread that uses it: call run() from one thread only, and never from inside a coroutine running on that same runner.

Verify: the list prints [0, 2, 4, 6, 8], and a client created in _connect() is still usable on the fifth call. Replace the runner with asyncio.run() per call and an async HTTP client will fail on the second call — that failure is what this step prevents.

3. Share a loop thread when many threads call in

Threaded web servers and worker pools call from many threads at once. One Runner cannot serve them, and a runner per thread multiplies connection pools by the thread count. Instead, run a single event loop on a dedicated daemon thread and submit coroutines to it with asyncio.run_coroutine_threadsafe(), which returns a concurrent.futures.Future the calling thread can block on.

import asyncio
import concurrent.futures
import threading
from typing import Any, Coroutine, TypeVar

T = TypeVar("T")


class LoopThread:
    """One event loop on a daemon thread, shared by synchronous callers."""

    def __init__(self) -> None:
        self._loop = asyncio.new_event_loop()
        self._ready = threading.Event()
        self._thread = threading.Thread(target=self._run, name="async-bridge", daemon=True)
        self._thread.start()
        self._ready.wait()

    def _run(self) -> None:
        asyncio.set_event_loop(self._loop)
        self._loop.call_soon(self._ready.set)
        self._loop.run_forever()

    def run(self, coro: Coroutine[Any, Any, T], timeout: float | None = None) -> T:
        if threading.current_thread() is self._thread:
            raise RuntimeError("LoopThread.run() called from its own loop: would deadlock")
        future = asyncio.run_coroutine_threadsafe(coro, self._loop)
        try:
            return future.result(timeout)
        except TimeoutError:
            future.cancel()                  # cancels the task on the loop thread too
            raise


bridge = LoopThread()


def sync_view(i: int) -> int:                # e.g. a Flask or Django view
    return bridge.run(fetch(i), timeout=2.0)


with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
    print(list(pool.map(sync_view, range(10))))

Every caller shares one loop, so an async client created on it has one connection pool for the whole process, and async concurrency limits such as an asyncio.Semaphore apply across all threads. The deadlock guard matters: a coroutine on the bridge loop that calls a synchronous helper which calls bridge.run() would block the only thread able to complete the future.

Verify: ten calls from eight threads return [0, 2, ..., 18], and threading.enumerate() shows exactly one async-bridge thread no matter how many requests have been served.

One request through the loop-thread bridge 5 stages from request thread to result. One request through the loop-thread bridge request thread sync view submit coroutine handed loop thread task runs async client shared pool result future.result() Many request threads, one loop, one connection pool.

4. Propagate timeouts and cancellation across the bridge

A synchronous caller that gives up must not leave a coroutine running forever on the loop thread. future.result(timeout) raises TimeoutError in the caller — since Python 3.11, concurrent.futures.TimeoutError is the built-in TimeoutError — and future.cancel() on a future from run_coroutine_threadsafe() schedules cancellation of the underlying task on the loop.

import asyncio


async def slow_call() -> str:
    try:
        await asyncio.sleep(10)
        return "done"
    except asyncio.CancelledError:
        print("slow_call cancelled on the loop thread")
        raise


try:
    bridge.run(slow_call(), timeout=0.05)
except TimeoutError:
    print("caller gave up after 50 ms")

Prefer putting the deadline inside the coroutine as well, with asyncio.timeout(), when the async side has cleanup to do: the async timeout lets the coroutine unwind on its own terms, while the outer result(timeout) remains a backstop for a coroutine that never yields. The same layering is described for services in propagating deadlines across async service calls.

Verify: the output shows caller gave up after 50 ms followed by slow_call cancelled on the loop thread, and asyncio.all_tasks(bridge._loop) no longer contains the task a moment later.

5. Shut the loop thread down cleanly

A daemon thread dies abruptly at interpreter exit, which skips finally blocks and leaves sockets unclosed. Close the bridge explicitly from your framework's shutdown hook — or atexit as a last resort — by cancelling remaining tasks on the loop, finalising async generators, stopping the loop, and joining the thread.

import asyncio
import atexit


def close_bridge(bridge: LoopThread, grace: float = 5.0) -> None:
    loop = bridge._loop
    if loop.is_closed():
        return

    async def _drain() -> None:
        tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
        for t in tasks:
            t.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)
        await loop.shutdown_asyncgens()

    asyncio.run_coroutine_threadsafe(_drain(), loop).result(grace)
    loop.call_soon_threadsafe(loop.stop)
    bridge._thread.join(grace)
    loop.close()


atexit.register(close_bridge, bridge)

Close async clients before draining — await client.aclose() submitted through the bridge — so connections are released gracefully rather than cancelled mid-request. The ordering mirrors shutting down async generators and executors cleanly.

Closing the bridge at exit 5 ordered steps. Closing the bridge at exit close async clients release connections gracefully cancel remaining tasks gather with return_exceptions shutdown_asyncgens() run generator finally blocks stop the loop call_soon_threadsafe(loop.stop) join thread, close loop bounded by a grace period A daemon thread killed at exit skips every one of these steps.

Verify: at exit, no "Task was destroyed but it is pending" or "unclosed client session" warnings are printed, and the process exits within the grace period.

Verification

The bridge between sync and async code is correct when:

  • No nested loops: there are no nest_asyncio patches and no asyncio.run() calls below the program's entry points.
  • Clients live on one loop: each async client is created on the loop that will use it and survives across calls.
  • One loop thread for many callers: in threaded servers, thread count and connection-pool count are independent.
  • Timeouts cancel work: a caller timeout cancels the coroutine on the loop thread instead of orphaning it.
  • Shutdown is explicit: the loop is drained, stopped and closed from a shutdown hook, without pending-task warnings.

Pitfalls & edge cases

  • nest_asyncio in production. Making the loop re-entrant lets a coroutine run another coroutine to completion inside one of its own steps, which breaks the assumption that nothing else runs between two awaits. Locks and state that were safe become racy.
  • Creating clients outside the loop that uses them. An async client created in one asyncio.run() and reused in another is bound to a closed loop. Create clients inside the runner or bridge loop.
  • Blocking the bridge loop. Every thread shares one loop, so one coroutine that calls blocking code stalls every request in the process. Offload blocking calls from the bridge with asyncio.to_thread().
  • Calling bridge.run() from async code. A coroutine that calls the synchronous bridge blocks its own loop waiting for work that needs that loop: a deadlock if it is the bridge loop, and a stalled loop if it is another one. Async callers should await directly.
  • Fork-based servers. A loop thread started before a pre-fork server forks does not exist in the children. Start the bridge lazily in each worker process, after the fork.

Frequently Asked Questions

How do I call an async function from a normal synchronous function?

At a program entry point, use asyncio.run(coro()). For repeated calls on one thread, keep an asyncio.Runner and call runner.run(coro()) each time so clients stay bound to one loop. When many threads call in, run one event loop on a background thread and submit coroutines with asyncio.run_coroutine_threadsafe, blocking on the returned future.

Why do I get asyncio.run() cannot be called from a running event loop?

The calling code is already running inside an event loop, for example in Jupyter, an async web framework, or a coroutine. Starting a second loop there is not allowed. The caller should await the coroutine directly, or the synchronous helper in between should be made async, rather than patching the loop with nest_asyncio.

Is nest_asyncio safe to use?

It is acceptable for interactive notebooks but risky in production. It makes the event loop re-entrant, so a coroutine can run other coroutines to completion inside one of its own steps. That breaks the guarantee that no other code runs between two awaits, which locks and shared state often rely on.

How do I use an async client from Flask or Django views?

Run a single event loop on a dedicated background thread at worker start, create the async client on that loop, and have each view submit coroutines with asyncio.run_coroutine_threadsafe and wait on the future with a timeout. All request threads then share one loop and one connection pool, and the loop is shut down from the worker's exit hook.