Create and Run Async Tasks in Python

Creating an async task is the moment your program transitions from doing one thing at a time to doing many things concurrently. asyncio.create_task() takes a coroutine and schedules it to run independently on the event loop, returning a Task object that represents the eventual result. You can await the task to get its return value, cancel it to stop it early, or inspect it to check whether it succeeded or failed.

If you have read about the async and await syntax, you know the basics. This article goes deeper into the task lifecycle, cancellation patterns, and the structured concurrency model introduced in Python 3.11.

Tasks are not just coroutines running in the background. They are first-class objects with state, callbacks, and a well-defined lifecycle. Understanding that lifecycle prevents the most common async bugs: tasks that are created but never awaited, exceptions that are silently lost, and cancellations that leave resources in an inconsistent state.

Creating and running a task

create_task() wraps a coroutine in a Task and schedules it on the running event loop. The task begins executing at the next opportunity, typically when the current coroutine yields control at an await.

python

import asyncio
 
async def fetch_data(source, delay):
    await asyncio.sleep(delay)
    return f"Data from {source}"
 
async def main():
    task_a = asyncio.create_task(fetch_data("API", 0.2))
    task_b = asyncio.create_task(fetch_data("DB", 0.1))
 
    print("Tasks created. Doing other work...")
    await asyncio.sleep(0.05)
    print("Other work done.")
 
    result_a = await task_a
    result_b = await task_b
    print(result_a)
    print(result_b)
 
asyncio.run(main())

Both tasks start running as soon as create_task() schedules them, so main() prints its own "doing other work" message while task_a and task_b are already sleeping in the background:

text

Tasks created. Doing other work...
Other work done.
Data from API
Data from DB

Both tasks run concurrently from the moment they are created. The await asyncio.sleep(0.05) in main() gives the event loop time to start both tasks before main() blocks waiting for task_a. By the time main() reaches await task_a, both tasks may already be partially or fully complete.

The task lifecycle

A task moves through distinct states during its lifetime:

python

import asyncio
 
async def slow_operation():
    await asyncio.sleep(0.1)
    return 42
 
async def main():
    task = asyncio.create_task(slow_operation())
 
    # PENDING: created but not yet done
    print(f"After create: done={task.done()}")
 
    await asyncio.sleep(0.05)
    # Still running (sleep hasn't finished)
    print(f"During sleep: done={task.done()}")
 
    result = await task
    # DONE: completed with a result
    print(f"After await: done={task.done()}, result={task.result()}")
 
asyncio.run(main())

The task reports done=False right after creation and still done=False midway through the sleep, since neither point gives the event loop enough time to finish the 0.1-second wait; only the final check after await task sees the completed result:

text

After create: done=False
During sleep: done=False
After await: done=True, result=42

A task is done() when it has either completed, raised an exception, or been cancelled. You can check task.result() after the task is done. Calling result() on a task that is not yet done raises asyncio.InvalidStateError.

Calling it on a task that raised an exception re-raises that exception.

task.exception() returns the exception if the task raised one, or None if it completed successfully. This lets you check for errors without a try/except:

python

async def main():
    task = asyncio.create_task(failing_operation())
    await asyncio.sleep(0.2)
 
    exc = task.exception()
    if exc is not None:
        print(f"Task failed with: {exc}")

Adding callbacks to tasks

You can attach callbacks that run when a task completes using task.add_done_callback(). The callback receives the task as its argument and runs synchronously in the event loop thread.

python

import asyncio
 
def on_complete(task):
    if task.exception() is None:
        print(f"Task completed with: {task.result()}")
    else:
        print(f"Task failed with: {task.exception()}")
 
async def worker(value):
    await asyncio.sleep(0.1)
    return value * 2
 
async def main():
    task = asyncio.create_task(worker(21))
    task.add_done_callback(on_complete)
    await task
 
asyncio.run(main())

Callbacks are useful for logging, metrics collection, and chaining work without blocking the current coroutine. For most coordination between tasks, awaiting or using synchronization primitives is clearer than callback chains.

Cancelling tasks

task.cancel() requests cancellation of the task. The event loop schedules a CancelledError to be raised inside the coroutine at its next await point. The coroutine can catch it for cleanup but should let it propagate or re-raise it for the task to be considered cancelled.

python

import asyncio
 
async def interruptible_work():
    try:
        print("Working...")
        await asyncio.sleep(10)
        print("Work completed")
    except asyncio.CancelledError:
        print("Cleanup before exit")
        raise
 
async def main():
    task = asyncio.create_task(interruptible_work())
    await asyncio.sleep(0.1)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Task was cancelled")
 
asyncio.run(main())

The cancel() call interrupts the coroutine while it is inside asyncio.sleep(10), so the try block never reaches "Work completed" and instead runs its cleanup before re-raising, which main() catches:

text

Working...
Cleanup before exit
Task was cancelled

A cancelled task is in the done state with its cancelled() method returning True. Calling result() on a cancelled task raises CancelledError. Calling exception() returns None because cancellation is not an exception on the task; it is a separate state.

Cancellation is cooperative. If a coroutine never awaits, it cannot be cancelled. A long CPU-bound computation with no await points will run to completion regardless of cancel() calls.

