Add delegation support to ContractFunc execute method and Wallet class

example_eip7702.py

@@ -0,0 +1,141 @@
+#!/usr/bin/env python3
+"""
+EIP-7702 Delegation Example: Counter Contract
+
+This example demonstrates how to use EIP-7702 delegation utilities to sponsor
+a delegated call to a Counter contract's increment method.
+
+Usage:
+    export SPONSOR_PRIVATE_KEY="0x..."
+    python example_eip7702.py
+
+The workflow:
+1. Sponsor wallet (with existing funds) pays gas for the transaction
+2. Delegate wallet (randomly created) authorizes setting its code to the Counter contract via EIP-7702
+3. Counter contract's increment method is executed on the delegate's account
+4. The delegate's account code is set to the Counter contract during execution
+"""
+
+import asyncio
+import sys
+import os
+from typing import Annotated
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "packages/eth_rpc/src"))
+
+from eth_rpc import TransactionReceipt
+from eth_rpc.contract import ProtocolBase, ContractFunc
+from eth_rpc.delegation import sponsor_delegation
+from eth_rpc.wallet import PrivateKeyWallet
+from eth_rpc.types import METHOD, Name, NoArgs
+from eth_typing import HexAddress, HexStr
+from eth_rpc.networks import Sepolia
+
+
+class Counter(ProtocolBase):
+    """
+    Example Counter contract with increment method.
+
+    Solidity equivalent:
+    contract Counter {
+        uint256 public count;
+        address public lastToEverUpdate;
+
+        function increment() external {
+            count++;
+            lastToEverUpdate = msg.sender;
+        }
+    }
+    """
+    increment: ContractFunc[NoArgs, None] = METHOD
+    number: ContractFunc[NoArgs, int] = METHOD
+    last_to_ever_update: Annotated[
+        ContractFunc[NoArgs, HexAddress],
+        Name("lastToEverUpdate"),
+    ] = METHOD
+
+
+async def main():
+    """Demonstrate EIP-7702 delegation workflow with Counter contract"""
+    print("🔗 EIP-7702 Delegation Example: Counter Contract")
+    print("=" * 50)
+
+    sponsor_private_key = os.getenv("SPONSOR_PRIVATE_KEY")
+
+    if not sponsor_private_key:
+        print("❌ Error: SPONSOR_PRIVATE_KEY environment variable not set")
+        print("Usage: export SPONSOR_PRIVATE_KEY='0x...' && python example_eip7702.py")
+        sys.exit(1)
+
+    print("\n1. Setting up wallets...")
+    sponsor_wallet = PrivateKeyWallet[Sepolia](private_key=HexStr(sponsor_private_key))
+    delegate_wallet = PrivateKeyWallet.create_new()
+
+    print(f"   Sponsor wallet: {sponsor_wallet.address} (has funds, pays gas fees)")
+    print(f"   Delegate wallet: {delegate_wallet.address} (randomly created, authorizes code setting)")
+
+    counter_address = HexAddress("0x0271297dcc0CceA3640bbaf34801025E6F63F448")
+    print(f"   Counter contract: {counter_address}")
+
+    print("\n2. Creating Counter contract instance...")
+    counter = Counter[Sepolia](address=counter_address)
+    print(f"   Counter.increment function: {counter.increment}")
+
+    print("\n3. Preparing increment call...")
+    increment_call_data = counter.increment().data
+    print(f"   Increment call data: {increment_call_data}")
+
+    # Create sponsored delegation transaction
+    print("\n4. Creating sponsored delegation transaction...")
+    print("   This transaction will:")
+    print("   - Be paid for by the sponsor wallet (which has ETH for gas)")
+    print("   - Set the delegate's account code to the Counter contract")
+    print("   - Execute the increment method within the same transaction")
+    print("   - Automatically handle network-aware nonce lookup")
+
+    print("   Using the enhanced execute method with delegation...")
+    tx_hash = await counter.increment().execute(
+        wallet=sponsor_wallet,
+        delegate_wallet=delegate_wallet,
+    )
+    print(f"   ✅ Transaction sent using enhanced execute method: {tx_hash}")
+
+    print(f"\n5. Demonstrating simple wallet delegation (without contract data)...")
+    print("   Using the wallet delegation utility method...")
+
+    simple_delegate = PrivateKeyWallet.create_new()
+    print(f"   New delegate wallet: {simple_delegate.address}")
+
+    delegation_tx_hash = await simple_delegate.delegate_to_contract(
+        sponsor_wallet=sponsor_wallet,
+        contract_address=counter_address,
+    )
+    print(f"   Delegation transaction sent: {delegation_tx_hash}")
+
+    print(f"\n🎉 EIP-7702 delegation workflow complete!")
+    print("   Both utility methods successfully demonstrated.")
+
+    print("Waiting for transaction to be mined...")
+    while True:
+        receipt = await TransactionReceipt[Sepolia].get_by_hash(tx_hash)
+        if receipt:
+            if receipt.status == 1:
+                print("Transaction mined successfully")
+                break
+            if receipt.status == 0:
+                raise Exception(f"Transaction failed: {receipt.status}")
+        await asyncio.sleep(4)
+
+    counter = Counter[Sepolia](address=delegate_wallet.address)
+    print(f"   Counter.number: {await counter.number().get()}")
+    print(f"   Counter.last_to_ever_update: {await counter.last_to_ever_update().get()}")
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except Exception as e:
+        print(f"\n❌ Example failed: {e}")
+        import traceback
+        traceback.print_exc()
+        sys.exit(1)

