Async Generators and Async Iterators in Python

Async generators are a way to lazily produce data in an async context. Perfect for streaming, pagination and processing large datasets.

Async generator — yield inside async def

import asyncio, httpx

async def paginate(client, base_url, pages=5):
    for page in range(1, pages + 1):
        r = await client.get(f"{base_url}?page={page}")
        data = r.json()
        for item in data['results']:
            yield item          # yield in async def = async generator

async def main():
    async with httpx.AsyncClient() as client:
        async for item in paginate(client, "https://api.example.com/posts"):
            print(item['title'])    # process one at a time, not everything at once

Without async generators you’d have to load all pages into memory. Here each item is processed as soon as it’s received.

Streaming LLM responses via async generator

from anthropic import AsyncAnthropic

async def stream_response(prompt: str):
    client = AsyncAnthropic()
    async with client.messages.stream(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    ) as stream:
        async for text in stream.text_stream:
            yield text              # each token is a separate yield

async def main():
    async for chunk in stream_response("Write a poem"):
        print(chunk, end="", flush=True)

Typing async generators

from typing import AsyncGenerator, AsyncIterator

# AsyncGenerator[YieldType, SendType]
async def number_stream(n: int) -> AsyncGenerator[int, None]:
    for i in range(n):
        await asyncio.sleep(0.1)
        yield i

# AsyncIterator — when SendType isn't needed
async def events() -> AsyncIterator[dict]:
    while True:
        event = await fetch_event()
        yield event

aiter and anext — custom iterator

class DatabaseCursor:
    def __init__(self, query: str):
        self.query = query
        self._rows = []
        self._index = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self._index >= len(self._rows):
            # Load next batch from DB
            self._rows = await fetch_batch(self.query, self._index)
            if not self._rows:
                raise StopAsyncIteration
        row = self._rows[self._index]
        self._index += 1
        return row