diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
index 3716767d..366a953b 100644
--- a/.github/workflows/tests.yml
+++ b/.github/workflows/tests.yml
@@ -40,7 +40,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        dependency: ["aiohttp", "requests", "websockets"]
+        dependency: ["aiohttp", "requests", "httpx", "websockets"]
 
     steps:
       - uses: actions/checkout@v3
diff --git a/Makefile b/Makefile
index 2275092c..59d08bac 100644
--- a/Makefile
+++ b/Makefile
@@ -3,7 +3,7 @@
 SRC_PYTHON := gql tests docs/code_examples
 
 dev-setup:
-	python pip install -e ".[test]"
+	python -m pip install -e ".[test]"
 
 tests:
 	pytest tests --cov=gql --cov-report=term-missing -vv
@@ -17,6 +17,9 @@ tests_aiohttp:
 tests_requests:
 	pytest tests --requests-only
 
+tests_httpx:
+	pytest tests --httpx-only
+
 tests_websockets:
 	pytest tests --websockets-only
 
diff --git a/docs/code_examples/httpx_async.py b/docs/code_examples/httpx_async.py
new file mode 100644
index 00000000..9a01232d
--- /dev/null
+++ b/docs/code_examples/httpx_async.py
@@ -0,0 +1,34 @@
+import asyncio
+
+from gql import Client, gql
+from gql.transport.httpx import HTTPXAsyncTransport
+
+
+async def main():
+
+    transport = HTTPXAsyncTransport(url="https://countries.trevorblades.com/graphql")
+
+    # Using `async with` on the client will start a connection on the transport
+    # and provide a `session` variable to execute queries on this connection
+    async with Client(
+        transport=transport,
+        fetch_schema_from_transport=True,
+    ) as session:
+
+        # Execute single query
+        query = gql(
+            """
+            query getContinents {
+              continents {
+                code
+                name
+              }
+            }
+        """
+        )
+
+        result = await session.execute(query)
+        print(result)
+
+
+asyncio.run(main())
diff --git a/docs/code_examples/httpx_sync.py b/docs/code_examples/httpx_sync.py
new file mode 100644
index 00000000..bd26f658
--- /dev/null
+++ b/docs/code_examples/httpx_sync.py
@@ -0,0 +1,20 @@
+from gql import Client, gql
+from gql.transport.httpx import HTTPXTransport
+
+transport = HTTPXTransport(url="https://countries.trevorblades.com/")
+
+client = Client(transport=transport, fetch_schema_from_transport=True)
+
+query = gql(
+    """
+    query getContinents {
+      continents {
+        code
+        name
+      }
+    }
+"""
+)
+
+result = client.execute(query)
+print(result)
diff --git a/docs/intro.rst b/docs/intro.rst
index bbe1cbf6..f7a4b71d 100644
--- a/docs/intro.rst
+++ b/docs/intro.rst
@@ -45,6 +45,10 @@ The corresponding between extra dependencies required and the GQL classes is:
 +---------------------+----------------------------------------------------------------+
 | requests            | :ref:`RequestsHTTPTransport <requests_transport>`              |
 +---------------------+----------------------------------------------------------------+
+| httpx               | :ref:`HTTPTXTransport <httpx_transport>`                       |
+|                     |                                                                |
+|                     | :ref:`HTTPXAsyncTransport <httpx_async_transport>`             |
++---------------------+----------------------------------------------------------------+
 | botocore            | :ref:`AppSyncIAMAuthentication <appsync_iam_auth>`             |
 +---------------------+----------------------------------------------------------------+
 
diff --git a/docs/modules/gql.rst b/docs/modules/gql.rst
index be6f904b..5f9edebe 100644
--- a/docs/modules/gql.rst
+++ b/docs/modules/gql.rst
@@ -26,6 +26,7 @@ Sub-Packages
    transport_exceptions
    transport_phoenix_channel_websockets
    transport_requests
+   transport_httpx
    transport_websockets
    transport_websockets_base
    dsl
diff --git a/docs/modules/transport_httpx.rst b/docs/modules/transport_httpx.rst
new file mode 100644
index 00000000..bf2da116
--- /dev/null
+++ b/docs/modules/transport_httpx.rst
@@ -0,0 +1,7 @@
+gql.transport.httpx
+===================
+
+.. currentmodule:: gql.transport.httpx
+
+.. automodule:: gql.transport.httpx
+    :member-order: bysource
diff --git a/docs/transports/async_transports.rst b/docs/transports/async_transports.rst
index df8c23cf..7d751df0 100644
--- a/docs/transports/async_transports.rst
+++ b/docs/transports/async_transports.rst
@@ -10,6 +10,7 @@ Async transports are transports which are using an underlying async library. The
    :maxdepth: 1
 
    aiohttp
+   httpx_async
    websockets
    phoenix
    appsync
diff --git a/docs/transports/httpx.rst b/docs/transports/httpx.rst
new file mode 100644
index 00000000..25796621
--- /dev/null
+++ b/docs/transports/httpx.rst
@@ -0,0 +1,13 @@
+.. _httpx_transport:
+
+HTTPXTransport
+==============
+
+The HTTPXTransport is a sync transport using the `httpx`_ library
+and allows you to send GraphQL queries using the HTTP protocol.
+
+Reference: :class:`gql.transport.httpx.HTTPXTransport`
+
+.. literalinclude:: ../code_examples/httpx_sync.py
+
+.. _httpx: https://www.python-httpx.org
diff --git a/docs/transports/httpx_async.rst b/docs/transports/httpx_async.rst
new file mode 100644
index 00000000..c09d0cdc
--- /dev/null
+++ b/docs/transports/httpx_async.rst
@@ -0,0 +1,39 @@
+.. _httpx_async_transport:
+
+HTTPXAsyncTransport
+===================
+
+This transport uses the `httpx`_ library and allows you to send GraphQL queries using the HTTP protocol.
+
+Reference: :class:`gql.transport.httpx.HTTPXAsyncTransport`
+
+.. note::
+
+    GraphQL subscriptions are not supported on the HTTP transport.
+    For subscriptions you should use the :ref:`websockets transport <websockets_transport>`.
+
+.. literalinclude:: ../code_examples/httpx_async.py
+
+Authentication
+--------------
+
+There are multiple ways to authenticate depending on the server configuration.
+
+1. Using HTTP Headers
+
+.. code-block:: python
+
+    transport = HTTPXAsyncTransport(
+        url='https://SERVER_URL:SERVER_PORT/graphql',
+        headers={'Authorization': 'token'}
+    )
+
+2. Using HTTP Cookies
+
+You can manually set the cookies which will be sent with each connection:
+
+.. code-block:: python
+
+    transport = HTTPXAsyncTransport(url=url, cookies={"cookie1": "val1"})
+
+.. _httpx: https://www.python-httpx.org
diff --git a/docs/transports/sync_transports.rst b/docs/transports/sync_transports.rst
index 3ed566d3..e0ec51a4 100644
--- a/docs/transports/sync_transports.rst
+++ b/docs/transports/sync_transports.rst
@@ -10,3 +10,4 @@ They cannot be used asynchronously.
    :maxdepth: 1
 
    requests
+   httpx
diff --git a/docs/usage/file_upload.rst b/docs/usage/file_upload.rst
index 8062f317..f3769d41 100644
--- a/docs/usage/file_upload.rst
+++ b/docs/usage/file_upload.rst
@@ -1,8 +1,9 @@
 File uploads
 ============
 
-GQL supports file uploads with the :ref:`aiohttp transport <aiohttp_transport>`
-and the :ref:`requests transport <requests_transport>`
+GQL supports file uploads with the :ref:`aiohttp transport <aiohttp_transport>`, the
+:ref:`requests transport <requests_transport>`, the :ref:`httpx transport <httpx_transport>`,
+and the :ref:`httpx async transport <httpx_async_transport>`,
 using the `GraphQL multipart request spec`_.
 
 .. _GraphQL multipart request spec: https://github.com/jaydenseric/graphql-multipart-request-spec
@@ -20,6 +21,8 @@ In order to upload a single file, you need to:
 
     transport = AIOHTTPTransport(url='YOUR_URL')
     # Or transport = RequestsHTTPTransport(url='YOUR_URL')
+    # Or transport = HTTPXTransport(url='YOUR_URL')
+    # Or transport = HTTPXAsyncTransport(url='YOUR_URL')
 
     client = Client(transport=transport)
 
@@ -48,6 +51,8 @@ It is also possible to upload multiple files using a list.
 
     transport = AIOHTTPTransport(url='YOUR_URL')
     # Or transport = RequestsHTTPTransport(url='YOUR_URL')
+    # Or transport = HTTPXTransport(url='YOUR_URL')
+    # Or transport = HTTPXAsyncTransport(url='YOUR_URL')
 
     client = Client(transport=transport)
 
