diff --git a/newsfragments/3093.change.rst b/newsfragments/3093.change.rst
new file mode 100644
index 000000000..bc3288ef2
--- /dev/null
+++ b/newsfragments/3093.change.rst
@@ -0,0 +1 @@
+Raise ``UnexpectedQueryResponseError`` for empty or non-JSON Cloud Query error responses.
diff --git a/spelling_private_dict.txt b/spelling_private_dict.txt
index 73d358a00..1cce267f5 100644
--- a/spelling_private_dict.txt
+++ b/spelling_private_dict.txt
@@ -22,6 +22,7 @@ TargetStatusNotSuccess
 TargetStatusProcessing
 TooManyRequests
 Ubuntu
+UnexpectedQueryResponse
 UnknownTarget
 admin
 api
diff --git a/src/vws/async_query.py b/src/vws/async_query.py
index bebd3089d..0c3bd353f 100644
--- a/src/vws/async_query.py
+++ b/src/vws/async_query.py
@@ -22,6 +22,7 @@
 from vws.exceptions.custom_exceptions import (
     RequestEntityTooLargeError,
     ServerError,
+    UnexpectedQueryResponseError,
 )
 from vws.include_target_data import CloudRecoIncludeTargetData
 from vws.reports import QueryResult
@@ -117,6 +118,8 @@ async def query(
                 given image is too large.
             ~vws.exceptions.custom_exceptions.ServerError: There is an
                 error with Vuforia's servers.
+            ~vws.exceptions.custom_exceptions.UnexpectedQueryResponseError:
+                The response body is empty or is not valid JSON.
 
         Returns:
             An ordered list of target details of matching
@@ -184,7 +187,12 @@ async def query(
         ):  # pragma: no cover
             raise ServerError(response=response)
 
-        result_code = json.loads(s=response.text)["result_code"]
+        try:
+            response_dict = json.loads(s=response.text)
+        except json.JSONDecodeError as exc:
+            raise UnexpectedQueryResponseError(response=response) from exc
+
+        result_code = response_dict["result_code"]
         if result_code != "Success":
             exception = {
                 "AuthenticationFailure": (AuthenticationFailureError),
@@ -194,9 +202,7 @@ async def query(
             }[result_code]
             raise exception(response=response)
 
-        result_list = list(
-            json.loads(s=response.text)["results"],
-        )
+        result_list = list(response_dict["results"])
         return [
             QueryResult.from_response_dict(response_dict=item)
             for item in result_list
diff --git a/src/vws/exceptions/custom_exceptions.py b/src/vws/exceptions/custom_exceptions.py
index cf3902264..19d6e180c 100644
--- a/src/vws/exceptions/custom_exceptions.py
+++ b/src/vws/exceptions/custom_exceptions.py
@@ -50,3 +50,25 @@ def __init__(self, response: Response) -> None:
     def response(self) -> Response:
         """The response returned by Vuforia which included this error."""
         return self._response
+
+
+@beartype
+class UnexpectedQueryResponseError(Exception):
+    """Exception raised when a Cloud Query response body is not valid JSON.
+
+    Vuforia documents that failed Cloud Query requests may return an
+    arbitrary body or no body at all.
+    """
+
+    def __init__(self, response: Response) -> None:
+        """
+        Args:
+            response: The response returned by Vuforia.
+        """
+        super().__init__(response.text)
+        self._response = response
+
+    @property
+    def response(self) -> Response:
+        """The response returned by Vuforia which included this error."""
+        return self._response
diff --git a/src/vws/query.py b/src/vws/query.py
index 44840d6f0..ae9f70d74 100644
--- a/src/vws/query.py
+++ b/src/vws/query.py
@@ -20,6 +20,7 @@
 from vws.exceptions.custom_exceptions import (
     RequestEntityTooLargeError,
     ServerError,
+    UnexpectedQueryResponseError,
 )
 from vws.include_target_data import CloudRecoIncludeTargetData
 from vws.reports import QueryResult
@@ -99,6 +100,8 @@ def query(
                 given image is too large.
             ~vws.exceptions.custom_exceptions.ServerError: There is an
                 error with Vuforia's servers.
+            ~vws.exceptions.custom_exceptions.UnexpectedQueryResponseError:
+                The response body is empty or is not valid JSON.
 
         Returns:
             An ordered list of target details of matching targets.
@@ -154,7 +157,12 @@ def query(
         ):  # pragma: no cover
             raise ServerError(response=response)
 
-        result_code = json.loads(s=response.text)["result_code"]
+        try:
+            response_dict = json.loads(s=response.text)
+        except json.JSONDecodeError as exc:
+            raise UnexpectedQueryResponseError(response=response) from exc
+
+        result_code = response_dict["result_code"]
         if result_code != "Success":
             exception = {
                 "AuthenticationFailure": AuthenticationFailureError,
@@ -164,7 +172,7 @@ def query(
             }[result_code]
             raise exception(response=response)
 
-        result_list = list(json.loads(s=response.text)["results"])
+        result_list = list(response_dict["results"])
         return [
             QueryResult.from_response_dict(response_dict=item)
             for item in result_list
diff --git a/tests/test_async_cloud_reco_exceptions.py b/tests/test_async_cloud_reco_exceptions.py
index cbe2b22cc..85b2f8406 100644
--- a/tests/test_async_cloud_reco_exceptions.py
+++ b/tests/test_async_cloud_reco_exceptions.py
@@ -19,7 +19,9 @@
 )
 from vws.exceptions.custom_exceptions import (
     RequestEntityTooLargeError,
+    UnexpectedQueryResponseError,
 )
+from vws.response import Response
 
 
 @pytest.mark.asyncio
@@ -118,3 +120,59 @@ async def test_inactive_project(
         response = exc.value.response
         assert response.status_code == HTTPStatus.FORBIDDEN
         assert response.tell_position != 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    argnames="response_text",
+    argvalues=["", "not-json"],
+)
+async def test_unexpected_query_response(
+    *,
+    high_quality_image: io.BytesIO,
+    response_text: str,
+) -> None:
+    """
+    An ``UnexpectedQueryResponseError`` is raised for empty or non-JSON
+    4xx Cloud Query responses.
+    """
+
+    class _NonJSONAsyncTransport:
+        """Async transport that returns a non-JSON 4xx response."""
+
+        async def aclose(self) -> None:
+            """Close the transport."""
+
+        async def __call__(
+            self,
+            *,
+            method: str,
+            url: str,
+            headers: dict[str, str],
+            data: bytes,
+            request_timeout: float | tuple[float, float],
+        ) -> Response:
+            """Return a non-JSON error response."""
+            del method, headers, request_timeout
+            return Response(
+                text=response_text,
+                url=url,
+                status_code=HTTPStatus.BAD_REQUEST,
+                headers={},
+                request_body=data,
+                tell_position=0,
+                content=response_text.encode(),
+            )
+
+    async with AsyncCloudRecoService(
+        client_access_key=uuid.uuid4().hex,
+        client_secret_key=uuid.uuid4().hex,
+        transport=_NonJSONAsyncTransport(),
+    ) as async_cloud_reco_client:
+        with pytest.raises(
+            expected_exception=UnexpectedQueryResponseError,
+        ) as exc:
+            await async_cloud_reco_client.query(image=high_quality_image)
+
+    assert exc.value.response.status_code == HTTPStatus.BAD_REQUEST
+    assert exc.value.response.text == response_text
diff --git a/tests/test_cloud_reco_exceptions.py b/tests/test_cloud_reco_exceptions.py
index 43a263d40..cc370c3b4 100644
--- a/tests/test_cloud_reco_exceptions.py
+++ b/tests/test_cloud_reco_exceptions.py
@@ -20,7 +20,9 @@
 )
 from vws.exceptions.custom_exceptions import (
     RequestEntityTooLargeError,
+    UnexpectedQueryResponseError,
 )
+from vws.response import Response
 
 
 def test_too_many_max_results(
@@ -126,3 +128,59 @@ def test_inactive_project(
         # We need one test which checks tell position
         # and so we choose this one almost at random.
         assert response.tell_position != 0
+
+
+@pytest.mark.parametrize(
+    argnames="response_text",
+    argvalues=["", "not-json"],
+)
+def test_unexpected_query_response(
+    *,
+    high_quality_image: io.BytesIO,
+    response_text: str,
+) -> None:
+    """
+    An ``UnexpectedQueryResponseError`` is raised for empty or non-JSON
+    4xx Cloud Query responses.
+    """
+
+    class _NonJSONTransport:
+        """Transport that returns a non-JSON 4xx response."""
+
+        def close(self) -> None:
+            """Close the transport."""
+
+        def __call__(
+            self,
+            *,
+            method: str,
+            url: str,
+            headers: dict[str, str],
+            data: bytes,
+            request_timeout: float | tuple[float, float],
+        ) -> Response:
+            """Return a non-JSON error response."""
+            del method, headers, request_timeout
+            return Response(
+                text=response_text,
+                url=url,
+                status_code=HTTPStatus.BAD_REQUEST,
+                headers={},
+                request_body=data,
+                tell_position=0,
+                content=response_text.encode(),
+            )
+
+    cloud_reco_client = CloudRecoService(
+        client_access_key=uuid.uuid4().hex,
+        client_secret_key=uuid.uuid4().hex,
+        transport=_NonJSONTransport(),
+    )
+
+    with pytest.raises(
+        expected_exception=UnexpectedQueryResponseError,
+    ) as exc:
+        cloud_reco_client.query(image=high_quality_image)
+
+    assert exc.value.response.status_code == HTTPStatus.BAD_REQUEST
+    assert exc.value.response.text == response_text