Skip to content

Reusing One Loop Across Calls with asyncio.Runner

asyncio.run() is built for a program with one async entry point: it creates a loop, runs one coroutine, cancels leftovers, shuts down async generators and the default executor, and closes the loop. Programs that need to enter async code several times from synchronous code — a CLI that runs a few async steps between prompts, a sync test helper, a migration script that alternates between sync and async libraries — either pay for a new loop each time or reach for the deprecated get_event_loop() + run_until_complete() idiom. Python 3.11 added asyncio.Runner for exactly this. In a test, a client created in one Runner.run() call was usable in the next, because both ran on the same loop; the same sequence with two asyncio.run() calls ran on two different loops, and the client from the first was unusable in the second.

Prerequisites

1. Replace repeated asyncio.run calls

import asyncio
import httpx


def main() -> None:
    with asyncio.Runner() as runner:
        client = runner.run(make_client())                 # created on the runner's loop
        user = runner.run(client.get("/me"))
        if input(f"Delete {user.json()['name']}? ") == "y":
            runner.run(client.delete("/me"))
        runner.run(client.aclose())


async def make_client() -> httpx.AsyncClient:
    return httpx.AsyncClient(base_url="https://api.example.com")

The input() call blocks between steps, which is fine: no loop is running at that point, so nothing is being starved. The connection pool inside the client survives across all three run() calls because the loop does. With asyncio.run() per step, the second call would fail with a loop-mismatch error the first time the pool tried to reuse a connection.

Leaving the with block does what asyncio.run() does on exit — cancels remaining tasks, shuts down async generators and the default executor, closes the loop.

Verify: runner.get_loop() returns the same object before and after each run().

One Runner, several entries into async code 3 lanes over time. One Runner, several entries into async code runner loop one loop for the whole with block async steps run(get) run(delete) run(aclose) sync code input() prompt more sync work time → The loop is idle, not closed, between run() calls, so loop-bound resources survive.

2. Know that context carries across runs

A Runner creates one contextvars.Context when it starts and runs every coroutine in it. A context variable set in one run() is visible in the next — verified: a value set in the first call was read back in the second.

import contextvars

tenant = contextvars.ContextVar("tenant", default=None)


async def set_tenant(t: str) -> None:
    tenant.set(t)


async def which() -> str | None:
    return tenant.get()


with asyncio.Runner() as r:
    r.run(set_tenant("acme"))
    print(r.run(which()))          # acme

This is usually what you want for a CLI session ("log in once, then run commands"), and occasionally a surprise in tests, where state from one test step leaks into the next. If each call should start clean, pass an explicit context: runner.run(coro, context=contextvars.copy_context()) — the per-call context argument was added for exactly this. The general rules for context isolation are in Context Variables & Request Context.

Verify: set a variable in one run() and read it in the next; then repeat with an explicit fresh context and confirm it is gone.

3. Pick the loop with loop_factory

Runner(loop_factory=...) replaces policy-based loop selection, which is deprecated as of Python 3.14. Use it for uvloop, for a selector loop on Windows, or for an instrumented loop subclass:

import asyncio
import sys


def loop_factory() -> asyncio.AbstractEventLoop:
    if sys.platform == "win32":
        return asyncio.SelectorEventLoop()        # e.g. for libraries needing add_reader
    try:
        import uvloop
        return uvloop.new_event_loop()
    except ImportError:
        return asyncio.new_event_loop()


with asyncio.Runner(loop_factory=loop_factory, debug=False) as runner:
    runner.run(main())

asyncio.run() accepts the same loop_factory argument since 3.12, so the choice is not specific to Runner. The debug flag is applied to the created loop — Runner(debug=True).get_loop().get_debug() returned True in testing — which is a clean way to enable debug mode for one entry point without the environment variable. Migration from policies is covered in migrating off event loop policies in Python 3.14.

Verify: type(runner.get_loop()) is the class your factory returned.

asyncio.run, Runner and run_until_complete A grid of 5 rows by 4 columns. asyncio.run, Runner and run_until_complete property asyncio.run asyncio.Runner run_until_complete loop reused across calls no yes yes context across calls fresh each time shared by default whatever is current cleanup on exit full full, on close yours to write Ctrl-C cancels main task yes yes, per run() no status recommended recommended legacy Runner is asyncio.run split into setup, any number of runs, and teardown.

