Repository: ethereum/web3.py

Issue:
Batch requests crash when an empty batch is requested against some Ethereum clients like Erigon
### What happened?

A batch request to an Ethereum RPC endpoint is responded with an array containing the responses to each request in the batch:
```sh
curl http://localhost:8545 -H "Content-Type: application/json" --data '[{"jsonrpc":"2.0", "id": 1, "method": "eth_blockNumber", "params": []}]'

[{"jsonrpc":"2.0","id":1,"result":"0x1410bb2"}]
```

So Web3.py assumes the response is an array while parsing it: https://github.com/ethereum/web3.py/blob/3c412f81d19f70d70b262ae8fefe6bd8411667e3/web3/providers/rpc/rpc.py#L187-L188

However, when an empty batch is executed, some RPC endpoint (mine is Erigon) returns an object specifying the error instead of an array. This breaks the parsing and result in error: `AttributeError: 'str' object has no attribute 'get'`
```sh
curl http://localhost:8545 -H "Content-Type: application/json" --data '[]'

{"jsonrpc":"2.0","id":null,"error":{"code":-32600,"message":"empty batch"}}
```

The same happens when the batch size exceeds the limit:
```json
{"jsonrpc":"2.0","id":null,"error":{"code":-32000,"message":"batch limit 100 exceeded (can increase by --rpc.batch.limit). Requested batch of size: 526"}}
```

### Code that produced the error

```python
from web3 import Web3, HTTPProvider
w3 = Web3(HTTPProvider("http://localhost:8545"))
with w3.batch_requests() as batch:
    batch.execute()
```


### Full error output

