Async Python

Async/Await Basics

Async And Await, From The Start

Agents spend almost all their time waiting. Waiting for a model, an API, a database. The work is not hard, it is slow, and it is slow somewhere else.

Async is Python's way of doing something useful while waiting. This page covers the syntax and the mental model; the concurrency page covers how to actually save time with it.

The problem, in one picture

Ordinary Python is blocking. When you call a slow thing, everything stops until it comes back.

import time

def fetch(name):
    time.sleep(2)
    return f"{name} done"

print(fetch("a"))
print(fetch("b"))
print(fetch("c"))

Six seconds, and for almost all of it your program is sitting idle waiting on someone else's server. Async lets that waiting overlap.

The two keywords

import asyncio

async def fetch(name):
    await asyncio.sleep(2)
    return f"{name} done"
  • async def makes it a coroutine — a function that is allowed to pause
  • await is a pause point: "wait for this, and let other work run meanwhile"

Note asyncio.sleep rather than time.sleep. That difference is the whole idea, and it is covered below.

Calling one does not run it

result = fetch("a")
print(result)
<coroutine object fetch at 0x...>
RuntimeWarning: coroutine 'fetch' was never awaited

Calling an async function builds a coroutine object. Nothing has happened yet.

To actually run it you must await it:

result = await fetch("a")

Forgetting the await is the most common async mistake, and the symptom is odd: no error, just a coroutine object where you expected a value. When something async "returns nothing sensible", check for a missing await first.

Where you can use await

await only works inside an async def. At the top level of a script you start the loop once:

import asyncio

async def main():
    result = await fetch("a")
    print(result)

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

asyncio.run starts the event loop, runs your coroutine, and shuts the loop down. Call it once, at the entry point.

In a notebook there is already a loop running, so asyncio.run fails with "this event loop is already running". In a Colab or Jupyter cell just await directly:

result = await fetch("a")

That is the one place await works outside a function, and it trips people who learned from scripts.

await on its own is still sequential

This is the thing to understand before anything else.

async def main():
    a = await fetch("a")
    b = await fetch("b")
    c = await fetch("c")

This still takes six seconds. await means "pause here until this finishes" — three pauses, one after another.

Writing async did not make it fast. It made it possible to be fast. Actually overlapping the waits needs asyncio.gather, which is the next page.

The event loop, briefly

There is one loop, on one thread, running one thing at a time. When it reaches an await, it parks that task and picks up another that is ready. When the first task's result arrives, it goes back to it.

Two consequences worth carrying:

  • Nothing runs in parallel. Python's async interleaves waiting, it does not use more cores. It helps with network calls and does not help with heavy computation.
  • There are no locks to worry about between your own coroutines. A whole category of threading bugs simply does not arise.

Never block the loop

Because there is one loop, one blocking call freezes everything.

async def bad():
    time.sleep(2)

async def good():
    await asyncio.sleep(2)

time.sleep stops the entire loop for two seconds — every other task included. asyncio.sleep yields, so other work continues.

The same applies to libraries:

BlockingAsync
time.sleepasyncio.sleep
requests.gethttpx.AsyncClient
open() for a large fileaiofiles
llm.invokellm.ainvoke
llm.streamllm.astream

Frameworks follow a naming habit worth knowing: the async version of a method is usually the same name with an a in front. invoke and ainvoke, stream and astream, run and arun.

Async is contagious

To await something you must be in an async def. So the async-ness spreads up your call stack to the entry point.

async def get_data():
    return await fetch("a")

async def process():
    data = await get_data()
    return data.upper()

async def main():
    print(await process())

asyncio.run(main())

This is why people say async is all-or-nothing. In practice, decide early: if your agent makes network calls and you care about latency, make the path async from the top. Retrofitting later is tedious.

If you are only ever making one call at a time and waiting anyway, plain synchronous code is perfectly fine and simpler. Use async when there is genuinely something to overlap.

Async generators

For streaming, the loop form gains a keyword:

async for chunk in llm.astream("Explain retries"):
    print(chunk.content, end="", flush=True)

Same idea as an ordinary generator, with waiting allowed between pieces.

Coming from Java or C#

C# will feel almost identical, because the two languages borrowed the same design.

ConceptJavaC#Python
Async functionCompletableFuture<T>async Task<T>async def
Wait for a result.join(), .get()awaitawait
Start the loopexecutor, virtual threadsruntime handles itasyncio.run
SleepThread.sleepTask.Delayasyncio.sleep
Async streamFluxIAsyncEnumerableasync for

The difference that matters: Python's async is single-threaded. C# tasks may run on a thread pool and genuinely execute in parallel. Python's coroutines never do — they take turns on one thread. So async solves waiting, not computation, and you get no data races between coroutines.

Common mistakes

  • Forgetting await, leaving you with a coroutine object.
  • Awaiting in sequence when the calls were independent — slow, and the reason for the next page.
  • time.sleep or requests inside async code, freezing the loop.
  • asyncio.run in a notebook, which fails because a loop is already running.
  • Calling asyncio.run more than once, or from inside a coroutine.
  • Making everything async when a single sequential call was all you needed.

Practise this

  • Write an async function that sleeps for one second and returns a string. Await it and print the result.
  • Call it without await and read the warning.
  • Await three of them in sequence and time it. Notice it is not faster.
  • Replace asyncio.sleep with time.sleep and observe that nothing overlaps at all.

Try It Yourself

Copy this into a new Colab notebook and run it. Nothing to install.

example.py
1import asyncio
2from typing import AsyncIterator
3
4# Basic async function
5async def fetch_response(prompt: str) -> str:
6 """Simulate an async API call."""
7 print(f"Sending: {prompt}")
8 await asyncio.sleep(1) # Simulate network delay
9 return f"Response to: {prompt}"
10
11# Async generator for streaming
12async def stream_response(prompt: str) -> AsyncIterator[str]:
13 """Simulate streaming response."""
14 words = ["Hello", "!", " I", " am", " an", " AI", " assistant", "."]
15 for word in words:
16 await asyncio.sleep(0.1) # Simulate token delay
17 yield word
18
19# Running async code
20async def main():
21 # Single async call
22 response = await fetch_response("What is Python?")
23 print(f"Got: {response}\n")
24
25 # Streaming response
26 print("Streaming: ", end="")
27 async for token in stream_response("Tell me a story"):
28 print(token, end="", flush=True)
29 print("\n")
30
31 # Concurrent calls
32 prompts = ["Question 1", "Question 2", "Question 3"]
33 tasks = [fetch_response(p) for p in prompts]
34 results = await asyncio.gather(*tasks)
35 print("Concurrent results:", results)
36
37# Run the async main function
38asyncio.run(main())