Build an Async HTTP Client with aiohttp

Making HTTP requests is a common task in Python. The classic requests library is simple and reliable.

But it has one big limit. It works one request at a time. When you need to fetch hundreds of URLs, that gets slow fast.

This is where aiohttp shines. It lets you send many requests at once using Python's asyncio.

In this guide, you will build a clean async HTTP client step by step. We will cover setup, sessions, concurrency, timeouts, and error handling.

Why Async HTTP Matters

Most of the time spent on an HTTP request is waiting. The server thinks, the network moves bytes, and your program sits idle.

With a synchronous client, that idle time is wasted. Ten requests take ten times as long as one.

With an async client, all ten can wait at the same time. The total time is close to the slowest single request.

This is a huge win for web scraping, API aggregation, and health checks.

Installing aiohttp

First, install the library. It is pure Python with a small C extension for speed.

pip install aiohttp

You also need Python 3.7 or newer. Async syntax has improved a lot in recent versions.

Your First Async Request

Let's start small. The code below fetches a single URL and prints the status code and body length.

import asyncio
import aiohttp

async def fetch(url):
    # Create a client session. Always use a session, not raw requests.
    async with aiohttp.ClientSession() as session:
        # Send a GET request and wait for the response.
        async with session.get(url) as response:
            # Read the full response body as text.
            text = await response.text()
            print(f"Status: {response.status}")
            print(f"Length: {len(text)} characters")
            return text

# Run the async function.
asyncio.run(fetch("https://example.com"))

Status: 200
Length: 1256 characters

Notice the two async with blocks. The first opens a session. The second opens a response.

Both must be closed properly. The async with syntax handles that for you.

Reuse One Session

Opening a new session for every request is wasteful. Each session keeps a connection pool and cookies.

The right pattern is one session for your whole program. Pass it into your fetch function.

import asyncio
import aiohttp

async def fetch(session, url):
    # Use the shared session passed in from the caller.
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        "https://example.com",
        "https://example.org",
        "https://example.net",
    ]
    # One session for all requests.
    async with aiohttp.ClientSession() as session:
        for url in urls:
            body = await fetch(session, url)
            print(f"{url} -> {len(body)} chars")

asyncio.run(main())

https://example.com -> 1256 chars
https://example.org -> 1256 chars
https://example.net -> 1256 chars

This works, but it is still sequential. Each request waits for the one before it.

To unlock real speed, we need to run them at the same time.

Run Requests Concurrently

Concurrency in asyncio comes from tasks. A task is a coroutine scheduled to run on the event loop.

We can use asyncio.gather to run many tasks and collect all results.

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as response:
        text = await response.text()
        return url, response.status, len(text)

async def main():
    urls = [
        "https://example.com",
        "https://example.org",
        "https://example.net",
    ]
    async with aiohttp.ClientSession() as session:
        # Build a task for each URL.
        tasks = [fetch(session, url) for url in urls]
        # Run all tasks at once and wait for every result.
        results = await asyncio.gather(*tasks)
        for url, status, length in results:
            print(f"{url} -> {status} ({length} chars)")

asyncio.run(main())

https://example.com -> 200 (1256 chars)
https://example.org -> 200 (1256 chars)
https://example.net -> 200 (1256 chars)

All three requests now overlap. The total time is close to one request, not three.

This is the core pattern of an async HTTP client. Session, tasks, and gather.

Control the Concurrency Level

Firing thousands of requests at once can overload servers. It can also exhaust your file descriptors.

A semaphore limits how many tasks run at the same time. It is a simple and effective guard.

import asyncio
import aiohttp

# Allow at most 10 requests at the same time.
sem = asyncio.Semaphore(10)

async def fetch(session, url):
    # Acquire a slot before sending the request.
    async with sem:
        async with session.get(url) as response:
            text = await response.text()
            return url, response.status, len(text)

async def main():
    urls = [f"https://example.com/?page={i}" for i in range(50)]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        print(f"Fetched {len(results)} URLs with a limit of 10")

asyncio.run(main())

Fetched 50 URLs with a limit of 10

Pick a limit that respects the target server. Ten to twenty is a safe starting point.

Add Timeouts

Some servers are slow. Some never respond. Without a timeout, your client can hang forever.

