Python async/await — Waiting Without Blocking

import asyncio
import httpx

async def fetch(client: httpx.AsyncClient, url: str) -> dict:
    response = await client.get(url, timeout=10)
    response.raise_for_status()
    return response.json()

async def main() -> None:
    urls = [
        "https://api.github.com/users/anthropic",
        "https://api.github.com/users/openai",
        "https://api.github.com/users/vercel",
    ]
    async with httpx.AsyncClient() as client:
        results = await asyncio.gather(*(fetch(client, u) for u in urls))
    for data in results:
        print(data["login"], "-", data.get("bio", ""))

asyncio.run(main())

import asyncio
from anthropic import AsyncAnthropic

client = AsyncAnthropic()
semaphore = asyncio.Semaphore(5)   # max 5 concurrent requests

async def summarize(text: str) -> str:
    async with semaphore:
        response = await client.messages.create(
            model="claude-opus-4-7",
            max_tokens=200,
            messages=[{"role": "user", "content": f"One sentence summary:\n{text}"}],
        )
        return response.content[0].text

async def main():
    articles = ["...text 1...", "...text 2...", "...text 20..."]
    summaries = await asyncio.gather(*(summarize(a) for a in articles))
    for s in summaries:
        print(s)

asyncio.run(main())

results = await asyncio.gather(
    *(fetch(client, u) for u in urls),
    return_exceptions=True,   # don't fail-fast
)
for url, result in zip(urls, results):
    if isinstance(result, Exception):
        print(f"Failed {url}: {result}")
    else:
        print(f"OK {url}")