asyncio.Runner provides a structured way to execute several top-level coroutines with the same event loop. It is useful for command-line programs, administrative tools, tests, integration scripts, and applications that must invoke asynchronous code more than once without rebuilding the complete runtime for every call. Although asyncio.run() remains the simplest choice for one asynchronous entry point, Runner gives applications more lifecycle control when they contain distinct phases.
This guide explains how to create a Runner, reuse its loop, preserve context variables, handle signals, enable debug mode, close resources, and avoid common mistakes. The goal is not merely to present syntax, but to show a dependable design for real applications.
Why asyncio.Runner exists
asyncio.run() creates a new event loop, executes one coroutine, finalizes asynchronous generators, and closes the loop. That behavior is ideal for a program with one main coroutine. Consider, however, a tool that loads configuration, synchronizes data, and then produces a report. Each phase is asynchronous, but synchronous orchestration may need to inspect the result before starting the next phase.
Calling asyncio.run() repeatedly creates separate loops. Resources, tasks, and context associated with one loop cannot be naturally reused in the next. Runner encapsulates a loop and allows multiple sequential run() calls while preserving its own context and lifecycle.
Basic use
import asyncio
async def load_config():
await asyncio.sleep(0.1)
return {"environment": "production"}
async def synchronize(config):
await asyncio.sleep(0.1)
return f"synced in {config['environment']}"
with asyncio.Runner() as runner:
config = runner.run(load_config())
result = runner.run(synchronize(config))
print(result)The context manager closes the Runner at the end of the block. Every call to run() receives an awaitable and returns its value or propagates its exception. The event loop remains available between calls.
Do not use Runner inside an active loop
Like asyncio.run(), Runner.run() cannot be invoked while another event loop is already running in the same thread. This situation is common in notebooks, asynchronous web servers, GUI integrations, and frameworks that own the loop. In those environments, use await directly or integrate the coroutine with the framework.
async def existing_flow():
config = await load_config()
return await synchronize(config)
# Inside async code, await coroutines.
# Do not create a nested Runner.This rule prevents nested loops and unpredictable scheduling. A reusable library should normally expose asynchronous functions and leave loop startup to the executable application.
Sharing ContextVar values
Context variables are useful for correlation IDs, tenant identifiers, tracing information, and request-scoped metadata. Runner maintains a context that can be reused across its calls. A specific context may also be passed to run().
import asyncio
import contextvars
job_id = contextvars.ContextVar("job_id", default="unknown")
async def log_phase(name):
print(name, job_id.get())
ctx = contextvars.copy_context()
ctx.run(job_id.set, "job-847")
with asyncio.Runner() as runner:
runner.run(log_phase("start"), context=ctx)
runner.run(log_phase("finish"), context=ctx)Do not place unnecessary secrets in context variables. Context propagation simplifies observability, but it is not an authorization boundary and does not replace explicit security controls.
Debug mode
Passing debug=True enables additional asyncio checks. Debug mode can reveal slow callbacks, coroutines that were created but never awaited, and operations called from an inappropriate thread.
with asyncio.Runner(debug=True) as runner:
runner.run(load_config())These checks are valuable in development and continuous integration. They add overhead and may generate verbose logs, so production systems should enable them deliberately through configuration and should avoid logging sensitive values.
Custom loop creation
The loop_factory argument controls how the event loop is created. It can be used for instrumentation, platform-specific setup, or an alternative event-loop implementation. The factory is responsible for creating and correctly registering the loop.
import asyncio
def create_loop():
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.set_debug(False)
return loop
with asyncio.Runner(loop_factory=create_loop) as runner:
runner.run(load_config())Use this feature only when a concrete requirement exists. An incorrect factory can leave resources open or conflict with libraries that assume standard behavior.
Ctrl+C and graceful interruption
Runner handles KeyboardInterrupt carefully. When the user presses Ctrl+C, the main task is cancelled so the coroutine can execute finally blocks and close resources. If the program does not respond, a subsequent interruption can stop it more directly.
async def temporary_server():
try:
while True:
await asyncio.sleep(1)
finally:
print("releasing resources")
with asyncio.Runner() as runner:
runner.run(temporary_server())Do not swallow CancelledError without a deliberate reason. Perform cleanup and normally re-raise cancellation. Suppressing it may prevent predictable shutdown and leave tasks running.
Closing tasks and executors
When closed, Runner finalizes asynchronous generators, shuts down the default executor, and closes the loop. This helps prevent leaked threads and file descriptors. The application must still manage resources it created, such as HTTP clients, database pools, files, queues, and temporary directories.
async def main():
client = create_client()
try:
return await client.fetch_data()
finally:
await client.aclose()Prefer asynchronous context managers when a library supports them. They make ownership visible, simplify tests, and ensure that cleanup remains close to resource creation.
Organizing multiple phases
A command-line application is a strong Runner use case. Separate phases can return values to synchronous orchestration while retaining the same loop and context.
async def validate():
return True
async def import_data():
return 120
async def write_report(total):
print(f"processed {total} items")
with asyncio.Runner() as runner:
if runner.run(validate()):
total = runner.run(import_data())
runner.run(write_report(total))Do not turn every small asynchronous function into a separate Runner call. If operations belong to one continuous asynchronous flow, create one main coroutine and use await internally. Multiple calls are most useful when genuinely separate phases are controlled by synchronous code.
Structured concurrency
Inside a coroutine executed by Runner, use asyncio.TaskGroup to coordinate related tasks. It propagates failures consistently and cancels sibling tasks when appropriate.
async def download(name):
await asyncio.sleep(0.1)
return name
async def batch():
async with asyncio.TaskGroup() as group:
group.create_task(download("a"))
group.create_task(download("b"))Avoid creating detached tasks without retaining references. An unsupervised task may fail without proper reporting or remain pending when the Runner closes.
Timeouts and resource limits
External operations need deadlines. Use asyncio.timeout() around a bounded unit of work and handle expiration at a layer that can decide whether to retry, report, or abort.
async def query_service():
async with asyncio.timeout(5):
return await remote_call()Combine deadlines with concurrency limits, exponential backoff, idempotency, and circuit-breaking where appropriate. Runner manages the loop lifecycle; it does not automatically protect a service from overload or unstable dependencies.
Error handling
An exception raised by the top-level awaitable leaves run() and can be handled by synchronous orchestration. Catch only exceptions that the current layer can meaningfully address. Record useful context, but do not hide programmer errors behind a generic success result.
with asyncio.Runner() as runner:
try:
runner.run(synchronize({"environment": "production"}))
except TimeoutError:
print("operation timed out")When several tasks fail in a TaskGroup, Python may raise an exception group. Handle specific subgroups carefully and preserve information about each failure.
Testing Runner-based code
For asynchronous unit tests, prefer the native async support provided by the test framework. Runner is especially helpful when testing a synchronous function that coordinates asynchronous phases. Every test should close its Runner and avoid global state shared between cases.
Cover success, ordinary exceptions, cancellation, timeout, resource cleanup, and pending-task detection. When signal behavior matters, add focused integration tests rather than making all unit tests depend on operating-system signals.
Threads and blocking work
Blocking calls freeze the event loop. Move short blocking functions to asyncio.to_thread(), or use a dedicated process for CPU-intensive work. Limit the number of concurrent jobs so the default executor does not become an uncontrolled queue.
async def read_legacy_file(path):
return await asyncio.to_thread(path.read_text, encoding="utf-8")Cancellation of the awaiting coroutine does not necessarily stop the underlying thread immediately. Design blocking functions with bounded duration and explicit cancellation strategies when required.
Production checklist
- Use
asyncio.run()for one simple asynchronous entry point. - Use Runner when synchronous orchestration needs several asynchronous phases.
- Never call Runner inside an active loop.
- Close clients, pools, and files explicitly.
- Propagate cancellation after cleanup.
- Use TaskGroup for related concurrent tasks.
- Apply timeouts and concurrency limits.
- Enable debug checks in development.
- Avoid a custom loop factory without a real requirement.
- Keep loop ownership in the application layer.
Continue learning with the Academify guides on Python asyncio, Python TaskGroup, Python contextvars, and asyncio timeouts.
Conclusion
asyncio.Runner is a high-level lifecycle tool for synchronous programs that execute several asynchronous phases. It reuses one event loop, preserves context, coordinates interruption handling, and closes the underlying runtime. Its strongest use cases are command-line applications and administrative tools with distinct phases. Inside a continuous asynchronous workflow, prefer one main coroutine, direct await, structured concurrency, and explicit resource ownership.
