Async in Python vs JavaScript

An async def function returns a coroutine — Python's version of a Promise. You await it to get the value. The one thing JS does for free that Python makes explicit: starting the event loop.

import asyncio

async def fetch_user(id: str) -> dict:     # `async def` ~ JS `async function`
    await asyncio.sleep(1)                  # await a coroutine ~ awaiting a Promise (here: fake I/O)
    return {"id": id, "name": "Pierre"}

# In JS you'd just call `await fetch_user(...)` at top level (in a module).
# In Python you enter the event loop once, at the program's edge:
async def main():
    user = await fetch_user("42")           # await inside an async function
    print(user)

asyncio.run(main())                         # starts the loop and runs main() to completion

Concurrency: asyncio.gather is Promise.all

Awaiting in sequence is serial. To run I/O concurrently, gather the coroutines — the direct analog of Promise.all.

# Serial: ~3 seconds total (1s each, one after another)
a = await fetch_user("1")
b = await fetch_user("2")
c = await fetch_user("3")

# Concurrent: ~1 second total — all three run at once
a, b, c = await asyncio.gather(          # asyncio.gather(...) ~ Promise.all([...])
    fetch_user("1"),
    fetch_user("2"),
    fetch_user("3"),
)

Real async HTTP with httpx

requests is synchronous (it blocks the loop). For async HTTP use httpx's AsyncClient — the pattern you'll use to call LLM and tool APIs concurrently.

import httpx

async def get_json(url: str) -> dict:
    async with httpx.AsyncClient() as client:   # `async with` = context manager that also awaits cleanup
        resp = await client.get(url)            # await the network call
        resp.raise_for_status()                 # throw on 4xx/5xx (like checking res.ok and throwing)
        return resp.json()                       # parse JSON body

# Fetch two endpoints at once:
weather, news = await asyncio.gather(get_json(URL_A), get_json(URL_B))

Building an API with FastAPI

FastAPI is the standard for Python AI backends: async-first, and it derives validation and docs from your type hints. Minimal app:

from fastapi import FastAPI

app = FastAPI()                    # the app instance (like `const app = express()`)

@app.get("/health")                # decorator registers a GET route (Express: app.get('/health', ...))
async def health():                # handlers can be async
    return {"status": "ok"}        # return a dict → FastAPI serialises it to JSON automatically

Run it with uvicorn app:app --reload (an ASGI server). FastAPI auto-generates interactive docs at /docs from your types — no extra work.

Path and query parameters

@app.get("/topics/{topic_id}")                 # {topic_id} is a path param (Express :topic_id)
async def get_topic(topic_id: str, verbose: bool = False):
    #                 ^ from the path            ^ from the query string (?verbose=true), typed + default
    return {"id": topic_id, "verbose": verbose}

FastAPI reads the types: topic_id comes from the path, and verbose (with a default) becomes an optional query parameter — automatically parsed and validated to a bool.

Typed request bodies with pydantic

A pydantic BaseModel validates and parses JSON request bodies — think zod, but the schema is the type. Invalid input yields a clean 422 error for free.

from pydantic import BaseModel

class AskRequest(BaseModel):         # like a zod schema / TS type that also validates at runtime
    question: str
    top_k: int = 4                   # optional with default
    temperature: float = 0.7

class AskResponse(BaseModel):
    answer: str
    used_chunks: int

@app.post("/ask")
async def ask(req: AskRequest) -> AskResponse:   # FastAPI parses+validates the JSON body into AskRequest
    # ... call your RAG/LLM here ...
    return AskResponse(answer="…", used_chunks=req.top_k)

If a client omits question or sends top_k: "four", pydantic rejects it with a 422 and a precise error — you never hand-write that validation.

Dependency injection

FastAPI's Depends injects shared resources (a config, a DB session, an API client, an auth check) into handlers — declared once, reused everywhere, and easy to override in tests.

from fastapi import Depends

def get_settings() -> dict:                    # a "dependency" — any callable that returns something
    return {"model": "gpt-4o-mini"}

@app.post("/ask")
async def ask(req: AskRequest, settings: dict = Depends(get_settings)):
    #                                     ^ FastAPI calls get_settings() and passes the result in
    model = settings["model"]
    ...

Because the dependency is injected (not imported directly), a test can swap get_settings for a fake — the same testability win as constructor injection.

Putting it together

import asyncio
from fastapi import FastAPI, Depends
from pydantic import BaseModel

app = FastAPI()

class AskRequest(BaseModel):
    question: str
    top_k: int = 4

def get_model() -> str:
    return "gpt-4o-mini"

async def call_llm(prompt: str, model: str) -> str:
    await asyncio.sleep(0.2)                    # stand-in for the real async API call
    return f"[{model}] answered: {prompt[:40]}"

@app.post("/ask")
async def ask(req: AskRequest, model: str = Depends(get_model)):
    answer = await call_llm(req.question, model)   # await the async LLM call
    return {"answer": answer, "top_k": req.top_k}

That's a complete, typed, async endpoint: validated input, injected config, awaited I/O, JSON out.

Recap

• async def returns a coroutine (a Promise); await it. Start the loop once with asyncio.run.
• asyncio.gather(...) is Promise.all — use it to run I/O concurrently; use httpx (not requests) for async HTTP. Async helps I/O-bound, not CPU-bound, work.
• FastAPI derives routes, validation, and docs from type hints. pydantic BaseModel validates request/response bodies like zod.
• Depends injects shared config/clients/auth into handlers and keeps them testable.

import asyncio, time

async def slow(label: str) -> str:
    await asyncio.sleep(1)
    return label
# serial:
async def main():
    a = await slow("a")
    b = await slow("b")
    print(a, b)

import asyncio, time

async def slow(label: str) -> str:
    await asyncio.sleep(1)
    return label

async def main():
    t = time.perf_counter()
    a, b = await asyncio.gather(slow("a"), slow("b"))   # both run at once ~ Promise.all
    print(a, b, f"{time.perf_counter() - t:.2f}s")       # ~1.00s, not 2.00s

asyncio.run(main())

import asyncio, httpx

async def get_json(client: httpx.AsyncClient, url: str) -> dict:
    resp = await client.get(url)
    resp.raise_for_status()
    return resp.json()

async def main():
    async with httpx.AsyncClient() as client:            # one client, reused for both calls
        a, b = await asyncio.gather(
            get_json(client, "https://httpbin.org/get?x=1"),
            get_json(client, "https://httpbin.org/get?x=2"),
        )
    print(a["args"], b["args"])

asyncio.run(main())

# file: app.py
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class AskRequest(BaseModel):
    question: str
    top_k: int = 4

@app.post("/ask")
async def ask(req: AskRequest):
    # pretend to answer; req is already validated
    return {"answer": f"You asked: {req.question}", "top_k": req.top_k}

uvicorn app:app --reload
# open http://localhost:8000/docs and POST to /ask

from fastapi import FastAPI, Depends
from pydantic import BaseModel

app = FastAPI()

class AskRequest(BaseModel):
    question: str
    top_k: int = 4

def get_settings() -> dict:
    return {"model": "gpt-4o-mini"}

@app.post("/ask")
async def ask(req: AskRequest, settings: dict = Depends(get_settings)):
    return {
        "answer": f"You asked: {req.question}",
        "model": settings["model"],      # injected, not imported
        "top_k": req.top_k,
    }