aiohttp oferece cliente e servidor HTTP para asyncio. No cliente, a regra central é reutilizar ClientSession, que mantém pool de conexões e deve ser fechada corretamente.

import asyncio
import aiohttp


async def buscar_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()


resultado = asyncio.run(buscar_status("https://api.example.com/status"))

Em uma aplicação persistente, crie a sessão no ciclo de vida e feche-a no encerramento, em vez de recriá-la dentro de cada função. Defina timeouts explícitos e limite tarefas simultâneas com semáforo ou fila. Concorrência sem limite pode sobrecarregar tanto o serviço remoto quanto sua máquina.

Valide status e conteúdo antes de confiar no JSON. Retry só deve cobrir falhas transitórias e operações idempotentes. Compare com HTTPX em Python e revise TaskGroup para grupos de tarefas.

A documentação oficial do cliente aiohttp, consultada em 22 de julho de 2026, detalha sessões, respostas, streaming e timeouts. Escolha a biblioteca pelo contrato e ecossistema, não apenas por ser assíncrona.

Instalação e modelo assíncrono

Instale em um ambiente virtual com python -m pip install aiohttp. Uma corrotina libera o loop de eventos enquanto espera rede, permitindo que outras tarefas avancem. Isso melhora utilização em cargas com muito I/O, mas não reduz por si só a latência do servidor remoto e não acelera trabalho pesado de CPU.

asyncio.run() é adequado no ponto de entrada de um script. Dentro de FastAPI, aiohttp, Jupyter ou outro ambiente que já possui loop, aguarde a corrotina com await; não inicie um loop aninhado. Bibliotecas devem expor funções assíncronas e deixar a decisão do loop para a aplicação.

Ciclo de vida de ClientSession

Uma sessão mantém pool de conexões, cookies, cabeçalhos padrão e configuração compartilhada. Criar uma por chamada desperdiça conexões e pode esgotar sockets. Em um script curto, use um único async with; em serviço persistente, crie a sessão no startup e feche no shutdown.

import aiohttp


class ClienteCatalogo:
    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()

Não registre cabeçalhos de autorização. Para credenciais persistentes, use o mecanismo de secrets da implantação. base_url simplifica rotas relativas, mas URLs absolutas passadas pelo chamador ainda precisam de política clara.

Timeouts são parte do contrato

Sem limites adequados, tarefas podem permanecer aguardando conexão ou corpo por tempo demais. ClientTimeout distingue tempo total, estabelecimento da conexão, aquisição no pool e leitura entre blocos. Escolha valores segundo o SLA e o tamanho esperado da resposta, não copie números sem medir.

Capture asyncio.TimeoutError separadamente de erros HTTP e dados inválidos. Um timeout indica resultado desconhecido: em uma operação de escrita, o servidor pode ter concluído o pedido antes de a resposta se perder. Repetir automaticamente uma compra ou criação pode duplicar efeitos.

import asyncio
import aiohttp


async def obter_json(session: aiohttp.ClientSession, caminho: str) -> dict:
    try:
        async with session.get(caminho) as response:
            response.raise_for_status()
            return await response.json(content_type="application/json")
    except asyncio.TimeoutError as erro:
        raise RuntimeError("A API excedeu o prazo") from erro
    except aiohttp.ClientResponseError as erro:
        raise RuntimeError(f"Resposta HTTP {erro.status}") from erro

Não inclua o corpo completo de uma resposta de erro em mensagens públicas ou logs indiscriminados. Ele pode conter dados pessoais ou detalhes internos.

Parâmetros, cabeçalhos e JSON

Use params para query strings e json para serialização do corpo. Isso evita concatenação manual e informa o tipo de conteúdo correto. data serve a formulários e bytes; misturar as opções pode gerar uma requisição diferente da esperada.

async with session.get("/produtos", params={"pagina": 2, "limite": 20}) as resposta:
    resposta.raise_for_status()
    produtos = await resposta.json()

async with session.post("/eventos", json={"tipo": "visualizacao", "item_id": 42}) as resposta:
    resposta.raise_for_status()

raise_for_status() transforma respostas 400 e 500 em ClientResponseError, mas o domínio pode precisar interpretar alguns estados. Um 404 pode significar ausência esperada; um 409 pode pedir reconciliação. Faça essa tradução em uma camada de cliente, mantendo detalhes de HTTP fora da regra de negócio.