```shell
AttributeError                            Traceback (most recent call last)
Cell In[222], line 5
      3 with w3.batch_requests() as batch:
----> 4     batch.execute()

File ~/.cache/pypoetry/virtualenvs/etherspect-dagster-jFkAMd9f-py3.12/lib/python3.12/site-packages/web3/_utils/batching.py:147, in RequestBatcher.execute(self)
    145 def execute(self) -> List["RPCResponse"]:
    146     self._validate_is_batching()
--> 147     responses = self.web3.manager._make_batch_request(self._requests_info)
    148     self._end_batching()
    149     return responses

File ~/.cache/pypoetry/virtualenvs/etherspect-dagster-jFkAMd9f-py3.12/lib/python3.12/site-packages/web3/manager.py:430, in RequestManager._make_batch_request(self, requests_info)
    426 provider = cast(JSONBaseProvider, self.provider)
    427 request_func = provider.batch_request_func(
    428     cast("Web3", self.w3), cast("MiddlewareOnion", self.middleware_onion)
    429 )
--> 430 responses = request_func(
    431     [
    432         (method, params)
    433         for (method, params), _response_formatters in requests_info
    434     ]
    435 )
    436 formatted_responses = [
    437     self._format_batched_response(info, resp)
    438     for info, resp in zip(requests_info, responses)
    439 ]
    440 return list(formatted_responses)

File ~/.cache/pypoetry/virtualenvs/etherspect-dagster-jFkAMd9f-py3.12/lib/python3.12/site-packages/web3/middleware/base.py:70, in Web3Middleware.wrap_make_batch_request.<locals>.middleware(requests_info)
     63 def middleware(
     64     requests_info: List[Tuple["RPCEndpoint", Any]]
     65 ) -> List["RPCResponse"]:
     66     req_processed = [
     67         self.request_processor(method, params)
     68         for (method, params) in requests_info
     69     ]
---> 70     responses = make_batch_request(req_processed)
     71     methods, _params = zip(*req_processed)
     72     formatted_responses = [
     73         self.response_processor(m, r) for m, r in zip(methods, responses)
     74     ]

File ~/.cache/pypoetry/virtualenvs/etherspect-dagster-jFkAMd9f-py3.12/lib/python3.12/site-packages/web3/middleware/base.py:70, in Web3Middleware.wrap_make_batch_request.<locals>.middleware(requests_info)
     63 def middleware(
     64     requests_info: List[Tuple["RPCEndpoint", Any]]
     65 ) -> List["RPCResponse"]:
     66     req_processed = [
     67         self.request_processor(method, params)
     68         for (method, params) in requests_info
     69     ]
---> 70     responses = make_batch_request(req_processed)
     71     methods, _params = zip(*req_processed)
     72     formatted_responses = [
     73         self.response_processor(m, r) for m, r in zip(methods, responses)
     74     ]

    [... skipping similar frames: Web3Middleware.wrap_make_batch_request.<locals>.middleware at line 70 (2 times)]

File ~/.cache/pypoetry/virtualenvs/etherspect-dagster-jFkAMd9f-py3.12/lib/python3.12/site-packages/web3/middleware/base.py:70, in Web3Middleware.wrap_make_batch_request.<locals>.middleware(requests_info)
     63 def middleware(
     64     requests_info: List[Tuple["RPCEndpoint", Any]]
     65 ) -> List["RPCResponse"]:
     66     req_processed = [
     67         self.request_processor(method, params)
     68         for (method, params) in requests_info
     69     ]
---> 70     responses = make_batch_request(req_processed)
     71     methods, _params = zip(*req_processed)
     72     formatted_responses = [
     73         self.response_processor(m, r) for m, r in zip(methods, responses)
     74     ]

File ~/.cache/pypoetry/virtualenvs/etherspect-dagster-jFkAMd9f-py3.12/lib/python3.12/site-packages/web3/providers/rpc/rpc.py:188, in HTTPProvider.make_batch_request(self, batch_requests)
    186 self.logger.debug("Received batch response HTTP.")
    187 responses_list = cast(List[RPCResponse], self.decode_rpc_response(raw_response))
--> 188 return sort_batch_response_by_response_ids(responses_list)

File ~/.cache/pypoetry/virtualenvs/etherspect-dagster-jFkAMd9f-py3.12/lib/python3.12/site-packages/web3/_utils/batching.py:203, in sort_batch_response_by_response_ids(responses)
    200 def sort_batch_response_by_response_ids(
    201     responses: List["RPCResponse"],
    202 ) -> List["RPCResponse"]:
--> 203     if all(response.get("id") is not None for response in responses):
    204         # If all responses have an `id`, sort them by `id, since the JSON-RPC 2.0 spec
    205         # doesn't guarantee order.
    206         return sorted(responses, key=lambda response: response["id"])
    207     else:
    208         # If any response is missing an `id`, which should only happen on particular
    209         # errors, return them in the order they were received and hope that the
    210         # provider is returning them in order. Issue a warning.

File ~/.cache/pypoetry/virtualenvs/etherspect-dagster-jFkAMd9f-py3.12/lib/python3.12/site-packages/web3/_utils/batching.py:203, in <genexpr>(.0)
    200 def sort_batch_response_by_response_ids(
    201     responses: List["RPCResponse"],
    202 ) -> List["RPCResponse"]:
--> 203     if all(response.get("id") is not None for response in responses):
    204         # If all responses have an `id`, sort them by `id, since the JSON-RPC 2.0 spec
    205         # doesn't guarantee order.
    206         return sorted(responses, key=lambda response: response["id"])
    207     else:
    208         # If any response is missing an `id`, which should only happen on particular
    209         # errors, return them in the order they were received and hope that the
    210         # provider is returning them in order. Issue a warning.

