Python asyncio: A Practical Guide to Asynchronous Programming

Python’s asyncio lets one thread make progress on other work while a coroutine waits for asynchronous I/O. Use it when a program has many independent network, socket, or subprocess waits. It does not make CPU-heavy synchronous code run in parallel, and a blocking call inside a coroutine can stall every task on that event loop.

The Python documentation describes asyncio as “a library to write concurrent code using the async/await syntax.” The examples below target Python 3.11 or later, where TaskGroup is available. Check the documentation for the exact Python version and platform you deploy; asyncio APIs and platform support can evolve. Python asyncio documentation

1. Start with an async entry point

Define asynchronous functions with async def, then ordinarily start the program with asyncio.run(). Calling an async function creates a coroutine object; it does not run the function by itself. Await it, or schedule it as a task.

import asyncio


async def greet(name: str) -> str:
    await asyncio.sleep(0.1)  # Demonstration of an asynchronous wait.
    return f"Hello, {name}!"


async def main() -> None:
    message = await greet("Ada")
    print(message)


if __name__ == "__main__":
    asyncio.run(main())

Save this as hello_async.py and run python hello_async.py. The sleep represents a wait that yields control; it is not a way to make CPU work faster. In real programs, use asynchronous library calls for the I/O you are waiting on.

2. Understand cooperative scheduling

An event loop runs ready tasks. A task continues until it returns, raises, or suspends at an await whose result is not ready. During that suspension, the loop can run other ready work. This is cooperative concurrency: tasks must yield for one another to make progress.

import asyncio


async def fetch_label(label: str, delay: float) -> str:
    print(f"starting {label}")
    await asyncio.sleep(delay)
    print(f"finished {label}")
    return label


async def main() -> None:
    first = asyncio.create_task(fetch_label("first", 1.0))
    second = asyncio.create_task(fetch_label("second", 0.5))
    results = await asyncio.gather(first, second)
    print(results)


asyncio.run(main())

Both tasks are scheduled before the parent awaits their results, so their waits can overlap. The order in which completion messages appear follows their waits in this example, but applications should not rely on task scheduling order unless they explicitly coordinate it.

Async waits versus blocking calls

An asynchronous network client can suspend while data is in transit, letting the loop run another task. A synchronous call such as time.sleep(), a blocking HTTP request, or a long CPU calculation occupies the event-loop thread. Other tasks on that loop cannot run until it returns.

Use asynchronous libraries inside async code. If a blocking I/O function has no async equivalent, consider moving it to a worker thread with asyncio.to_thread(). For CPU-heavy Python work, use a process pool or another parallel execution approach when parallelism is appropriate; moving it to a thread does not generally remove the Python interpreter’s CPU execution constraints.

import asyncio
import time


def blocking_read() -> str:
    time.sleep(1)
    return "finished blocking I/O"


async def main() -> None:
    result = await asyncio.to_thread(blocking_read)
    print(result)


asyncio.run(main())

3. Choose a way to run concurrent work

Prefer a TaskGroup when a set of related tasks should have a clear shared lifetime. Use gather() when its result-collection behavior suits the work. Use create_task() for a task you will explicitly retain, await, cancel, and otherwise manage.

API	Good fit	Key behavior
await coroutine()	One operation that must finish before continuing	Runs the coroutine as part of the current task; no separate task is needed.
asyncio.TaskGroup	Related child tasks with one owner and bounded lifetime	On an unhandled child failure, cancels remaining group tasks and reports failures as an exception group.
asyncio.gather()	Collecting results from a known set of awaitables	Returns results in input order. By default, the first exception is propagated; other submitted awaitables are not automatically canceled just because that exception was propagated.
asyncio.create_task()	A task that needs an explicit handle or independent scheduling	Keep a reference and ensure it is eventually awaited, canceled, or otherwise supervised.

Related tasks with TaskGroup

import asyncio


async def load_record(record_id: int) -> str:
    await asyncio.sleep(0.1)
    return f"record-{record_id}"


