Skip to main content
Version: Next

Build a custom HTTP client

Copy for LLM

This guide implements a custom HttpClientAsync with aiohttp and a custom HttpClient with requests. Neither library satisfies the HttpResponse protocol, so both examples also show how to adapt a foreign response API.

For an overview of the architecture and the built-in Impit and HTTPX implementations, see HTTP clients.

Installation

Install the transport alongside the Apify client. Neither aiohttp nor requests is an apify-client extra:

pip install apify-client aiohttp   # for the asynchronous client
pip install apify-client requests  # for the synchronous client

Implementation

Each example has three parts:

  1. The response adapter, AiohttpResponse or RequestsResponse, maps the library's own response onto the HttpResponse protocol that resource clients expect.
  2. The client, AiohttpHttpClient or RequestsHttpClient, implements the transport, error-classification, and lifecycle hooks. It inherits request preparation, retry handling, timeout growth, and API error conversion from its base class. Only the requests client also overrides timeout classification. requests.Timeout doesn't derive from Python's TimeoutError, while aiohttp's timeout errors do, so the inherited default already recognizes them.
  3. with_custom_http_client() connects the implementation to the resource clients and applies the API token. The context manager closes the session at shutdown.
from __future__ import annotations

import asyncio
import json as jsonlib
from typing import TYPE_CHECKING, Any

import aiohttp
from typing_extensions import override

from apify_client import ApifyClientAsync
from apify_client.http_clients import HttpClientAsync, HttpResponse

if TYPE_CHECKING:
    from collections.abc import AsyncIterator, Iterator, Mapping

TOKEN = 'MY-APIFY-TOKEN'


class AiohttpResponse:
    """Adapt an aiohttp response to the Apify client's HttpResponse protocol."""

    def __init__(self, response: aiohttp.ClientResponse) -> None:
        self._response = response
        self._body: bytes | None = None

    @property
    def status_code(self) -> int:
        return self._response.status

    @property
    def headers(self) -> Mapping[str, str]:
        return self._response.headers

    @property
    def content(self) -> bytes:
        if self._body is None:
            raise RuntimeError(
                'The streamed response has not been read yet; call aread() first'
            )
        return self._body

    @property
    def text(self) -> str:
        encoding = self._response.charset or 'utf-8'
        return self.content.decode(encoding, errors='replace')

    def json(self) -> Any:
        return jsonlib.loads(self.text)

    def read(self) -> bytes:
        return self.content

    async def aread(self) -> bytes:
        if self._body is None:
            self._body = await self._response.read()
        return self._body

    def close(self) -> None:
        self._response.close()

    async def aclose(self) -> None:
        await self._response.wait_for_close()

    def iter_bytes(self) -> Iterator[bytes]:
        body = self.content
        if body:
            yield body

    async def aiter_bytes(self) -> AsyncIterator[bytes]:
        if self._body is not None:
            if self._body:
                yield self._body
            return
        async for chunk in self._response.content.iter_chunked(64 * 1024):
            yield chunk


class AiohttpHttpClient(HttpClientAsync):
    """Minimal custom asynchronous HTTP client backed by aiohttp."""

    def __init__(self) -> None:
        super().__init__()
        self._session = aiohttp.ClientSession()

    @override
    def is_retryable_transport_error(self, exc: Exception) -> bool:
        return isinstance(exc, (TimeoutError, aiohttp.ClientError))

    @override
    async def aclose(self) -> None:
        await self._session.close()

    @override
    async def send_request(
        self,
        *,
        method: str,
        url: str,
        headers: dict[str, str],
        content: bytes | None,
        timeout: float | None,
        stream: bool,
    ) -> HttpResponse:
        response = await self._session.request(
            method=method,
            url=url,
            headers=headers,
            data=content,
            timeout=aiohttp.ClientTimeout(total=timeout),
        )
        adapted_response = AiohttpResponse(response)

        if not stream:
            await adapted_response.aread()

        return adapted_response


async def main() -> None:
    async with AiohttpHttpClient() as http_client:
        client = ApifyClientAsync.with_custom_http_client(
            token=TOKEN,
            http_client=http_client,
        )
        actor = await client.actor('apify/hello-world').get()
        print(actor)


if __name__ == '__main__':
    asyncio.run(main())
warning

These examples are compact integrations, not a replacement for all built-in client behavior. A production custom client should account for transport-specific details such as proxy configuration, TLS settings, redirects, and response resource cleanup. Timeout semantics differ per transport too: the aiohttp example passes the value as a budget for the whole request, while requests applies it to each socket read. Both example sessions also keep a shared cookie jar, which replays server cookies on later API requests. The built-in HTTPX client clears it instead.