Async Python

HTTP Clients

HTTP Clients

Every model provider is an HTTP API underneath. Frameworks hide that most of the time, but you will need to call an API directly whenever you build a tool — fetching a ticket, querying an internal service, calling something your company already runs.

This page is about doing that properly.

The one to use

httpx handles both synchronous and async code with one API. requests is the older standard and is still everywhere, but it cannot do async, which rules it out for agent work.

pip install httpx

A simple GET

import httpx

response = httpx.get("https://api.example.com/tickets/4471")

print(response.status_code)
print(response.json())
  • .status_code — 200 means success, 4xx is your mistake, 5xx is theirs
  • .json() — parses the body into a dictionary
  • .text — the raw body, useful when the response is not JSON

A POST with JSON and a key

This is the shape of every model API call.

import os
import httpx

response = httpx.post(
    "https://api.deepseek.com/chat/completions",
    headers={"Authorization": f"Bearer {os.getenv('DEEPSEEK_API_KEY')}"},
    json={
        "model": "deepseek-chat",
        "messages": [{"role": "user", "content": "Hello"}],
    },
    timeout=30,
)

data = response.json()
print(data["choices"][0]["message"]["content"])

Points worth noticing:

  • json= sets the body and the content-type header for you. Do not build the JSON string yourself
  • the key comes from the environment, never from the code
  • timeout is set explicitly, which matters more than people expect

That nested data["choices"][0]["message"]["content"] is the standard chat-completions shape. Work left to right: a dictionary, a list, a dictionary, a dictionary.

Always set a timeout

httpx.get(url, timeout=10)

Without one, a request can hang far longer than you would ever want. In an agent that means one unresponsive service stalls the whole run, and the user sees nothing at all.

Ten to thirty seconds is reasonable for a model call. Shorter for internal services.

Check the status

httpx does not raise on a 404 or a 500 by default. You get a response object with a bad status and, often, .json() failing confusingly a line later.

response = httpx.get(url, timeout=10)
response.raise_for_status()
data = response.json()

raise_for_status() turns any 4xx or 5xx into an exception, which is almost always what you want.

try:
    response = httpx.get(url, timeout=10)
    response.raise_for_status()
except httpx.HTTPStatusError as e:
    print("server said:", e.response.status_code, e.response.text)
except httpx.TimeoutException:
    print("timed out")
except httpx.RequestError as e:
    print("could not reach it:", e)

Three different failures, three different responses. A 401 means your key is wrong and retrying is pointless. A timeout is worth retrying.

Reuse the client

Creating a client per request re-does the TCP and TLS handshake every time. For anything in a loop, make one and keep it.

with httpx.Client(
    base_url="https://api.example.com",
    headers={"Authorization": f"Bearer {token}"},
    timeout=30,
) as client:
    a = client.get("/tickets/1").json()
    b = client.get("/tickets/2").json()

base_url and shared headers also stop you repeating yourself, and the with block closes the connections properly.

The async version

Same library, three changes: AsyncClient, async with, and await.

import httpx

async def fetch_ticket(client, ticket_id):
    response = await client.get(f"/tickets/{ticket_id}")
    response.raise_for_status()
    return response.json()

async def main():
    async with httpx.AsyncClient(base_url="https://api.example.com", timeout=30) as client:
        ticket = await fetch_ticket(client, 4471)
        print(ticket)

Note that httpx.AsyncClient is what makes async worthwhile at all. Using requests inside an async def blocks the event loop and quietly removes every benefit.

Fetching several at once

This is the payoff, and the reason to bother with async.

import asyncio
import httpx

async def main(ids):
    async with httpx.AsyncClient(base_url="https://api.example.com", timeout=30) as client:
        results = await asyncio.gather(
            *[fetch_ticket(client, i) for i in ids],
            return_exceptions=True,
        )
    return results

Ten tickets in the time of the slowest one, rather than ten times the average. return_exceptions=True keeps the nine that worked when one fails.

