AsyncClient: many API calls at once

Tickets sent one after another wait for each answer in turn. httpx2.AsyncClient sends them together, using asyncio.gather from Python for AI.

import asyncio

import httpx2
from model_api import app


async def ask(client, text):
    body = {"model": "local", "messages": [{"role": "user", "content": text}]}
    response = await client.post("/v1/chat/completions", json=body)
    response.raise_for_status()
    return response.json()["choices"][0]["message"]["content"]

ask is the call from lesson 9, written with async def. await client.post sends the request and lets other work run while the answer is on its way.

async def main():
    texts = ["I was charged twice", "My parcel has not arrived", "How do I change my password?"]
    transport = httpx2.ASGITransport(app=app)
    headers = {"Authorization": "Bearer sk-local-123"}
    async with httpx2.AsyncClient(transport=transport, base_url="http://model", headers=headers) as client:
        answers = await asyncio.gather(*[ask(client, text) for text in texts])
    for text, answer in zip(texts, answers):
        print(f"{text} -> {answer}")


asyncio.run(main())

ASGITransport is the async counterpart of the test client: it hands requests to the app directly. async with opens the client and closes its connections when the block ends. zip pairs each text with its answer, and gather keeps them in order.

python batch.py

Against a real API, replace the transport with the address: httpx2.AsyncClient(base_url="https://api.example.com", headers=headers). Everything else stays.

One client, not one per call

Opening a connection takes time. A client keeps connections open and reuses them, so make one client for the whole batch and pass it to each call, as main does. A new client inside ask would work and be slower for every ticket.