packages/eth_rpc/src/eth_rpc/__init__.py

@@ -13,6 +13,7 @@
 from .block import Block
 from .codegen import codegen
 from .contract import Contract, ContractFunc, EthResponse, FuncSignature, ProtocolBase
+from .delegation import create_authorization_item, prepare_delegation_transaction, sponsor_delegation
 from .event import Event
 from .log import Log
 from .models import EventData
@@ -47,12 +48,15 @@
     "add_middleware",
     "codegen",
     "configure_rpc_from_env",
+    "create_authorization_item",
     "get_current_network",
     "get_selected_wallet",
+    "prepare_delegation_transaction",
     "set_alchemy_key",
     "set_default_network",
-    "set_selected_wallet",
-    "set_transport",
     "set_rpc_timeout",
     "set_rpc_url",
+    "set_selected_wallet",
+    "set_transport",
+    "sponsor_delegation",
 ]

packages/eth_rpc/src/eth_rpc/contract/function.py

@@ -21,6 +21,7 @@
 from .._transport import _force_get_global_rpc
 from ..block import Block
 from ..constants import ADDRESS_ZERO
+from ..delegation import sponsor_delegation
 from ..rpc.core import RPC
 from ..transaction import PreparedTransaction
 from ..utils import run
@@ -530,8 +531,23 @@ async def _execute(
         max_fee_per_gas: Optional[int] = None,
         max_priority_fee_per_gas: Optional[int] = None,
         use_access_list: bool = False,
+        delegate_wallet: Optional["BaseWallet"] = None,
+        chain_id: Optional[int] = None,
+        gas: Optional[int] = None,
         sync: bool = True,
     ) -> HexStr:
