# Build a custom HTTP client

Copy for LLM

This guide implements a custom [`HttpClientAsync`](https://docs.apify.com/api/client/python/api/client/python/reference/class/HttpClientAsync.md) with [aiohttp](https://docs.aiohttp.org/) and a custom [`HttpClient`](https://docs.apify.com/api/client/python/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/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 HTTPX implementations, see [HTTP clients](https://docs.apify.com/api/client/python/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/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.

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