aiohttp
aiohttp is the async HTTP workhorse for Python's asyncio: one package that is both an HTTP client (ClientSession, with connection pooling, streaming, and timeouts) and a web server framework (aiohttp.web, with middleware and pluggable routing), plus WebSocket support on both sides. It sits lower level than requests or FastAPI: you write async with blocks and manage session lifetimes yourself, and in exchange you get fine control and high throughput on concurrency-heavy workloads like crawlers, aggregators, and long-lived socket services.
The proven pick when raw async HTTP throughput or WebSockets matter, client side or server side. For everyday API servers use FastAPI, and for simple scripts use requests or httpx; aiohttp rewards the workloads that actually need it.
Use it if
- You need thousands of concurrent HTTP requests from one process (crawler, API aggregator, proxy) where blocking clients would need a thread per call
- You need WebSocket client or server support inside asyncio without adding a second dependency
- You want client and server in one library, for example a service that both serves HTTP and calls upstream APIs on the same event loop
- You want control over pooling, per-request timeouts, and DNS behavior (the README recommends aiodns for speed)
- Your codebase is synchronous: requests or sync httpx is far simpler, and rebuilding around asyncio for a few HTTP calls is not worth it
- You want requests-style ergonomics with async as an option: httpx has a near-identical API in both modes plus HTTP/2
- You are picking a framework for a typical JSON API: FastAPI gives you validation and OpenAPI docs; aiohttp's server layer is deliberately minimal
- You need HTTP/2 or HTTP/3: the aiohttp client and server speak HTTP/1.1 only
Setup reality
pip install aiohttp is quick, and wheels ship for CPython 3.10+ so you rarely compile anything, though unusual platforms may build the C extensions of aiohttp and its multidict/yarl/frozenlist dependencies from source. The real cost is discipline, not installation: everything lives inside async with blocks, a ClientSession must be created once inside a running loop and reused (per-request sessions defeat pooling and leak sockets), and a session you forget to close greets you with the Unclosed client session warning at shutdown.
Patterns
Fetch a page with ClientSessionclient-get
import aiohttp
import asyncio
async def main():
async with aiohttp.ClientSession() as session:
async with session.get('https://python.org') as response:
print('Status:', response.status)
html = await response.text()
asyncio.run(main())Both context managers matter: the session owns the connection pool and the response releases its connection back to it.
POST JSON and read a JSON responseclient-post-json
async with session.post(url, json={'name': 'demo'}) as resp:
resp.raise_for_status()
data = await resp.json()resp.json() raises ContentTypeError when the server sends a non-JSON content type; pass content_type=None to skip that check.
Create one session for the whole appreuse-session
class ApiClient:
def __init__(self) -> None:
self._session: aiohttp.ClientSession | None = None
async def start(self) -> None:
self._session = aiohttp.ClientSession()
async def close(self) -> None:
if self._session:
await self._session.close()One ClientSession per application, created inside a running event loop; per-request sessions are the top aiohttp performance mistake.
Set client timeoutsset-timeout
timeout = aiohttp.ClientTimeout(total=30, connect=5)
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get(url) as resp:
body = await resp.text()The default total timeout is 5 minutes; set your own or a slow upstream quietly stalls your workers.
Fetch many URLs concurrentlyconcurrent-requests
async def fetch(session, url):
async with session.get(url) as resp:
return await resp.text()
async def main(urls):
async with aiohttp.ClientSession() as session:
pages = await asyncio.gather(
*(fetch(session, u) for u in urls)
)Bound concurrency with an asyncio.Semaphore for large URL lists, or you exhaust sockets and trip rate limits.
Stream a large download to diskstream-download
async with session.get(url) as resp:
with open('big.bin', 'wb') as f:
async for chunk in resp.content.iter_chunked(1 << 16):
f.write(chunk)await resp.read() buffers the entire body in memory; iter_chunked is the way to handle files bigger than RAM.
Connect to a WebSocket as a clientwebsocket-client
async with session.ws_connect('wss://example.com/ws') as ws:
await ws.send_str('hello')
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
print(msg.data)
elif msg.type == aiohttp.WSMsgType.ERROR:
breakReceiving means iterating the socket; the loop ends when the connection closes, so branch on msg.type for ERROR and CLOSE.
Serve HTTP with aiohttp.webserver-basic
from aiohttp import web
async def handle(request):
name = request.match_info.get('name', 'Anonymous')
return web.Response(text=f'Hello, {name}')
app = web.Application()
app.add_routes([
web.get('/', handle),
web.get('/{name}', handle),
])
if __name__ == '__main__':
web.run_app(app)web.run_app blocks and manages the event loop itself; to embed the server in an existing loop use AppRunner instead.
Handle WebSockets on the serverserver-websocket
from aiohttp import web
async def wshandle(request):
ws = web.WebSocketResponse()
await ws.prepare(request)
async for msg in ws:
if msg.type == web.WSMsgType.text:
await ws.send_str(f'Hello, {msg.data}')
elif msg.type == web.WSMsgType.close:
break
return wsCall prepare(request) before any send, and return the WebSocketResponse from the handler when the loop exits.
Add error-handling middlewareserver-middleware
from aiohttp import web
@web.middleware
async def error_mw(request, handler):
try:
return await handler(request)
except web.HTTPException:
raise
except Exception:
return web.json_response({'error': 'internal'}, status=500)
app = web.Application(middlewares=[error_mw])Re-raise HTTPException subclasses: aiohttp uses them as control flow (redirects, 404s), not as errors.