4. Rely on its Ctrl-C handling

Runner.run() installs a temporary SIGINT handler that cancels the main task on the first Ctrl-C, so finally blocks and async with exits run before KeyboardInterrupt surfaces. The bare run_until_complete() idiom instead raises KeyboardInterrupt wherever the loop happens to be, which can be in the middle of a callback, skipping cleanup in whatever task was running.

async def long_job() -> None:
    try:
        await asyncio.sleep(3600)
    finally:
        print("cleanup ran")                 # runs on Ctrl-C under Runner


with asyncio.Runner() as runner:
    try:
        runner.run(long_job())
    except KeyboardInterrupt:
        print("interrupted, loop still open for cleanup steps")
        runner.run(flush_state())            # the loop is still usable here

The ability to run more async code after an interrupt — inside the with block — is something asyncio.run() cannot offer, and is handy for CLIs that want to persist progress before exiting. More detail on interrupt handling is in handling Ctrl-C in asyncio scripts.

Verify: press Ctrl-C during long_job; "cleanup ran" prints before the KeyboardInterrupt handler runs.

5. Use it for sync test helpers and notebooks-adjacent code

Synchronous test suites that need to call async code are the other common user. A session-scoped runner avoids a new loop per call and keeps loop-bound fixtures valid:

import asyncio
import pytest


@pytest.fixture(scope="session")
def runner():
    with asyncio.Runner() as r:
        yield r


@pytest.fixture(scope="session")
def db(runner):
    pool = runner.run(create_pool(DSN))
    yield pool
    runner.run(pool.close())


def test_user_count(runner, db):
    assert runner.run(db.fetchval("select count(*) from users")) >= 0

The constraint is the same as for asyncio.run(): Runner.run() raises RuntimeError if called while another loop is running in the same thread. That rules it out inside Jupyter, where a loop is already running and you should use top-level await instead, as described in running asyncio code inside Jupyter notebooks.

Verify: the pool is created once per session, and the suite leaves no "Event loop is closed" errors at teardown.

Which entry point fits this program? A decision on How does sync code enter async code with 3 outcomes. Which entry point fits this program? How does sync code enter async code? several times, sharing resources asyncio.Runner one loop, one context once, at program start asyncio.run simplest, full cleanup a loop is already running top-level await Jupyter, async REPL Runner is for repeated entry; it is never a way to nest loops.

Verification

The Runner is used correctly when:

  • Loop-bound resources created in one run() work in later ones.
  • The runner is closed — by the with block or an explicit close() — so tasks, generators and the executor are cleaned up.
  • Context sharing is intentional: shared for sessions, an explicit fresh context where isolation matters.
  • No asyncio.get_event_loop() deprecation warnings remain in the code it replaced.

Diagnostic Hook: in debug runs, enable Runner(debug=True) and watch for "Executing … took" warnings between steps — they show which async step of a CLI is blocking. In tests, assert len(asyncio.all_tasks(runner.get_loop())) == 0 after each test to catch tasks leaking from one run() into the next.

Pitfalls & edge cases

  • Calling run() from inside a coroutine. Nested runs are not supported and raise RuntimeError.
  • Using the runner from several threads. A runner and its loop belong to the thread that created them.
  • Forgetting to close it. Without the with block or close(), pending tasks and the executor are never shut down.
  • Expecting a fresh context per call. The default shares one; pass context= when isolation matters.
  • Creating the runner at import time. It creates a loop lazily, but closing it at interpreter exit is not guaranteed; tie it to an explicit scope.

Frequently Asked Questions

What is asyncio.Runner?

A context manager added in Python 3.11 that owns one event loop and one context and lets you call run() several times. It is asyncio.run split into setup, any number of runs, and teardown, with the same cleanup on close.

When should I use asyncio.Runner instead of asyncio.run?

When synchronous code needs to enter async code more than once and share loop-bound resources such as an HTTP client or a database pool between those calls — CLIs, sync test helpers and scripts that mix sync and async libraries.

Does asyncio.Runner share context variables between run calls?

Yes. It runs every coroutine in one context created at startup, so a ContextVar set in one call is visible in the next. Pass context=contextvars.copy_context() to run() for an isolated call.

Can I use asyncio.Runner in Jupyter?

No. Jupyter already runs an event loop in the same thread, and Runner.run raises RuntimeError when a loop is running. Use top-level await in notebook cells instead.