Feature/web3 v7 support

CHANGELOG.md

@@ -0,0 +1,20 @@
+## [3.0.0] – 2025-05-06
+
+### Breaking Changes
+- Renamed all `.rawTransaction` properties to `.raw_transaction` (Web3.py v7 naming).
+- Completely refactored middleware for Web3.py v7: now a class-based middleware using `wrap_make_request`.
+- Overhauled `FlashbotProvider`:
+  - Replaced internal Web3 HTTP utils with `requests.post`.
+  - Switched to EIP-191 signing via `eth_account.messages.encode_defunct`.
+  - Removed deprecated imports from `eth_account._utils`.
+
+### Added
+- Official compatibility with Web3.py v7 (7.x series).
+- New helper `get_transaction_type()` to detect legacy, access list, and EIP-1559 tx types.
+- Expanded tests covering `_parse_signed_tx` for legacy, EIP-2930 (type=1), and EIP-1559 (type=2) transactions.
+
+### Fixed
+- Ensured numeric RLP fields (bytes) are converted to `int`.
+- Added recovery of `chainId` for legacy transactions signed under EIP-155.
+- Updated examples (`examples/simple.py`) to use `.raw_transaction` and new middleware/provider APIs.
+

examples/simple.py

@@ -161,7 +161,7 @@ def main() -> None:
 
     tx1_signed = w3.eth.account.sign_transaction(tx1, private_key=sender.key)
     bundle = [
-        {"signed_transaction": tx1_signed.rawTransaction},
+        {"signed_transaction": tx1_signed.raw_transaction},
         {"transaction": tx2, "signer": sender},
     ]
 

flashbots/flashbots.py

@@ -6,14 +6,8 @@
 import rlp
 from eth_account import Account
 from eth_account._utils.legacy_transactions import (
-    Transaction,
-    encode_transaction,
-    serializable_unsigned_transaction_from_dict,
-)
-from eth_account._utils.typed_transactions import (
-    AccessListTransaction,
-    DynamicFeeTransaction,
-)
+    Transaction, encode_transaction,
+    serializable_unsigned_transaction_from_dict)
 from eth_typing import HexStr
 from hexbytes import HexBytes
 from toolz import dissoc
@@ -23,17 +17,21 @@
 from web3.module import Module
 from web3.types import Nonce, RPCEndpoint, TxParams
 
-from .types import (
-    FlashbotsBundleDictTx,
-    FlashbotsBundleRawTx,
-    FlashbotsBundleTx,
-    FlashbotsOpts,
-    SignedTxAndHash,
-    TxReceipt,
-)
+from .types import (FlashbotsBundleDictTx, FlashbotsBundleRawTx,
+                    FlashbotsBundleTx, FlashbotsOpts, SignedTxAndHash,
+                    TxReceipt)
 
 SECONDS_PER_BLOCK = 12
 
+def get_transaction_type(tx: Dict[str, Any]) -> str:
+    tx_type = tx.get("type", "0x0")
+    if tx_type in ("0x0", 0, None):
+        return "legacy"
+    elif tx_type in ("0x1", 1):
+        return "access_list"
+    elif tx_type in ("0x2", 2):
+        return "eip1559"
+    return "unknown"
 
 class FlashbotsRPC:
     eth_sendBundle = RPCEndpoint("eth_sendBundle")
