aiohttp provides HTTP client and server APIs for asyncio. On the client side, reuse ClientSession, which owns a connection pool and must be closed correctly.

import asyncio
import aiohttp


async def fetch_status(url: str) -> dict:
    timeout = aiohttp.ClientTimeout(total=10)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.get(url) as response:
            response.raise_for_status()
            return await response.json()


result = asyncio.run(fetch_status("https://api.example.com/status"))

In a long-running app, create the session during startup and close it during shutdown instead of rebuilding it in every function. Set explicit timeouts and bound simultaneous tasks with a semaphore or queue. Unlimited concurrency can overload both the remote service and your process.

Validate status and content before trusting JSON. Retry only transient failures and safe operations. Compare the Python HTTPX guide and review TaskGroup for task groups.

The official aiohttp client documentation, accessed July 22, 2026, covers sessions, responses, streaming, and timeouts. Choose based on API and ecosystem needs, not merely the async label.

Installation and the async model

Install the library in a virtual environment with python -m pip install aiohttp. A coroutine yields control while waiting for network I/O, allowing other tasks to make progress. This improves utilization for I/O-heavy workloads, but does not reduce remote-server latency or accelerate CPU-heavy work.

asyncio.run() belongs at a script entry point. Inside FastAPI, aiohttp, Jupyter, or another environment with a running event loop, use await instead of starting a nested loop. Libraries should expose async functions and leave loop ownership to the application.

Manage the ClientSession lifecycle

A session owns a connection pool, cookies, default headers, and shared settings. Creating one for every call wastes established connections and can exhaust sockets. Use one async with in a short script; in a persistent service, create the session at startup and close it during shutdown.

import aiohttp


class CatalogClient:
    def __init__(self, base_url: str, token: str):
        self._base_url = base_url
        self._token = token
        self._session: aiohttp.ClientSession | None = None

    async def __aenter__(self):
        timeout = aiohttp.ClientTimeout(total=15, connect=3, sock_read=10)
        self._session = aiohttp.ClientSession(
            base_url=self._base_url,
            timeout=timeout,
            headers={"Authorization": f"Bearer {self._token}"},
        )
        return self

    async def __aexit__(self, exc_type, exc, tb):
        assert self._session is not None
        await self._session.close()

Do not log authorization headers. Store long-lived credentials in the deployment's secret mechanism. A base URL simplifies relative routes, but callers still need a clear policy for any absolute URL.

Timeouts are part of the contract

Without useful limits, tasks may wait too long for a connection or response body. ClientTimeout distinguishes total time, connection establishment, pool acquisition, and time between body chunks. Choose values from the service-level objective and expected response size rather than copying arbitrary numbers.

Handle asyncio.TimeoutError separately from HTTP failures and invalid data. A timeout means the outcome may be unknown. For a write, the server could have completed the operation before its response was lost. Blindly retrying a purchase or creation can duplicate effects.

import asyncio
import aiohttp


async def get_json(session: aiohttp.ClientSession, path: str) -> dict:
    try:
        async with session.get(path) as response:
            response.raise_for_status()
            return await response.json(content_type="application/json")
    except asyncio.TimeoutError as error:
        raise RuntimeError("The API exceeded its deadline") from error
    except aiohttp.ClientResponseError as error:
        raise RuntimeError(f"HTTP response {error.status}") from error

Avoid placing an entire error body in public messages or indiscriminate logs because it may contain personal or internal data.

Query parameters, headers, and JSON

Use params for the query string and json for body serialization. This avoids manual concatenation and supplies the appropriate content type. data serves form fields and bytes; mixing these options can send a different request than intended.

async with session.get("/products", params={"page": 2, "limit": 20}) as response:
    response.raise_for_status()
    products = await response.json()

async with session.post("/events", json={"type": "view", "item_id": 42}) as response:
    response.raise_for_status()

raise_for_status() maps 400 and 500 responses to ClientResponseError, but domain logic may interpret certain states. A 404 can represent expected absence; a 409 may require reconciliation. Translate these cases inside a client layer so transport details do not spread through business logic.

