A Python library for interacting with Ethereum blockchain
Overall
score
88%
Evaluation — 88%
↑ 1.01xAgent success when using this tile
Smart contract deployment, interaction, and management with ABI-based function calls, event filtering, transaction handling, and comprehensive type support for contract development.
Create contract instances for deployment or interaction with existing contracts.
class Contract:
def __init__(
self,
address: Optional[AnyAddress] = None,
abi: Optional[ABI] = None
):
"""
Initialize contract instance.
Parameters:
- address: Contract address (for existing contracts)
- abi: Contract Application Binary Interface
"""
@classmethod
def constructor(
cls,
**constructor_kwargs: Any
) -> ContractConstructor:
"""
Create contract constructor for deployment.
Parameters:
- constructor_kwargs: Constructor arguments
"""
@property
def address(self) -> ChecksumAddress:
"""Get contract address."""
@property
def abi(self) -> ABI:
"""Get contract ABI."""
@property
def functions(self) -> ContractFunctions:
"""Access contract functions."""
@property
def events(self) -> ContractEvents:
"""Access contract events."""
@property
def fallback(self) -> ContractFunction:
"""Access fallback function."""
@property
def receive(self) -> ContractFunction:
"""Access receive function."""
def caller(self, transaction: Optional[TxParams] = None) -> ContractCaller:
"""
Create contract caller for read-only operations.
Parameters:
- transaction: Transaction parameters for call context
"""Async version of contract operations with the same interface.
class AsyncContract:
def __init__(
self,
address: Optional[AnyAddress] = None,
abi: Optional[ABI] = None
):
"""
Initialize async contract instance.
Parameters:
- address: Contract address
- abi: Contract ABI
"""
@classmethod
async def constructor(
cls,
**constructor_kwargs: Any
) -> AsyncContractConstructor:
"""Async contract constructor."""
@property
def address(self) -> ChecksumAddress:
"""Get contract address."""
@property
def abi(self) -> ABI:
"""Get contract ABI."""
@property
def functions(self) -> AsyncContractFunctions:
"""Access async contract functions."""
@property
def events(self) -> AsyncContractEvents:
"""Access async contract events."""
def caller(self, transaction: Optional[TxParams] = None) -> AsyncContractCaller:
"""Create async contract caller."""Interface for calling contract functions and sending transactions.
class ContractFunction:
def call(
self,
transaction: Optional[TxParams] = None,
block_identifier: BlockIdentifier = "latest"
) -> Any:
"""
Call function without creating transaction.
Parameters:
- transaction: Transaction parameters for call context
- block_identifier: Block for execution
"""
def transact(self, transaction: Optional[TxParams] = None) -> HexBytes:
"""
Execute function as transaction.
Parameters:
- transaction: Transaction parameters
Returns:
Transaction hash
"""
def estimate_gas(
self,
transaction: Optional[TxParams] = None,
block_identifier: BlockIdentifier = "latest"
) -> int:
"""
Estimate gas for function execution.
Parameters:
- transaction: Transaction parameters
- block_identifier: Block for estimation
"""
def build_transaction(self, transaction: Optional[TxParams] = None) -> TxParams:
"""
Build transaction data for function call.
Parameters:
- transaction: Base transaction parameters
"""Interface for filtering and processing contract events.
class ContractEvent:
def get_logs(
self,
from_block: BlockIdentifier = "earliest",
to_block: BlockIdentifier = "latest",
argument_filters: Optional[Dict[str, Any]] = None
) -> List[EventData]:
"""
Get historical event logs.
Parameters:
- from_block: Starting block
- to_block: Ending block
- argument_filters: Filter by event arguments
"""
def create_filter(
self,
from_block: BlockIdentifier = "latest",
to_block: BlockIdentifier = "latest",
argument_filters: Optional[Dict[str, Any]] = None
) -> LogFilter:
"""
Create event filter for new events.
Parameters:
- from_block: Starting block
- to_block: Ending block
- argument_filters: Filter by event arguments
"""
def process_log(self, log: LogReceipt) -> EventData:
"""
Process raw log into event data.
Parameters:
- log: Raw log receipt
"""
def process_receipt(self, receipt: TxReceipt) -> List[EventData]:
"""
Process transaction receipt for events.
Parameters:
- receipt: Transaction receipt
"""Interface for contract deployment.
class ContractConstructor:
def transact(self, transaction: Optional[TxParams] = None) -> HexBytes:
"""
Deploy contract.
Parameters:
- transaction: Deployment transaction parameters
Returns:
Transaction hash
"""
def estimate_gas(
self,
transaction: Optional[TxParams] = None,
block_identifier: BlockIdentifier = "latest"
) -> int:
"""
Estimate gas for contract deployment.
Parameters:
- transaction: Deployment parameters
- block_identifier: Block for estimation
"""
def build_transaction(self, transaction: Optional[TxParams] = None) -> TxParams:
"""
Build deployment transaction.
Parameters:
- transaction: Base transaction parameters
"""Interface for read-only contract operations.
class ContractCaller:
def __init__(self, contract_functions: ContractFunctions):
"""
Initialize contract caller.
Parameters:
- contract_functions: Contract functions interface
"""
def __getattr__(self, function_name: str) -> Callable:
"""Access contract function for calling."""Contract-related type definitions.
ABI = List[ABIElement]
class ABIElement(TypedDict):
type: Literal["function", "constructor", "event", "fallback", "receive"]
name: NotRequired[str]
inputs: NotRequired[List[ABIInput]]
outputs: NotRequired[List[ABIOutput]]
stateMutability: NotRequired[Literal["pure", "view", "nonpayable", "payable"]]
class ABIInput(TypedDict):
name: str
type: str
indexed: NotRequired[bool]
class ABIOutput(TypedDict):
name: str
type: str
class EventData(TypedDict):
event: str
logIndex: int
transactionIndex: int
transactionHash: Hash32
address: ChecksumAddress
blockHash: Hash32
blockNumber: BlockNumber
args: Dict[str, Any]
class TxReceipt(TypedDict):
transactionHash: Hash32
transactionIndex: int
blockHash: Hash32
blockNumber: BlockNumber
from_: ChecksumAddress
to: ChecksumAddress
gasUsed: int
status: int
logs: List[LogReceipt]
contractAddress: Optional[ChecksumAddress]from web3 import Web3
import json
w3 = Web3(Web3.HTTPProvider('http://localhost:8545'))
# Load contract ABI and bytecode
with open('contract.json') as f:
contract_data = json.load(f)
abi = contract_data['abi']
bytecode = contract_data['bytecode']
# Create contract factory
contract = w3.eth.contract(abi=abi, bytecode=bytecode)
# Deploy contract
constructor_tx = contract.constructor(
initial_value=100,
owner=w3.eth.accounts[0]
).build_transaction({
'from': w3.eth.accounts[0],
'gas': 2000000,
'gasPrice': w3.to_wei(20, 'gwei')
})
# Send deployment transaction
tx_hash = w3.eth.send_transaction(constructor_tx)
receipt = w3.eth.wait_for_transaction_receipt(tx_hash)
print(f"Contract deployed at: {receipt.contractAddress}")from web3 import Web3
w3 = Web3(Web3.HTTPProvider('http://localhost:8545'))
# Connect to existing contract
contract_address = '0x123...'
contract_abi = [...] # Contract ABI
contract = w3.eth.contract(address=contract_address, abi=contract_abi)
# Call read-only function
balance = contract.functions.balanceOf('0x742d35Cc6635C0532925a3b8D5c0d9E3C4B3c8').call()
print(f"Balance: {balance}")
# Send transaction to contract function
tx_hash = contract.functions.transfer(
'0x456...',
w3.to_wei(10, 'ether')
).transact({
'from': w3.eth.accounts[0],
'gas': 100000
})
receipt = w3.eth.wait_for_transaction_receipt(tx_hash)
print(f"Transfer completed in block: {receipt.blockNumber}")from web3 import Web3
w3 = Web3(Web3.HTTPProvider('http://localhost:8545'))
contract = w3.eth.contract(address='0x123...', abi=abi)
# Get historical events
transfer_events = contract.events.Transfer.get_logs(
from_block=17000000,
to_block=17001000,
argument_filters={'from': '0x742d35Cc6635C0532925a3b8D5c0d9E3C4B3c8'}
)
for event in transfer_events:
print(f"Transfer: {event.args['from']} -> {event.args['to']}, Amount: {event.args['value']}")
# Create filter for new events
event_filter = contract.events.Transfer.create_filter(
from_block='latest',
argument_filters={'to': '0x456...'}
)
# Poll for new events
while True:
for event in event_filter.get_new_entries():
print(f"New transfer to monitored address: {event.args}")
time.sleep(2)from web3 import Web3
w3 = Web3(Web3.HTTPProvider('http://localhost:8545'))
contract = w3.eth.contract(address='0x123...', abi=abi)
# Use caller for read-only operations with custom context
caller = contract.caller({
'from': '0x742d35Cc6635C0532925a3b8D5c0d9E3C4B3c8',
'block_identifier': 17000000
})
# Call functions through caller
balance = caller.balanceOf('0x456...')
allowance = caller.allowance('0x742d35Cc6635C0532925a3b8D5c0d9E3C4B3c8', '0x456...')
print(f"Historical balance: {balance}")
print(f"Historical allowance: {allowance}")import asyncio
from web3 import AsyncWeb3
async def main():
w3 = AsyncWeb3(AsyncWeb3.AsyncHTTPProvider('http://localhost:8545'))
contract = w3.eth.contract(address='0x123...', abi=abi)
# Async contract call
balance = await contract.functions.balanceOf('0x742d35Cc6635C0532925a3b8D5c0d9E3C4B3c8').call()
print(f"Balance: {balance}")
# Async transaction
tx_hash = await contract.functions.transfer(
'0x456...',
w3.to_wei(1, 'ether')
).transact({'from': w3.eth.accounts[0]})
receipt = await w3.eth.wait_for_transaction_receipt(tx_hash)
print(f"Transaction completed: {receipt.transactionHash.hex()}")
asyncio.run(main())from web3 import Web3
w3 = Web3(Web3.HTTPProvider('http://localhost:8545'))
contract = w3.eth.contract(address='0x123...', abi=abi)
# Send transaction
tx_hash = contract.functions.someMethod().transact({'from': w3.eth.accounts[0]})
receipt = w3.eth.wait_for_transaction_receipt(tx_hash)
# Process events from receipt
events = contract.events.SomeEvent.process_receipt(receipt)
for event in events:
print(f"Event data: {event.args}")
print(f"Event address: {event.address}")
print(f"Block number: {event.blockNumber}")Install with Tessl CLI
npx tessl i tessl/pypi-web3docs
evals
scenario-1
scenario-2
scenario-3
scenario-4
scenario-5
scenario-6
scenario-7
scenario-8
scenario-9
scenario-10