This is another reason async is for I/O-bound work, not CPU-bound work.

Shielding critical sections from cancellation

asyncio.shield() protects an awaitable from cancellation. The shielded coroutine continues even if the task that awaits it is cancelled.

python

import asyncio
 
async def critical_operation():
    try:
        print("Critical operation started")
        await asyncio.sleep(0.5)
        print("Critical operation completed")
        return "result"
    except asyncio.CancelledError:
        print("Critical operation was NOT cancelled (shielded)")
        raise
 
async def main():
    task = asyncio.create_task(critical_operation())
    shielded = asyncio.shield(critical_operation())
    await asyncio.sleep(0.1)
    shielded.cancel()
    try:
        await shielded
    except asyncio.CancelledError:
        pass
    print("Shield demonstration complete")
 
asyncio.run(main())

shield() is useful when a task performs an operation that must not be interrupted, like committing a database transaction or writing a file. The outer task can still be cancelled, but the shielded inner operation runs to completion.

TaskGroup: structured concurrency

Python 3.11 introduced asyncio.TaskGroup, which treats a group of tasks as a single unit. When the async with TaskGroup block exits, all tasks in the group have completed. If any task raises an exception, all remaining tasks are cancelled.

python

import asyncio
 
async def worker(name, delay, should_fail=False):
    await asyncio.sleep(delay)
    if should_fail:
        raise ValueError(f"{name} failed")
    return f"{name} done"
 
async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(worker("A", 0.2))
            tg.create_task(worker("B", 0.1, should_fail=True))
            tg.create_task(worker("C", 0.3))
    except ExceptionGroup as eg:
        print(f"Group failed: {eg.exceptions}")
 
asyncio.run(main())

Worker A and worker C get cancelled as soon as worker B raises, so their own results never surface; only worker B's failure is wrapped and re-raised as the group's ExceptionGroup:

text

Group failed: (ValueError('B failed'),)

When worker B raises ValueError, the task group cancels workers A and C, waits for them to complete, and raises an ExceptionGroup containing the original exception. If multiple tasks fail, the ExceptionGroup contains all the exceptions.

TaskGroup replaces raw create_task() for most use cases because it guarantees cleanup. With raw create_task(), you must manually track every task and ensure they are all awaited or cancelled. A forgotten task with an unhandled exception produces a warning but otherwise fails silently, potentially leaking resources or producing incorrect results.

Practical patterns for task management

Pattern 1: Fire and forget with error logging

When you need a background task that runs independently, ensure exceptions are logged:

python

import asyncio
import logging
 
async def background_work():
    while True:
        try:
            await asyncio.sleep(60)
            await check_health()
        except Exception:
            logging.exception("Background health check failed")
 
async def main():
    task = asyncio.create_task(background_work())
    # Main program continues. task will be cancelled when the loop closes.
    await asyncio.sleep(300)

Pattern 2: Timeout on any task

A request that hangs longer than expected can tie up resources indefinitely if nothing bounds how long you are willing to wait. Wrap the awaitable in asyncio.wait_for() to enforce a deadline, and catch the timeout so the rest of your program can decide what to do next:

python

async def fetch_with_timeout(url):
    try:
        return await asyncio.wait_for(actual_fetch(url), timeout=5)
    except asyncio.TimeoutError:
        return f"Timeout fetching {url}"

Pattern 3: First result wins

When you send the same request to two redundant sources and only care about whichever answers first, use asyncio.wait() with FIRST_COMPLETED to get the fastest result and cancel the rest:

python

import asyncio
 
async def fetch_from(source, delay):
    await asyncio.sleep(delay)
    return f"Result from {source}"
 
async def main():
    tasks = [
        asyncio.create_task(fetch_from("Primary API", 0.3)),
        asyncio.create_task(fetch_from("Backup API", 0.1)),
    ]
 
    done, pending = await asyncio.wait(
        tasks, return_when=asyncio.FIRST_COMPLETED
    )
 
    for task in pending:
        task.cancel()
 
    result = done.pop().result()
    print(f"Got fastest result: {result}")
 
asyncio.run(main())

This pattern is useful for redundant requests where you want the fastest response and can discard slower ones.

Common task mistakes

Forgetting to await a task. If you create a task and never await it, the event loop still runs it, but you never get the result or exception. Python issues a warning if the task is garbage collected with an unhandled exception, but the warning does not appear until cleanup, which may be too late. Always await tasks or store them and check for exceptions later.

Cancelling without cleanup. If a coroutine holds a resource like a file handle or a database connection when cancelled, it should release that resource in a finally block or a CancelledError handler.

python

async def safe_operation():
    resource = await acquire_resource()
    try:
        await do_work(resource)
    except asyncio.CancelledError:
        await resource.rollback()
        raise
    finally:
        await resource.release()

Creating tasks in a loop without limiting concurrency. If you create a task for every item in a large iterable without limiting how many run at once, you can overwhelm the event loop or the remote service you are calling. Use a semaphore to cap concurrent tasks. The next article covers running multiple tasks concurrently with concurrency limiting patterns.