+        # If delegate_wallet is provided, use sponsored delegation
+        if delegate_wallet is not None:
+            return await self._execute_sponsored(
+                sponsor_wallet=wallet,
+                delegate_wallet=delegate_wallet,
+                chain_id=chain_id,
+                nonce=nonce,
+                value=value,
+                gas=gas or 100000,
+                sync=sync,
+            )
+
         if sync is True:
             prepared_tx = self.prepare(
                 wallet,
@@ -575,6 +591,9 @@ def execute(
         max_fee_per_gas: Optional[int] = ...,
         max_priority_fee_per_gas: Optional[int] = ...,
         use_access_list: bool = ...,
+        delegate_wallet: Optional["BaseWallet"] = ...,
+        chain_id: Optional[int] = ...,
+        gas: Optional[int] = ...,
     ) -> HexStr: ...
 
     @overload
@@ -588,6 +607,9 @@ def execute(
         max_fee_per_gas: Optional[int] = ...,
         max_priority_fee_per_gas: Optional[int] = ...,
         use_access_list: bool = ...,
+        delegate_wallet: Optional["BaseWallet"] = ...,
+        chain_id: Optional[int] = ...,
+        gas: Optional[int] = ...,
     ) -> Awaitable[HexStr]: ...
 
     def execute(
@@ -600,6 +622,9 @@ def execute(
         max_fee_per_gas: Optional[int] = None,
         max_priority_fee_per_gas: Optional[int] = None,
         use_access_list: bool = False,
+        delegate_wallet: Optional["BaseWallet"] = None,
+        chain_id: Optional[int] = None,
+        gas: Optional[int] = None,
         sync: bool = False,
     ) -> MaybeAwaitable[HexStr]:
         return run(
@@ -611,6 +636,88 @@ def execute(
             max_fee_per_gas=max_fee_per_gas,
             max_priority_fee_per_gas=max_priority_fee_per_gas,
             use_access_list=use_access_list,
+            delegate_wallet=delegate_wallet,
+            chain_id=chain_id,
+            gas=gas,
+            sync=sync,
+        )
+
+    async def _execute_sponsored(
+        self,
+        sponsor_wallet: "BaseWallet",
+        delegate_wallet: "BaseWallet",
+        *,
+        chain_id: Optional[int] = None,
+        nonce: Optional[int] = None,
+        value: int = 0,
+        gas: int = 100000,
+        sync: bool = False,
+    ) -> HexStr:
+        sponsored_tx = await sponsor_delegation(
+            sponsor_wallet=sponsor_wallet,
+            delegate_wallet=delegate_wallet,
+            contract_address=self.address,
+            chain_id=chain_id,
+            nonce=nonce,
+            value=value,
+            data=self.data,
+            gas=gas,
+        )
+        signed_tx = sponsor_wallet.sign_transaction(sponsored_tx)
+        if sync:
+            return (
+                sponsor_wallet[self._network]
+                .send_raw_transaction(HexStr("0x" + signed_tx.raw_transaction))
+                .sync
+            )
+        return await sponsor_wallet[self._network].send_raw_transaction(
+            HexStr("0x" + signed_tx.raw_transaction)
+        )
+
+    @overload
+    def execute_sponsored(
+        self,
+        sponsor_wallet: "BaseWallet",
+        delegate_wallet: "BaseWallet",
+        *,
+        sync: Literal[True],
+        chain_id: Optional[int] = ...,
+        nonce: Optional[int] = ...,
+        value: int = ...,
+        gas: int = ...,
+    ) -> HexStr: ...
+
+    @overload
+    def execute_sponsored(
+        self,
+        sponsor_wallet: "BaseWallet",
+        delegate_wallet: "BaseWallet",
+        *,
+        chain_id: Optional[int] = ...,
+        nonce: Optional[int] = ...,
+        value: int = ...,
+        gas: int = ...,
+    ) -> Awaitable[HexStr]: ...
+
+    def execute_sponsored(
+        self,
+        sponsor_wallet: "BaseWallet",
+        delegate_wallet: "BaseWallet",
+        *,
+        chain_id: Optional[int] = None,
+        nonce: Optional[int] = None,
+        value: int = 0,
+        gas: int = 100000,
+        sync: bool = False,
+    ) -> MaybeAwaitable[HexStr]:
+        return run(
+            self._execute_sponsored,
+            sponsor_wallet,
+            delegate_wallet,
+            chain_id=chain_id,
+            nonce=nonce,
+            value=value,
+            gas=gas,
             sync=sync,
         )
 
@@ -710,6 +817,9 @@ def execute(  # type: ignore
         max_fee_per_gas: Optional[int] = None,
         max_priority_fee_per_gas: Optional[int] = None,
         use_access_list: bool = False,
+        delegate_wallet: Optional[BaseWallet] = None,
+        chain_id: Optional[int] = None,
+        gas: Optional[int] = None,
     ) -> HexStr:
         return super().execute(
             wallet,
@@ -718,5 +828,28 @@ def execute(  # type: ignore
             max_fee_per_gas=max_fee_per_gas,
             max_priority_fee_per_gas=max_priority_fee_per_gas,
             use_access_list=use_access_list,
+            delegate_wallet=delegate_wallet,
+            chain_id=chain_id,
+            gas=gas,
+            sync=self.SYNC,
+        )
+
+    def execute_sponsored(  # type: ignore
+        self,
+        sponsor_wallet: BaseWallet,
+        delegate_wallet: BaseWallet,
+        *,
+        chain_id: Optional[int] = None,
+        nonce: Optional[int] = None,
+        value: int = 0,
+        gas: int = 100000,
+    ) -> HexStr:
+        return super().execute_sponsored(
+            sponsor_wallet,
+            delegate_wallet,
+            chain_id=chain_id,
+            nonce=nonce,
+            value=value,
+            gas=gas,
             sync=self.SYNC,
         )