diff --git a/gql/transport/httpx.py b/gql/transport/httpx.py
new file mode 100644
index 00000000..6e844775
--- /dev/null
+++ b/gql/transport/httpx.py
@@ -0,0 +1,306 @@
+import io
+import json
+import logging
+from typing import (
+    Any,
+    AsyncGenerator,
+    Callable,
+    Dict,
+    List,
+    Optional,
+    Tuple,
+    Type,
+    Union,
+    cast,
+)
+
+import httpx
+from graphql import DocumentNode, ExecutionResult, print_ast
+
+from ..utils import extract_files
+from . import AsyncTransport, Transport
+from .exceptions import (
+    TransportAlreadyConnected,
+    TransportClosed,
+    TransportProtocolError,
+    TransportServerError,
+)
+
+log = logging.getLogger(__name__)
+
+
+class _HTTPXTransport:
+    file_classes: Tuple[Type[Any], ...] = (io.IOBase,)
+
+    reponse_headers: Optional[httpx.Headers] = None
+
+    def __init__(
+        self,
+        url: Union[str, httpx.URL],
+        json_serialize: Callable = json.dumps,
+        **kwargs,
+    ):
+        """Initialize the transport with the given httpx parameters.
+
+        :param url: The GraphQL server URL. Example: 'https://server.com:PORT/path'.
+        :param json_serialize: Json serializer callable.
+                By default json.dumps() function.
+        :param kwargs: Extra args passed to the `httpx` client.
+        """
+        self.url = url
+        self.json_serialize = json_serialize
+        self.kwargs = kwargs
+
+    def _prepare_request(
+        self,
+        document: DocumentNode,
+        variable_values: Optional[Dict[str, Any]] = None,
+        operation_name: Optional[str] = None,
+        extra_args: Optional[Dict[str, Any]] = None,
+        upload_files: bool = False,
+    ) -> Dict[str, Any]:
+        query_str = print_ast(document)
+
+        payload: Dict[str, Any] = {
+            "query": query_str,
+        }
+
+        if operation_name:
+            payload["operationName"] = operation_name
+
+        if upload_files:
+            # If the upload_files flag is set, then we need variable_values
+            assert variable_values is not None
+
+            post_args = self._prepare_file_uploads(variable_values, payload)
+        else:
+            if variable_values:
+                payload["variables"] = variable_values
+
+            post_args = {"json": payload}
+
+        # Log the payload
+        if log.isEnabledFor(logging.DEBUG):
+            log.debug(">>> %s", self.json_serialize(payload))
+
+        # Pass post_args to httpx post method
+        if extra_args:
+            post_args.update(extra_args)
+
+        return post_args
+
+    def _prepare_file_uploads(self, variable_values, payload) -> Dict[str, Any]:
+        # If we upload files, we will extract the files present in the
+        # variable_values dict and replace them by null values
+        nulled_variable_values, files = extract_files(
+            variables=variable_values,
+            file_classes=self.file_classes,
+        )
+
+        # Save the nulled variable values in the payload
+        payload["variables"] = nulled_variable_values
+
+        # Prepare to send multipart-encoded data
+        data: Dict[str, Any] = {}
+        file_map: Dict[str, List[str]] = {}
+        file_streams: Dict[str, Tuple[str, Any]] = {}
+
+        for i, (path, val) in enumerate(files.items()):
+            key = str(i)
+
+            # Generate the file map
+            # path is nested in a list because the spec allows multiple pointers
+            # to the same file. But we don't support that.
+            # Will generate something like {"0": ["variables.file"]}
+            file_map[key] = [path]
+
+            # Generate the file streams
+            # Will generate something like
+            # {"0": ("variables.file", <_io.BufferedReader ...>)}
+            filename = cast(str, getattr(val, "name", key))
+            file_streams[key] = (filename, val)
+
+        # Add the payload to the operations field
+        operations_str = self.json_serialize(payload)
+        log.debug("operations %s", operations_str)
+        data["operations"] = operations_str
+
+        # Add the file map field
+        file_map_str = self.json_serialize(file_map)
+        log.debug("file_map %s", file_map_str)
+        data["map"] = file_map_str
+
+        return {"data": data, "files": file_streams}
+
+    def _prepare_result(self, response: httpx.Response) -> ExecutionResult:
+        # Save latest response headers in transport
+        self.response_headers = response.headers
+
+        if log.isEnabledFor(logging.DEBUG):
+            log.debug("<<< %s", response.text)
+
+        try:
+            result: Dict[str, Any] = response.json()
+
+        except Exception:
+            self._raise_response_error(response, "Not a JSON answer")
+
+        if "errors" not in result and "data" not in result:
+            self._raise_response_error(response, 'No "data" or "errors" keys in answer')
+
+        return ExecutionResult(
+            errors=result.get("errors"),
+            data=result.get("data"),
+            extensions=result.get("extensions"),
+        )
+
+    def _raise_response_error(self, response: httpx.Response, reason: str):
+        # We raise a TransportServerError if the status code is 400 or higher
+        # We raise a TransportProtocolError in the other cases
+
+        try:
+            # Raise a HTTPError if response status is 400 or higher
+            response.raise_for_status()
+        except httpx.HTTPStatusError as e:
+            raise TransportServerError(str(e), e.response.status_code) from e
+
+        raise TransportProtocolError(
+            f"Server did not return a GraphQL result: " f"{reason}: " f"{response.text}"
+        )
+
+
+class HTTPXTransport(Transport, _HTTPXTransport):
+    """:ref:`Sync Transport <sync_transports>` used to execute GraphQL queries
+    on remote servers.
+
+    The transport uses the httpx library to send HTTP POST requests.
+    """
+
+    client: Optional[httpx.Client] = None
+
+    def connect(self):
+        if self.client:
+            raise TransportAlreadyConnected("Transport is already connected")
+
+        log.debug("Connecting transport")
+
+        self.client = httpx.Client(**self.kwargs)
+
+    def execute(  # type: ignore
+        self,
+        document: DocumentNode,
+        variable_values: Optional[Dict[str, Any]] = None,
+        operation_name: Optional[str] = None,
+        extra_args: Optional[Dict[str, Any]] = None,
+        upload_files: bool = False,
+    ) -> ExecutionResult:
+        """Execute GraphQL query.
+
+        Execute the provided document AST against the configured remote server. This
+        uses the httpx library to perform a HTTP POST request to the remote server.
+
+        :param document: GraphQL query as AST Node object.
+        :param variable_values: Dictionary of input parameters (Default: None).
+        :param operation_name: Name of the operation that shall be executed.
+            Only required in multi-operation documents (Default: None).
+        :param extra_args: additional arguments to send to the httpx post method
+        :param upload_files: Set to True if you want to put files in the variable values
+        :return: The result of execution.
+            `data` is the result of executing the query, `errors` is null
+            if no errors occurred, and is a non-empty array if an error occurred.
+        """
+        if not self.client:
+            raise TransportClosed("Transport is not connected")
+
+        post_args = self._prepare_request(
+            document,
+            variable_values,
+            operation_name,
+            extra_args,
+            upload_files,
+        )
+
+        response = self.client.post(self.url, **post_args)
+
+        return self._prepare_result(response)
+
+    def close(self):
+        """Closing the transport by closing the inner session"""
+        if self.client:
+            self.client.close()
+            self.client = None
+
+
+class HTTPXAsyncTransport(AsyncTransport, _HTTPXTransport):
+    """:ref:`Async Transport <async_transports>` used to execute GraphQL queries
+    on remote servers.
+
+    The transport uses the httpx library with anyio.
+    """
+
+    client: Optional[httpx.AsyncClient] = None
+
+    async def connect(self):
+        if self.client:
+            raise TransportAlreadyConnected("Transport is already connected")
+
+        log.debug("Connecting transport")
+
+        self.client = httpx.AsyncClient(**self.kwargs)
+
+    async def execute(
+        self,
+        document: DocumentNode,
+        variable_values: Optional[Dict[str, Any]] = None,
+        operation_name: Optional[str] = None,
+        extra_args: Optional[Dict[str, Any]] = None,
+        upload_files: bool = False,
+    ) -> ExecutionResult:
+        """Execute GraphQL query.
+
+        Execute the provided document AST against the configured remote server. This
+        uses the httpx library to perform a HTTP POST request asynchronously to the
+        remote server.
+
+        :param document: GraphQL query as AST Node object.
+        :param variable_values: Dictionary of input parameters (Default: None).
+        :param operation_name: Name of the operation that shall be executed.
+            Only required in multi-operation documents (Default: None).
+        :param extra_args: additional arguments to send to the httpx post method
+        :param upload_files: Set to True if you want to put files in the variable values
+        :return: The result of execution.
+            `data` is the result of executing the query, `errors` is null
+            if no errors occurred, and is a non-empty array if an error occurred.
+        """
+        if not self.client:
+            raise TransportClosed("Transport is not connected")
+
+        post_args = self._prepare_request(
+            document,
+            variable_values,
+            operation_name,
+            extra_args,
+            upload_files,
+        )
+
+        response = await self.client.post(self.url, **post_args)
+
+        return self._prepare_result(response)
+
+    async def close(self):
+        """Closing the transport by closing the inner session"""
+        if self.client:
+            await self.client.aclose()
+            self.client = None
+
+    def subscribe(
+        self,
+        document: DocumentNode,
+        variable_values: Optional[Dict[str, Any]] = None,
+        operation_name: Optional[str] = None,
+    ) -> AsyncGenerator[ExecutionResult, None]:
+        """Subscribe is not supported on HTTP.
+
+        :meta private:
+        """
+        raise NotImplementedError("The HTTP transport does not support subscriptions")
diff --git a/setup.py b/setup.py
index 9615906f..30817ec4 100644
--- a/setup.py
+++ b/setup.py
@@ -47,6 +47,10 @@
     "urllib3>=1.26",
 ]
 