AttributeError: 'str' object has no attribute 'get'
```


### Fill this section in if you know how this could or should be fixed

Revise the response parsing behavior in `web3/providers/rpc/rpc.py:make_batch_request()` and any other relevant files

### web3 Version

7.4.0

### Python Version

3.12.5

### Operating System

linux

### Output from `pip freeze`

```shell
aiohappyeyeballs==2.4.3
aiohttp==3.10.10
aiosignal==1.3.1
alembic==1.13.3
annotated-types==0.7.0
anyio==4.6.2.post1
asttokens==2.4.1
attrs==24.2.0
backoff==2.2.1
bitarray==3.0.0
certifi==2024.8.30
charset-normalizer==3.4.0
ckzg==2.0.1
click==8.1.7
coloredlogs==14.0
croniter==3.0.3
cytoolz==1.0.0
dagster==1.8.12
dagster-graphql==1.8.12
dagster-pipes==1.8.12
dagster-webserver==1.8.12
decorator==5.1.1
docstring_parser==0.16
eth-account==0.13.4
eth-hash==0.7.0
eth-keyfile==0.8.1
eth-keys==0.6.0
eth-rlp==2.1.0
eth-typing==5.0.1
eth-utils==5.1.0
eth_abi==5.1.0
executing==2.1.0
filelock==3.16.1
frozenlist==1.5.0
fsspec==2024.10.0
gql==3.5.0
graphene==3.4
graphql-core==3.2.5
graphql-relay==3.2.0
greenlet==3.1.1
grpcio==1.67.0
grpcio-health-checking==1.62.3
h11==0.14.0
hexbytes==1.2.1
httptools==0.6.4
humanfriendly==10.0
idna==3.10
iniconfig==2.0.0
ipdb==0.13.13
ipython==8.28.0
jedi==0.19.1
Jinja2==3.1.4
Mako==1.3.5
markdown-it-py==3.0.0
MarkupSafe==3.0.2
matplotlib-inline==0.1.7
mdurl==0.1.2
more-itertools==10.5.0
multidict==6.1.0
mypy==1.12.1
mypy-extensions==1.0.0
numpy==2.1.2
packaging==24.1
pandas==2.2.3
pandas-stubs==2.2.3.241009
parsimonious==0.10.0
parso==0.8.4
pexpect==4.9.0
pluggy==1.5.0
prompt_toolkit==3.0.48
propcache==0.2.0
protobuf==4.25.5
ptyprocess==0.7.0
pure_eval==0.2.3
pycryptodome==3.21.0
pydantic==2.9.2
pydantic_core==2.23.4
Pygments==2.18.0
pytest==8.3.3
python-dateutil==2.9.0.post0
python-dotenv==1.0.1
pytz==2024.2
pyunormalize==16.0.0
PyYAML==6.0.2
regex==2024.9.11
requests==2.32.3
requests-toolbelt==1.0.0
rich==13.9.2
rlp==4.0.1
setuptools==75.2.0
six==1.16.0
sniffio==1.3.1
SQLAlchemy==2.0.36
stack-data==0.6.3
starlette==0.41.0
structlog==24.4.0
tabulate==0.9.0
tomli==2.0.2
toolz==1.0.0
toposort==1.10
tqdm==4.66.5
traitlets==5.14.3
types-pytz==2024.2.0.20241003
types-requests==2.32.0.20241016
typing_extensions==4.12.2
tzdata==2024.2
universal_pathlib==0.2.5
urllib3==2.2.3
uvicorn==0.32.0
uvloop==0.21.0
watchdog==5.0.3
watchfiles==0.24.0
wcwidth==0.2.13
web3==7.4.0
websockets==13.1
yarl==1.15.5
```


Discussion:

Patch:
diff --git a/newsfragments/3585.bugfix.rst b/newsfragments/3585.bugfix.rst
new file mode 100644
index 00000000..2c3babd2
--- /dev/null
+++ b/newsfragments/3585.bugfix.rst
@@ -0,0 +1,1 @@
+Handle the case when a single RPC error response is returned for a batch request, instead of always expecting a list of responses.
diff --git a/web3/manager.py b/web3/manager.py
index 8e2f2b64..2a3e6d11 100644
--- a/web3/manager.py
+++ b/web3/manager.py
@@ -8,6 +8,7 @@ from typing import (
     Coroutine,
     Dict,
     List,
+    NoReturn,
     Optional,
     Sequence,
     Tuple,
@@ -266,6 +267,30 @@ def _validate_response(
         _raise_bad_response_format(response)
 
 
+def _raise_error_for_batch_response(
+    response: RPCResponse,
+    logger: Optional[logging.Logger] = None,
+) -> NoReturn:
+    error = response.get("error")
+    if error is None:
+        _raise_bad_response_format(
+            response,
+            "Batch response must be formatted as a list of responses or "
+            "as a single JSON-RPC error response.",
+        )
+    _validate_response(
+        response,
+        None,
+        is_subscription_response=False,
+        logger=logger,
+        params=[],
+    )
+    # This should not be reached, but if it is, raise a generic `BadResponseFormat`
+    raise BadResponseFormat(
+        "Batch response was in an unexpected format and unable to be parsed."
+    )
+
+
 class RequestManager:
     logger = logging.getLogger("web3.manager.RequestManager")
 
@@ -436,17 +461,23 @@ class RequestManager:
         request_func = provider.batch_request_func(
             cast("Web3", self.w3), cast("MiddlewareOnion", self.middleware_onion)
         )
-        responses = request_func(
+        response = request_func(
             [
                 (method, params)
                 for (method, params), _response_formatters in requests_info
             ]
         )
-        formatted_responses = [
-            self._format_batched_response(info, resp)
-            for info, resp in zip(requests_info, responses)
-        ]
-        return list(formatted_responses)
+
+        if isinstance(response, list):
+            # expected format
+            formatted_responses = [
+                self._format_batched_response(info, cast(RPCResponse, resp))
+                for info, resp in zip(requests_info, response)
+            ]
+            return list(formatted_responses)
+        else:
+            # expect a single response with an error
+            _raise_error_for_batch_response(response, self.logger)
 
     async def _async_make_batch_request(
         self,
@@ -465,25 +496,31 @@ class RequestManager:
         # since we add items to the batch without awaiting, we unpack the coroutines
         # and await them all here
         unpacked_requests_info = await asyncio.gather(*requests_info)
-        responses = await request_func(
+        response = await request_func(
             [
                 (method, params)
                 for (method, params), _response_formatters in unpacked_requests_info
             ]
         )
 
-        if isinstance(self.provider, PersistentConnectionProvider):
-            # call _process_response for each response in the batch
-            return [
-                cast(RPCResponse, await self._process_response(resp))
-                for resp in responses
+        if isinstance(response, list):
+            # expected format
+            response = cast(List[RPCResponse], response)
+            if isinstance(self.provider, PersistentConnectionProvider):
+                # call _process_response for each response in the batch
+                return [
+                    cast(RPCResponse, await self._process_response(resp))
+                    for resp in response
+                ]
+
+            formatted_responses = [
+                self._format_batched_response(info, resp)
+                for info, resp in zip(unpacked_requests_info, response)
             ]
-
-        formatted_responses = [
-            self._format_batched_response(info, resp)
-            for info, resp in zip(unpacked_requests_info, responses)
-        ]
-        return list(formatted_responses)
+            return list(formatted_responses)
+        else:
+            # expect a single response with an error
+            _raise_error_for_batch_response(response, self.logger)
 
     def _format_batched_response(
         self,
@@ -491,6 +528,13 @@ class RequestManager:
         response: RPCResponse,
     ) -> RPCResponse:
         result_formatters, error_formatters, null_result_formatters = requests_info[1]
+        _validate_response(
+            response,
+            error_formatters,
+            is_subscription_response=False,
+            logger=self.logger,
+            params=requests_info[0][1],
+        )
         return apply_result_formatters(
             result_formatters,
             self.formatted_response(
diff --git a/web3/middleware/base.py b/web3/middleware/base.py
index 590fce58..5f968f82 100644
--- a/web3/middleware/base.py
+++ b/web3/middleware/base.py
@@ -62,15 +62,19 @@ class Web3Middleware:
     ) -> "MakeBatchRequestFn":
         def middleware(
             requests_info: List[Tuple["RPCEndpoint", Any]]
-        ) -> List["RPCResponse"]:
+        ) -> Union[List["RPCResponse"], "RPCResponse"]:
             req_processed = [
                 self.request_processor(method, params)
                 for (method, params) in requests_info
             ]
-            responses = make_batch_request(req_processed)
+            response = make_batch_request(req_processed)
+            if not isinstance(response, list):
+                # RPC errors return only one response with the error object
+                return response
+
             methods, _params = zip(*req_processed)
             formatted_responses = [
-                self.response_processor(m, r) for m, r in zip(methods, responses)
+                self.response_processor(m, r) for m, r in zip(methods, response)
             ]
             return formatted_responses
 
@@ -103,16 +107,20 @@ class Web3Middleware:
     ) -> "AsyncMakeBatchRequestFn":
         async def middleware(
             requests_info: List[Tuple["RPCEndpoint", Any]]
-        ) -> List["RPCResponse"]:
+        ) -> Union[List["RPCResponse"], "RPCResponse"]:
             req_processed = [
                 await self.async_request_processor(method, params)
                 for (method, params) in requests_info
             ]
-            responses = await make_batch_request(req_processed)
+            response = await make_batch_request(req_processed)
+            if not isinstance(response, list):
+                # RPC errors return only one response with the error object
+                return response
+
             methods, _params = zip(*req_processed)
             formatted_responses = [
                 await self.async_response_processor(m, r)
-                for m, r in zip(methods, responses)
+                for m, r in zip(methods, response)
             ]
             return formatted_responses
 
diff --git a/web3/providers/async_base.py b/web3/providers/async_base.py
index 55a4a8ee..56c4fe71 100644
--- a/web3/providers/async_base.py
+++ b/web3/providers/async_base.py
@@ -77,7 +77,8 @@ class AsyncBaseProvider:
 
     _is_batching: bool = False
     _batch_request_func_cache: Tuple[
-        Tuple[Middleware, ...], Callable[..., Coroutine[Any, Any, List[RPCResponse]]]
+        Tuple[Middleware, ...],
+        Callable[..., Coroutine[Any, Any, Union[List[RPCResponse], RPCResponse]]],
     ] = (None, None)
 
     is_async = True
@@ -119,7 +120,7 @@ class AsyncBaseProvider:
 
     async def batch_request_func(
         self, async_w3: "AsyncWeb3", middleware_onion: MiddlewareOnion
-    ) -> Callable[..., Coroutine[Any, Any, List[RPCResponse]]]:
+    ) -> Callable[..., Coroutine[Any, Any, Union[List[RPCResponse], RPCResponse]]]:
         middleware: Tuple[Middleware, ...] = middleware_onion.as_tuple_of_middleware()
 
         cache_key = self._batch_request_func_cache[0]
@@ -141,8 +142,8 @@ class AsyncBaseProvider:
 
     async def make_batch_request(
         self, requests: List[Tuple[RPCEndpoint, Any]]
-    ) -> List[RPCResponse]:
-        raise NotImplementedError("Only AsyncHTTPProvider supports this method")
+    ) -> Union[List[RPCResponse], RPCResponse]:
+        raise NotImplementedError("Providers must implement this method")
 
     async def is_connected(self, show_traceback: bool = False) -> bool:
         raise NotImplementedError("Providers must implement this method")
diff --git a/web3/providers/base.py b/web3/providers/base.py
index e1606967..36a1ca9f 100644
--- a/web3/providers/base.py
+++ b/web3/providers/base.py
@@ -118,7 +118,7 @@ class JSONBaseProvider(BaseProvider):
 
     _is_batching: bool = False
     _batch_request_func_cache: Tuple[
-        Tuple[Middleware, ...], Callable[..., List[RPCResponse]]
+        Tuple[Middleware, ...], Callable[..., Union[List[RPCResponse], RPCResponse]]
     ] = (None, None)
 
     def __init__(self, **kwargs: Any) -> None:
@@ -168,7 +168,7 @@ class JSONBaseProvider(BaseProvider):
 
     def batch_request_func(
         self, w3: "Web3", middleware_onion: MiddlewareOnion
-    ) -> Callable[..., List[RPCResponse]]:
+    ) -> Callable[..., Union[List[RPCResponse], RPCResponse]]:
         middleware: Tuple[Middleware, ...] = middleware_onion.as_tuple_of_middleware()
 
         cache_key = self._batch_request_func_cache[0]
@@ -199,5 +199,5 @@ class JSONBaseProvider(BaseProvider):
 
     def make_batch_request(
         self, requests: List[Tuple[RPCEndpoint, Any]]
-    ) -> List[RPCResponse]:
+    ) -> Union[List[RPCResponse], RPCResponse]:
         raise NotImplementedError("Providers must implement this method")
diff --git a/web3/providers/persistent/request_processor.py b/web3/providers/persistent/request_processor.py
index 9d5bf30a..65d4d998 100644
--- a/web3/providers/persistent/request_processor.py
+++ b/web3/providers/persistent/request_processor.py
@@ -6,6 +6,7 @@ from typing import (
     Callable,
     Dict,
     Generic,
+    List,
     Optional,
     Tuple,
     TypeVar,
@@ -270,6 +271,15 @@ class RequestProcessor:
 
     # raw response cache
 
+    def _is_batch_response(
+        self, raw_response: Union[List[RPCResponse], RPCResponse]
+    ) -> bool:
+        return isinstance(raw_response, list) or (
+            isinstance(raw_response, dict)
+            and raw_response.get("id") is None
+            and self._provider._is_batching
+        )
+
     async def cache_raw_response(
         self, raw_response: Any, subscription: bool = False
     ) -> None:
@@ -296,7 +306,7 @@ class RequestProcessor:
                 # otherwise, put it in the subscription response queue so a response
                 # can be yielded by the message stream
                 await self._subscription_response_queue.put(raw_response)
-        elif isinstance(raw_response, list):
+        elif self._is_batch_response(raw_response):
             # Since only one batch should be in the cache at all times, we use a
             # constant cache key for the batch response.
             cache_key = generate_cache_key(BATCH_REQUEST_ID)
diff --git a/web3/providers/rpc/async_rpc.py b/web3/providers/rpc/async_rpc.py
index bc8ef17d..85e27c33 100644
--- a/web3/providers/rpc/async_rpc.py
+++ b/web3/providers/rpc/async_rpc.py
@@ -168,15 +168,20 @@ class AsyncHTTPProvider(AsyncJSONBaseProvider):
 
     async def make_batch_request(
         self, batch_requests: List[Tuple[RPCEndpoint, Any]]
-    ) -> List[RPCResponse]:
+    ) -> Union[List[RPCResponse], RPCResponse]:
         self.logger.debug(f"Making batch request HTTP - uri: `{self.endpoint_uri}`")
         request_data = self.encode_batch_rpc_request(batch_requests)
         raw_response = await self._request_session_manager.async_make_post_request(
             self.endpoint_uri, request_data, **self.get_request_kwargs()
         )
         self.logger.debug("Received batch response HTTP.")
-        responses_list = cast(List[RPCResponse], self.decode_rpc_response(raw_response))
-        return sort_batch_response_by_response_ids(responses_list)
+        response = self.decode_rpc_response(raw_response)
+        if not isinstance(response, list):
+            # RPC errors return only one response with the error object
+            return response
+        return sort_batch_response_by_response_ids(
+            cast(List[RPCResponse], sort_batch_response_by_response_ids(response))
+        )
 
     async def disconnect(self) -> None:
         cache = self._request_session_manager.session_cache
diff --git a/web3/providers/rpc/rpc.py b/web3/providers/rpc/rpc.py
index 7dde0416..4c41d46e 100644
--- a/web3/providers/rpc/rpc.py
+++ b/web3/providers/rpc/rpc.py
@@ -176,12 +176,17 @@ class HTTPProvider(JSONBaseProvider):
 
     def make_batch_request(
         self, batch_requests: List[Tuple[RPCEndpoint, Any]]
-    ) -> List[RPCResponse]:
+    ) -> Union[List[RPCResponse], RPCResponse]:
         self.logger.debug(f"Making batch request HTTP, uri: `{self.endpoint_uri}`")
         request_data = self.encode_batch_rpc_request(batch_requests)
         raw_response = self._request_session_manager.make_post_request(
             self.endpoint_uri, request_data, **self.get_request_kwargs()
         )
         self.logger.debug("Received batch response HTTP.")
-        responses_list = cast(List[RPCResponse], self.decode_rpc_response(raw_response))
-        return sort_batch_response_by_response_ids(responses_list)
+        response = self.decode_rpc_response(raw_response)
+        if not isinstance(response, list):
+            # RPC errors return only one response with the error object
+            return response
+        return sort_batch_response_by_response_ids(
+            cast(List[RPCResponse], sort_batch_response_by_response_ids(response))
+        )
diff --git a/web3/types.py b/web3/types.py
index f7d0acc7..5a4fe72b 100644
--- a/web3/types.py
+++ b/web3/types.py
@@ -301,10 +301,13 @@ class CreateAccessListResponse(TypedDict):
 
 
 MakeRequestFn = Callable[[RPCEndpoint, Any], RPCResponse]
-MakeBatchRequestFn = Callable[[List[Tuple[RPCEndpoint, Any]]], List[RPCResponse]]
+MakeBatchRequestFn = Callable[
+    [List[Tuple[RPCEndpoint, Any]]], Union[List[RPCResponse], RPCResponse]
+]
 AsyncMakeRequestFn = Callable[[RPCEndpoint, Any], Coroutine[Any, Any, RPCResponse]]
 AsyncMakeBatchRequestFn = Callable[
-    [List[Tuple[RPCEndpoint, Any]]], Coroutine[Any, Any, List[RPCResponse]]
+    [List[Tuple[RPCEndpoint, Any]]],
+    Coroutine[Any, Any, Union[List[RPCResponse], RPCResponse]],
 ]
 
 


Test patch:
diff --git a/tests/core/providers/test_async_http_provider.py b/tests/core/providers/test_async_http_provider.py
index 558d260b..4ae96224 100644
--- a/tests/core/providers/test_async_http_provider.py
+++ b/tests/core/providers/test_async_http_provider.py
@@ -1,4 +1,8 @@
 import pytest
+from unittest.mock import (
+    AsyncMock,
+    patch,
+)
 
 from aiohttp import (
     ClientSession,
@@ -13,6 +17,7 @@ from web3.eth import (
 )
 from web3.exceptions import (
     ProviderConnectionError,
+    Web3RPCError,
 )
 from web3.geth import (
     AsyncGeth,
@@ -107,3 +112,21 @@ def test_get_request_headers(provider):
         headers["User-Agent"] == f"web3.py/{web3py_version}/"
         f"{AsyncHTTPProvider.__module__}.{AsyncHTTPProvider.__qualname__}"
     )
+
+
+@patch(
+    "web3._utils.http_session_manager.HTTPSessionManager.async_make_post_request",
+    new_callable=AsyncMock,
+)
+@pytest.mark.asyncio
+async def test_async_http_empty_batch_response(mock_async_post):
+    mock_async_post.return_value = (
+        b'{"jsonrpc":"2.0","id":null,"error":{"code":-32600,"message":"empty batch"}}'
+    )
+    async_w3 = AsyncWeb3(AsyncHTTPProvider())
+    async with async_w3.batch_requests() as batch:
+        with pytest.raises(Web3RPCError, match="empty batch"):
+            await batch.async_execute()
+
+    # assert that even though there was an error, we have reset the batching state
+    assert not async_w3.provider._is_batching
diff --git a/tests/core/providers/test_async_ipc_provider.py b/tests/core/providers/test_async_ipc_provider.py
index c559a445..132d3a1a 100644
--- a/tests/core/providers/test_async_ipc_provider.py
+++ b/tests/core/providers/test_async_ipc_provider.py
@@ -25,6 +25,7 @@ from web3.datastructures import (
 )
 from web3.exceptions import (
     ReadBufferLimitReached,
+    Web3RPCError,
 )
 from web3.providers import (
     AsyncIPCProvider,
@@ -366,3 +367,24 @@ async def test_async_ipc_provider_write_messages_end_with_new_line_delimiter(
 
         request_data = b'{"id": 0, "jsonrpc": "2.0", "method": "method", "params": []}'
         w3.provider._writer.write.assert_called_with(request_data + b"\n")
+
+
+@pytest.mark.asyncio
+async def test_persistent_connection_provider_empty_batch_response(
+    simple_ipc_server,
+    jsonrpc_ipc_pipe_path,
+):
+    async with AsyncWeb3(
+        AsyncIPCProvider(pathlib.Path(jsonrpc_ipc_pipe_path))
+    ) as async_w3:
+        async_w3.provider._reader.readline = AsyncMock()
+        async_w3.provider._reader.readline.return_value = (
+            b'{"jsonrpc": "2.0","id":null,"error": {"code": -32600, "message": '
+            b'"empty batch"}}\n'
+        )
+        async with async_w3.batch_requests() as batch:
+            with pytest.raises(Web3RPCError, match="empty batch"):
+                await batch.async_execute()
+
+        # assert that even though there was an error, we have reset the batching state
+        assert not async_w3.provider._is_batching
diff --git a/tests/core/providers/test_http_provider.py b/tests/core/providers/test_http_provider.py
index 82a982de..49b1b2ba 100644
--- a/tests/core/providers/test_http_provider.py
+++ b/tests/core/providers/test_http_provider.py
@@ -1,4 +1,8 @@
 import pytest
+from unittest.mock import (
+    Mock,
+    patch,
+)
 
 from requests import (
     Session,
@@ -16,6 +20,7 @@ from web3.eth import (
 )
 from web3.exceptions import (
     ProviderConnectionError,
+    Web3RPCError,
 )
 from web3.geth import (
     Geth,
@@ -110,3 +115,20 @@ def test_get_request_headers(provider):
         headers["User-Agent"] == f"web3.py/{web3py_version}/"
         f"{HTTPProvider.__module__}.{HTTPProvider.__qualname__}"
     )
+
+
+@patch(
+    "web3._utils.http_session_manager.HTTPSessionManager.make_post_request",
+    new_callable=Mock,
+)
+def test_http_empty_batch_response(mock_post):
+    mock_post.return_value = (
+        b'{"jsonrpc":"2.0","id":null,"error":{"code":-32600,"message":"empty batch"}}'
+    )
+    w3 = Web3(HTTPProvider())
+    with w3.batch_requests() as batch:
+        with pytest.raises(Web3RPCError, match="empty batch"):
+            batch.execute()
+
+    # assert that even though there was an error, we have reset the batching state
+    assert not w3.provider._is_batching
diff --git a/tests/core/providers/test_websocket_provider.py b/tests/core/providers/test_websocket_provider.py
index 7df7fbd8..c3205670 100644
--- a/tests/core/providers/test_websocket_provider.py
+++ b/tests/core/providers/test_websocket_provider.py
@@ -27,6 +27,7 @@ from web3._utils.module_testing.module_testing_utils import (
 )
 from web3.exceptions import (
     TimeExhausted,
+    Web3RPCError,
 )
 from web3.providers.persistent import (
     WebSocketProvider,
@@ -432,3 +433,24 @@ async def test_listener_task_breaks_out_of_handle_subscriptions_when_cancelled()
     # up the exception.
     with pytest.raises(asyncio.CancelledError):
         await async_w3.subscription_manager.handle_subscriptions(run_forever=True)
+
+
+@pytest.mark.asyncio
+async def test_persistent_connection_provider_empty_batch_response():
+    with patch(
+        "web3.providers.persistent.websocket.connect",
+        new=lambda *_1, **_2: _mocked_ws_conn(),
+    ):
+        async with AsyncWeb3(WebSocketProvider("ws://mocked")) as async_w3:
+            async_w3.provider._ws.recv = AsyncMock()
+            async_w3.provider._ws.recv.return_value = (
+                b'{"jsonrpc": "2.0","id":null,"error": {"code": -32600, "message": '
+                b'"empty batch"}}\n'
+            )
+            async with async_w3.batch_requests() as batch:
+                with pytest.raises(Web3RPCError, match="empty batch"):
+                    await batch.async_execute()
+
+            # assert that even though there was an error, we have reset the batching
+            # state
+            assert not async_w3.provider._is_batching