# Build a custom HTTP client

Copy for LLM

This guide implements a custom [`HttpClientAsync`](https://docs.apify.com/api/client/python/reference/class/HttpClientAsync.md) with [aiohttp](https://docs.aiohttp.org/) and a custom [`HttpClient`](https://docs.apify.com/api/client/python/reference/class/HttpClient.md) with [requests](https://requests.readthedocs.io/). Neither library satisfies the [`HttpResponse`](https://docs.apify.com/api/client/python/reference/class/HttpResponse.md) 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 HTTPX2 implementations, see [HTTP clients](https://docs.apify.com/api/client/python/docs/concepts/custom-http-clients.md).

## Installation[](#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[](#implementation)

Each example has three parts:

1. The response adapter, `AiohttpResponse` or `RequestsResponse`, maps the library's own response onto the [`HttpResponse`](https://docs.apify.com/api/client/python/reference/class/HttpResponse.md) 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.

Both libraries accept an iterator of chunks as the request body, a synchronous one in `requests` and an asynchronous one in aiohttp, so [streamed uploads](https://docs.apify.com/api/client/python/docs/concepts/streaming-resources.md#streaming-uploads) work through these clients as they are.

* Async client
* Sync client

```
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 | AsyncIterator[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())
```

```
from __future__ import annotations



import json as jsonlib

from typing import TYPE_CHECKING, Any



import requests

from typing_extensions import override



from apify_client import ApifyClient

from apify_client.http_clients import HttpClient, HttpResponse



if TYPE_CHECKING:

    from collections.abc import AsyncIterator, Iterator, Mapping



TOKEN = 'MY-APIFY-TOKEN'





class RequestsResponse:

    """Adapt a requests response to the Apify client's HttpResponse protocol."""



    def __init__(self, response: requests.Response) -> None:

        self._response = response

        self._body: bytes | None = None



    @property

    def status_code(self) -> int:

        return self._response.status_code



    @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 read() first'

            )

        return self._body



    @property

    def text(self) -> str:

        encoding = self._response.encoding or 'utf-8'

        return self.content.decode(encoding, errors='replace')



    def json(self) -> Any:

        return jsonlib.loads(self.text)



    def read(self) -> bytes:

        if self._body is None:

            self._body = self._response.content

        return self._body



    async def aread(self) -> bytes:

        return self.read()



    def close(self) -> None:

        self._response.close()



    async def aclose(self) -> None:

        self.close()



    def iter_bytes(self) -> Iterator[bytes]:

        if self._body is not None:

            if self._body:

                yield self._body

            return

        yield from self._response.iter_content(64 * 1024)



    async def aiter_bytes(self) -> AsyncIterator[bytes]:

        for chunk in self.iter_bytes():

            yield chunk





class RequestsHttpClient(HttpClient):

    """Minimal custom synchronous HTTP client backed by requests."""



    def __init__(self) -> None:

        super().__init__()

        self._session = requests.Session()



    @override

    def is_timeout_error(self, exc: Exception) -> bool:

        return super().is_timeout_error(exc) or isinstance(exc, requests.Timeout)



    @override

    def is_retryable_transport_error(self, exc: Exception) -> bool:

        return isinstance(

            exc,

            (

                requests.ConnectionError,

                requests.Timeout,

                requests.exceptions.ChunkedEncodingError,

            ),

        )



    @override

    def close(self) -> None:

        self._session.close()



    @override

    def send_request(

        self,

        *,

        method: str,

        url: str,

        headers: dict[str, str],

        content: bytes | Iterator[bytes] | None,

        timeout: float | None,

        stream: bool,

    ) -> HttpResponse:

        response = self._session.request(

            method=method,

            url=url,

            headers=headers,

            data=content,

            timeout=timeout,

            stream=stream,

        )

        adapted_response = RequestsResponse(response)



        if not stream:

            adapted_response.read()



        return adapted_response





def main() -> None:

    with RequestsHttpClient() as http_client:

        client = ApifyClient.with_custom_http_client(

            token=TOKEN,

            http_client=http_client,

        )

        actor = client.actor('apify/hello-world').get()

        print(actor)





if __name__ == '__main__':

    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 HTTPX2 client clears it instead.