Antes de confiar em JSON, confira status, limite de corpo e estrutura. O servidor pode enviar HTML de proxy com status inesperado. Validação de schema ajuda a detectar mudanças de contrato, mas deve produzir erros que não vazem o payload inteiro.

Concorrência limitada

asyncio.gather() inicia todas as corrotinas fornecidas. Em uma lista grande, isso pode criar milhares de requisições simultâneas. Um semáforo limita a região que ocupa recursos externos.

import asyncio


async def buscar_um(session, url, limite):
    async with limite:
        async with session.get(url) as resposta:
            resposta.raise_for_status()
            return await resposta.json()


async def buscar_todos(session, urls):
    limite = asyncio.Semaphore(10)
    tarefas = [buscar_um(session, url, limite) for url in urls]
    return await asyncio.gather(*tarefas)

O conector também permite limitar conexões com aiohttp.TCPConnector(limit=...). O semáforo expressa concorrência da operação; o conector protege o pool. Use filas com workers para coleções enormes, pois criar milhões de objetos de tarefa também consome memória. Respeite limites publicados pela API e respostas 429.

Retentativas responsáveis

Retente somente falhas transitórias, com número máximo, espera exponencial e jitter. GET, HEAD e algumas operações com chave de idempotência costumam ser candidatas. POST não é automaticamente seguro. Estados 400, 401, 403 e erros de validação normalmente exigem correção, não repetição.

Se o serviço retorna Retry-After, interprete-o dentro de um limite razoável. Cancelamento deve interromper a espera e a requisição; não capture CancelledError como se fosse uma falha comum. Uma política de retry pertence a uma função central para permanecer consistente e observável.

Streaming sem explodir memória

await response.read() e await response.json() acumulam o corpo. Para arquivos grandes, leia blocos de response.content e aplique limite. Grave em destino temporário e publique apenas após completar e, quando disponível, verificar hash.

from pathlib import Path


async def baixar(session, url: str, destino: Path, max_bytes=20_000_000):
    total = 0
    async with session.get(url) as resposta:
        resposta.raise_for_status()
        with destino.open("wb") as arquivo:
            async for bloco in resposta.content.iter_chunked(64 * 1024):
                total += len(bloco)
                if total > max_bytes:
                    raise ValueError("Download excedeu o limite")
                arquivo.write(bloco)

Em código de produção, remova o temporário quando houver erro. Não aceite um nome de arquivo remoto como caminho local. Gere o destino na aplicação e proteja contra sobrescrita.

URLs externas e SSRF

Um cliente que busca uma URL fornecida pelo usuário pode virar um vetor de SSRF, acessando serviços internos, metadados de nuvem ou localhost. Prefira uma lista de hosts e esquemas permitidos. Bloqueie redirecionamentos para destinos fora da política e revalide após cada redirecionamento.

Validar apenas a string do hostname pode ser insuficiente diante de DNS e endereços alternativos. Aplicações de alto risco precisam de controles de rede, resolução e proxy além da validação no código. Nunca envie o cabeçalho de autorização de uma API confiável para um host arbitrário.

Testes, observabilidade e encerramento

Teste sucesso, status de erro, JSON inválido, conexão recusada, timeout, corpo acima do limite, cancelamento e 429. Prefira um servidor HTTP local de teste ou mecanismo oficial de mock que reproduza streaming e atrasos; mocks superficiais podem ocultar erros no uso do contexto assíncrono.

Registre método, host aprovado, rota normalizada, status, duração, tentativas e tamanho, evitando query strings sensíveis e tokens. Métricas de latência e erro ajudam a ajustar timeout e concorrência com evidência.

Antes de implantar, confirme uma sessão por ciclo de vida, fechamento garantido, timeouts explícitos, concorrência limitada, resposta liberada pelo async with, retries apenas para casos seguros e validação de destinos externos. O ganho do aiohttp vem de coordenar espera de rede com disciplina, não de lançar o máximo possível de tarefas.

Defina também uma versão clara do contrato da API e teste os cabeçalhos de negociação de conteúdo. Uma integração resiliente falha de forma compreensível quando recebe um formato ou versão incompatível, em vez de tratar qualquer dicionário JSON como válido.