async def main() -> None:
    tasks: list[asyncio.Task[str]] = []
    async with asyncio.TaskGroup() as group:
        for record_id in range(3):
            tasks.append(group.create_task(load_record(record_id)))

    # Exiting the group waits for all children to finish.
    print([task.result() for task in tasks])


asyncio.run(main())

If a child raises an exception other than cancellation, a task group cancels unfinished siblings, waits for them to finish their cleanup, and raises an exception group. Handle failures with except* SomeError when you need to handle matching members of that group. Verify these details against the task documentation for your selected Python release. Python task groups

Gathering results

import asyncio


async def lookup(item: str) -> str:
    await asyncio.sleep(0.1)
    return item.upper()


async def main() -> None:
    results = await asyncio.gather(
        lookup("alpha"),
        lookup("beta"),
    )
    print(results)  # ['ALPHA', 'BETA']


asyncio.run(main())

With return_exceptions=True, gather() places exceptions in the returned result list instead of raising them immediately. Check each item’s type before treating it as a successful result. Do not assume this option provides the same failure and cancellation behavior as a task group.

4. Handle cancellation, timeouts, and cleanup

Cancellation is how asyncio asks a task to stop. task.cancel() requests cancellation; CancelledError is raised into the coroutine at a cancellation point. Use try/finally to release resources, and normally let cancellation propagate after cleanup.

import asyncio


async def worker() -> None:
    resource = await open_resource()
    try:
        await use_resource(resource)
    finally:
        await resource.close()


async def open_resource():
    ...


async def use_resource(resource) -> None:
    ...


async def main() -> None:
    task = asyncio.create_task(worker())
    try:
        await task
    except asyncio.CancelledError:
        # Cleanup in worker's finally block has run.
        raise


asyncio.run(main())

The resource functions above are placeholders; replace them with the async resource API used by your application. Avoid swallowing CancelledError accidentally. Structured-concurrency tools use cancellation internally, so suppressing it without a deliberate reason can break their guarantees.

Use asyncio.timeout() to bound how long a group of operations may take. It is available in Python 3.11 and later.

import asyncio


async def fetch_data() -> str:
    await asyncio.sleep(10)
    return "data"


async def main() -> None:
    try:
        async with asyncio.timeout(2):
            result = await fetch_data()
    except TimeoutError:
        print("The operation exceeded two seconds")
    else:
        print(result)


asyncio.run(main())

A timeout cancels the work inside its scope and raises TimeoutError outside it. The actual time to finish can exceed the limit if cancellation cleanup takes time. For an older Python version, consult that version’s documentation for timeout options such as asyncio.wait_for().

5. Use high-level APIs for common jobs

Network streams

For basic TCP client or server code, asyncio streams provide reader and writer abstractions. Always close the writer and await its closure.

import asyncio


async def main() -> None:
    reader, writer = await asyncio.open_connection("example.com", 80)
    try:
        writer.write(
            b"GET / HTTP/1.1\r\nHost: example.com\r\nConnection: close\r\n\r\n"
        )
        await writer.drain()
        response = await reader.read(4096)
        print(response.decode("latin-1", errors="replace"))
    finally:
        writer.close()
        await writer.wait_closed()


asyncio.run(main())

This small example demonstrates a plain HTTP request over a stream, not a production HTTP client. Real clients need correct protocol handling, TLS, redirects, connection pooling, response-size limits, and error policy. Prefer a maintained asynchronous HTTP library for application HTTP traffic.

Queues and synchronization

asyncio.Queue coordinates producers and consumers. Async locks, events, conditions, and semaphores coordinate tasks on an event loop; they are not substitutes for thread synchronization primitives when sharing state with OS threads.

import asyncio


async def producer(queue: asyncio.Queue[int | None]) -> None:
    for value in range(3):
        await queue.put(value)
    await queue.put(None)  # Sentinel tells the consumer to stop.


async def consumer(queue: asyncio.Queue[int | None]) -> None:
    while True:
        value = await queue.get()
        try:
            if value is None:
                return
            print(value)
        finally:
            queue.task_done()