+install_httpx_requires = [
+    "httpx>=0.23.1,<1",
+]
+
 install_websockets_requires = [
     "websockets>=10,<11",
 ]
@@ -56,7 +60,7 @@
 ]
 
 install_all_requires = (
-    install_aiohttp_requires + install_requests_requires + install_websockets_requires + install_botocore_requires
+    install_aiohttp_requires + install_requests_requires + install_httpx_requires + install_websockets_requires + install_botocore_requires
 )
 
 # Get version from __version__.py file
@@ -100,6 +104,7 @@
         "dev": install_all_requires + dev_requires,
         "aiohttp": install_aiohttp_requires,
         "requests": install_requests_requires,
+        "httpx": install_httpx_requires,
         "websockets": install_websockets_requires,
         "botocore": install_botocore_requires,
     },
diff --git a/tests/conftest.py b/tests/conftest.py
index 518d0d3a..b880cff4 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -14,7 +14,7 @@
 
 from gql import Client
 
-all_transport_dependencies = ["aiohttp", "requests", "websockets", "botocore"]
+all_transport_dependencies = ["aiohttp", "requests", "httpx", "websockets", "botocore"]
 
 
 def pytest_addoption(parser):
@@ -55,6 +55,7 @@ def pytest_collection_modifyitems(config, items):
 
     # --aiohttp-only
     # --requests-only
+    # --httpx-only
     # --websockets-only
     for transport in all_transport_dependencies:
 
@@ -119,6 +120,7 @@ async def ssl_aiohttp_server():
     "gql.transport.appsync",
     "gql.transport.phoenix_channel_websockets",
     "gql.transport.requests",
+    "gql.transport.httpx",
     "gql.transport.websockets",
     "gql.dsl",
     "gql.utilities.parse_result",