aiohttp lets you set a total timeout on the session. You can also set a connect timeout.

import asyncio
import aiohttp

# Total request timeout of 5 seconds.
timeout = aiohttp.ClientTimeout(total=5, connect=2)

async def fetch(session, url):
    async with session.get(url, timeout=timeout) as response:
        return await response.text()

async def main():
    async with aiohttp.ClientSession(timeout=timeout) as session:
        try:
            body = await fetch(session, "https://example.com")
            print(f"Got {len(body)} chars")
        except asyncio.TimeoutError:
            print("Request timed out")

asyncio.run(main())

Got 1256 chars

Always set a timeout in production code. It turns a hang into a clean error.

Handle Errors Gracefully

Networks fail. DNS breaks. Servers return 500s. Your client must survive all of it.

Catch aiohttp.ClientError for request problems. Catch asyncio.TimeoutError for slow servers.

If you use asyncio.gather with return_exceptions=True, one failure will not cancel the rest.

import asyncio
import aiohttp

async def fetch(session, url):
    try:
        async with session.get(url) as response:
            text = await response.text()
            return url, response.status, len(text)
    except asyncio.TimeoutError:
        return url, "timeout", 0
    except aiohttp.ClientError as e:
        # Covers connection errors, invalid URLs, and more.
        return url, f"error: {type(e).__name__}", 0

async def main():
    urls = ["https://example.com", "https://bad.invalid"]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        for url, status, length in results:
            print(f"{url} -> {status} ({length} chars)")

asyncio.run(main())

https://example.com -> 200 (1256 chars)
https://bad.invalid -> error: ClientConnectorError (0 chars)

Now a single bad URL will not crash your whole batch. That is essential for real workloads.

Send POST Requests and JSON

GET is not the only method. aiohttp handles POST, headers, and JSON bodies with ease.

import asyncio
import aiohttp

async def post_json(session, url, payload):
    # Send a JSON body and parse the JSON response.
    async with session.post(url, json=payload) as response:
        data = await response.json()
        return response.status, data

async def main():
    async with aiohttp.ClientSession() as session:
        status, data = await post_json(
            session,
            "https://httpbin.org/post",
            {"name": "async", "ok": True},
        )
        print(f"Status: {status}")
        print(f"Echoed JSON: {data['json']}")

asyncio.run(main())

Status: 200
Echoed JSON: {'name': 'async', 'ok': True}

Use the json= argument to send data. Use response.json() to read it back.

Set Custom Headers

Many APIs need an API key or a custom user agent. You can set headers per request or per session.

import asyncio
import aiohttp

async def main():
    headers = {
        "User-Agent": "MyAsyncClient/1.0",
        "Authorization": "Bearer TOKEN",
    }
    async with aiohttp.ClientSession(headers=headers) as session:
        async with session.get("https://httpbin.org/headers") as response:
            data = await response.json()
            print(data["headers"]["User-Agent"])

asyncio.run(main())

MyAsyncClient/1.0

Session-level headers are applied to every request. That keeps your code clean and consistent.

Best Practices Checklist

Follow these rules and your async client will be fast and safe.

Reuse one session for the whole program. Do not create a session per request.

Limit concurrency with a semaphore. Respect the servers you call.

Always set timeouts so a slow server cannot freeze your app.

Handle exceptions per task. One failure should not kill the batch.

Close the session with async with or by calling session.close().

Do not block the loop. Avoid synchronous calls like time.sleep inside coroutines.

Common Pitfalls to Avoid

Beginners often hit the same few problems. Here is how to dodge them.

Creating a new session inside a loop is slow and leaks connections. Create it once outside.

Forgetting await returns a coroutine object, not a response. You will see errors like "coroutine was never awaited".

Using asyncio.run inside an already running loop raises an error. Use it only at the top level.

Ignoring rate limits gets your IP blocked. Add a small delay or a semaphore when scraping.

Conclusion

Building an async HTTP client with aiohttp is not hard. The pattern is small and repeatable.

Create one session, build tasks, and gather the results. Add a semaphore for control and timeouts for safety.

Wrap each request in a try block so failures stay local. Then your client can handle real-world traffic.

Start with the single-request example. Then move to asyncio.gather for concurrency.

Once you master these basics, you can fetch hundreds of URLs in seconds. That is the real power of async Python.