Before trusting JSON, verify status, enforce a body limit, and validate structure. A proxy may return HTML unexpectedly. Schema validation can detect contract changes, but error handling should not leak the entire payload.

Bound concurrency

asyncio.gather() schedules every supplied coroutine. Given a large list, it can start thousands of requests at once. A semaphore bounds the resource-intensive region.

import asyncio


async def fetch_one(session, url, limit):
    async with limit:
        async with session.get(url) as response:
            response.raise_for_status()
            return await response.json()


async def fetch_all(session, urls):
    limit = asyncio.Semaphore(10)
    tasks = [fetch_one(session, url, limit) for url in urls]
    return await asyncio.gather(*tasks)

The connector can also constrain connections through aiohttp.TCPConnector(limit=...). A semaphore expresses operation concurrency; the connector protects its pool. For enormous collections, use a fixed worker queue because creating millions of task objects consumes memory as well. Honor published API quotas and 429 responses.

Retry responsibly

Retry only transient failures, with a maximum attempt count, exponential delay, and jitter. GET, HEAD, and selected operations carrying an idempotency key are common candidates. POST is not inherently safe. Statuses 400, 401, 403, and validation failures usually require correction rather than repetition.

If the service supplies Retry-After, interpret it within a reasonable ceiling. Cancellation should stop both waiting and requests; do not swallow CancelledError as an ordinary failure. Keep retry policy in one function so behavior and telemetry remain consistent.

Stream without unbounded memory

await response.read() and await response.json() buffer a body. For large files, iterate over response.content chunks and enforce a byte limit. Write to a temporary destination and publish it only after completion and, where available, hash verification.

from pathlib import Path


async def download(session, url: str, target: Path, max_bytes=20_000_000):
    total = 0
    async with session.get(url) as response:
        response.raise_for_status()
        with target.open("wb") as file:
            async for chunk in response.content.iter_chunked(64 * 1024):
                total += len(chunk)
                if total > max_bytes:
                    raise ValueError("Download exceeded its limit")
                file.write(chunk)

Production code should remove the temporary file on failure. Never turn a remote filename into a local path. Generate destinations within the application and protect against overwriting.

External URLs and SSRF

A client that fetches a user-supplied URL can become an SSRF vector, reaching internal services, cloud metadata, or localhost. Prefer an allowlist of schemes and hosts. Disable redirects or revalidate every redirect destination against the same policy.

String-only hostname checks can be insufficient in the presence of DNS changes and alternate address forms. High-risk applications need network, resolver, or proxy controls in addition to code validation. Never forward a trusted API authorization header to an arbitrary host.

Testing, observability, and shutdown

Test success, error statuses, invalid JSON, refused connections, timeout, oversized bodies, cancellation, and 429 behavior. A local test server or purpose-built aiohttp fixture can reproduce streaming and delays more accurately than a shallow mock.

Record method, approved host, normalized route, status, duration, attempt count, and byte size while excluding sensitive query strings and tokens. Latency and error metrics let you tune timeout and concurrency from evidence.

Before deployment, confirm one session per lifecycle, guaranteed closure, explicit timeouts, bounded concurrency, response release through async with, retries only for safe cases, and validation of external destinations. aiohttp's benefit comes from coordinating network waits with discipline, not from launching the largest possible number of tasks.

Define an explicit API contract version and test content-negotiation headers. A resilient integration fails clearly when it receives an incompatible media type or schema version instead of treating any JSON dictionary as valid. Keep parsing models separate from transport code so a contract migration can be reviewed independently.

Connection behavior also depends on DNS caching, TLS verification, proxies, and the connector. Keep certificate verification enabled. If a private certificate authority is required, configure a deliberate SSL context rather than disabling checks. Proxy credentials and URLs are secrets and should not appear in exception messages.

When shutting down a service, stop accepting new work, allow bounded in-flight operations to finish, and then close the session. This order avoids Unclosed client session warnings and partially written work. Place an upper bound on graceful shutdown so an unhealthy upstream cannot keep deployment termination open forever.