diff --git a/tests/test_httpx.py b/tests/test_httpx.py
new file mode 100644
index 00000000..13f487dd
--- /dev/null
+++ b/tests/test_httpx.py
@@ -0,0 +1,850 @@
+from typing import Mapping
+
+import pytest
+
+from gql import Client, gql
+from gql.transport.exceptions import (
+    TransportAlreadyConnected,
+    TransportClosed,
+    TransportProtocolError,
+    TransportQueryError,
+    TransportServerError,
+)
+from tests.conftest import TemporaryFile
+
+# Marking all tests in this file with the httpx marker
+pytestmark = pytest.mark.httpx
+
+query1_str = """
+    query getContinents {
+      continents {
+        code
+        name
+      }
+    }
+"""
+
+query1_server_answer = (
+    '{"data":{"continents":['
+    '{"code":"AF","name":"Africa"},{"code":"AN","name":"Antarctica"},'
+    '{"code":"AS","name":"Asia"},{"code":"EU","name":"Europe"},'
+    '{"code":"NA","name":"North America"},{"code":"OC","name":"Oceania"},'
+    '{"code":"SA","name":"South America"}]}}'
+)
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_query(event_loop, aiohttp_server, run_sync_test):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXTransport
+
+    async def handler(request):
+        return web.Response(
+            text=query1_server_answer,
+            content_type="application/json",
+            headers={"dummy": "test1234"},
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    def test_code():
+        transport = HTTPXTransport(url=url)
+
+        with Client(transport=transport) as session:
+
+            query = gql(query1_str)
+
+            # Execute query synchronously
+            result = session.execute(query)
+
+            continents = result["continents"]
+
+            africa = continents[0]
+
+            assert africa["code"] == "AF"
+
+            # Checking response headers are saved in the transport
+            assert hasattr(transport, "response_headers")
+            assert isinstance(transport.response_headers, Mapping)
+            assert transport.response_headers["dummy"] == "test1234"
+
+    await run_sync_test(event_loop, server, test_code)
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_cookies(event_loop, aiohttp_server, run_sync_test):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXTransport
+
+    async def handler(request):
+        assert "COOKIE" in request.headers
+        assert "cookie1=val1" == request.headers["COOKIE"]
+
+        return web.Response(text=query1_server_answer, content_type="application/json")
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    def test_code():
+        transport = HTTPXTransport(url=url, cookies={"cookie1": "val1"})
+
+        with Client(transport=transport) as session:
+
+            query = gql(query1_str)
+
+            # Execute query synchronously
+            result = session.execute(query)
+
+            continents = result["continents"]
+
+            africa = continents[0]
+
+            assert africa["code"] == "AF"
+
+    await run_sync_test(event_loop, server, test_code)
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_error_code_401(event_loop, aiohttp_server, run_sync_test):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXTransport
+
+    async def handler(request):
+        # Will generate http error code 401
+        return web.Response(
+            text='{"error":"Unauthorized","message":"401 Client Error: Unauthorized"}',
+            content_type="application/json",
+            status=401,
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    def test_code():
+        transport = HTTPXTransport(url=url)
+
+        with Client(transport=transport) as session:
+
+            query = gql(query1_str)
+
+            with pytest.raises(TransportServerError) as exc_info:
+                session.execute(query)
+
+            assert "Client error '401 Unauthorized'" in str(exc_info.value)
+
+    await run_sync_test(event_loop, server, test_code)
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_error_code_429(event_loop, aiohttp_server, run_sync_test):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXTransport
+
+    async def handler(request):
+        # Will generate http error code 429
+        return web.Response(
+            text="""
+<html>
+  <head>
+     <title>Too Many Requests</title>
+  </head>
+  <body>
+     <h1>Too Many Requests</h1>
+     <p>I only allow 50 requests per hour to this Web site per
+        logged in user.  Try again soon.</p>
+  </body>
+</html>""",
+            content_type="text/html",
+            status=429,
+            headers={"Retry-After": "3600"},
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    def test_code():
+        transport = HTTPXTransport(url=url)
+
+        with Client(transport=transport) as session:
+
+            query = gql(query1_str)
+
+            with pytest.raises(TransportServerError) as exc_info:
+                session.execute(query)
+
+            assert "429, message='Too Many Requests'" in str(exc_info.value)
+
+        # Checking response headers are saved in the transport
+        assert hasattr(transport, "response_headers")
+        assert isinstance(transport.response_headers, Mapping)
+        assert transport.response_headers["Retry-After"] == "3600"
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_error_code_500(event_loop, aiohttp_server, run_sync_test):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXTransport
+
+    async def handler(request):
+        # Will generate http error code 500
+        raise Exception("Server error")
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    def test_code():
+        transport = HTTPXTransport(url=url)
+
+        with Client(transport=transport) as session:
+
+            query = gql(query1_str)
+
+            with pytest.raises(TransportServerError):
+                session.execute(query)
+
+    await run_sync_test(event_loop, server, test_code)
+
+
+query1_server_error_answer = '{"errors": ["Error 1", "Error 2"]}'
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_error_code(event_loop, aiohttp_server, run_sync_test):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXTransport
+
+    async def handler(request):
+        return web.Response(
+            text=query1_server_error_answer, content_type="application/json"
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    def test_code():
+        transport = HTTPXTransport(url=url)
+
+        with Client(transport=transport) as session:
+
+            query = gql(query1_str)
+
+            with pytest.raises(TransportQueryError):
+                session.execute(query)
+
+    await run_sync_test(event_loop, server, test_code)
+
+
+invalid_protocol_responses = [
+    "{}",
+    "qlsjfqsdlkj",
+    '{"not_data_or_errors": 35}',
+]
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+@pytest.mark.parametrize("response", invalid_protocol_responses)
+async def test_httpx_invalid_protocol(
+    event_loop, aiohttp_server, response, run_sync_test
+):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXTransport
+
+    async def handler(request):
+        return web.Response(text=response, content_type="application/json")
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    def test_code():
+        transport = HTTPXTransport(url=url)
+
+        with Client(transport=transport) as session:
+
+            query = gql(query1_str)
+
+            with pytest.raises(TransportProtocolError):
+                session.execute(query)
+
+    await run_sync_test(event_loop, server, test_code)
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_cannot_connect_twice(event_loop, aiohttp_server, run_sync_test):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXTransport
+
+    async def handler(request):
+        return web.Response(text=query1_server_answer, content_type="application/json")
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    def test_code():
+        transport = HTTPXTransport(url=url)
+
+        with Client(transport=transport) as session:
+
+            with pytest.raises(TransportAlreadyConnected):
+                session.transport.connect()
+
+    await run_sync_test(event_loop, server, test_code)
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_cannot_execute_if_not_connected(
+    event_loop, aiohttp_server, run_sync_test
+):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXTransport
+
+    async def handler(request):
+        return web.Response(text=query1_server_answer, content_type="application/json")
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    def test_code():
+        transport = HTTPXTransport(url=url)
+
+        query = gql(query1_str)
+
+        with pytest.raises(TransportClosed):
+            transport.execute(query)
+
+    await run_sync_test(event_loop, server, test_code)
+
+
+query1_server_answer_with_extensions = (
+    '{"data":{"continents":['
+    '{"code":"AF","name":"Africa"},{"code":"AN","name":"Antarctica"},'
+    '{"code":"AS","name":"Asia"},{"code":"EU","name":"Europe"},'
+    '{"code":"NA","name":"North America"},{"code":"OC","name":"Oceania"},'
+    '{"code":"SA","name":"South America"}]},'
+    '"extensions": {"key1": "val1"}'
+    "}"
+)
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_query_with_extensions(event_loop, aiohttp_server, run_sync_test):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXTransport
+
+    async def handler(request):
+        return web.Response(
+            text=query1_server_answer_with_extensions, content_type="application/json"
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    def test_code():
+        transport = HTTPXTransport(url=url)
+
+        with Client(transport=transport) as session:
+
+            query = gql(query1_str)
+
+            execution_result = session.execute(query, get_execution_result=True)
+
+            assert execution_result.extensions["key1"] == "val1"
+
+    await run_sync_test(event_loop, server, test_code)
+
+
+file_upload_server_answer = '{"data":{"success":true}}'
+
+file_upload_mutation_1 = """
+    mutation($file: Upload!) {
+      uploadFile(input:{ other_var:$other_var, file:$file }) {
+        success
+      }
+    }
+"""
+
+file_upload_mutation_1_operations = (
+    '{"query": "mutation ($file: Upload!) {\\n  uploadFile(input: { other_var: '
+    '$other_var, file: $file }) {\\n    success\\n  }\\n}", "variables": '
+    '{"file": null, "other_var": 42}}'
+)
+
+file_upload_mutation_1_map = '{"0": ["variables.file"]}'
+
+file_1_content = """
+This is a test file
+This file will be sent in the GraphQL mutation
+"""
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_file_upload(event_loop, aiohttp_server, run_sync_test):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXTransport
+
+    async def single_upload_handler(request):
+        from aiohttp import web
+
+        reader = await request.multipart()
+
+        field_0 = await reader.next()
+        assert field_0.name == "operations"
+        field_0_text = await field_0.text()
+        assert field_0_text == file_upload_mutation_1_operations
+
+        field_1 = await reader.next()
+        assert field_1.name == "map"
+        field_1_text = await field_1.text()
+        assert field_1_text == file_upload_mutation_1_map
+
+        field_2 = await reader.next()
+        assert field_2.name == "0"
+        field_2_text = await field_2.text()
+        assert field_2_text == file_1_content
+
+        field_3 = await reader.next()
+        assert field_3 is None
+
+        return web.Response(
+            text=file_upload_server_answer, content_type="application/json"
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", single_upload_handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    def test_code():
+        transport = HTTPXTransport(url=url)
+
+        with TemporaryFile(file_1_content) as test_file:
+            with Client(transport=transport) as session:
+                query = gql(file_upload_mutation_1)
+
+                file_path = test_file.filename
+
+                with open(file_path, "rb") as f:
+
+                    params = {"file": f, "other_var": 42}
+                    execution_result = session._execute(
+                        query, variable_values=params, upload_files=True
+                    )
+
+                    assert execution_result.data["success"]
+
+    await run_sync_test(event_loop, server, test_code)
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_file_upload_additional_headers(
+    event_loop, aiohttp_server, run_sync_test
+):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXTransport
+
+    async def single_upload_handler(request):
+        from aiohttp import web
+
+        assert request.headers["X-Auth"] == "foobar"
+
+        reader = await request.multipart()
+
+        field_0 = await reader.next()
+        assert field_0.name == "operations"
+        field_0_text = await field_0.text()
+        assert field_0_text == file_upload_mutation_1_operations
+
+        field_1 = await reader.next()
+        assert field_1.name == "map"
+        field_1_text = await field_1.text()
+        assert field_1_text == file_upload_mutation_1_map
+
+        field_2 = await reader.next()
+        assert field_2.name == "0"
+        field_2_text = await field_2.text()
+        assert field_2_text == file_1_content
+
+        field_3 = await reader.next()
+        assert field_3 is None
+
+        return web.Response(
+            text=file_upload_server_answer, content_type="application/json"
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", single_upload_handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    def test_code():
+        transport = HTTPXTransport(url=url, headers={"X-Auth": "foobar"})
+
+        with TemporaryFile(file_1_content) as test_file:
+            with Client(transport=transport) as session:
+                query = gql(file_upload_mutation_1)
+
+                file_path = test_file.filename
+
+                with open(file_path, "rb") as f:
+
+                    params = {"file": f, "other_var": 42}
+                    execution_result = session._execute(
+                        query, variable_values=params, upload_files=True
+                    )
+
+                    assert execution_result.data["success"]
+
+    await run_sync_test(event_loop, server, test_code)
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_binary_file_upload(event_loop, aiohttp_server, run_sync_test):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXTransport
+
+    # This is a sample binary file content containing all possible byte values
+    binary_file_content = bytes(range(0, 256))
+
+    async def binary_upload_handler(request):
+
+        from aiohttp import web
+
+        reader = await request.multipart()
+
+        field_0 = await reader.next()
+        assert field_0.name == "operations"
+        field_0_text = await field_0.text()
+        assert field_0_text == file_upload_mutation_1_operations
+
+        field_1 = await reader.next()
+        assert field_1.name == "map"
+        field_1_text = await field_1.text()
+        assert field_1_text == file_upload_mutation_1_map
+
+        field_2 = await reader.next()
+        assert field_2.name == "0"
+        field_2_binary = await field_2.read()
+        assert field_2_binary == binary_file_content
+
+        field_3 = await reader.next()
+        assert field_3 is None
+
+        return web.Response(
+            text=file_upload_server_answer, content_type="application/json"
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", binary_upload_handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    transport = HTTPXTransport(url=url)
+
+    def test_code():
+        with TemporaryFile(binary_file_content) as test_file:
+            with Client(transport=transport) as session:
+
+                query = gql(file_upload_mutation_1)
+
+                file_path = test_file.filename
+
+                with open(file_path, "rb") as f:
+
+                    params = {"file": f, "other_var": 42}
+
+                    execution_result = session._execute(
+                        query, variable_values=params, upload_files=True
+                    )
+
+                    assert execution_result.data["success"]
+
+    await run_sync_test(event_loop, server, test_code)
+
+
+file_upload_mutation_2_operations = (
+    '{"query": "mutation ($file1: Upload!, $file2: Upload!) {\\n  '
+    'uploadFile(input: { file1: $file, file2: $file }) {\\n    success\\n  }\\n}", '
+    '"variables": {"file1": null, "file2": null}}'
+)
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_file_upload_two_files(event_loop, aiohttp_server, run_sync_test):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXTransport
+
+    file_upload_mutation_2 = """
+    mutation($file1: Upload!, $file2: Upload!) {
+      uploadFile(input:{file1:$file, file2:$file}) {
+        success
+      }
+    }
+    """
+
+    file_upload_mutation_2_map = '{"0": ["variables.file1"], "1": ["variables.file2"]}'
+
+    file_2_content = """
+    This is a second test file
+    This file will also be sent in the GraphQL mutation
+    """
+
+    async def handler(request):
+
+        reader = await request.multipart()
+
+        field_0 = await reader.next()
+        assert field_0.name == "operations"
+        field_0_text = await field_0.text()
+        assert field_0_text == file_upload_mutation_2_operations
+
+        field_1 = await reader.next()
+        assert field_1.name == "map"
+        field_1_text = await field_1.text()
+        assert field_1_text == file_upload_mutation_2_map
+
+        field_2 = await reader.next()
+        assert field_2.name == "0"
+        field_2_text = await field_2.text()
+        assert field_2_text == file_1_content
+
+        field_3 = await reader.next()
+        assert field_3.name == "1"
+        field_3_text = await field_3.text()
+        assert field_3_text == file_2_content
+
+        field_4 = await reader.next()
+        assert field_4 is None
+
+        return web.Response(
+            text=file_upload_server_answer, content_type="application/json"
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    def test_code():
+        transport = HTTPXTransport(url=url)
+
+        with TemporaryFile(file_1_content) as test_file_1:
+            with TemporaryFile(file_2_content) as test_file_2:
+
+                with Client(transport=transport) as session:
+
+                    query = gql(file_upload_mutation_2)
+
+                    file_path_1 = test_file_1.filename
+                    file_path_2 = test_file_2.filename
+
+                    f1 = open(file_path_1, "rb")
+                    f2 = open(file_path_2, "rb")
+
+                    params = {
+                        "file1": f1,
+                        "file2": f2,
+                    }
+
+                    execution_result = session._execute(
+                        query, variable_values=params, upload_files=True
+                    )
+
+                    assert execution_result.data["success"]
+
+                    f1.close()
+                    f2.close()
+
+    await run_sync_test(event_loop, server, test_code)
+
+
+file_upload_mutation_3_operations = (
+    '{"query": "mutation ($files: [Upload!]!) {\\n  uploadFiles'
+    "(input: { files: $files })"
+    ' {\\n    success\\n  }\\n}", "variables": {"files": [null, null]}}'
+)
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_file_upload_list_of_two_files(
+    event_loop, aiohttp_server, run_sync_test
+):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXTransport
+
+    file_upload_mutation_3 = """
+    mutation($files: [Upload!]!) {
+      uploadFiles(input:{files:$files}) {
+        success
+      }
+    }
+    """
+
+    file_upload_mutation_3_map = (
+        '{"0": ["variables.files.0"], "1": ["variables.files.1"]}'
+    )
+
+    file_2_content = """
+    This is a second test file
+    This file will also be sent in the GraphQL mutation
+    """
+
+    async def handler(request):
+
+        reader = await request.multipart()
+
+        field_0 = await reader.next()
+        assert field_0.name == "operations"
+        field_0_text = await field_0.text()
+        assert field_0_text == file_upload_mutation_3_operations
+
+        field_1 = await reader.next()
+        assert field_1.name == "map"
+        field_1_text = await field_1.text()
+        assert field_1_text == file_upload_mutation_3_map
+
+        field_2 = await reader.next()
+        assert field_2.name == "0"
+        field_2_text = await field_2.text()
+        assert field_2_text == file_1_content
+
+        field_3 = await reader.next()
+        assert field_3.name == "1"
+        field_3_text = await field_3.text()
+        assert field_3_text == file_2_content
+
+        field_4 = await reader.next()
+        assert field_4 is None
+
+        return web.Response(
+            text=file_upload_server_answer, content_type="application/json"
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    def test_code():
+        transport = HTTPXTransport(url=url)
+
+        with TemporaryFile(file_1_content) as test_file_1:
+            with TemporaryFile(file_2_content) as test_file_2:
+                with Client(transport=transport) as session:
+
+                    query = gql(file_upload_mutation_3)
+
+                    file_path_1 = test_file_1.filename
+                    file_path_2 = test_file_2.filename
+
+                    f1 = open(file_path_1, "rb")
+                    f2 = open(file_path_2, "rb")
+
+                    params = {"files": [f1, f2]}
+
+                    execution_result = session._execute(
+                        query, variable_values=params, upload_files=True
+                    )
+
+                    assert execution_result.data["success"]
+
+                    f1.close()
+                    f2.close()
+
+    await run_sync_test(event_loop, server, test_code)
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_error_fetching_schema(event_loop, aiohttp_server, run_sync_test):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXTransport
+
+    error_answer = """
+{
+    "errors": [
+        {
+            "errorType": "UnauthorizedException",
+            "message": "Permission denied"
+        }
+    ]
+}
+"""
+
+    async def handler(request):
+        return web.Response(
+            text=error_answer,
+            content_type="application/json",
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    def test_code():
+        transport = HTTPXTransport(url=url)
+
+        with pytest.raises(TransportQueryError) as exc_info:
+            with Client(transport=transport, fetch_schema_from_transport=True):
+                pass
+
+        expected_error = (
+            "Error while fetching schema: "
+            "{'errorType': 'UnauthorizedException', 'message': 'Permission denied'}"
+        )
+
+        assert expected_error in str(exc_info.value)
+        assert transport.client is None
+
+    await run_sync_test(event_loop, server, test_code)
diff --git a/tests/test_httpx_async.py b/tests/test_httpx_async.py
new file mode 100644
index 00000000..362875de
--- /dev/null
+++ b/tests/test_httpx_async.py
@@ -0,0 +1,1391 @@
+import io
+import json
+from typing import Mapping
+
+import pytest
+
+from gql import Client, gql
+from gql.cli import get_parser, main
+from gql.transport.exceptions import (
+    TransportAlreadyConnected,
+    TransportClosed,
+    TransportProtocolError,
+    TransportQueryError,
+    TransportServerError,
+)
+
+from .conftest import TemporaryFile, get_localhost_ssl_context
+
+query1_str = """
+    query getContinents {
+      continents {
+        code
+        name
+      }
+    }
+"""
+
+query1_server_answer_data = (
+    '{"continents":['
+    '{"code":"AF","name":"Africa"},{"code":"AN","name":"Antarctica"},'
+    '{"code":"AS","name":"Asia"},{"code":"EU","name":"Europe"},'
+    '{"code":"NA","name":"North America"},{"code":"OC","name":"Oceania"},'
+    '{"code":"SA","name":"South America"}]}'
+)
+
+
+query1_server_answer = f'{{"data":{query1_server_answer_data}}}'
+
+# Marking all tests in this file with the httpx marker
+pytestmark = pytest.mark.httpx
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_query(event_loop, aiohttp_server):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    async def handler(request):
+        return web.Response(
+            text=query1_server_answer,
+            content_type="application/json",
+            headers={"dummy": "test1234"},
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    transport = HTTPXAsyncTransport(url=url, timeout=10)
+
+    async with Client(transport=transport) as session:
+
+        query = gql(query1_str)
+
+        # Execute query asynchronously
+        result = await session.execute(query)
+
+        continents = result["continents"]
+
+        africa = continents[0]
+
+        assert africa["code"] == "AF"
+
+        # Checking response headers are saved in the transport
+        assert hasattr(transport, "response_headers")
+        assert isinstance(transport.response_headers, Mapping)
+        assert transport.response_headers["dummy"] == "test1234"
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_ignore_backend_content_type(event_loop, aiohttp_server):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    async def handler(request):
+        return web.Response(text=query1_server_answer, content_type="text/plain")
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    transport = HTTPXAsyncTransport(url=url, timeout=10)
+
+    async with Client(transport=transport) as session:
+
+        query = gql(query1_str)
+
+        result = await session.execute(query)
+
+        continents = result["continents"]
+
+        africa = continents[0]
+
+        assert africa["code"] == "AF"
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_cookies(event_loop, aiohttp_server):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    async def handler(request):
+        assert "COOKIE" in request.headers
+        assert "cookie1=val1" == request.headers["COOKIE"]
+
+        return web.Response(text=query1_server_answer, content_type="application/json")
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    transport = HTTPXAsyncTransport(url=url, cookies={"cookie1": "val1"})
+
+    async with Client(transport=transport) as session:
+
+        query = gql(query1_str)
+
+        # Execute query asynchronously
+        result = await session.execute(query)
+
+        continents = result["continents"]
+
+        africa = continents[0]
+
+        assert africa["code"] == "AF"
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_error_code_401(event_loop, aiohttp_server):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    async def handler(request):
+        # Will generate http error code 401
+        return web.Response(
+            text='{"error":"Unauthorized","message":"401 Client Error: Unauthorized"}',
+            content_type="application/json",
+            status=401,
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    transport = HTTPXAsyncTransport(url=url)
+
+    async with Client(transport=transport) as session:
+
+        query = gql(query1_str)
+
+        with pytest.raises(TransportServerError) as exc_info:
+            await session.execute(query)
+
+        assert "Client error '401 Unauthorized'" in str(exc_info.value)
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_error_code_429(event_loop, aiohttp_server):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    async def handler(request):
+        # Will generate http error code 429
+        return web.Response(
+            text="""
+<html>
+  <head>
+     <title>Too Many Requests</title>
+  </head>
+  <body>
+     <h1>Too Many Requests</h1>
+     <p>I only allow 50 requests per hour to this Web site per
+        logged in user.  Try again soon.</p>
+  </body>
+</html>""",
+            content_type="text/html",
+            status=429,
+            headers={"Retry-After": "3600"},
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    transport = HTTPXAsyncTransport(url=url)
+
+    async with Client(transport=transport) as session:
+
+        query = gql(query1_str)
+
+        with pytest.raises(TransportServerError) as exc_info:
+            await session.execute(query)
+
+        assert "Client error '429 Too Many Requests'" in str(exc_info.value)
+
+        # Checking response headers are saved in the transport
+        assert hasattr(transport, "response_headers")
+        assert isinstance(transport.response_headers, Mapping)
+        assert transport.response_headers["Retry-After"] == "3600"
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_error_code_500(event_loop, aiohttp_server):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    async def handler(request):
+        # Will generate http error code 500
+        raise Exception("Server error")
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    transport = HTTPXAsyncTransport(url=url)
+
+    async with Client(transport=transport) as session:
+
+        query = gql(query1_str)
+
+        with pytest.raises(TransportServerError) as exc_info:
+            await session.execute(query)
+
+        assert "Server error '500 Internal Server Error'" in str(exc_info.value)
+
+
+transport_query_error_responses = [
+    '{"errors": ["Error 1", "Error 2"]}',
+    '{"errors": {"error_1": "Something"}}',
+    '{"errors": 5}',
+]
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+@pytest.mark.parametrize("query_error", transport_query_error_responses)
+async def test_httpx_error_code(event_loop, aiohttp_server, query_error):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    async def handler(request):
+        return web.Response(text=query_error, content_type="application/json")
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    transport = HTTPXAsyncTransport(url=url)
+
+    async with Client(transport=transport) as session:
+
+        query = gql(query1_str)
+
+        with pytest.raises(TransportQueryError):
+            await session.execute(query)
+
+
+invalid_protocol_responses = [
+    {
+        "response": "{}",
+        "expected_exception": (
+            "Server did not return a GraphQL result: "
+            'No "data" or "errors" keys in answer: {}'
+        ),
+    },
+    {
+        "response": "qlsjfqsdlkj",
+        "expected_exception": (
+            "Server did not return a GraphQL result: Not a JSON answer: qlsjfqsdlkj"
+        ),
+    },
+    {
+        "response": '{"not_data_or_errors": 35}',
+        "expected_exception": (
+            "Server did not return a GraphQL result: "
+            'No "data" or "errors" keys in answer: {"not_data_or_errors": 35}'
+        ),
+    },
+    {
+        "response": "",
+        "expected_exception": (
+            "Server did not return a GraphQL result: Not a JSON answer: "
+        ),
+    },
+]
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+@pytest.mark.parametrize("param", invalid_protocol_responses)
+async def test_httpx_invalid_protocol(event_loop, aiohttp_server, param):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    response = param["response"]
+
+    async def handler(request):
+        return web.Response(text=response, content_type="application/json")
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    transport = HTTPXAsyncTransport(url=url)
+
+    async with Client(transport=transport) as session:
+
+        query = gql(query1_str)
+
+        with pytest.raises(TransportProtocolError) as exc_info:
+            await session.execute(query)
+
+        assert param["expected_exception"] in str(exc_info.value)
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_subscribe_not_supported(event_loop, aiohttp_server):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    async def handler(request):
+        return web.Response(text="does not matter", content_type="application/json")
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    transport = HTTPXAsyncTransport(url=url)
+
+    async with Client(transport=transport) as session:
+
+        query = gql(query1_str)
+
+        with pytest.raises(NotImplementedError):
+            async for result in session.subscribe(query):
+                pass
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_cannot_connect_twice(event_loop, aiohttp_server):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    async def handler(request):
+        return web.Response(text=query1_server_answer, content_type="application/json")
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    transport = HTTPXAsyncTransport(url=url, timeout=10)
+
+    async with Client(transport=transport) as session:
+
+        with pytest.raises(TransportAlreadyConnected):
+            await session.transport.connect()
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_cannot_execute_if_not_connected(event_loop, aiohttp_server):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    async def handler(request):
+        return web.Response(text=query1_server_answer, content_type="application/json")
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    transport = HTTPXAsyncTransport(url=url, timeout=10)
+
+    query = gql(query1_str)
+
+    with pytest.raises(TransportClosed):
+        await transport.execute(query)
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_extra_args(event_loop, aiohttp_server):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+    import httpx
+
+    async def handler(request):
+        return web.Response(text=query1_server_answer, content_type="application/json")
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    # passing extra arguments to httpx.AsyncClient
+    transport = httpx.AsyncHTTPTransport(retries=2)
+    transport = HTTPXAsyncTransport(url=url, max_redirects=2, transport=transport)
+
+    async with Client(transport=transport) as session:
+
+        query = gql(query1_str)
+
+        # Passing extra arguments to the post method of aiohttp
+        result = await session.execute(query, extra_args={"follow_redirects": True})
+
+        continents = result["continents"]
+
+        africa = continents[0]
+
+        assert africa["code"] == "AF"
+
+
+query2_str = """
+    query getEurope ($code: ID!) {
+      continent (code: $code) {
+        name
+      }
+    }
+"""
+
+query2_server_answer = '{"data": {"continent": {"name": "Europe"}}}'
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_query_variable_values(event_loop, aiohttp_server):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    async def handler(request):
+        return web.Response(text=query2_server_answer, content_type="application/json")
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    transport = HTTPXAsyncTransport(url=url, timeout=10)
+
+    async with Client(transport=transport) as session:
+
+        params = {"code": "EU"}
+
+        query = gql(query2_str)
+
+        # Execute query asynchronously
+        result = await session.execute(
+            query, variable_values=params, operation_name="getEurope"
+        )
+
+        continent = result["continent"]
+
+        assert continent["name"] == "Europe"
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_query_variable_values_fix_issue_292(event_loop, aiohttp_server):
+    """Allow to specify variable_values without keyword.
+
+    See https://github.com/graphql-python/gql/issues/292"""
+
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    async def handler(request):
+        return web.Response(text=query2_server_answer, content_type="application/json")
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    transport = HTTPXAsyncTransport(url=url, timeout=10)
+
+    async with Client(transport=transport) as session:
+
+        params = {"code": "EU"}
+
+        query = gql(query2_str)
+
+        # Execute query asynchronously
+        result = await session.execute(query, params, operation_name="getEurope")
+
+        continent = result["continent"]
+
+        assert continent["name"] == "Europe"
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_execute_running_in_thread(
+    event_loop, aiohttp_server, run_sync_test
+):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    async def handler(request):
+        return web.Response(text=query1_server_answer, content_type="application/json")
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    def test_code():
+        transport = HTTPXAsyncTransport(url=url)
+
+        client = Client(transport=transport)
+
+        query = gql(query1_str)
+
+        client.execute(query)
+
+    await run_sync_test(event_loop, server, test_code)
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_subscribe_running_in_thread(
+    event_loop, aiohttp_server, run_sync_test
+):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    async def handler(request):
+        return web.Response(text=query1_server_answer, content_type="application/json")
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    def test_code():
+        transport = HTTPXAsyncTransport(url=url)
+
+        client = Client(transport=transport)
+
+        query = gql(query1_str)
+
+        # Note: subscriptions are not supported on the httpx transport
+        # But we add this test in order to have 100% code coverage
+        # It is to check that we will correctly set an event loop
+        # in the subscribe function if there is none (in a Thread for example)
+        # We cannot test this with the websockets transport because
+        # the websockets transport will set an event loop in its init
+
+        with pytest.raises(NotImplementedError):
+            for result in client.subscribe(query):
+                pass
+
+    await run_sync_test(event_loop, server, test_code)
+
+
+file_upload_server_answer = '{"data":{"success":true}}'
+
+file_upload_mutation_1 = """
+    mutation($file: Upload!) {
+      uploadFile(input:{ other_var:$other_var, file:$file }) {
+        success
+      }
+    }
+"""
+
+file_upload_mutation_1_operations = (
+    '{"query": "mutation ($file: Upload!) {\\n  uploadFile(input: { other_var: '
+    '$other_var, file: $file }) {\\n    success\\n  }\\n}", "variables": '
+    '{"file": null, "other_var": 42}}'
+)
+
+file_upload_mutation_1_map = '{"0": ["variables.file"]}'
+
+file_1_content = """
+This is a test file
+This file will be sent in the GraphQL mutation
+"""
+
+
+async def single_upload_handler(request):
+
+    from aiohttp import web
+
+    reader = await request.multipart()
+
+    field_0 = await reader.next()
+    assert field_0.name == "operations"
+    field_0_text = await field_0.text()
+    assert field_0_text == file_upload_mutation_1_operations
+
+    field_1 = await reader.next()
+    assert field_1.name == "map"
+    field_1_text = await field_1.text()
+    assert field_1_text == file_upload_mutation_1_map
+
+    field_2 = await reader.next()
+    assert field_2.name == "0"
+    field_2_text = await field_2.text()
+    assert field_2_text == file_1_content
+
+    field_3 = await reader.next()
+    assert field_3 is None
+
+    return web.Response(text=file_upload_server_answer, content_type="application/json")
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_file_upload(event_loop, aiohttp_server):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    app = web.Application()
+    app.router.add_route("POST", "/", single_upload_handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    transport = HTTPXAsyncTransport(url=url, timeout=10)
+
+    with TemporaryFile(file_1_content) as test_file:
+
+        async with Client(transport=transport) as session:
+
+            query = gql(file_upload_mutation_1)
+
+            file_path = test_file.filename
+
+            with open(file_path, "rb") as f:
+
+                params = {"file": f, "other_var": 42}
+
+                # Execute query asynchronously
+                result = await session.execute(
+                    query, variable_values=params, upload_files=True
+                )
+
+            success = result["success"]
+
+            assert success
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_file_upload_without_session(
+    event_loop, aiohttp_server, run_sync_test
+):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    app = web.Application()
+    app.router.add_route("POST", "/", single_upload_handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    def test_code():
+        transport = HTTPXAsyncTransport(url=url, timeout=10)
+
+        with TemporaryFile(file_1_content) as test_file:
+
+            client = Client(transport=transport)
+
+            query = gql(file_upload_mutation_1)
+
+            file_path = test_file.filename
+
+            with open(file_path, "rb") as f:
+
+                params = {"file": f, "other_var": 42}
+
+                result = client.execute(
+                    query, variable_values=params, upload_files=True
+                )
+
+                success = result["success"]
+
+                assert success
+
+    await run_sync_test(event_loop, server, test_code)
+
+
+# This is a sample binary file content containing all possible byte values
+binary_file_content = bytes(range(0, 256))
+
+
+async def binary_upload_handler(request):
+
+    from aiohttp import web
+
+    reader = await request.multipart()
+
+    field_0 = await reader.next()
+    assert field_0.name == "operations"
+    field_0_text = await field_0.text()
+    assert field_0_text == file_upload_mutation_1_operations
+
+    field_1 = await reader.next()
+    assert field_1.name == "map"
+    field_1_text = await field_1.text()
+    assert field_1_text == file_upload_mutation_1_map
+
+    field_2 = await reader.next()
+    assert field_2.name == "0"
+    field_2_binary = await field_2.read()
+    assert field_2_binary == binary_file_content
+
+    field_3 = await reader.next()
+    assert field_3 is None
+
+    return web.Response(text=file_upload_server_answer, content_type="application/json")
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_binary_file_upload(event_loop, aiohttp_server):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    app = web.Application()
+    app.router.add_route("POST", "/", binary_upload_handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    transport = HTTPXAsyncTransport(url=url, timeout=10)
+
+    with TemporaryFile(binary_file_content) as test_file:
+
+        async with Client(transport=transport) as session:
+
+            query = gql(file_upload_mutation_1)
+
+            file_path = test_file.filename
+
+            with open(file_path, "rb") as f:
+
+                params = {"file": f, "other_var": 42}
+
+                # Execute query asynchronously
+                result = await session.execute(
+                    query, variable_values=params, upload_files=True
+                )
+
+            success = result["success"]
+
+            assert success
+
+
+file_upload_mutation_2 = """
+    mutation($file1: Upload!, $file2: Upload!) {
+      uploadFile(input:{file1:$file, file2:$file}) {
+        success
+      }
+    }
+"""
+
+file_upload_mutation_2_operations = (
+    '{"query": "mutation ($file1: Upload!, $file2: Upload!) {\\n  '
+    'uploadFile(input: { file1: $file, file2: $file }) {\\n    success\\n  }\\n}", '
+    '"variables": {"file1": null, "file2": null}}'
+)
+
+file_upload_mutation_2_map = '{"0": ["variables.file1"], "1": ["variables.file2"]}'
+
+file_2_content = """
+This is a second test file
+This file will also be sent in the GraphQL mutation
+"""
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_file_upload_two_files(event_loop, aiohttp_server):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    async def handler(request):
+
+        reader = await request.multipart()
+
+        field_0 = await reader.next()
+        assert field_0.name == "operations"
+        field_0_text = await field_0.text()
+        assert field_0_text == file_upload_mutation_2_operations
+
+        field_1 = await reader.next()
+        assert field_1.name == "map"
+        field_1_text = await field_1.text()
+        assert field_1_text == file_upload_mutation_2_map
+
+        field_2 = await reader.next()
+        assert field_2.name == "0"
+        field_2_text = await field_2.text()
+        assert field_2_text == file_1_content
+
+        field_3 = await reader.next()
+        assert field_3.name == "1"
+        field_3_text = await field_3.text()
+        assert field_3_text == file_2_content
+
+        field_4 = await reader.next()
+        assert field_4 is None
+
+        return web.Response(
+            text=file_upload_server_answer, content_type="application/json"
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    transport = HTTPXAsyncTransport(url=url, timeout=10)
+
+    with TemporaryFile(file_1_content) as test_file_1:
+        with TemporaryFile(file_2_content) as test_file_2:
+
+            async with Client(transport=transport) as session:
+
+                query = gql(file_upload_mutation_2)
+
+                file_path_1 = test_file_1.filename
+                file_path_2 = test_file_2.filename
+
+                f1 = open(file_path_1, "rb")
+                f2 = open(file_path_2, "rb")
+
+                params = {
+                    "file1": f1,
+                    "file2": f2,
+                }
+
+                result = await session.execute(
+                    query, variable_values=params, upload_files=True
+                )
+
+                f1.close()
+                f2.close()
+
+                success = result["success"]
+
+                assert success
+
+
+file_upload_mutation_3 = """
+    mutation($files: [Upload!]!) {
+      uploadFiles(input:{ files:$files }) {
+        success
+      }
+    }
+"""
+
+file_upload_mutation_3_operations = (
+    '{"query": "mutation ($files: [Upload!]!) {\\n  uploadFiles('
+    "input: { files: $files })"
+    ' {\\n    success\\n  }\\n}", "variables": {"files": [null, null]}}'
+)
+
+file_upload_mutation_3_map = '{"0": ["variables.files.0"], "1": ["variables.files.1"]}'
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_file_upload_list_of_two_files(event_loop, aiohttp_server):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    async def handler(request):
+
+        reader = await request.multipart()
+
+        field_0 = await reader.next()
+        assert field_0.name == "operations"
+        field_0_text = await field_0.text()
+        assert field_0_text == file_upload_mutation_3_operations
+
+        field_1 = await reader.next()
+        assert field_1.name == "map"
+        field_1_text = await field_1.text()
+        assert field_1_text == file_upload_mutation_3_map
+
+        field_2 = await reader.next()
+        assert field_2.name == "0"
+        field_2_text = await field_2.text()
+        assert field_2_text == file_1_content
+
+        field_3 = await reader.next()
+        assert field_3.name == "1"
+        field_3_text = await field_3.text()
+        assert field_3_text == file_2_content
+
+        field_4 = await reader.next()
+        assert field_4 is None
+
+        return web.Response(
+            text=file_upload_server_answer, content_type="application/json"
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    transport = HTTPXAsyncTransport(url=url, timeout=10)
+
+    with TemporaryFile(file_1_content) as test_file_1:
+        with TemporaryFile(file_2_content) as test_file_2:
+
+            async with Client(transport=transport) as session:
+
+                query = gql(file_upload_mutation_3)
+
+                file_path_1 = test_file_1.filename
+                file_path_2 = test_file_2.filename
+
+                f1 = open(file_path_1, "rb")
+                f2 = open(file_path_2, "rb")
+
+                params = {"files": [f1, f2]}
+
+                # Execute query asynchronously
+                result = await session.execute(
+                    query, variable_values=params, upload_files=True
+                )
+
+                f1.close()
+                f2.close()
+
+                success = result["success"]
+
+                assert success
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_using_cli(event_loop, aiohttp_server, monkeypatch, capsys):
+    from aiohttp import web
+
+    async def handler(request):
+        return web.Response(text=query1_server_answer, content_type="application/json")
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    parser = get_parser(with_examples=True)
+    args = parser.parse_args([url, "--verbose"])
+
+    # Monkeypatching sys.stdin to simulate getting the query
+    # via the standard input
+    monkeypatch.setattr("sys.stdin", io.StringIO(query1_str))
+
+    exit_code = await main(args)
+
+    assert exit_code == 0
+
+    # Check that the result has been printed on stdout
+    captured = capsys.readouterr()
+    captured_out = str(captured.out).strip()
+
+    expected_answer = json.loads(query1_server_answer_data)
+    print(f"Captured: {captured_out}")
+    received_answer = json.loads(captured_out)
+
+    assert received_answer == expected_answer
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+@pytest.mark.script_launch_mode("subprocess")
+async def test_httpx_using_cli_ep(
+    event_loop, aiohttp_server, monkeypatch, script_runner, run_sync_test
+):
+    from aiohttp import web
+
+    async def handler(request):
+        return web.Response(text=query1_server_answer, content_type="application/json")
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    def test_code():
+
+        monkeypatch.setattr("sys.stdin", io.StringIO(query1_str))
+
+        ret = script_runner.run(
+            "gql-cli", url, "--verbose", stdin=io.StringIO(query1_str)
+        )
+
+        assert ret.success
+
+        # Check that the result has been printed on stdout
+        captured_out = str(ret.stdout).strip()
+
+        expected_answer = json.loads(query1_server_answer_data)
+        print(f"Captured: {captured_out}")
+        received_answer = json.loads(captured_out)
+
+        assert received_answer == expected_answer
+
+    await run_sync_test(event_loop, server, test_code)
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_using_cli_invalid_param(
+    event_loop, aiohttp_server, monkeypatch, capsys
+):
+    from aiohttp import web
+
+    async def handler(request):
+        return web.Response(text=query1_server_answer, content_type="application/json")
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    parser = get_parser(with_examples=True)
+    args = parser.parse_args([url, "--variables", "invalid_param"])
+
+    # Monkeypatching sys.stdin to simulate getting the query
+    # via the standard input
+    monkeypatch.setattr("sys.stdin", io.StringIO(query1_str))
+
+    # Check that the exit_code is an error
+    exit_code = await main(args)
+    assert exit_code == 1
+
+    # Check that the error has been printed on stdout
+    captured = capsys.readouterr()
+    captured_err = str(captured.err).strip()
+    print(f"Captured: {captured_err}")
+
+    expected_error = "Error: Invalid variable: invalid_param"
+
+    assert expected_error in captured_err
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_using_cli_invalid_query(
+    event_loop, aiohttp_server, monkeypatch, capsys
+):
+    from aiohttp import web
+
+    async def handler(request):
+        return web.Response(text=query1_server_answer, content_type="application/json")
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    parser = get_parser(with_examples=True)
+    args = parser.parse_args([url])
+
+    # Send invalid query on standard input
+    monkeypatch.setattr("sys.stdin", io.StringIO("BLAHBLAH"))
+
+    exit_code = await main(args)
+
+    assert exit_code == 1
+
+    # Check that the error has been printed on stdout
+    captured = capsys.readouterr()
+    captured_err = str(captured.err).strip()
+    print(f"Captured: {captured_err}")
+
+    expected_error = "Syntax Error: Unexpected Name 'BLAHBLAH'"
+
+    assert expected_error in captured_err
+
+
+query1_server_answer_with_extensions = (
+    f'{{"data":{query1_server_answer_data}, "extensions":{{"key1": "val1"}}}}'
+)
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_query_with_extensions(event_loop, aiohttp_server):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    async def handler(request):
+        return web.Response(
+            text=query1_server_answer_with_extensions, content_type="application/json"
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    transport = HTTPXAsyncTransport(url=url, timeout=10)
+
+    async with Client(transport=transport) as session:
+
+        query = gql(query1_str)
+
+        execution_result = await session.execute(query, get_execution_result=True)
+
+        assert execution_result.extensions["key1"] == "val1"
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_query_https(event_loop, ssl_aiohttp_server):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    async def handler(request):
+        return web.Response(text=query1_server_answer, content_type="application/json")
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await ssl_aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    assert url.startswith("https://")
+
+    cert, _ = get_localhost_ssl_context()
+
+    transport = HTTPXAsyncTransport(url=url, timeout=10, verify=cert.decode())
+
+    async with Client(transport=transport) as session:
+
+        query = gql(query1_str)
+
+        # Execute query asynchronously
+        result = await session.execute(query)
+
+        continents = result["continents"]
+
+        africa = continents[0]
+
+        assert africa["code"] == "AF"
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_error_fetching_schema(event_loop, aiohttp_server):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    error_answer = """
+{
+    "errors": [
+        {
+            "errorType": "UnauthorizedException",
+            "message": "Permission denied"
+        }
+    ]
+}
+"""
+
+    async def handler(request):
+        return web.Response(
+            text=error_answer,
+            content_type="application/json",
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    transport = HTTPXAsyncTransport(url=url, timeout=10)
+
+    with pytest.raises(TransportQueryError) as exc_info:
+        async with Client(transport=transport, fetch_schema_from_transport=True):
+            pass
+
+    expected_error = (
+        "Error while fetching schema: "
+        "{'errorType': 'UnauthorizedException', 'message': 'Permission denied'}"
+    )
+
+    assert expected_error in str(exc_info.value)
+    assert transport.client is None
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_reconnecting_session(event_loop, aiohttp_server):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    async def handler(request):
+        return web.Response(
+            text=query1_server_answer,
+            content_type="application/json",
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    transport = HTTPXAsyncTransport(url=url, timeout=10)
+
+    client = Client(transport=transport)
+
+    session = await client.connect_async(reconnecting=True)
+
+    query = gql(query1_str)
+
+    # Execute query asynchronously
+    result = await session.execute(query)
+
+    continents = result["continents"]
+
+    africa = continents[0]
+
+    assert africa["code"] == "AF"
+
+    await client.close_async()
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+@pytest.mark.parametrize("retries", [False, lambda e: e])
+async def test_httpx_reconnecting_session_retries(event_loop, aiohttp_server, retries):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    async def handler(request):
+        return web.Response(
+            text=query1_server_answer,
+            content_type="application/json",
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    transport = HTTPXAsyncTransport(url=url, timeout=10)
+
+    client = Client(transport=transport)
+
+    session = await client.connect_async(
+        reconnecting=True, retry_execute=retries, retry_connect=retries
+    )
+
+    assert session._execute_with_retries == session._execute_once
+    assert session._connect_with_retries == session.transport.connect
+
+    await client.close_async()
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_reconnecting_session_start_connecting_task_twice(
+    event_loop, aiohttp_server, caplog
+):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    async def handler(request):
+        return web.Response(
+            text=query1_server_answer,
+            content_type="application/json",
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    transport = HTTPXAsyncTransport(url=url, timeout=10)
+
+    client = Client(transport=transport)
+
+    session = await client.connect_async(reconnecting=True)
+
+    await session.start_connecting_task()
+
+    print(f"Captured log: {caplog.text}")
+
+    expected_warning = "connect task already started!"
+    assert expected_warning in caplog.text
+
+    await client.close_async()
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_httpx_json_serializer(event_loop, aiohttp_server, caplog):
+    from aiohttp import web
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    async def handler(request):
+
+        request_text = await request.text()
+        print(f"Received on backend: {request_text}")
+
+        return web.Response(
+            text=query1_server_answer,
+            content_type="application/json",
+        )
+
+    app = web.Application()
+    app.router.add_route("POST", "/", handler)
+    server = await aiohttp_server(app)
+
+    url = str(server.make_url("/"))
+
+    transport = HTTPXAsyncTransport(
+        url=url,
+        timeout=10,
+        json_serialize=lambda e: json.dumps(e, separators=(",", ":")),
+    )
+
+    async with Client(transport=transport) as session:
+
+        query = gql(query1_str)
+
+        # Execute query asynchronously
+        result = await session.execute(query)
+
+        continents = result["continents"]
+
+        africa = continents[0]
+
+        assert africa["code"] == "AF"
+
+    # Checking that there is no space after the colon in the log
+    expected_log = '"query":"query getContinents'
+    assert expected_log in caplog.text
diff --git a/tests/test_httpx_online.py b/tests/test_httpx_online.py
new file mode 100644
index 00000000..ee08e2b1
--- /dev/null
+++ b/tests/test_httpx_online.py
@@ -0,0 +1,148 @@
+import asyncio
+import sys
+from typing import Dict
+
+import pytest
+
+from gql import Client, gql
+from gql.transport.exceptions import TransportQueryError
+
+
+@pytest.mark.httpx
+@pytest.mark.online
+@pytest.mark.asyncio
+@pytest.mark.parametrize("protocol", ["http", "https"])
+async def test_httpx_simple_query(event_loop, protocol):
+
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    # Create http or https url
+    url = f"{protocol}://countries.trevorblades.com/graphql"
+
+    # Get transport
+    sample_transport = HTTPXAsyncTransport(url=url)
+
+    # Instanciate client
+    async with Client(transport=sample_transport) as session:
+
+        query = gql(
+            """
+            query getContinents {
+              continents {
+                code
+                name
+              }
+            }
+        """
+        )
+
+        # Fetch schema
+        await session.fetch_schema()
+
+        # Execute query
+        result = await session.execute(query)
+
+        # Verify result
+        assert isinstance(result, Dict)
+
+        print(result)
+
+        continents = result["continents"]
+
+        africa = continents[0]
+
+        assert africa["code"] == "AF"
+
+
+@pytest.mark.httpx
+@pytest.mark.online
+@pytest.mark.asyncio
+async def test_httpx_invalid_query(event_loop):
+
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    sample_transport = HTTPXAsyncTransport(
+        url="https://countries.trevorblades.com/graphql"
+    )
+
+    async with Client(transport=sample_transport) as session:
+
+        query = gql(
+            """
+            query getContinents {
+              continents {
+                code
+                bloh
+              }
+            }
+        """
+        )
+
+        with pytest.raises(TransportQueryError):
+            await session.execute(query)
+
+
+@pytest.mark.httpx
+@pytest.mark.online
+@pytest.mark.skipif(sys.version_info < (3, 8), reason="requires python3.8 or higher")
+@pytest.mark.asyncio
+async def test_httpx_two_queries_in_parallel_using_two_tasks(event_loop):
+
+    from gql.transport.httpx import HTTPXAsyncTransport
+
+    sample_transport = HTTPXAsyncTransport(
+        url="https://countries.trevorblades.com/graphql",
+    )
+
+    # Instanciate client
+    async with Client(transport=sample_transport) as session:
+
+        query1 = gql(
+            """
+            query getContinents {
+              continents {
+                code
+              }
+            }
+        """
+        )
+
+        query2 = gql(
+            """
+            query getContinents {
+              continents {
+                name
+              }
+            }
+        """
+        )
+
+        async def query_task1():
+            result = await session.execute(query1)
+
+            assert isinstance(result, Dict)
+
+            print(result)
+
+            continents = result["continents"]
+
+            africa = continents[0]
+            assert africa["code"] == "AF"
+
+        async def query_task2():
+            result = await session.execute(query2)
+
+            assert isinstance(result, Dict)
+
+            print(result)
+
+            continents = result["continents"]
+
+            africa = continents[0]
+            assert africa["name"] == "Africa"
+
+        task1 = asyncio.create_task(query_task1())
+        task2 = asyncio.create_task(query_task2())
+
+        await task1
+        await task2