Be careful with large lists — see the concurrency page for capping how many run at once.

Streaming a response

For a model streaming tokens back:

async with httpx.AsyncClient(timeout=None) as client:
    async with client.stream("POST", url, json=payload) as response:
        async for line in response.aiter_lines():
            if line.startswith("data: "):
                print(line[6:])

timeout=None because a stream stays open by design. The data: prefix is server-sent events, which is how the major providers stream.

You will not often write this by hand — frameworks give you .astream() — but it is useful to know what is underneath when a stream misbehaves.

Retrying

import asyncio, httpx

async def get_with_retry(client, url, attempts=3):
    for i in range(attempts):
        try:
            response = await client.get(url)
            response.raise_for_status()
            return response.json()
        except (httpx.TimeoutException, httpx.HTTPStatusError) as e:
            status = getattr(getattr(e, "response", None), "status_code", None)
            if status and status < 500 and status != 429:
                raise
            if i == attempts - 1:
                raise
            await asyncio.sleep(2 ** i)

The rule encoded there: retry 429 and 5xx, do not retry other 4xx. A 401 or a 400 will fail identically every time; retrying wastes time and money. A 429 means slow down, and backing off is exactly right.

Coming from Java or C#

ConceptJavaC#Python
ClientHttpClient, OkHttpHttpClienthttpx.Client
AsyncsendAsyncawait SendAsynchttpx.AsyncClient
JSON bodyJackson + builderPostAsJsonAsyncjson=
Parse JSONreadValueReadFromJsonAsync.json()
Throw on error statusmanualEnsureSuccessStatusCoderaise_for_status()
Reuserequired, HttpClient is expensiverequiredrecommended

If you are used to HttpClient, the habits transfer directly: create one, reuse it, set timeouts, check the status. raise_for_status() is EnsureSuccessStatusCode with a different name.

Common mistakes

  • No timeout, so a request hangs indefinitely.
  • Not checking the status, then getting a confusing JSON error instead of the real one.
  • A new client per request inside a loop.
  • requests inside async code, blocking the loop.
  • Retrying a 401. It will never succeed.
  • The API key in the code rather than the environment.

Practise this

  • Call a public API with httpx.get, print the status code, and parse the JSON.
  • Add raise_for_status() and trigger it by requesting a URL that 404s.
  • Rewrite it with AsyncClient and fetch three URLs with asyncio.gather.
  • Set a one-second timeout against a slow endpoint and catch the TimeoutException.

Try It Yourself

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

example.py
1import asyncio
2# Install with: pip install httpx
3
4# Simulated HTTP client behavior for demo
5class MockAsyncClient:
6 """Mock client for demonstration."""
7
8 async def __aenter__(self):
9 return self
10
11 async def __aexit__(self, *args):
12 pass
13
14 async def post(self, url: str, json: dict, timeout: float = 30.0):
15 await asyncio.sleep(0.5) # Simulate network
16
17 class Response:
18 status_code = 200
19 def json(self):
20 return {
21 "choices": [{
22 "message": {
23 "content": f"Response to: {json.get('messages', [{}])[-1].get('content', '')}"
24 }
25 }]
26 }
27 return Response()
28
29async def chat_completion(messages: list[dict]) -> str:
30 """Make a chat completion request."""
31 async with MockAsyncClient() as client:
32 response = await client.post(
33 "https://api.deepseek.com/chat/completions",
34 json={
35 "model": "deepseek-chat",
36 "messages": messages
37 },
38 timeout=30.0
39 )
40
41 if response.status_code == 200:
42 data = response.json()
43 return data["choices"][0]["message"]["content"]
44 else:
45 raise Exception(f"API error: {response.status_code}")
46
47async def main():
48 messages = [
49 {"role": "system", "content": "You are helpful."},
50 {"role": "user", "content": "Hello!"}
51 ]
52
53 response = await chat_completion(messages)
54 print(f"AI: {response}")
55
56asyncio.run(main())