@@ -156,7 +154,7 @@ def sign_bundle(
                     tx["gas"] = self.w3.eth.estimate_gas(tx)
 
                 signed_tx = signer.sign_transaction(tx)
-                signed_transactions.append(signed_tx.rawTransaction)
+                signed_transactions.append(signed_tx.raw_transaction)
 
             elif all(key in tx for key in ["v", "r", "s"]):  # FlashbotsBundleDictTx
                 v, r, s = (
@@ -198,12 +196,6 @@ def sign_bundle(
 
         return signed_transactions
 
-    def to_hex(self, signed_transaction: bytes) -> str:
-        tx_hex = signed_transaction.hex()
-        if tx_hex[0:2] != "0x":
-            tx_hex = f"0x{tx_hex}"
-        return tx_hex
-
     def send_raw_bundle_munger(
         self,
         signed_bundled_transactions: List[HexBytes],
@@ -218,7 +210,7 @@ def send_raw_bundle_munger(
         # convert to hex
         return [
             {
-                "txs": list(map(lambda x: self.to_hex(x), signed_bundled_transactions)),
+                "txs": list(map(lambda x: x.to_0x_hex(), signed_bundled_transactions)),
                 "blockNumber": hex(target_block_number),
                 "minTimestamp": opts["minTimestamp"] if "minTimestamp" in opts else 0,
                 "maxTimestamp": opts["maxTimestamp"] if "maxTimestamp" in opts else 0,
@@ -295,11 +287,11 @@ def simulate(
         )
 
         # sets evm params
-        evm_block_number = self.w3.to_hex(block_number)
+        evm_block_number = block_number.to_0x_hex()
         evm_block_state_number = (
-            self.w3.to_hex(state_block_tag)
+            state_block_tag.to_0x_hex()
             if state_block_tag is not None
-            else self.w3.to_hex(block_number - 1)
+            else (block_number - 1).to_0x_hex()
         )
         evm_timestamp = (
             block_timestamp
@@ -348,7 +340,7 @@ def call_bundle_munger(
         """Given a raw signed bundle, it packages it up with the block number and the timestamps"""
         inpt = [
             {
-                "txs": list(map(lambda x: x.hex(), signed_bundled_transactions)),
+                "txs": list(map(lambda x: x.to_0x_hex(), signed_bundled_transactions)),
                 "blockNumber": evm_block_number,
                 "stateBlockNumber": evm_block_state_number,
                 "timestamp": evm_timestamp,
@@ -421,7 +413,7 @@ def send_private_transaction_munger(
             current_block = self.w3.eth.block_number
             max_block_number = current_block + 25
         params = {
-            "tx": self.to_hex(signed_transaction),
+            "tx": signed_transaction.to_0x_hex(),
             "maxBlockNumber": max_block_number,
         }
         self.response = FlashbotsPrivateTransactionResponse(
@@ -463,24 +455,51 @@ def cancel_private_transaction_munger(
 
 
 def _parse_signed_tx(signed_tx: HexBytes) -> TxParams:
-    # decode tx params based on its type
+    """
+    Parse any signed Ethereum transaction (legacy, EIP-2930, EIP-1559).
+    """
     tx_type = signed_tx[0]
-    if tx_type > int("0x7f", 16):
-        # legacy and EIP-155 transactions
+
+    if tx_type > 0x7f:
+        # Legacy transaction
         decoded_tx = rlp.decode(signed_tx, Transaction).as_dict()
+        # If transaction was signed EIP-155, extract chainId from v
+        if "chainId" not in decoded_tx:
+            v = decoded_tx.get("v", 0)
+            if v > 28:
+                # EIP-155: chainId = (v - 35) // 2
+                decoded_tx["chainId"] = (v - 35) // 2
+        # Some RLP impls use 'chain_id' key
+        if "chainId" not in decoded_tx and "chain_id" in decoded_tx:
+            decoded_tx["chainId"] = decoded_tx.pop("chain_id")
     else:
-        # typed transactions (EIP-2718)
+        # Typed transaction: skip the type byte and decode the RLP payload
+        payload = signed_tx[1:]
+        tx_data = rlp.decode(payload)
         if tx_type == 1:
-            # EIP-2930
-            sedes = AccessListTransaction._signed_transaction_serializer
+            # EIP-2930 Access List transaction
+            keys = [
+                "chainId", "nonce", "gasPrice", "gas", "to", "value", "data",
+                "accessList", "v", "r", "s"
+            ]
         elif tx_type == 2:
-            # EIP-1559
-            sedes = DynamicFeeTransaction._signed_transaction_serializer
+            # EIP-1559 Dynamic Fee transaction
+            keys = [
+                "chainId", "nonce", "maxPriorityFeePerGas", "maxFeePerGas",
+                "gas", "to", "value", "data", "accessList", "v", "r", "s"
+            ]
         else:
-            raise ValueError(f"Unknown transaction type: {tx_type}.")
-        decoded_tx = rlp.decode(signed_tx[1:], sedes).as_dict()
+            raise ValueError(f"Unsupported transaction type {tx_type}")
+        decoded_tx = dict(zip(keys, tx_data))
 
-    # recover sender address and remove signature fields
+    # Automatically convert byte fields to int for numeric fields
+    for field in ("chainId", "nonce", "gas", "value", "maxFeePerGas", "maxPriorityFeePerGas"):
+        if field in decoded_tx and isinstance(decoded_tx[field], (bytes, HexBytes)):
+            decoded_tx[field] = int.from_bytes(decoded_tx[field], byteorder="big")
+
+    # Add the 'from' field by recovering the sender address
+    # Note: Account.recover_transaction handles both legacy and typed transactions
     decoded_tx["from"] = Account.recover_transaction(signed_tx)
     decoded_tx = dissoc(decoded_tx, "v", "r", "s")
     return decoded_tx
+

flashbots/middleware.py

@@ -1,7 +1,6 @@
-from typing import Any, Callable
+from typing import Any, Type
 
 from web3 import Web3
-from web3.middleware import Middleware
 from web3.types import RPCEndpoint, RPCResponse
 
 from .provider import FlashbotProvider
@@ -21,25 +20,23 @@
 
 def construct_flashbots_middleware(
     flashbots_provider: FlashbotProvider,
-) -> Middleware:
-    """Captures Flashbots RPC requests and sends them to the Flashbots endpoint
-    while also injecting the required authorization headers
-
-    Keyword arguments:
-    flashbots_provider -- An HTTP provider instantiated with any authorization headers
-    required
+) -> Type:
     """
-
-    def flashbots_middleware(
-        make_request: Callable[[RPCEndpoint, Any], Any], w3: Web3
-    ) -> Callable[[RPCEndpoint, Any], RPCResponse]:
-        def middleware(method: RPCEndpoint, params: Any) -> RPCResponse:
-            if method not in FLASHBOTS_METHODS:
+    Returns a Web3.py v7-compatible middleware class.
+    Inject it using:
+        w3.middleware_onion.add(construct_flashbots_middleware(provider))
+    """
+    class FlashbotsMiddleware:
+        def __init__(self, w3: Web3):
+            self.w3 = w3
+            self.flashbots_provider = flashbots_provider
+
+        def wrap_make_request(self, make_request):
+            # This method is called by combine_middleware
+            def middleware(method: RPCEndpoint, params: Any) -> RPCResponse:
+                if method in FLASHBOTS_METHODS:
+                    return self.flashbots_provider.make_request(method, params)
                 return make_request(method, params)
-            else:
-                # otherwise intercept it and POST it
-                return flashbots_provider.make_request(method, params)
-
-        return middleware
+            return middleware
 
-    return flashbots_middleware
+    return FlashbotsMiddleware

flashbots/provider.py

@@ -1,14 +1,17 @@
+import json
 import logging
 import os
 from typing import Any, Dict, Optional, Union
 
+import requests
 from eth_account import Account, messages
 from eth_account.signers.local import LocalAccount
 from eth_typing import URI
 from web3 import HTTPProvider, Web3
-from web3._utils.request import make_post_request
 from web3.types import RPCEndpoint, RPCResponse
 
+logger = logging.getLogger(__name__)
+
 
 def get_default_endpoint() -> URI:
     return URI(
@@ -40,9 +43,10 @@ class FlashbotProvider(HTTPProvider):
 
     logger = logging.getLogger("web3.providers.FlashbotProvider")
 
+
     def __init__(
         self,
-        signature_account: LocalAccount,
+        signature_account: Union[str, LocalAccount],
         endpoint_uri: Optional[Union[URI, str]] = None,
         request_kwargs: Optional[Dict[str, Any]] = None,
         session: Optional[Any] = None,
@@ -55,19 +59,23 @@ def __init__(
         :param request_kwargs: Additional keyword arguments for requests.
         :param session: The session object to use for requests.
         """
+
+        if isinstance(signature_account, str):
+            signature_account = Account.from_key(signature_account)
+        self.signature_account = signature_account
+
         _endpoint_uri = endpoint_uri or get_default_endpoint()
         super().__init__(_endpoint_uri, request_kwargs, session)
-        self.signature_account = signature_account
 
     def _get_flashbots_headers(self, request_data: bytes) -> Dict[str, str]:
         message = messages.encode_defunct(
-            text=Web3.keccak(text=request_data.decode("utf-8")).hex()
+            text=Web3.keccak(text=request_data.decode("utf-8")).to_0x_hex()
         )
         signed_message = Account.sign_message(
             message, private_key=self.signature_account._private_key
         )
         return {
-            "X-Flashbots-Signature": f"{self.signature_account.address}:{signed_message.signature.hex()}"
+            "X-Flashbots-Signature": f"{self.signature_account.address}:{signed_message.signature.to_0x_hex()}"
         }
 
     def make_request(self, method: RPCEndpoint, params: Any) -> RPCResponse:
@@ -78,19 +86,25 @@ def make_request(self, method: RPCEndpoint, params: Any) -> RPCResponse:
         :param params: The parameters for the RPC method.
         :return: The RPC response.
         """
+
         self.logger.debug(
             f"Making request HTTP. URI: {self.endpoint_uri}, Method: {method}"
         )
         request_data = self.encode_rpc_request(method, params)
 
-        raw_response = make_post_request(
-            self.endpoint_uri,
-            request_data,
-            headers=self.get_request_headers()
-            | self._get_flashbots_headers(request_data),
-        )
-        response = self.decode_rpc_response(raw_response)
-        self.logger.debug(
-            f"Getting response HTTP. URI: {self.endpoint_uri}, Method: {method}, Response: {response}"
-        )
-        return response
+        try:
+            raw_response = requests.post(
+                self.endpoint_uri,
+                data=request_data,
+                headers=self.get_request_headers()
+                    | self._get_flashbots_headers(request_data),
+                timeout=10
+            )
+            response = self.decode_rpc_response(raw_response)
+            self.logger.debug(
+                f"Getting response HTTP. URI: {self.endpoint_uri}, Method: {method}, Response: {response}"
+            )
+            return response 
+        except Exception as e:
+            logger.exception("FlashbotProvider request failed")
+            raise e