async def main() -> None:
    queue: asyncio.Queue[int | None] = asyncio.Queue(maxsize=10)
    async with asyncio.TaskGroup() as group:
        group.create_task(producer(queue))
        group.create_task(consumer(queue))


asyncio.run(main())

A bounded queue applies backpressure: put() waits when the queue is full. In a larger worker pool, make shutdown signaling and the number of sentinels match the number of consumers, and define how queued work is handled after errors.

Subprocesses

Async subprocess APIs let a program wait for a child process without blocking the event-loop thread. Use argument lists rather than shell command strings when possible, and consider output size and process cleanup.

import asyncio


async def main() -> None:
    process = await asyncio.create_subprocess_exec(
        "python", "--version",
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout, stderr = await process.communicate()
    print("exit:", process.returncode)
    print(stdout.decode().strip())
    if stderr:
        print(stderr.decode().strip())


asyncio.run(main())

Platform support and subprocess behavior can vary. Check the asyncio platform notes for the Python version and operating system you use.

When you need lower-level control

The event loop, futures, transports, and protocols are lower-level building blocks. They are mainly useful for framework authors, library authors, and integrations that need direct control over scheduling or I/O. Application code should generally start with coroutines, tasks, streams, queues, and other high-level APIs.

6. Capture web pages from async Python

A screenshot request is a useful example of asynchronous network I/O: while one request waits for a remote page to render, a program can schedule other requests. The standard library does not provide a high-level async HTTP client, so the example below uses aiohttp. Install it with python -m pip install aiohttp.

import asyncio
import aiohttp

API_URL = "https://api.screenshotneo.com/v1/shot"


async def capture(session: aiohttp.ClientSession, url: str) -> bytes:
    params = {"access_key": "YOUR_API_KEY", "url": url}
    async with session.get(API_URL, params=params) as response:
        response.raise_for_status()
        return await response.read()


async def main() -> None:
    timeout = aiohttp.ClientTimeout(total=90)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        image = await capture(session, "https://stripe.com")
    with open("shot.webp", "wb") as output:
        output.write(image)


if __name__ == "__main__":
    asyncio.run(main())

For multiple captures, reuse one ClientSession so connections can be reused, and bound concurrency rather than launching an unbounded number of requests. Keep API keys out of source control; load them from your deployment’s secret configuration. Check the ScreenshotNeo API documentation for request parameters and response details.

7. Debug common asyncio problems

Symptom	Likely cause	Fix
RuntimeWarning: coroutine was never awaited	An async function was called, but its coroutine was neither awaited nor scheduled.	Use await function(), or create and supervise a task.
asyncio.run() cannot be called from a running event loop	An event loop is already running, common in notebooks and async frameworks.	In an async function, use await main(). Let the framework own its loop instead of calling asyncio.run() inside it.
Other tasks appear frozen during a request or delay	A synchronous blocking function is running on the event-loop thread.	Use an async library, or move blocking I/O to asyncio.to_thread(). Use process-based execution where CPU parallelism is required.
A task’s exception appears late or is reported as never retrieved	A task was created but its lifetime and result were not managed.	Keep the task handle and await it, or put related work in a TaskGroup.
A task group raises an exception group	One or more child tasks failed; remaining tasks may have been canceled.	Inspect the grouped exceptions and handle expected failures with except*. Keep cleanup in finally.
Code works alone but fails when scheduled from another thread	Most asyncio objects are not thread-safe.	Use loop.call_soon_threadsafe() for callbacks or asyncio.run_coroutine_threadsafe() for a coroutine submitted to a loop from another OS thread.
Timeout appears longer than configured	Cancellation cleanup took additional time, or the timed operation did not reach a cancellation point promptly.	Make operations cancellation-aware, keep cleanup bounded where possible, and inspect blocking calls.

Enable debug mode

For a script, use asyncio.run(main(), debug=True). You can also enable debug mode with the PYTHONASYNCIODEBUG=1 environment variable or development mode. Debugging can reveal forgotten awaits, wrong-thread calls, and slow callbacks. Treat slow callback reports as a prompt to find blocking work, not as a benchmark. Python asyncio development guide

8. Performance, reliability, and cost

• Performance: Asyncio can improve throughput for workloads that spend much of their time waiting on I/O, especially when tasks can overlap those waits. It adds no automatic speedup to CPU-bound work. Measure with your actual libraries, workload, and limits; no universal speed advantage follows from using async.
• Concurrency limits: Bound simultaneous requests and queued work. Unbounded tasks can exhaust sockets, memory, remote service limits, or file descriptors. Use a semaphore, bounded queue, or a client’s connection limits.
• Reliability: Give tasks an owner, use timeouts for operations that must finish within a limit, clean up in finally, and make retries selective. A timeout or retry policy should account for whether repeating the operation is safe.
• Cost: Asyncio is part of Python; there is no separate asyncio license charge. Network services, compute, and infrastructure used by your program may have their own costs. Async concurrency can reduce idle waiting but does not by itself reduce provider charges.

Screenshot request billing

ScreenshotNeo bills only clean shots. Bot checks or CAPTCHAs, blank pages, timeouts, failed loads, and cache hits cost nothing, and each response reports the page verdict and billing status in X-Page-Verdict and X-Billed headers. Its plans are Free: 1,000 shots per month; Starter: $5 for 3,000; Growth: $15 for 15,000; Pro: $39 for 60,000; Scale: $99 for 250,000; and Business: $249 for 1,000,000. Yearly billing gives two months free, and every feature is available on every plan. See ScreenshotNeo for product details.

9. Or skip the browser setup

If your goal is a rendered website image rather than managing browser infrastructure, ScreenshotNeo accepts a URL in one GET request. The API can return PNG, JPEG, WebP, or PDF; see the ScreenshotNeo API documentation.

curl -G "https://api.screenshotneo.com/v1/shot" -d access_key=YOUR_API_KEY --data-urlencode url=https://stripe.com -o shot.webp

import requests

r = requests.get(
    "https://api.screenshotneo.com/v1/shot",
    params={"access_key": "YOUR_API_KEY", "url": "https://stripe.com"},
    timeout=90,
)
r.raise_for_status()
open("shot.webp", "wb").write(r.content)

const q = new URLSearchParams({ access_key: 'YOUR_API_KEY', url: 'https://stripe.com' });
const res = await fetch(`https://api.screenshotneo.com/v1/shot?${q}`);
if (!res.ok) throw new Error(`Screenshot request failed: ${res.status}`);
await Bun.write('shot.webp', new Uint8Array(await res.arrayBuffer()));

Use the Node.js example in an environment that provides fetch and Bun.write; in Node.js, write the response bytes with node:fs/promises instead. ScreenshotNeo removes cookie banners, popups, and chat widgets before the shot. Bot checks, blank pages, and failed loads are never billed. Its MCP server lets AI agents use take_screenshot, get_page_info, and capture_pdf. The free plan includes 1,000 screenshots a month with no card, and paid plans start at $5 for 3,000.

Sign up for ScreenshotNeo’s free plan and get 1,000 screenshots a month with no card.

10. Frequently asked questions

Does asyncio require multiple CPU cores?

No. Asyncio commonly runs many I/O-waiting tasks on one thread. It is concurrency through cooperative scheduling, not automatic multicore execution.

Can I use asyncio from a notebook?

Many notebook environments already run an event loop. In a notebook cell, await a coroutine directly when supported, rather than calling asyncio.run() inside the running loop.

Do I need to rewrite an entire application as async?

No. Asyncio can be used at a boundary where an async library or framework owns the event loop. Mixing synchronous and asynchronous components is possible, but take care that blocking calls do not stall the loop.

Should I use asyncio for every network request?

No. For a small script with a few sequential requests, synchronous code may be simpler. Asyncio is useful when overlapping waits or integrating with an async framework justifies the added task and cancellation management.